import uuid from unittest.mock import patch from django.contrib import admin from django.contrib.auth import get_user_model from django.contrib.auth.models import AnonymousUser, Group, Permission from django.core.exceptions import ValidationError from django.db import IntegrityError, transaction from django.db.models.deletion import ProtectedError from django.test import TestCase from django.urls import reverse from apps.organization import services as organization_services from apps.organization import assignment_services from apps.organization.models import Company from apps.organization.assignments import UserOrganizationAssignment from apps.organization.test_lifecycle import make_tree from .models import Role, UserRoleAssignment from .queries import business_permissions_for_user from . import services class AccessFixture(TestCase): @classmethod def setUpTestData(cls): cls.user = get_user_model().objects.create_user(username="rbac-user") cls.other_user = get_user_model().objects.create_user(username="rbac-other") cls.company, cls.region, cls.center, cls.department = make_tree("CCARE") cls.scope = assignment_services.create_assignment(user=cls.user, company=cls.company, region=cls.region, service_center=cls.center, department=cls.department, is_primary=True) cls.other_scope = assignment_services.create_assignment(user=cls.user, company=cls.company) cls.foreign_scope = assignment_services.create_assignment(user=cls.other_user, company=cls.company) cls.role = Role.objects.create(code=" sample ", name="Sample") cls.other_role = Role.objects.create(code="SECOND", name="Second") cls.permission = Permission.objects.get(content_type__app_label="organization", codename="view_servicecenter") cls.other_permission = Permission.objects.get(content_type__app_label="organization", codename="view_company") def assign(self, **kwargs): values = dict(user=self.user, role=self.role, organization_assignment=self.scope) values.update(kwargs) return services.create_role_assignment(**values) def assert_db_rejects(self, operation): with self.assertRaises(IntegrityError), transaction.atomic(): operation() class RoleTests(AccessFixture): def test_role_uuid_timestamps_defaults_and_normalization(self): self.assertIsInstance(self.role.pk, uuid.UUID) self.assertIsNotNone(self.role.created_at) self.assertIsNotNone(self.role.updated_at) self.assertTrue(self.role.is_active) self.assertEqual(self.role.code, "SAMPLE") self.assertEqual(self.role.description, "") def test_duplicate_and_case_equivalent_codes_rejected(self): for code in ["SAMPLE", " sample ", "Sample"]: with self.subTest(code=code), self.assertRaises(ValidationError): Role.objects.create(code=code, name="Duplicate") self.assert_db_rejects(lambda: Role.objects.bulk_create([Role(code="SAMPLE", name="Duplicate")])) def test_database_canonical_code_protection(self): for code in ["sample", " SAMPLE ", "", "BAD CODE"]: with self.subTest(code=code): self.assert_db_rejects(lambda: Role.objects.filter(pk=self.role.pk).update(code=code)) def test_required_code_and_name(self): for fields in [{"code": "", "name": "Name"}, {"code": "VALID", "name": ""}]: with self.assertRaises(ValidationError): Role.objects.create(**fields) def test_names_not_normalized_and_codes_can_be_corrected(self): self.role.name = " Display " self.role.code = " corrected " created, updated = self.role.created_at, self.role.updated_at self.role.save(update_fields=["code", "name"]) self.role.refresh_from_db() self.assertEqual(self.role.code, "CORRECTED") self.assertEqual(self.role.name, " Display ") self.assertEqual(self.role.created_at, created) self.assertGreater(self.role.updated_at, updated) def test_permission_relationship_and_atomic_replacement(self): services.set_role_permissions(role=self.role, permissions=[self.permission, self.other_permission]) self.assertEqual(self.role.permissions.count(), 2) services.set_role_permissions(role=self.role, permissions=[self.other_permission]) self.assertEqual(list(self.role.permissions.all()), [self.other_permission]) services.set_role_permissions(role=self.role, permissions=[]) self.assertEqual(self.role.permissions.count(), 0) def test_nonexistent_permission_rejected_without_losing_membership(self): self.role.permissions.add(self.permission) with self.assertRaises(ValueError): services.set_role_permissions(role=self.role, permissions=[Permission(pk=2147483647)]) self.assertEqual(list(self.role.permissions.all()), [self.permission]) def test_permission_replacement_rolls_back_on_failure(self): self.role.permissions.add(self.permission) with patch.object(Role, "save", side_effect=RuntimeError("simulated failure")): with self.assertRaises(RuntimeError): services.set_role_permissions(role=self.role, permissions=[self.other_permission]) self.assertEqual(list(self.role.permissions.all()), [self.permission]) def test_deactivation_only_ends_its_assignments(self): affected = self.assign() unrelated = self.assign(role=self.other_role) services.deactivate_role(role=self.role) affected.refresh_from_db() unrelated.refresh_from_db() self.role.refresh_from_db() self.assertFalse(self.role.is_active) self.assertFalse(affected.is_active) self.assertTrue(unrelated.is_active) services.reactivate_role(role=self.role) affected.refresh_from_db() self.assertFalse(affected.is_active) def test_direct_role_deactivation_rejected_without_hidden_cascade(self): obj = self.assign() self.role.is_active = False with self.assertRaises(ValidationError): self.role.save() obj.refresh_from_db() self.assertTrue(obj.is_active) def test_role_cascade_rolls_back(self): obj = self.assign() with patch.object(Role, "save", side_effect=RuntimeError("simulated failure")): with self.assertRaises(RuntimeError): services.deactivate_role(role=self.role) obj.refresh_from_db() self.assertTrue(obj.is_active) def test_representation_without_queries(self): with self.assertNumQueries(0): self.assertEqual(str(self.role), "SAMPLE - Sample") class RoleAssignmentTests(AccessFixture): def test_creation_uuid_timestamps_and_relationships(self): obj = self.assign() self.assertIsInstance(obj.pk, uuid.UUID) self.assertIsNotNone(obj.created_at) self.assertIsNotNone(obj.updated_at) self.assertTrue(obj.is_active) self.assertEqual(obj.user, self.user) self.assertEqual(obj.role, self.role) self.assertEqual(obj.organization_assignment, self.scope) def test_multiple_roles_and_same_role_across_scopes(self): self.assign() self.assign(role=self.other_role) self.assign(organization_assignment=self.other_scope) self.assertEqual(UserRoleAssignment.objects.for_user(self.user).active().count(), 3) def test_user_mismatch_rejected(self): with self.assertRaises(ValidationError): self.assign(organization_assignment=self.foreign_scope) def test_inactive_dependencies_rejected(self): for parent in [self.user, self.role, self.scope]: parent.is_active = False if isinstance(parent, UserOrganizationAssignment): parent.is_primary = False parent.save() with self.subTest(model=type(parent).__name__), self.assertRaises(ValidationError): self.assign() parent.is_active = True parent.save() def test_duplicate_active_model_and_database(self): self.assign() with self.assertRaises(ValidationError): self.assign() self.assert_db_rejects(lambda: UserRoleAssignment.objects.bulk_create([ UserRoleAssignment(user=self.user, role=self.role, organization_assignment=self.scope)])) def test_inactive_history_can_repeat_and_reference_inactive_parents(self): services.deactivate_role(role=self.role) assignment_services.deactivate_assignment(assignment=self.scope) self.user.is_active = False self.user.save() self.assign(is_active=False) self.assign(is_active=False) self.assertEqual(UserRoleAssignment.objects.count(), 2) def test_inactive_history_still_rejects_user_mismatch(self): with self.assertRaises(ValidationError): self.assign(organization_assignment=self.foreign_scope, is_active=False) def test_reactivation_requires_valid_dependencies_and_no_duplicate(self): active = self.assign() historical = self.assign(is_active=False) with self.assertRaises(ValidationError): services.reactivate_role_assignment(assignment=historical) services.deactivate_role_assignment(assignment=active) restored = services.reactivate_role_assignment(assignment=historical) self.assertTrue(restored.is_active) def test_user_deactivation_has_no_hidden_cascade(self): obj = self.assign() self.user.is_active = False self.user.save() obj.refresh_from_db() self.scope.refresh_from_db() self.assertTrue(obj.is_active) self.assertTrue(self.scope.is_active) with self.assertRaises(ValidationError): self.assign(role=self.other_role) def test_user_transfer_rejected(self): obj = self.assign() obj.user, obj.organization_assignment = self.other_user, self.foreign_scope with self.assertRaises(ValidationError): obj.save() def test_partial_update_validates_actual_state(self): obj = self.assign() services.deactivate_role(role=self.other_role) obj.role = self.other_role obj.is_active = False # Not persisted by this write. with self.assertRaises(ValidationError): obj.save(update_fields=["role"]) obj.refresh_from_db() self.assertEqual(obj.role, self.role) def test_deletion_protects_history(self): self.assign(is_active=False) for obj in [self.user, self.role, self.scope]: with self.subTest(model=type(obj).__name__), self.assertRaises(ProtectedError): obj.delete() class IntegrationTests(AccessFixture): def test_assignment_deactivation_ends_roles_and_clears_primary(self): obj = self.assign() unrelated = self.assign(organization_assignment=self.other_scope) assignment_services.deactivate_assignment(assignment=self.scope) obj.refresh_from_db() unrelated.refresh_from_db() self.scope.refresh_from_db() self.assertFalse(obj.is_active) self.assertFalse(self.scope.is_primary) self.assertTrue(unrelated.is_active) assignment_services.reactivate_assignment(assignment=self.scope) obj.refresh_from_db() self.assertFalse(obj.is_active) def test_all_organization_cascades_reach_role_assignments(self): # Independent savepoints allow checking every lifecycle against the same fixture. for operation, kwargs in [ (organization_services.deactivate_company, {"company": self.company}), (organization_services.deactivate_region, {"region": self.region}), (organization_services.deactivate_service_center, {"service_center": self.center}), (organization_services.deactivate_department, {"department": self.department}), ]: with self.subTest(operation=operation.__name__), transaction.atomic(): obj = self.assign() operation(**kwargs) obj.refresh_from_db() self.scope.refresh_from_db() self.assertFalse(obj.is_active) self.assertFalse(self.scope.is_active) self.assertFalse(self.scope.is_primary) transaction.set_rollback(True) def test_unrelated_organization_roles_untouched(self): obj = self.assign() unrelated = self.assign(organization_assignment=self.other_scope) organization_services.deactivate_department(department=self.department) obj.refresh_from_db() unrelated.refresh_from_db() self.assertFalse(obj.is_active) self.assertTrue(unrelated.is_active) def test_direct_assignment_deactivation_rejected(self): obj = self.assign() self.scope.is_active, self.scope.is_primary = False, False with self.assertRaises(ValidationError): self.scope.save() obj.refresh_from_db() self.assertTrue(obj.is_active) def test_organization_cascade_rolls_back_roles(self): obj = self.assign() with patch.object(Company, "save", side_effect=RuntimeError("simulated failure")): with self.assertRaises(RuntimeError): organization_services.deactivate_company(company=self.company) obj.refresh_from_db() self.scope.refresh_from_db() self.assertTrue(obj.is_active) self.assertTrue(self.scope.is_active) self.assertTrue(self.scope.is_primary) def test_assignment_cascade_rolls_back_roles(self): obj = self.assign() with patch.object(UserOrganizationAssignment, "save", side_effect=RuntimeError("simulated failure")): with self.assertRaises(RuntimeError): assignment_services.deactivate_assignment(assignment=self.scope) obj.refresh_from_db() self.assertTrue(obj.is_active) class PermissionTests(AccessFixture): def test_active_path_collects_native_permissions(self): self.role.permissions.add(self.permission) self.assign() self.assertEqual(list(business_permissions_for_user(self.user)), [self.permission]) def test_inactive_paths_are_excluded_even_after_bulk_changes(self): self.role.permissions.add(self.permission) obj = self.assign() for parent in [self.user, self.role, obj, self.scope]: with self.subTest(model=type(parent).__name__), transaction.atomic(): updates = {"is_active": False} if isinstance(parent, UserOrganizationAssignment): updates["is_primary"] = False type(parent).objects.filter(pk=parent.pk).update(**updates) self.assertFalse(business_permissions_for_user(self.user).exists()) transaction.set_rollback(True) def test_permissions_deduplicated_and_aliases_match_same_assignment(self): self.role.permissions.add(self.permission) self.other_role.permissions.add(self.permission, self.other_permission) self.assign() self.assign(role=self.other_role) self.assertEqual(set(business_permissions_for_user(self.user)), {self.permission, self.other_permission}) self.assertEqual(business_permissions_for_user(self.user).count(), 2) self.assertFalse(business_permissions_for_user(self.other_user).exists()) def test_raw_user_mismatch_is_excluded_from_collection(self): self.role.permissions.add(self.permission) UserRoleAssignment.objects.bulk_create([ UserRoleAssignment(user=self.user, role=self.role, organization_assignment=self.foreign_scope)]) self.assertFalse(business_permissions_for_user(self.user).exists()) def test_anonymous_and_unassigned_superuser_get_no_role_inventory(self): self.assertFalse(business_permissions_for_user(AnonymousUser()).exists()) superuser = get_user_model().objects.create_superuser(username="super", password="test-only-password") self.assertFalse(business_permissions_for_user(superuser).exists()) self.assertTrue(superuser.has_perm("organization.view_servicecenter")) def test_django_groups_direct_permissions_and_has_perm_remain_independent(self): self.role.permissions.add(self.permission) self.assign() self.assertFalse(self.user.has_perm("organization.view_servicecenter")) self.user.user_permissions.add(self.permission) fresh = get_user_model().objects.get(pk=self.user.pk) self.assertTrue(fresh.has_perm("organization.view_servicecenter")) group = Group.objects.create(name="Django test group") group.permissions.add(self.other_permission) self.user.groups.add(group) fresh = get_user_model().objects.get(pk=self.user.pk) self.assertTrue(fresh.has_perm("organization.view_company")) self.assertEqual(list(business_permissions_for_user(fresh)), [self.permission]) class AccessAdminTests(AccessFixture): def setUp(self): self.admin_user = get_user_model().objects.create_superuser(username="access-admin", password="test-only-password") self.client.force_login(self.admin_user) def test_models_registered_and_pages_load(self): for model in [Role, UserRoleAssignment]: self.assertIn(model, admin.site._registry) self.assertEqual(self.client.get(reverse(f"admin:access_{model._meta.model_name}_changelist")).status_code, 200) def test_role_admin_selects_existing_permissions(self): response = self.client.post(reverse("admin:access_role_add"), { "code": " from-admin ", "name": "Admin", "is_active": "on", "permissions": [self.permission.pk], "_save": "Save", }) self.assertEqual(response.status_code, 302) self.assertEqual(list(Role.objects.get(code="FROM-ADMIN").permissions.all()), [self.permission]) def test_assignment_admin_rejects_mismatch(self): response = self.client.post(reverse("admin:access_userroleassignment_add"), { "user": self.user.pk, "role": self.role.pk, "organization_assignment": self.foreign_scope.pk, "is_active": "on", "_save": "Save", }) self.assertContains(response, "must belong to the selected user") self.assertFalse(UserRoleAssignment.objects.exists()) def test_assignment_admin_rejects_inactive_dependency_and_duplicate(self): self.assign() response = self.client.post(reverse("admin:access_userroleassignment_add"), { "user": self.user.pk, "role": self.role.pk, "organization_assignment": self.scope.pk, "is_active": "on", "_save": "Save", }) self.assertEqual(response.status_code, 200) self.assertEqual(UserRoleAssignment.objects.count(), 1) services.deactivate_role(role=self.role) response = self.client.post(reverse("admin:access_userroleassignment_add"), { "user": self.user.pk, "role": self.role.pk, "organization_assignment": self.scope.pk, "is_active": "on", "_save": "Save", }) self.assertContains(response, "requires an active dependency") def test_role_admin_action_uses_lifecycle_service(self): obj = self.assign() with patch.object(services, "deactivate_role", wraps=services.deactivate_role) as operation: response = self.client.post(reverse("admin:access_role_changelist"), { "action": "deactivate_selected", "_selected_action": [str(self.role.pk)], }) self.assertEqual(response.status_code, 302) operation.assert_called_once() obj.refresh_from_db() self.assertFalse(obj.is_active) def test_admin_cannot_delete_role_assignment_history(self): obj = self.assign() response = self.client.post(reverse("admin:access_userroleassignment_delete", args=[obj.pk]), {"post": "yes"}) self.assertEqual(response.status_code, 403)