"""RBAC v2 Slice 1: capability registry, source-bound resolver, method-aware DRF enforcement, seed backfill, and /me capability publication.""" import pytest from django.contrib.auth import get_user_model from django.core.management import call_command from rest_framework.test import APIClient, APIRequestFactory from rest_framework.viewsets import ViewSet from apps.companies.models import Organization, OrganizationMembership from apps.core.authorization import ( AUTHENTICATED_PERSONAL, DENY, PUBLIC, RequiresCapabilities, baseline_capability_codes, effective_capabilities, entitlements, has_capability, registry, ) from apps.core.authorization.drf import unmapped_action_methods, validate_capability_map from apps.core.authorization.resolver import ( SCOPE_ORGANIZATION, SCOPE_OWN, SOURCE_DIRECTOR, SOURCE_MEMBERSHIP, SOURCE_ROLE, ) from apps.core.models import Permission, Role, UserRole User = get_user_model() BASELINE = { "procurement.requisition.read", "procurement.requisition.create", "procurement.requisition.update", "procurement.requisition.submit", } def _org(name="Authz Test Org"): return Organization.objects.create(name=name) def _member(email, organization): user = User.objects.create_user(email=email, password="pass123") OrganizationMembership.objects.create(organization=organization, user=user) return user def _capability_role(slug, *codes): call_command("seed_rbac") role = Role.objects.create(name=f"Role {slug}", slug=slug) role.permissions.set(Permission.objects.filter(code__in=codes)) return role # ---- registry ---- def test_registry_baseline_is_exactly_the_four_requisition_codes(): assert baseline_capability_codes() == frozenset(BASELINE) def test_registry_prerequisites_reference_registered_codes(): for definition in registry.all_capabilities(): for prereq in definition.prerequisites: assert registry.is_registered(prereq) def test_registry_codes_follow_grammar(): for definition in registry.all_capabilities(): module, resource, verb = definition.code.split(".") assert definition.module == module assert definition.resource == resource assert definition.verb == verb def test_validate_role_capability_codes_rejects_wildcards_unknowns_and_missing_prereqs(): problems = registry.validate_role_capability_codes([ "procurement.*", "procurement.nonsense.read", "procurement.payment.release", # missing procurement.payment.read ]) text = " ".join(problems) assert "Wildcard" in text assert "Unknown capability" in text assert "procurement.payment.release requires procurement.payment.read" in text assert registry.validate_role_capability_codes( ["procurement.payment.read", "procurement.payment.release"] ) == [] # ---- resolver ---- @pytest.mark.django_db def test_membership_baseline_granted_to_active_member_only(): org = _org() member = _member("member@x.com", org) outsider = User.objects.create_user(email="outsider@x.com", password="pass123") assert effective_capabilities(member, org) == frozenset(BASELINE) assert effective_capabilities(outsider, org) == frozenset() @pytest.mark.django_db def test_role_assignment_grants_only_registered_v2_codes(): org = _org() member = _member("officer@x.com", org) role = _capability_role( "procurement-officer-x", "procurement.vendor.read", "procurement.vendor.manage", ) UserRole.objects.create(user=member, role=role, organization=org) caps = effective_capabilities(member, org) assert "procurement.vendor.manage" in caps assert caps == frozenset(BASELINE) | { "procurement.vendor.read", "procurement.vendor.manage", } @pytest.mark.django_db def test_legacy_wildcard_and_domain_permissions_never_expand_to_v2_codes(): org = _org() member = _member("legacy-manager@x.com", org) call_command("seed_rbac") manager = Role.objects.get(slug="manager") # legacy "*" tier UserRole.objects.create(user=member, role=manager, organization=org) caps = effective_capabilities(member, org) assert caps == frozenset(BASELINE) # baseline only, no specialist codes assert not has_capability(member, org, "procurement.payment.release") # v1 keeps working during dual-read assert member.has_menu_perm("procurements.update", org) @pytest.mark.django_db def test_capabilities_are_organization_isolated(): org_a, org_b = _org("Org A"), _org("Org B") member = _member("multi@x.com", org_a) OrganizationMembership.objects.create(organization=org_b, user=member) role = _capability_role("sourcing-x", "procurement.sourcing.read") UserRole.objects.create(user=member, role=role, organization=org_a) assert "procurement.sourcing.read" in effective_capabilities(member, org_a) fresh = User.objects.get(pk=member.pk) # avoid per-instance memo assert "procurement.sourcing.read" not in effective_capabilities(fresh, org_b) @pytest.mark.django_db def test_director_gets_all_registered_capabilities_and_org_scope(): org = _org() director = _member("director@x.com", org) director.organization_role = "director" director.save(update_fields=["organization_role"]) caps = effective_capabilities(director, org) assert caps == frozenset(d.code for d in registry.all_capabilities()) sources = {e.source: e for e in entitlements(director, org)} assert sources[SOURCE_DIRECTOR].scope == SCOPE_ORGANIZATION @pytest.mark.django_db def test_inactive_user_loses_all_capabilities(): org = _org() member = _member("suspended@x.com", org) member.is_active = False member.save(update_fields=["is_active"]) assert effective_capabilities(member, org) == frozenset() assert entitlements(member, org) == [] @pytest.mark.django_db def test_entitlements_are_source_bound(): org = _org() member = _member("bound@x.com", org) role = _capability_role("invoicer-x", "procurement.invoice.read") assignment = UserRole.objects.create(user=member, role=role, organization=org) by_source = {e.source: e for e in entitlements(member, org)} assert by_source[SOURCE_MEMBERSHIP].scope == SCOPE_OWN assert by_source[SOURCE_MEMBERSHIP].capabilities == frozenset(BASELINE) assert by_source[SOURCE_ROLE].role_assignment_id == assignment.pk assert "procurement.invoice.read" in by_source[SOURCE_ROLE].capabilities # Baseline capabilities are not folded into the role entitlement. assert "procurement.requisition.update" not in by_source[SOURCE_ROLE].capabilities @pytest.mark.django_db def test_role_permission_change_invalidates_effective_access(): org = _org() member = _member("cached@x.com", org) role = _capability_role("granted-x", "procurement.report.read") UserRole.objects.create(user=member, role=role, organization=org) assert has_capability(member, org, "procurement.report.read") admin = User.objects.create_superuser(email="authz-admin@x.com", password="p") client = APIClient() client.force_authenticate(user=admin) response = client.post(f"/api/rbac/roles/{role.pk}/permissions/", {"permission_ids": []}, format="json") assert response.status_code == 200 fresh = User.objects.get(pk=member.pk) assert not has_capability(fresh, org, "procurement.report.read") # ---- DRF enforcement ---- class _ProbeViewSet(ViewSet): permission_classes = [RequiresCapabilities] capability_map = { ("list", "GET"): "procurement.vendor.read", ("create", "POST"): "procurement.vendor.manage", ("workspace", "GET"): "procurement.requisition.read", ("destroy", "DELETE"): DENY, ("ping", "GET"): PUBLIC, ("profile", "GET"): AUTHENTICATED_PERSONAL, } def list(self, request): # pragma: no cover - permission gate only pass def create(self, request): # pragma: no cover pass def _check(user, action, method): factory = APIRequestFactory() request = factory.generic(method, "/probe/") request.user = user request.session = {} view = _ProbeViewSet() view.action = action return RequiresCapabilities().has_permission(request, view) @pytest.mark.django_db def test_requires_capabilities_is_method_aware_and_fails_closed(): org = _org() member = _member("drf@x.com", org) role = _capability_role("vendor-reader-x", "procurement.vendor.read") UserRole.objects.create(user=member, role=role, organization=org) assert _check(member, "list", "GET") is True # has vendor.read assert _check(member, "create", "POST") is False # lacks vendor.manage assert _check(member, "workspace", "GET") is True # baseline read assert _check(member, "workspace", "POST") is False # unmapped method: closed assert _check(member, "unmapped_action", "GET") is False assert _check(member, "destroy", "DELETE") is False # explicit DENY assert _check(member, "profile", "GET") is True # authenticated personal @pytest.mark.django_db def test_requires_capabilities_public_marker_allows_anonymous(): from django.contrib.auth.models import AnonymousUser anonymous = AnonymousUser() assert _check(anonymous, "ping", "GET") is True assert _check(anonymous, "list", "GET") is False def test_capability_map_meta_helpers(): assert validate_capability_map(_ProbeViewSet) == [] assert unmapped_action_methods(_ProbeViewSet) == [] class Unmapped(_ProbeViewSet): capability_map = {("list", "GET"): "procurement.vendor.read"} assert ("create", "POST") in unmapped_action_methods(Unmapped) class BadCode(_ProbeViewSet): capability_map = {("list", "GET"): "procurement.vendor.frobnicate"} assert any("unknown capability" in p for p in validate_capability_map(BadCode)) # ---- seed + role editor validation ---- @pytest.mark.django_db def test_seed_backfills_legacy_codes_and_seeds_canonical_capabilities(): call_command("seed_rbac") assert not Permission.objects.filter(code__isnull=True).exists() assert Permission.objects.filter(code="procurement.payment.release").exists() legacy = Permission.objects.get(domain="procurements", action="view") assert legacy.code == "legacy.procurements.view" # Wildcard tier roles never absorb v2 capability rows. manager = Role.objects.get(slug="manager") assert not manager.permissions.exclude(code__startswith="legacy.").exists() # Idempotent re-run. total = Permission.objects.count() call_command("seed_rbac") assert Permission.objects.count() == total @pytest.mark.django_db def test_role_update_rejects_capability_set_missing_prerequisites(): call_command("seed_rbac") role = Role.objects.create(name="Bad Prereqs", slug="bad-prereqs") release_only = Permission.objects.get(code="procurement.payment.release") admin = User.objects.create_superuser(email="prereq-admin@x.com", password="p") client = APIClient() client.force_authenticate(user=admin) response = client.patch( f"/api/rbac/roles/{role.pk}/", {"permission_ids": [release_only.pk]}, format="json", ) assert response.status_code == 400 assert "requires procurement.payment.read" in str(response.json()) ok = client.patch( f"/api/rbac/roles/{role.pk}/", {"permission_ids": list( Permission.objects.filter( code__in=["procurement.payment.read", "procurement.payment.release"] ).values_list("pk", flat=True) )}, format="json", ) assert ok.status_code == 200 # ---- /me contract ---- @pytest.mark.django_db def test_me_publishes_sorted_capabilities_and_access_version(): org = _org() member = _member("me-contract@x.com", org) role = _capability_role("po-reader-x", "procurement.purchase_order.read") UserRole.objects.create(user=member, role=role, organization=org) client = APIClient() client.force_authenticate(user=member) body = client.get("/api/auth/me/").json() assert body["access_version"] == 2 assert body["capabilities"] == sorted( list(BASELINE) + ["procurement.purchase_order.read"] ) # Deprecated alias serves legacy codenames only — the v2 row on this # role must not leak into it as "procurement.read"-style junk. assert body["permissions"] == []