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.
242 lines
11 KiB
242 lines
11 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="[email protected]",
|
|
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": "[email protected]",
|
|
"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"], "[email protected]")
|
|
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": "[email protected]",
|
|
"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": "[email protected]",
|
|
"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": "[email protected]"
|
|
}, 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": "[email protected]"
|
|
}, 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": "[email protected]",
|
|
"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": "[email protected]",
|
|
"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="[email protected]",
|
|
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")
|
|
|