from django.contrib.auth import get_user_model from rest_framework import status from rest_framework.test import APITestCase from checkpoint.models import CheckPoint, FacilityType from users.models import AuditLog User = get_user_model() class UserListCheckpointScopeTests(APITestCase): endpoint = "/api/v1/users/" def setUp(self): facility_type = FacilityType.objects.create(name="Border Point") self.checkpoint_one = CheckPoint.objects.create( name="Checkpoint One", location="Kakarbhitta", facility_type=facility_type, ) self.checkpoint_two = CheckPoint.objects.create( name="Checkpoint Two", location="Birgunj", facility_type=facility_type, ) self.staff_one = User.objects.create_user( email="staff1@example.com", password="password123", name="Staff One", checkpoint=self.checkpoint_one, role=User.Role.STAFF, ) self.staff_one_peer = User.objects.create_user( email="staff1b@example.com", password="password123", name="Staff One Peer", checkpoint=self.checkpoint_one, role=User.Role.STAFF, ) self.staff_two = User.objects.create_user( email="staff2@example.com", password="password123", name="Staff Two", checkpoint=self.checkpoint_two, role=User.Role.STAFF, ) self.admin_user = User.objects.create_user( email="admin@example.com", password="password123", name="Admin User", role=User.Role.ADMIN, ) @staticmethod def _result_ids(response): results = response.data.get("results", response.data) return {item["id"] for item in results} def test_staff_lists_only_users_from_own_checkpoint(self): self.client.force_authenticate(user=self.staff_one) response = self.client.get(self.endpoint) self.assertEqual(response.status_code, status.HTTP_200_OK) self.assertSetEqual( self._result_ids(response), {str(self.staff_one.id), str(self.staff_one_peer.id)}, ) def test_staff_cannot_filter_other_checkpoint_users(self): self.client.force_authenticate(user=self.staff_one) response = self.client.get( self.endpoint, {"checkpoint": str(self.checkpoint_two.id)} ) self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST) details = response.data["error"]["details"] self.assertIn("checkpoint", [detail.get("field") for detail in details]) def test_admin_lists_all_non_superusers(self): self.client.force_authenticate(user=self.admin_user) response = self.client.get(self.endpoint) self.assertEqual(response.status_code, status.HTTP_200_OK) self.assertSetEqual( self._result_ids(response), { str(self.staff_one.id), str(self.staff_one_peer.id), str(self.staff_two.id), str(self.admin_user.id), }, ) class AdminUserStatusUpdateTests(APITestCase): endpoint = "/api/v1/users/" def setUp(self): facility_type = FacilityType.objects.create(name="Border Point") self.checkpoint = CheckPoint.objects.create( name="Checkpoint One", location="Kakarbhitta", facility_type=facility_type, ) self.staff_user = User.objects.create_user( email="managed.staff@example.com", password="password123", name="Managed Staff", checkpoint=self.checkpoint, role=User.Role.STAFF, ) self.admin_user = User.objects.create_user( email="status.admin@example.com", password="password123", name="Status Admin", role=User.Role.ADMIN, ) def detail_endpoint(self, user): return f"{self.endpoint}{user.id}/" def test_admin_can_suspend_staff_user(self): self.client.force_authenticate(user=self.admin_user) response = self.client.patch( self.detail_endpoint(self.staff_user), {"status": User.Status.SUSPENDED}, format="json", ) self.assertEqual(response.status_code, status.HTTP_200_OK) self.assertEqual(response.data["status"], User.Status.SUSPENDED) self.assertFalse(response.data["is_active"]) self.staff_user.refresh_from_db() self.assertEqual(self.staff_user.status, User.Status.SUSPENDED) self.assertFalse(self.staff_user.is_active) def test_admin_can_reactivate_suspended_staff_user(self): self.staff_user.status = User.Status.SUSPENDED self.staff_user.is_active = False self.staff_user.save(update_fields=["status", "is_active", "updated_at"]) self.client.force_authenticate(user=self.admin_user) response = self.client.patch( self.detail_endpoint(self.staff_user), {"status": User.Status.ACTIVE}, format="json", ) self.assertEqual(response.status_code, status.HTTP_200_OK) self.assertEqual(response.data["status"], User.Status.ACTIVE) self.assertTrue(response.data["is_active"]) self.staff_user.refresh_from_db() self.assertEqual(self.staff_user.status, User.Status.ACTIVE) self.assertTrue(self.staff_user.is_active) def test_staff_cannot_update_user_status(self): self.client.force_authenticate(user=self.staff_user) response = self.client.patch( self.detail_endpoint(self.admin_user), {"status": User.Status.SUSPENDED}, format="json", ) self.assertEqual(response.status_code, status.HTTP_403_FORBIDDEN) self.admin_user.refresh_from_db() self.assertEqual(self.admin_user.status, User.Status.ACTIVE) class UserStatusAuthenticationTests(APITestCase): login_endpoint = "/api/v1/users/auth/login/" profile_endpoint = "/api/v1/users/me/" def setUp(self): facility_type = FacilityType.objects.create(name="Border Point") self.checkpoint = CheckPoint.objects.create( name="Checkpoint One", location="Kakarbhitta", facility_type=facility_type, ) self.user = User.objects.create_user( email="auth.staff@example.com", password="StrongPass123!", name="Auth Staff", checkpoint=self.checkpoint, role=User.Role.STAFF, ) def test_suspended_user_cannot_login_even_when_is_active_true(self): self.user.status = User.Status.SUSPENDED self.user.is_active = True self.user.save(update_fields=["status", "is_active", "updated_at"]) response = self.client.post( self.login_endpoint, {"email": self.user.email, "password": "StrongPass123!"}, format="json", ) self.assertEqual(response.status_code, status.HTTP_401_UNAUTHORIZED) def test_existing_token_stops_working_after_user_is_suspended(self): login_response = self.client.post( self.login_endpoint, {"email": self.user.email, "password": "StrongPass123!"}, format="json", ) access_token = login_response.data["access"] self.user.status = User.Status.SUSPENDED self.user.is_active = True self.user.save(update_fields=["status", "is_active", "updated_at"]) self.client.credentials(HTTP_AUTHORIZATION=f"Bearer {access_token}") response = self.client.get(self.profile_endpoint) self.assertEqual(response.status_code, status.HTTP_401_UNAUTHORIZED) def test_suspended_user_cannot_refresh_token(self): login_response = self.client.post( self.login_endpoint, {"email": self.user.email, "password": "StrongPass123!"}, format="json", ) refresh_token = login_response.data["refresh"] self.user.status = User.Status.SUSPENDED self.user.is_active = True self.user.save(update_fields=["status", "is_active", "updated_at"]) response = self.client.post( "/api/v1/users/auth/refresh/", {"refresh": refresh_token}, format="json", ) self.assertEqual(response.status_code, status.HTTP_401_UNAUTHORIZED) class UserProfileUpdatePermissionsTests(APITestCase): endpoint = "/api/v1/users/me/" def setUp(self): facility_type = FacilityType.objects.create(name="Border Point") self.checkpoint = CheckPoint.objects.create( name="Checkpoint One", location="Kakarbhitta", facility_type=facility_type, ) self.user = User.objects.create_user( email="profile.staff@example.com", password="StrongPass123!", name="Profile Staff", checkpoint=self.checkpoint, role=User.Role.STAFF, ) def test_profile_patch_cannot_change_role_or_status(self): self.client.force_authenticate(user=self.user) response = self.client.patch( self.endpoint, { "name": "Updated Staff", "role": User.Role.ADMIN, "status": User.Status.SUSPENDED, }, format="json", ) self.assertEqual(response.status_code, status.HTTP_200_OK) self.user.refresh_from_db() self.assertEqual(self.user.name, "Updated Staff") self.assertEqual(self.user.role, User.Role.STAFF) self.assertEqual(self.user.status, User.Status.ACTIVE) class UserRegistrationCheckpointRequirementTests(APITestCase): endpoint = "/api/v1/users/auth/register/" def setUp(self): facility_type = FacilityType.objects.create(name="Border Point") self.checkpoint = CheckPoint.objects.create( name="Checkpoint One", location="Kakarbhitta", facility_type=facility_type, ) def test_registration_requires_checkpoint(self): response = self.client.post( self.endpoint, { "email": "new.user@example.com", "name": "New User", "password": "StrongPass123!", "password_confirm": "StrongPass123!", "role": User.Role.STAFF, }, format="json", ) self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST) details = response.data["error"]["details"] self.assertIn("checkpoint", [detail.get("field") for detail in details]) def test_registration_succeeds_with_checkpoint(self): response = self.client.post( self.endpoint, { "email": "new.user@example.com", "name": "New User", "password": "StrongPass123!", "password_confirm": "StrongPass123!", "checkpoint": str(self.checkpoint.id), "role": User.Role.STAFF, }, format="json", ) self.assertEqual(response.status_code, status.HTTP_201_CREATED) created_user = User.objects.get(email="new.user@example.com") self.assertEqual(created_user.checkpoint_id, self.checkpoint.id) class UserEmailCaseSensitivityTests(APITestCase): login_endpoint = "/api/v1/users/auth/login/" register_endpoint = "/api/v1/users/auth/register/" def setUp(self): facility_type = FacilityType.objects.create(name="Border Point") self.checkpoint = CheckPoint.objects.create( name="Checkpoint One", location="Kakarbhitta", facility_type=facility_type, ) self.user = User.objects.create_user( email="case.user@example.com", password="StrongPass123!", name="Case User", checkpoint=self.checkpoint, role=User.Role.STAFF, ) def test_login_accepts_mixed_case_email(self): response = self.client.post( self.login_endpoint, {"email": "Case.User@Example.COM", "password": "StrongPass123!"}, format="json", ) self.assertEqual(response.status_code, status.HTTP_200_OK) self.assertEqual(response.data["user"]["id"], str(self.user.id)) def test_registration_rejects_duplicate_email_with_different_case(self): response = self.client.post( self.register_endpoint, { "email": "Case.User@Example.COM", "name": "Duplicate User", "password": "StrongPass123!", "password_confirm": "StrongPass123!", "checkpoint": str(self.checkpoint.id), "role": User.Role.STAFF, }, format="json", ) self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST) details = response.data["error"]["details"] self.assertIn("email", [detail.get("field") for detail in details]) class AuditLogAPITests(APITestCase): endpoint = "/api/v1/users/audit-logs/" def setUp(self): self.staff_user = User.objects.create_user( email="activity.staff@example.com", password="password123", name="Activity Staff", role=User.Role.STAFF, ) self.other_staff_user = User.objects.create_user( email="other.activity.staff@example.com", password="password123", name="Other Activity Staff", role=User.Role.STAFF, ) self.admin_user = User.objects.create_user( email="activity.admin@example.com", password="password123", name="Activity Admin", role=User.Role.ADMIN, ) AuditLog.objects.create( user=self.other_staff_user, action=AuditLog.ActionChoices.UPDATE, action_description="Updated Export Submission TRK-2083-002", method="PATCH", path="/api/v1/record/export-submissions/", status_code=status.HTTP_200_OK, target_model="ExportSubmission", response_summary={"tracking_code": "TRK-2083-002"}, ) self.audit_log = AuditLog.objects.create( user=self.staff_user, action=AuditLog.ActionChoices.CREATE, action_description="Created Import Submission TRK-2083-001", method="POST", path="/api/v1/record/import-submissions/", status_code=status.HTTP_201_CREATED, target_model="ImportSubmission", response_summary={"tracking_code": "TRK-2083-001"}, ) @staticmethod def _results(response): return response.data.get("results", response.data) def test_admin_can_list_audit_logs(self): self.client.force_authenticate(user=self.admin_user) response = self.client.get(self.endpoint) self.assertEqual(response.status_code, status.HTTP_200_OK) item = next( result for result in self._results(response) if result["user_name"] == "Activity Staff" ) self.assertEqual(item["user_name"], "Activity Staff") self.assertEqual(item["action"], AuditLog.ActionChoices.CREATE) self.assertEqual( item["action_description"], "Created Import Submission TRK-2083-001", ) self.assertEqual(item["target_model"], "ImportSubmission") self.assertIn("created_at", item) def test_admin_can_filter_audit_logs_by_staff(self): self.client.force_authenticate(user=self.admin_user) response = self.client.get(self.endpoint, {"staff": str(self.staff_user.id)}) self.assertEqual(response.status_code, status.HTTP_200_OK) results = self._results(response) self.assertEqual(len(results), 1) self.assertEqual(results[0]["user_name"], "Activity Staff") def test_audit_staff_filter_rejects_invalid_id(self): self.client.force_authenticate(user=self.admin_user) response = self.client.get(self.endpoint, {"staff": "not-a-user-id"}) self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST) def test_non_admin_cannot_list_audit_logs(self): self.client.force_authenticate(user=self.staff_user) response = self.client.get(self.endpoint) self.assertEqual(response.status_code, status.HTTP_403_FORBIDDEN) def test_activity_logs_endpoint_is_not_exposed(self): self.client.force_authenticate(user=self.admin_user) response = self.client.get("/api/v1/users/activity-logs/") self.assertEqual(response.status_code, status.HTTP_404_NOT_FOUND) def test_login_activity_message_is_friendly(self): log = AuditLog.objects.create( user=self.staff_user, action=AuditLog.ActionChoices.LOGIN, action_description="User logged in", method="POST", path="/api/v1/users/auth/login/", status_code=status.HTTP_200_OK, target_model="User", target_object_id=self.staff_user.id, ) self.assertEqual( log.activity_message, "Activity Staff logged in to the system.", )