from django.test import TestCase from django.core.files.uploadedfile import SimpleUploadedFile from django.contrib.auth import get_user_model from django.contrib.auth.tokens import default_token_generator from rest_framework.test import APIClient from rest_framework import status from apps.account.models.verification import VerificationDocument User = get_user_model() class Phase1AuthAndIdentityTests(TestCase): """ Automated test suite for Phase 1: - User Registration & Immediate JWT Issuance - JWT Token Lifecycle (Obtain, Refresh, Verify) - User Profile (/api/v1/auth/me/) Retrieval & Patch Updates - Verification Document Upload & Retrieval - Password Recovery & Reset - Role-Based Access Control (RBAC) """ def setUp(self): self.client = APIClient() self.user_password = "SecurePassword123!" self.user = User.objects.create_user( email="representative@mashhadcenter.org", password=self.user_password, fullname="Hassan Razavi", user_type=User.UserType.INSTITUTION_ADMIN, country="Iran", city="Mashhad", languages=["fa", "ar", "en"], skills=["Diplomacy", "Arabic Translation"] ) def test_user_registration_success(self): payload = { "email": "new.member@shiahub.org", "password": "StrongMemberPassword123!", "fullname": "Fatima Al-Zahra", "user_type": "editor", "languages": ["ar", "en"], "skills": ["Content Management", "Graphic Design"], "country": "Lebanon", "city": "Beirut" } response = self.client.post("/api/v1/auth/register/", payload, format="json") self.assertEqual(response.status_code, status.HTTP_201_CREATED) self.assertIn("user", response.data) self.assertIn("access", response.data) self.assertIn("refresh", response.data) self.assertEqual(response.data["user"]["email"], "new.member@shiahub.org") self.assertEqual(response.data["user"]["fullname"], "Fatima Al-Zahra") self.assertEqual(response.data["user"]["user_type"], "editor") self.assertIn("editor", response.data["user"]["roles"]) def test_user_registration_duplicate_email(self): payload = { "email": "representative@mashhadcenter.org", "password": "AnotherPassword123!", "fullname": "Duplicate User" } response = self.client.post("/api/v1/auth/register/", payload, format="json") self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST) def test_jwt_token_lifecycle(self): # 1. Obtain Pair obtain_resp = self.client.post("/api/v1/auth/token/", { "email": "representative@mashhadcenter.org", "password": self.user_password }, format="json") self.assertEqual(obtain_resp.status_code, status.HTTP_200_OK) access_token = obtain_resp.data["access"] refresh_token = obtain_resp.data["refresh"] # 2. Verify verify_resp = self.client.post("/api/v1/auth/token/verify/", { "token": access_token }, format="json") self.assertEqual(verify_resp.status_code, status.HTTP_200_OK) # 3. Refresh refresh_resp = self.client.post("/api/v1/auth/token/refresh/", { "refresh": refresh_token }, format="json") self.assertEqual(refresh_resp.status_code, status.HTTP_200_OK) self.assertIn("access", refresh_resp.data) def test_user_me_unauthenticated(self): response = self.client.get("/api/v1/auth/me/") self.assertEqual(response.status_code, status.HTTP_401_UNAUTHORIZED) def test_user_me_authenticated(self): self.client.force_authenticate(user=self.user) response = self.client.get("/api/v1/auth/me/") self.assertEqual(response.status_code, status.HTTP_200_OK) self.assertEqual(response.data["email"], self.user.email) self.assertEqual(response.data["fullname"], "Hassan Razavi") self.assertIn("institution_admin", response.data["roles"]) self.assertIn("manage_institution_profile", response.data["permissions"]) self.assertEqual(response.data["verification_status"], "unverified") self.assertFalse(response.data["is_verified"]) def test_user_me_patch_update(self): self.client.force_authenticate(user=self.user) patch_payload = { "fullname": "Hassan M. Razavi", "bio": "Senior Director of International Relations & Cultural Affairs.", "city": "Tehran", "skills": ["Cultural Diplomacy", "Conference Management"], "languages": ["fa", "ar", "en", "ur"] } response = self.client.patch("/api/v1/auth/me/", patch_payload, format="json") self.assertEqual(response.status_code, status.HTTP_200_OK) self.assertEqual(response.data["fullname"], "Hassan M. Razavi") self.assertEqual(response.data["bio"], patch_payload["bio"]) self.assertEqual(response.data["city"], "Tehran") self.assertEqual(response.data["languages"], ["fa", "ar", "en", "ur"]) # Check DB update self.user.refresh_from_db() self.assertEqual(self.user.fullname, "Hassan M. Razavi") self.assertEqual(self.user.info, patch_payload["bio"]) def test_verification_document_upload_and_status(self): self.client.force_authenticate(user=self.user) # 1. Upload Document test_file = SimpleUploadedFile( "institution_license.pdf", b"%PDF-1.4 Mock License Content", content_type="application/pdf" ) upload_resp = self.client.post("/api/v1/auth/documents/", { "document_type": "institution_license", "title": "Official Cultural Center License", "document_file": test_file }, format="multipart") self.assertEqual(upload_resp.status_code, status.HTTP_201_CREATED) self.assertEqual(upload_resp.data["status"], "pending") self.assertEqual(upload_resp.data["document_type"], "institution_license") # 2. List Documents list_resp = self.client.get("/api/v1/auth/documents/") self.assertEqual(list_resp.status_code, status.HTTP_200_OK) self.assertEqual(len(list_resp.data), 1) # 3. Check User Me Verification Status me_resp = self.client.get("/api/v1/auth/me/") self.assertEqual(me_resp.data["verification_status"], "pending") # 4. Admin Approves Document doc = VerificationDocument.objects.get(id=upload_resp.data["id"]) doc.status = VerificationDocument.Status.VERIFIED doc.save() me_resp_after = self.client.get("/api/v1/auth/me/") self.assertEqual(me_resp_after.data["verification_status"], "verified") self.assertTrue(me_resp_after.data["is_verified"]) def test_password_recovery_and_reset_flow(self): # 1. Recovery request for existing user rec_resp = self.client.post("/api/v1/auth/recover-password/", { "email": "representative@mashhadcenter.org" }, format="json") self.assertEqual(rec_resp.status_code, status.HTTP_200_OK) # 2. Recovery request for non-existing user (silent 200 for security) rec_non_exist = self.client.post("/api/v1/auth/recover-password/", { "email": "unknown@example.com" }, format="json") self.assertEqual(rec_non_exist.status_code, status.HTTP_200_OK) # 3. Generate token & reset password token = default_token_generator.make_token(self.user) reset_resp = self.client.post("/api/v1/auth/reset-password/", { "email": "representative@mashhadcenter.org", "token": token, "new_password": "NewBrandSecurePass123!" }, format="json") self.assertEqual(reset_resp.status_code, status.HTTP_200_OK) # 4. Verify login with new password login_resp = self.client.post("/api/v1/auth/token/", { "email": "representative@mashhadcenter.org", "password": "NewBrandSecurePass123!" }, format="json") self.assertEqual(login_resp.status_code, status.HTTP_200_OK) self.assertIn("access", login_resp.data) def test_rbac_user_properties(self): self.assertTrue(self.user.is_institution_admin) self.assertTrue(self.user.is_editor) self.assertFalse(self.user.is_super_admin) super_admin = User.objects.create_superuser( email="superadmin@razavi.org", password="SuperPassword123!" ) self.assertTrue(super_admin.is_super_admin) self.assertTrue(super_admin.is_regional_admin) self.assertTrue(super_admin.is_institution_admin) self.assertTrue(super_admin.is_editor) def test_user_profile_extended_fields_crud(self): self.client.force_authenticate(user=self.user) payload = { "first_name": "Ali", "last_name": "Rezaei", "title": "Dr", "rank": "Professor", "institution": "Ferdowsi University", "website": "https://example.com/scholar", "phone": "09123456789", "address": "Mashhad, Iran" } # PATCH update response = self.client.patch("/api/v1/auth/me/", payload, format="json") self.assertEqual(response.status_code, status.HTTP_200_OK) self.assertEqual(response.data["first_name"], "Ali") self.assertEqual(response.data["last_name"], "Rezaei") self.assertEqual(response.data["fullname"], "Ali Rezaei") self.assertEqual(response.data["title"], "Dr") self.assertEqual(response.data["rank"], "Professor") self.assertEqual(response.data["institution"], "Ferdowsi University") self.assertEqual(response.data["website"], "https://example.com/scholar") self.assertEqual(response.data["phone"], "09123456789") self.assertEqual(response.data["address"], "Mashhad, Iran") # GET retrieval get_resp = self.client.get("/api/v1/auth/me/") self.assertEqual(get_resp.status_code, status.HTTP_200_OK) self.assertEqual(get_resp.data["first_name"], "Ali") self.assertEqual(get_resp.data["last_name"], "Rezaei") self.assertEqual(get_resp.data["title"], "Dr") self.assertEqual(get_resp.data["rank"], "Professor") self.assertEqual(get_resp.data["institution"], "Ferdowsi University") self.assertEqual(get_resp.data["website"], "https://example.com/scholar") self.assertEqual(get_resp.data["phone"], "09123456789") self.assertEqual(get_resp.data["address"], "Mashhad, Iran") class TwoFactorAuthenticationTests(TestCase): """ Automated test suite for Two-Factor Authentication (2FA): - User self-toggle ON/OFF (/api/v1/auth/2fa/toggle/ and /api/v1/auth/me/) - Email OTP 2FA challenge & verification (/api/v1/auth/token/ -> /api/v1/auth/2fa/verify/) - Authenticator App (TOTP RFC 6238) setup, QR code, and verification - Single-use 8-character Backup Recovery Codes verification & consumption - Admin Force 2FA permission (`require_2fa` on User.custom_permissions and Role.require_2fa) preventing login without 2FA and blocking user from disabling 2FA """ def setUp(self): self.client = APIClient() self.password = "StrongPass2FA#2026" self.user = User.objects.create_user( email="scholar2fa@mashhadcenter.org", password=self.password, fullname="Dr. Mahdi Ansari", user_type=User.UserType.CLIENT, ) self.admin = User.objects.create_superuser( email="admin2fa@mashhadcenter.org", password=self.password, fullname="System Security Admin", ) def test_user_self_toggle_email_2fa_and_login_flow(self): # 1. Initially 2FA is not required; login issues JWT tokens directly login_initial = self.client.post("/api/v1/auth/token/", { "email": self.user.email, "password": self.password, }, format="json") self.assertEqual(login_initial.status_code, status.HTTP_200_OK) self.assertIn("access", login_initial.data) self.assertFalse(login_initial.data.get("requires_2fa", False)) # 2. User enables Email 2FA for themselves self.client.force_authenticate(user=self.user) toggle_on = self.client.post("/api/v1/auth/2fa/toggle/", { "enabled": True, "method": "email", }, format="json") self.assertEqual(toggle_on.status_code, status.HTTP_200_OK) self.assertTrue(toggle_on.data["two_factor_enabled"]) self.assertTrue(toggle_on.data["is_2fa_required"]) self.assertFalse(toggle_on.data["is_2fa_forced_by_admin"]) # 3. Now logging in with password returns 2FA challenge instead of JWT tokens self.client.force_authenticate(user=None) login_2fa = self.client.post("/api/v1/auth/token/", { "email": self.user.email, "password": self.password, }, format="json") self.assertEqual(login_2fa.status_code, status.HTTP_200_OK) self.assertTrue(login_2fa.data.get("requires_2fa")) self.assertNotIn("access", login_2fa.data) temp_token = login_2fa.data["temp_token"] # Retrieve the generated OTP from RedisManager from utils.redis import RedisManager redis_mgr = RedisManager() otp_code = redis_mgr.get_2fa_email_otp(temp_token) self.assertIsNotNone(otp_code) self.assertEqual(len(otp_code), 6) # 4. Wrong OTP fails with 400 bad_verify = self.client.post("/api/v1/auth/2fa/verify/", { "temp_token": temp_token, "code": "000000", "method": "email", }, format="json") self.assertEqual(bad_verify.status_code, status.HTTP_400_BAD_REQUEST) # 5. Correct OTP succeeds and issues JWT access + refresh tokens good_verify = self.client.post("/api/v1/auth/2fa/verify/", { "temp_token": temp_token, "code": otp_code, "method": "email", }, format="json") self.assertEqual(good_verify.status_code, status.HTTP_200_OK) self.assertIn("access", good_verify.data) self.assertIn("refresh", good_verify.data) self.assertEqual(good_verify.data["user"]["email"], self.user.email) # 6. User can toggle 2FA OFF for themselves when not forced by admin self.client.force_authenticate(user=self.user) toggle_off = self.client.post("/api/v1/auth/2fa/toggle/", { "enabled": False, }, format="json") self.assertEqual(toggle_off.status_code, status.HTTP_200_OK) self.assertFalse(toggle_off.data["two_factor_enabled"]) self.assertFalse(toggle_off.data["is_2fa_required"]) def test_admin_force_2fa_locks_user_and_enforces_2fa_on_login(self): # 1. Admin forces 2FA on user via AdminUserDetailView PATCH self.client.force_authenticate(user=self.admin) admin_patch = self.client.patch(f"/api/v1/account/admin/users/{self.user.id}/", { "require_2fa": True, }, format="json") self.assertEqual(admin_patch.status_code, status.HTTP_200_OK) self.assertTrue(admin_patch.data["require_2fa"]) self.assertTrue(admin_patch.data["is_2fa_forced_by_admin"]) self.assertTrue(admin_patch.data["is_2fa_required"]) self.user.refresh_from_db() self.assertTrue(self.user.is_2fa_forced_by_admin()) self.assertTrue(self.user.is_2fa_required()) # 2. User cannot log in without completing 2FA self.client.force_authenticate(user=None) login_resp = self.client.post("/api/v1/auth/token/", { "email": self.user.email, "password": self.password, }, format="json") self.assertEqual(login_resp.status_code, status.HTTP_200_OK) self.assertTrue(login_resp.data["requires_2fa"]) self.assertTrue(login_resp.data["is_forced_by_admin"]) self.assertNotIn("access", login_resp.data) # 3. User cannot disable 2FA via /api/v1/auth/2fa/toggle/ (403 Forbidden) self.client.force_authenticate(user=self.user) disable_toggle = self.client.post("/api/v1/auth/2fa/toggle/", { "enabled": False, }, format="json") self.assertEqual(disable_toggle.status_code, status.HTTP_403_FORBIDDEN) # 4. User cannot disable 2FA via /api/v1/auth/2fa/disable/ (403 Forbidden) disable_view = self.client.post("/api/v1/auth/2fa/disable/", { "password": self.password, }, format="json") self.assertEqual(disable_view.status_code, status.HTTP_403_FORBIDDEN) # 5. User cannot disable 2FA via PATCH /api/v1/auth/me/ (403 PermissionDenied / 400 ValidationError) me_patch = self.client.patch("/api/v1/auth/me/", { "two_factor_enabled": False, }, format="json") self.assertIn(me_patch.status_code, (status.HTTP_400_BAD_REQUEST, status.HTTP_403_FORBIDDEN)) # 6. Admin un-forces require_2fa -> user can now toggle 2FA OFF self.client.force_authenticate(user=self.admin) admin_unforce = self.client.patch(f"/api/v1/account/admin/users/{self.user.id}/", { "require_2fa": False, }, format="json") self.assertEqual(admin_unforce.status_code, status.HTTP_200_OK) self.assertFalse(admin_unforce.data["is_2fa_forced_by_admin"]) self.user.refresh_from_db() self.client.force_authenticate(user=self.user) user_toggle_off = self.client.post("/api/v1/auth/2fa/toggle/", { "enabled": False, }, format="json") self.assertEqual(user_toggle_off.status_code, status.HTTP_200_OK) self.assertFalse(user_toggle_off.data["is_2fa_required"]) def test_totp_setup_confirm_and_backup_codes_login(self): from utils.totp import get_totp_code self.client.force_authenticate(user=self.user) # 1. Initiate TOTP setup init_resp = self.client.post("/api/v1/auth/2fa/setup/initiate/", { "method": "totp", }, format="json") self.assertEqual(init_resp.status_code, status.HTTP_200_OK) self.assertIn("secret", init_resp.data) self.assertIn("qr_code_data_url", init_resp.data) secret = init_resp.data["secret"] # 2. Confirm TOTP setup with live RFC 6238 code live_totp = get_totp_code(secret) confirm_resp = self.client.post("/api/v1/auth/2fa/setup/confirm/", { "method": "totp", "code": live_totp, }, format="json") self.assertEqual(confirm_resp.status_code, status.HTTP_200_OK) self.assertTrue(confirm_resp.data["two_factor_enabled"]) self.assertEqual(confirm_resp.data["two_factor_method"], "totp") backup_codes = confirm_resp.data["backup_codes"] self.assertEqual(len(backup_codes), 8) # 3. Login with TOTP code self.client.force_authenticate(user=None) login_challenge = self.client.post("/api/v1/auth/token/", { "email": self.user.email, "password": self.password, }, format="json") self.assertTrue(login_challenge.data["requires_2fa"]) self.assertEqual(login_challenge.data["method"], "totp") verify_totp = self.client.post("/api/v1/auth/2fa/verify/", { "temp_token": login_challenge.data["temp_token"], "code": get_totp_code(secret), "method": "totp", }, format="json") self.assertEqual(verify_totp.status_code, status.HTTP_200_OK) self.assertIn("access", verify_totp.data) # 4. Login with a single-use Backup Code login_challenge_2 = self.client.post("/api/v1/auth/token/", { "email": self.user.email, "password": self.password, }, format="json") first_backup_code = backup_codes[0] verify_backup = self.client.post("/api/v1/auth/2fa/verify/", { "temp_token": login_challenge_2.data["temp_token"], "code": first_backup_code, "method": "backup_code", }, format="json") self.assertEqual(verify_backup.status_code, status.HTTP_200_OK) self.assertIn("access", verify_backup.data) # 5. Re-using the same consumed backup code fails login_challenge_3 = self.client.post("/api/v1/auth/token/", { "email": self.user.email, "password": self.password, }, format="json") reuse_backup = self.client.post("/api/v1/auth/2fa/verify/", { "temp_token": login_challenge_3.data["temp_token"], "code": first_backup_code, "method": "backup_code", }, format="json") self.assertEqual(reuse_backup.status_code, status.HTTP_400_BAD_REQUEST)