You can not select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
 
 

204 lines
8.8 KiB

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)