from django.contrib.auth import get_user_model from django.contrib.auth.models import AnonymousUser, Group, Permission from django.contrib.contenttypes.models import ContentType from django.core.exceptions import PermissionDenied from django.db import transaction from django.test import TestCase from django.utils.functional import SimpleLazyObject from apps.organization.models import Company, Region, ServiceCenter, Department, UserOrganizationAssignment from apps.organization.assignment_services import create_assignment from apps.organization import services as organization_services from .models import Role, UserRoleAssignment from .services import create_role_assignment from .queries import business_permissions_for_user from .authorization import authorized_queryset, is_authorized, require_permission, scope_contains class AuthorizationTests(TestCase): @classmethod def setUpTestData(cls): cls.user = get_user_model().objects.create_user(username="scope-user") cls.other_user = get_user_model().objects.create_user(username="other-scope-user") cls.superuser = get_user_model().objects.create_superuser(username="scope-super", password="test-only-password") cls.company = Company.objects.create(code="A", name="Company A") cls.foreign_company = Company.objects.create(code="B", name="Company B") cls.region = Region.objects.create(company=cls.company, code="A1", name="A1") cls.sibling_region = Region.objects.create(company=cls.company, code="A2", name="A2") cls.foreign_region = Region.objects.create(company=cls.foreign_company, code="B1", name="B1") cls.center = ServiceCenter.objects.create(company=cls.company, region=cls.region, code="A1", name="A1", center_type="OWN") cls.sibling_center = ServiceCenter.objects.create(company=cls.company, region=cls.region, code="A2", name="A2", center_type="OWN") cls.other_center = ServiceCenter.objects.create(company=cls.company, region=cls.sibling_region, code="A3", name="A3", center_type="OWN") cls.foreign_center = ServiceCenter.objects.create(company=cls.foreign_company, region=cls.foreign_region, code="B1", name="B1", center_type="OWN") cls.department = Department.objects.create(company=cls.company, code="A1", name="A1") cls.other_department = Department.objects.create(company=cls.company, code="A2", name="A2") cls.foreign_department = Department.objects.create(company=cls.foreign_company, code="B1", name="B1") cls.view = Permission.objects.get(content_type__app_label="organization", codename="view_servicecenter") cls.change = Permission.objects.get(content_type__app_label="organization", codename="change_servicecenter") cls.role = Role.objects.create(code="VIEW", name="View") cls.role.permissions.add(cls.view) cls.change_role = Role.objects.create(code="CHANGE", name="Change") cls.change_role.permissions.add(cls.change) cls.permission = "organization.view_servicecenter" def grant(self, *, user=None, role=None, **scope): source = create_assignment(user=user or self.user, company=scope.pop("company", self.company), **scope) path = create_role_assignment(user=user or self.user, role=role or self.role, organization_assignment=source) return source, path def allowed(self, target, *, permission=None, user=None): return is_authorized(user=user or self.user, permission=permission or self.permission, target=target) def test_scope_matrix_and_queryset_parity(self): targets = [self.company, self.region, self.sibling_region, self.center, self.sibling_center, self.other_center, self.department, self.other_department, self.foreign_company, self.foreign_region, self.foreign_center, self.foreign_department] cases = [ ({}, targets[:8]), ({"region": self.region}, [self.region, self.center, self.sibling_center]), ({"region": self.region, "service_center": self.center}, [self.center]), ({"department": self.department}, [self.department]), ({"region": self.region, "service_center": self.center, "department": self.department}, [self.center, self.department]), ({"region": self.region, "department": self.department}, [self.region, self.department]), ] for scope, expected in cases: with self.subTest(scope=list(scope)), transaction.atomic(): source, _ = self.grant(**scope) for target in targets: with self.subTest(target=str(target)): self.assertEqual(self.allowed(target), target in expected) self.assertEqual(scope_contains(assignment=source, target=target), target in expected) for model in [Company, Region, ServiceCenter, Department]: visible = authorized_queryset(user=self.user, permission=self.permission, queryset=model.objects.all()) self.assertEqual(set(visible), {obj for obj in expected if isinstance(obj, model)}) transaction.set_rollback(True) def test_scope_and_permission_cannot_be_borrowed_from_different_paths(self): self.grant(role=self.change_role, region=self.region, service_center=self.center) self.grant(region=self.sibling_region, service_center=self.other_center) self.assertTrue(self.allowed(self.center, permission="organization.change_servicecenter")) self.assertFalse(self.allowed(self.other_center, permission="organization.change_servicecenter")) self.assertTrue(self.allowed(self.other_center)) self.assertFalse(self.allowed(self.center)) self.assertEqual(list(authorized_queryset(user=self.user, permission="organization.change_servicecenter", queryset=ServiceCenter.objects.all())), [self.center]) def test_multiple_assignments_nonprimary_can_authorize(self): self.grant(region=self.region, service_center=self.center, is_primary=True) self.grant(region=self.sibling_region, service_center=self.other_center) self.assertTrue(self.allowed(self.center)) self.assertTrue(self.allowed(self.other_center)) self.assertFalse(self.allowed(self.sibling_center)) def test_multiple_roles_any_valid_role_can_grant(self): source, _ = self.grant(role=self.change_role, region=self.region) create_role_assignment(user=self.user, role=self.role, organization_assignment=source) self.assertTrue(self.allowed(self.center)) self.assertTrue(self.allowed(self.center, permission="organization.change_servicecenter")) def test_separate_company_path_required(self): self.grant() for target in [self.foreign_company, self.foreign_region, self.foreign_center, self.foreign_department]: self.assertFalse(self.allowed(target)) self.grant(company=self.foreign_company) for target in [self.foreign_company, self.foreign_region, self.foreign_center, self.foreign_department]: self.assertTrue(self.allowed(target)) def test_other_user_paths_never_grant(self): self.grant(user=self.other_user) self.assertFalse(self.allowed(self.company)) def test_anonymous_unsaved_and_missing_users_denied(self): self.grant() unsaved = get_user_model()(id=self.user.pk, username="unsaved") missing = get_user_model().objects.create_user(username="deleted") missing_id = missing.pk missing.delete() missing.pk = missing_id for user in [None, AnonymousUser(), object(), unsaved, missing]: self.assertFalse(is_authorized(user=user, permission=self.permission, target=self.company)) def test_inactive_user_and_inactive_superuser_denied(self): self.grant() for user in [self.user, self.superuser]: get_user_model().objects.filter(pk=user.pk).update(is_active=False) self.assertFalse(self.allowed(self.company, user=user)) self.assertFalse(authorized_queryset(user=user, permission=self.permission, queryset=Company.objects.all()).exists()) def test_active_superuser_has_full_supported_scope_without_assignments(self): organization_services.deactivate_company(company=self.foreign_company) self.assertTrue(self.allowed(self.foreign_center, user=self.superuser)) self.assertEqual(set(authorized_queryset(user=self.superuser, permission=self.permission, queryset=ServiceCenter.objects.all())), set(ServiceCenter.objects.all())) def test_stale_superuser_flag_cannot_grant(self): get_user_model().objects.filter(pk=self.superuser.pk).update(is_superuser=False) self.assertFalse(self.allowed(self.company, user=self.superuser)) def test_staff_groups_and_direct_permissions_do_not_create_business_scope(self): self.user.is_staff = True self.user.save() self.user.user_permissions.add(self.view) group = Group.objects.create(name="Native permission group") group.permissions.add(self.view) self.user.groups.add(group) self.assertTrue(self.user.has_perm(self.permission)) self.assertFalse(self.allowed(self.company)) create_assignment(user=self.user, company=self.company) self.assertFalse(self.allowed(self.company)) def test_role_without_permission_denied(self): self.grant(role=self.change_role) self.assertFalse(self.allowed(self.company)) def test_permission_inputs_fail_closed_for_every_user(self): self.grant() for permission in [None, 1, [], "", "view_servicecenter", "organization.", ".view", "a.b.c", " organization.view_servicecenter", "organization.view_servicecenter ", "organization.view_servicecenter\n", "missing.nonexistent", "organization.nonexistent"]: for user in [self.user, self.superuser]: self.assertFalse(is_authorized(user=user, permission=permission, target=self.company)) def test_ambiguous_permission_denied(self): Permission.objects.create(content_type=ContentType.objects.get_for_model(Region), codename="view_servicecenter", name="Ambiguous test permission") self.grant() self.assertFalse(self.allowed(self.company)) self.assertFalse(self.allowed(self.company, user=self.superuser)) def test_inactive_role_and_assignments_denied_even_after_raw_changes(self): source, path = self.grant() for model, pk, updates in [(Role, self.role.pk, {"is_active": False}), (UserRoleAssignment, path.pk, {"is_active": False}), (UserOrganizationAssignment, source.pk, {"is_active": False})]: with self.subTest(model=model.__name__), transaction.atomic(): model.objects.filter(pk=pk).update(**updates) self.assertFalse(self.allowed(self.company)) transaction.set_rollback(True) def test_all_source_hierarchy_states_are_checked(self): source, _ = self.grant(region=self.region, service_center=self.center, department=self.department) for parent in [self.company, self.region, self.center, self.department]: with self.subTest(model=type(parent).__name__), transaction.atomic(): type(parent).objects.filter(pk=parent.pk).update(is_active=False) self.assertFalse(self.allowed(self.center)) self.assertFalse(self.allowed(self.department)) self.assertFalse(scope_contains(assignment=source, target=self.center)) transaction.set_rollback(True) def test_corrupt_source_cross_company_dimensions_denied(self): source, _ = self.grant(region=self.region, service_center=self.center, department=self.department) for updates in [{"company": self.foreign_company}, {"region": self.foreign_region}, {"service_center": self.foreign_center}, {"department": self.foreign_department}, {"region": self.sibling_region}]: with self.subTest(fields=list(updates)), transaction.atomic(): UserOrganizationAssignment.objects.filter(pk=source.pk).update(**updates) self.assertFalse(self.allowed(self.center)) self.assertFalse(self.allowed(self.department)) transaction.set_rollback(True) def test_corrupt_source_missing_center_region_denied(self): # Same-row constraint prevents the missing-region form even for bulk writes; # a mismatched center's own parent is a cross-table corruption to defend against. self.grant(region=self.region, service_center=self.center) ServiceCenter.objects.filter(pk=self.center.pk).update(region=self.foreign_region) self.assertFalse(self.allowed(self.center)) def test_corrupt_role_assignment_user_mismatch_denied(self): source = create_assignment(user=self.other_user, company=self.company) UserRoleAssignment.objects.bulk_create([UserRoleAssignment(user=self.user, role=self.role, organization_assignment=source)]) self.assertFalse(self.allowed(self.company)) def test_target_active_hierarchy_and_integrity_checked(self): self.grant() for model, pk, updates in [ (Region, self.region.pk, {"is_active": False}), (ServiceCenter, self.center.pk, {"is_active": False}), (Department, self.department.pk, {"is_active": False}), (ServiceCenter, self.center.pk, {"company": self.foreign_company}), (ServiceCenter, self.center.pk, {"region": self.foreign_region}), ]: with self.subTest(model=model.__name__, fields=list(updates)), transaction.atomic(): model.objects.filter(pk=pk).update(**updates) target = {Region: self.region, ServiceCenter: self.center, Department: self.department}[model] self.assertFalse(self.allowed(target)) self.assertNotIn(target, authorized_queryset(user=self.user, permission=self.permission, queryset=model.objects.all())) transaction.set_rollback(True) def test_target_scope_reads_database_not_unsaved_fields(self): self.grant(region=self.region, service_center=self.center) self.foreign_center.company = self.company self.foreign_center.region = self.region self.assertFalse(self.allowed(self.foreign_center)) self.center.company = self.foreign_company self.assertTrue(self.allowed(self.center)) def test_unsupported_unsaved_and_deleted_targets_denied(self): self.grant() unsaved = Company(id=self.company.pk, code="FAKE", name="Fake") deleted = Company.objects.create(code="DELETED", name="Deleted") pk = deleted.pk deleted.delete() deleted.pk = pk for target in [None, object(), self.user, self.role, unsaved, deleted]: for user in [self.user, self.superuser]: self.assertFalse(is_authorized(user=user, permission=self.permission, target=target)) def test_assignment_target_intersection_preserves_every_dimension(self): source, _ = self.grant(region=self.region, service_center=self.center, department=self.department) exact = create_assignment(user=self.other_user, company=self.company, region=self.region, service_center=self.center, department=self.department) center_only = create_assignment(user=self.other_user, company=self.company, region=self.region, service_center=self.center) dept_only = create_assignment(user=self.other_user, company=self.company, department=self.department) self.assertTrue(self.allowed(exact)) self.assertFalse(self.allowed(center_only)) self.assertFalse(self.allowed(dept_only)) self.assertTrue(scope_contains(assignment=source, target=exact)) self.assertEqual(set(authorized_queryset(user=self.user, permission=self.permission, queryset=UserOrganizationAssignment.objects.all())), {source, exact}) def test_region_contains_descendant_assignment_but_not_company_department_scope(self): self.grant(region=self.region) descendant = create_assignment(user=self.other_user, company=self.company, region=self.region, service_center=self.center, department=self.department) department = create_assignment(user=self.other_user, company=self.company, department=self.department) self.assertTrue(self.allowed(descendant)) self.assertFalse(self.allowed(department)) def test_company_contains_assignments_but_inactive_target_owner_denied(self): self.grant() target = create_assignment(user=self.other_user, company=self.company) self.assertTrue(self.allowed(target)) get_user_model().objects.filter(pk=self.other_user.pk).update(is_active=False) self.assertFalse(self.allowed(target)) def test_inventory_is_not_authorization(self): self.grant(region=self.region, service_center=self.center) self.assertIn(self.view, business_permissions_for_user(self.user)) self.assertFalse(self.allowed(self.other_center)) def test_queryset_is_lazy_and_one_query_without_target_materialization(self): self.grant(region=self.region) with self.assertNumQueries(0): result = authorized_queryset(user=self.user, permission=self.permission, queryset=ServiceCenter.objects.order_by("-code")) with self.assertNumQueries(1): self.assertEqual(list(result), [self.sibling_center, self.center]) def test_point_and_containment_queries_are_bounded(self): source, _ = self.grant() with self.assertNumQueries(1): self.assertTrue(self.allowed(self.center)) with self.assertNumQueries(1): self.assertTrue(scope_contains(assignment=source, target=self.center)) def test_many_paths_dont_duplicate_targets_or_add_queries(self): source, _ = self.grant() for number in range(8): role = Role.objects.create(code=f"EXTRA_{number}", name="Extra") role.permissions.add(self.view) create_role_assignment(user=self.user, role=role, organization_assignment=source) with self.assertNumQueries(1): result = list(authorized_queryset(user=self.user, permission=self.permission, queryset=ServiceCenter.objects.all())) self.assertEqual(set(result), {self.center, self.sibling_center, self.other_center}) self.assertEqual(len(result), 3) def test_queryset_preserves_caller_filters_and_superuser_order(self): self.grant() query = ServiceCenter.objects.filter(region=self.region).order_by("-code") for user in [self.user, self.superuser]: self.assertEqual(list(authorized_queryset(user=user, permission=self.permission, queryset=query)), [self.sibling_center, self.center]) def test_unsupported_sliced_and_combined_querysets_deny(self): self.grant() for query in [Role.objects.all(), ServiceCenter.objects.all()[:1], ServiceCenter.objects.all().union(ServiceCenter.objects.all())]: self.assertFalse(authorized_queryset(user=self.user, permission=self.permission, queryset=query).exists()) with self.assertRaises(TypeError): authorized_queryset(user=self.user, permission=self.permission, queryset=[]) def test_strict_helper_and_lazy_request_user(self): self.grant() user = SimpleLazyObject(lambda: self.user) self.assertIsNone(require_permission(user=user, permission=self.permission, target=self.center)) with self.assertRaisesMessage(PermissionDenied, "Permission denied."): require_permission(user=user, permission=self.permission, target=self.foreign_center) def test_revocation_is_seen_without_permission_cache(self): self.grant() self.assertTrue(self.allowed(self.center)) pending = authorized_queryset(user=self.user, permission=self.permission, queryset=ServiceCenter.objects.all()) self.role.permissions.clear() self.assertFalse(self.allowed(self.center)) self.assertFalse(pending.exists()) def test_permission_content_type_is_independent_of_scope_object(self): self.grant() self.assertTrue(self.allowed(self.company)) self.assertTrue(self.allowed(self.region))