GuilanCE/guilan-ace-backend
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

test: migrate backend suites to pytest with module coverage gate

Browse Source 
Split backend tests into unit and integration suites, add isolated fixtures and coverage enforcement, and update CI and test documentation. Cover current authentication, content, notification, payment, media, and admin flows; fix behaviors revealed by the new tests. Validate both suites and the per-module gate against local PostgreSQL.
This commit is contained in:
Amirhossein Khalili
2026-09-26 16:50:29 +03:30
parent 3ec28638d1
commit ff99130938
76 changed files with 3271 additions and 1357 deletions
Download Patch File Download Diff File Expand all files Collapse all files

171
apps/users/tests/integration/test_api_flows.py Normal file
View File

@@ -0,0 +1,171 @@
import json
from io import BytesIO
import pytest
from django.core.files.uploadedfile import SimpleUploadedFile
from PIL import Image
from apps.users.models import Major, University, User
from apps.users.services import auth
from core.authentication import create_jwt_token
pytestmark = [pytest.mark.integration, pytest.mark.django_db]
def _post(client, path, payload, headers=None):
return client.post(path, data=json.dumps(payload), content_type="application/json", **(headers or {}))
def test_otp_registration_login_refresh_and_reset(client, fake_redis, monkeypatch, mocker):
monkeypatch.setattr(auth, "_redis", lambda: fake_redis)
sms = mocker.patch("apps.users.services.auth.send_critical_sms.delay")
mobile = "09197770001"
assert _post(client, "/api/auth/otp/send", {"mobile": "bad", "mode": "register"}).status_code == 400
sent = _post(client, "/api/auth/otp/send", {"mobile": mobile, "mode": "register"})
assert sent.status_code == 200, sent.content
code = fake_redis.get(auth._otp_key("register", mobile)).decode()
sms.assert_called_with(mobile, auth.SMS_KIND_REGISTER, code)
assert _post(client, "/api/auth/otp/verify-register", {"mobile": mobile, "code": "00000"}).status_code == 400
assert _post(client, "/api/auth/otp/verify-register", {"mobile": mobile, "code": code}).status_code == 200
assert fake_redis.get(auth._otp_key("register", mobile)) is None
registration = {"username": "otp_member", "mobile": mobile, "password": "StrongPass123!", "code": code}
assert _post(client, "/api/auth/register", registration).status_code == 201
member = User.objects.get(username="otp_member")
assert member.is_mobile_verified
assert _post(client, "/api/auth/register", registration).status_code == 400
assert _post(client, "/api/auth/login", {"identifier": mobile, "password": "wrong"}).status_code == 401
tokens = _post(client, "/api/auth/login", {"identifier": mobile, "password": "StrongPass123!"}).json()
assert tokens["access_token"] and tokens["refresh_token"]
assert _post(client, "/api/auth/refresh", {"refresh_token": tokens["access_token"]}).status_code == 401
assert _post(client, "/api/auth/refresh", {"refresh_token": tokens["refresh_token"]}).status_code == 200
assert _post(client, "/api/auth/otp/send", {"mobile": mobile, "mode": "reset_password"}).status_code == 200
reset_code = fake_redis.get(auth._otp_key("reset_password", mobile)).decode()
reset = _post(client, "/api/auth/reset-password", {"mobile": mobile, "code": reset_code, "new_password": "NewStrongPass123!"})
assert reset.status_code == 200, reset.content
member.refresh_from_db()
assert member.check_password("NewStrongPass123!")
def test_profile_and_authorization_roles(client, user_factory, auth_headers):
admin = user_factory(is_superuser=True, is_staff=True)
member = user_factory(email="member@example.com")
headers = auth_headers(member)
assert client.get("/api/auth/profile", **headers).json()["username"] == member.username
invalid = client.put("/api/auth/profile", data=json.dumps({"major": "missing-code"}),
content_type="application/json", **headers)
assert invalid.status_code == 400
assert client.put("/api/auth/profile", data=json.dumps({"first_name": "New", "email": " NEW@EXAMPLE.COM "}),
content_type="application/json", **headers).status_code == 200
member.refresh_from_db()
assert member.first_name == "New" and member.email == "new@example.com"
assert client.get("/api/auth/users", **headers).status_code == 403
admin_headers = auth_headers(admin)
assert client.get("/api/auth/users", {"search": member.username}, **admin_headers).json()[0]["id"] == member.id
assert client.get(f"/api/auth/users/{member.pk}", **admin_headers).json()["id"] == member.pk
assert client.get("/api/auth/roles", **headers).status_code == 403
assert client.get("/api/auth/roles", **admin_headers).status_code == 200
assert client.put(f"/api/auth/users/{admin.pk}/authorization", data=json.dumps({"groups": [], "is_staff": False}),
content_type="application/json", **admin_headers).status_code == 400
changed = client.put(f"/api/auth/users/{member.pk}/authorization",
data=json.dumps({"groups": ["blog_editor"], "is_staff": True}),
content_type="application/json", **admin_headers)
assert changed.status_code == 200, changed.content
member.refresh_from_db()
assert member.is_staff and member.groups.filter(name="blog_editor").exists()
assert client.get("/api/auth/check-username", {"username": member.username}).json()["exists"]
assert client.get("/api/auth/check-mobile", {"mobile": member.mobile}).json()["exists"]
def test_profile_picture_upload_validation_and_delete(client, user_factory, auth_headers, settings, tmp_path):
settings.MEDIA_ROOT = tmp_path
user = user_factory()
headers = auth_headers(user)
path = "/api/auth/profile/picture"
assert client.post(path, **headers).status_code == 400
invalid = client.post(path, {"file": SimpleUploadedFile("bad.txt", b"text", content_type="text/plain")}, **headers)
assert invalid.status_code == 400
image_bytes = BytesIO()
Image.new("RGB", (20, 20), "blue").save(image_bytes, format="PNG")
uploaded = client.post(path, {"file": SimpleUploadedFile("avatar.png", image_bytes.getvalue(), content_type="image/png")}, **headers)
assert uploaded.status_code == 200, uploaded.content
user.refresh_from_db()
assert user.profile_picture
assert client.delete(path, **headers).status_code == 200
user.refresh_from_db()
assert not user.profile_picture
def test_user_admin_filters_soft_delete_and_restore(client, user_factory, auth_headers):
admin = user_factory(is_superuser=True, is_staff=True)
staff = user_factory(is_staff=True)
member = user_factory(username="filter_member", student_id="1234567890", is_active=False)
headers = auth_headers(admin)
path = "/api/auth/users"
assert [row["id"] for row in client.get(path, {"search": "filter_member"}, **headers).json()] == [member.pk]
assert [row["id"] for row in client.get(path, {"student_id": "1234567890"}, **headers).json()] == [member.pk]
assert all(row["is_staff"] for row in client.get(path, {"role": "staff"}, **headers).json())
assert all(row["is_superuser"] for row in client.get(path, {"role": "superuser"}, **headers).json())
assert [row["id"] for row in client.get(path, {"is_active": "false"}, **headers).json()] == [member.pk]
assert client.get(path + "/deleted", **auth_headers(staff)).status_code == 200
member.delete()
assert client.get(path + "/deleted", **auth_headers(staff)).json()[0]["id"] == member.pk
restored = client.post(f"{path}/{member.pk}/restore", **auth_headers(staff))
assert restored.status_code == 200, restored.content
assert User.objects.filter(pk=member.pk).exists()
assert client.post(f"{path}/{member.pk}/restore", **auth_headers(staff)).status_code == 400
def test_mobile_otp_login_and_verification_api(client, user_factory, auth_headers, fake_redis, monkeypatch, mocker):
monkeypatch.setattr(auth, "_redis", lambda: fake_redis)
mocker.patch("apps.users.services.auth.send_critical_sms.delay")
user = user_factory(is_mobile_verified=False)
assert _post(client, "/api/auth/login/otp", {"mobile": user.mobile, "code": "12345"}).status_code == 400
assert _post(client, "/api/auth/otp/send", {"mobile": user.mobile, "mode": "login"}).status_code == 200
code = fake_redis.get(auth._otp_key("login", user.mobile)).decode()
assert _post(client, "/api/auth/login/otp", {"mobile": user.mobile, "code": code}).status_code == 200
user.refresh_from_db()
assert user.is_mobile_verified
new_mobile = "09197770078"
headers = auth_headers(user)
assert _post(client, "/api/auth/mobile/send-otp", {"mobile": new_mobile}, headers).status_code == 200
verify_code = fake_redis.get(auth._otp_key("verify_mobile", new_mobile)).decode()
assert _post(client, "/api/auth/mobile/verify", {"mobile": new_mobile, "code": verify_code}, headers).status_code == 200
user.refresh_from_db()
assert user.mobile == new_mobile
def test_profile_rejects_duplicate_email_and_resolves_education_codes(client, user_factory, auth_headers):
user = user_factory()
another = user_factory(email="taken@example.com")
major = Major.objects.create(code="TEST_PROFILE_MAJOR", name="Profile Major")
university = University.objects.create(code="TEST_PROFILE_UNI", name="Profile University")
headers = auth_headers(user)
path = "/api/auth/profile"
duplicate = client.put(path, data=json.dumps({"email": another.email}),
content_type="application/json", **headers)
assert duplicate.status_code == 400
user.refresh_from_db()
assert user.email != another.email
changed = client.put(path, data=json.dumps({"major": major.code, "university": university.code}),
content_type="application/json", **headers)
assert changed.status_code == 200, changed.content
user.refresh_from_db()
assert user.major == major and user.university == university
assert client.put(path, data=json.dumps({"major": None, "university": None}),
content_type="application/json", **headers).status_code == 200
user.refresh_from_db()
assert user.major is None and user.university is None
def test_authorization_rejects_unknown_group_and_refresh_bad_token(client, user_factory, auth_headers):
admin = user_factory(is_superuser=True, is_staff=True)
member = user_factory()
headers = auth_headers(admin)
path = f"/api/auth/users/{member.pk}/authorization"
invalid = client.put(path, data=json.dumps({"groups": ["unknown-role"], "is_staff": True}),
content_type="application/json", **headers)
assert invalid.status_code == 400
member.refresh_from_db()
assert not member.is_staff and not member.groups.exists()
assert _post(client, "/api/auth/refresh", {"refresh_token": "malformed"}).status_code == 401

20
apps/users/tests/integration/test_audit_command.py Normal file
View File

@@ -0,0 +1,20 @@
from io import StringIO
import pytest
from django.core.management import call_command
from apps.users.models import User
pytestmark = [pytest.mark.integration, pytest.mark.django_db]
def test_legacy_identity_audit_reports_placeholder_and_email_only_users(user_factory):
first = user_factory(email="old@noemail.local")
second = user_factory(email="person@example.com")
User.objects.filter(pk__in=[first.pk, second.pk]).update(mobile=None)
output = StringIO()
call_command("audit_legacy_email_identities", stdout=output)
report = output.getvalue()
assert "Placeholder emails: 1" in report
assert "Email-only users needing mobile bind: 2" in report

48
apps/users/tests/integration/test_auth_flows.py Normal file
View File

@@ -0,0 +1,48 @@
import pytest
from apps.users.models import User, UserSocialAccount
from apps.users.services import auth
from apps.users.services import google_oauth as google
pytestmark = [pytest.mark.integration, pytest.mark.django_db]
def test_registration_consumes_verified_mobile_and_creates_account(fake_redis, monkeypatch):
monkeypatch.setattr(auth, "_redis", lambda: fake_redis)
fake_redis.set(auth._verified_otp_key("register", "09190000001"), "1")
payload = auth.RegistrationPayload(
username="new_member", mobile="09190000001", password="StrongPass123!", code="12345",
email=" NEW@EXAMPLE.COM ",
)
user = auth.register_user(payload)
assert user.email == "new@example.com"
assert user.is_mobile_verified
assert user.check_password("StrongPass123!")
assert fake_redis.get(auth._verified_otp_key("register", "09190000001")) is None
def test_registration_rejects_duplicate_username(user_factory, fake_redis, monkeypatch):
user_factory(username="taken")
monkeypatch.setattr(auth, "_redis", lambda: fake_redis)
payload = auth.RegistrationPayload(username="taken", mobile="09190000002", password="StrongPass123!", code="12345")
with pytest.raises(auth.AuthServiceError) as error:
auth.register_user(payload)
assert error.value.field == "username"
assert User.objects.count() == 1
def test_google_claim_creates_account_and_link_without_live_gateway(mocker):
verify = mocker.patch("apps.users.services.google_oauth.verify_otp_code")
profile = google.GoogleProfile("google-1", "person@example.com", True, "Person", "Example", "")
flow = google.create_google_flow({
"status": "claim_required", "resolution": "new_account", "mobile": "09190000001",
"username": "google_member", "google_profile": google._profile_payload(profile),
})
result = google.verify_google_claim(flow, "12345")
assert result["status"] == "authenticated"
verify.assert_called_once()
user = User.objects.get(username="google_member")
assert user.is_mobile_verified
assert not user.has_usable_password()
assert UserSocialAccount.objects.filter(user=user, provider_user_id="google-1").exists()

89
apps/users/tests/integration/test_auth_service_edges.py Normal file
View File

@@ -0,0 +1,89 @@
import pytest
from apps.users.models import Major, University, User
from apps.users.services import auth
pytestmark = [pytest.mark.integration, pytest.mark.django_db]
def test_otp_send_modes_and_login_otp(user_factory, fake_redis, monkeypatch, mocker):
monkeypatch.setattr(auth, "_redis", lambda: fake_redis)
sms = mocker.patch("apps.users.services.auth.send_critical_sms.delay")
user = user_factory(is_mobile_verified=False)
with pytest.raises(auth.AuthServiceError):
auth.generate_and_send_otp("bad", "register")
with pytest.raises(auth.AuthServiceError):
auth.generate_and_send_otp(user.mobile, "register")
with pytest.raises(auth.AuthServiceError):
auth.generate_and_send_otp("09197770061", "login")
response = auth.generate_and_send_otp(user.mobile, "login")
assert response["expires_in_seconds"] == auth.OTP_EXPIRY_SECONDS
code = fake_redis.get(auth._otp_key("login", user.mobile)).decode()
sms.assert_called_with(user.mobile, auth.SMS_KIND_LOGIN, code)
tokens = auth.login_with_otp(user.mobile, code)
assert tokens["access_token"] and tokens["refresh_token"]
user.refresh_from_db()
assert user.is_mobile_verified
with pytest.raises(auth.AuthServiceError):
auth.login_with_otp(user.mobile, code)
def test_mobile_claim_conflict_and_verification(user_factory, fake_redis, monkeypatch, mocker):
monkeypatch.setattr(auth, "_redis", lambda: fake_redis)
sms = mocker.patch("apps.users.services.auth.send_critical_sms.delay")
user = user_factory()
another = user_factory()
with pytest.raises(auth.AuthServiceError):
auth.send_authenticated_mobile_otp(user, another.mobile)
new_mobile = "09197770062"
auth.send_authenticated_mobile_otp(user, new_mobile)
code = fake_redis.get(auth._otp_key("verify_mobile", new_mobile)).decode()
sms.assert_called_with(new_mobile, auth.SMS_KIND_VERIFY_MOBILE, code)
assert auth.verify_authenticated_mobile_otp(user, new_mobile, code).mobile == new_mobile
user.refresh_from_db()
assert user.mobile == new_mobile and user.is_mobile_verified
assert auth.lookup_mobile_registration_state(new_mobile) == {"exists": True, "has_password": True}
assert auth.lookup_mobile_registration_state("09197770063") == {"exists": False, "has_password": False}
def test_registration_validates_profile_fields_and_duplicate_email(user_factory, fake_redis, monkeypatch):
monkeypatch.setattr(auth, "_redis", lambda: fake_redis)
existing = user_factory(email="used@example.com")
payload = auth.RegistrationPayload(username="new_profile", mobile="09197770064", password="StrongPass123!", code="12345")
fake_redis.set(auth._verified_otp_key("register", payload.mobile), "1")
for changes, field in [({"student_id": "123"}, "student_id"),
({"email": existing.email}, "email"),
({"major": "invalid"}, "major"),
({"university": "invalid"}, "university")]:
fake_redis.set(auth._verified_otp_key("register", payload.mobile), "1")
current = auth.RegistrationPayload(**{**payload.__dict__, **changes})
with pytest.raises(auth.AuthServiceError) as error:
auth.register_user(current)
assert error.value.field == field
major = Major.objects.create(code="TEST_AUTH_MAJOR", name="Auth Major")
university = University.objects.create(code="TEST_AUTH_UNI", name="Auth University")
current = auth.RegistrationPayload(**{**payload.__dict__, "major": major.code, "university": university.code,
"student_id": "1234567890"})
fake_redis.set(auth._verified_otp_key("register", payload.mobile), "1")
created = auth.register_user(current)
assert created.major == major and created.university == university
assert created.student_id == "1234567890"
assert User.objects.filter(username=payload.username).count() == 1
def test_password_reset_rejects_reuse_and_inactive_login(user_factory, fake_redis, monkeypatch):
monkeypatch.setattr(auth, "_redis", lambda: fake_redis)
user = user_factory(is_active=False)
with pytest.raises(auth.AuthServiceError) as error:
auth.login_with_password(user.mobile, "TestPass123!")
assert error.value.status_code == 401
user.is_active = True
user.save(update_fields=["is_active"])
with pytest.raises(auth.AuthServiceError):
auth.reset_password_with_otp("09197770065", "12345", "NewStrongPass123!")
fake_redis.set(auth._otp_key("reset_password", user.mobile), "12345")
with pytest.raises(auth.AuthServiceError) as error:
auth.reset_password_with_otp(user.mobile, "12345", "TestPass123!")
assert error.value.field == "password"
assert user.check_password("TestPass123!")

62
apps/users/tests/integration/test_google_api.py Normal file
View File

@@ -0,0 +1,62 @@
import json
from urllib.parse import parse_qs, urlparse
import pytest
from apps.users.models import UserSocialAccount
from apps.users.services import google_oauth as google
pytestmark = [pytest.mark.integration, pytest.mark.django_db]
def _flow_from_redirect(response):
return parse_qs(urlparse(response.url).query)["flow"][0]
def test_google_oauth_start_callback_and_flow_lookup(client, settings, mocker):
settings.GOOGLE_OAUTH_CLIENT_ID = "client-id"
settings.GOOGLE_OAUTH_REDIRECT_URI = "https://backend.test/callback"
settings.GOOGLE_OAUTH_FRONTEND_CALLBACK_URL = "https://frontend.test/google"
start = client.get("/api/auth/oauth/google/start")
assert start.status_code == 302 and "accounts.google.com" in start.url
state = parse_qs(urlparse(start.url).query)["state"][0]
profile = google.GoogleProfile("google-555", "new@example.com", True, "New", "Person", "")
exchange = mocker.patch("apps.users.api.views.exchange_code_for_google_profile", return_value=profile)
callback = client.get("/api/auth/oauth/google/callback", {"state": state, "code": "auth-code"})
assert callback.status_code == 302
exchange.assert_called_once_with("auth-code")
flow = _flow_from_redirect(callback)
response = client.get("/api/auth/oauth/google/flow", {"flow": flow})
assert response.status_code == 200
assert response.json()["resolution"] == "new_account"
assert client.get("/api/auth/oauth/google/flow", {"flow": "unknown"}).status_code == 400
assert client.get("/api/auth/oauth/google/callback", {"state": state, "code": "replay"}).status_code == 302
cancelled = client.get("/api/auth/oauth/google/callback", {"error": "access_denied"})
assert google.get_google_flow(_flow_from_redirect(cancelled))["status"] == "error"
def test_google_api_complete_claim_and_linked_login(client, settings, mocker, user_factory):
settings.GOOGLE_OAUTH_FRONTEND_CALLBACK_URL = "https://frontend.test/google"
profile = google.GoogleProfile("google-556", "person@example.com", True, "Google", "Person", "")
flow = google.create_google_flow(google.build_pending_google_flow_payload(profile))
mocker.patch("apps.users.services.google_oauth.generate_and_send_otp")
mocker.patch("apps.users.services.google_oauth.verify_otp_code")
complete = client.post("/api/auth/oauth/google/complete", data=json.dumps({"flow": flow, "mobile": "09197770056", "username": "google_api"}),
content_type="application/json")
assert complete.status_code == 200, complete.content
assert complete.json()["status"] == "claim_required"
assert client.post("/api/auth/oauth/google/claim/send-otp", data=json.dumps({"flow": flow}),
content_type="application/json").status_code == 200
verified = client.post("/api/auth/oauth/google/claim/verify", data=json.dumps({"flow": flow, "code": "12345"}),
content_type="application/json")
assert verified.status_code == 200, verified.content
assert verified.json()["status"] == "authenticated"
social = UserSocialAccount.objects.get(provider_user_id="google-556")
assert social.user.username == "google_api"
mocker.patch("apps.users.api.views.consume_google_state")
mocker.patch("apps.users.api.views.exchange_code_for_google_profile", return_value=profile)
callback = client.get("/api/auth/oauth/google/callback", {"state": "fresh", "code": "code"})
assert callback.status_code == 302
linked = google.get_google_flow(_flow_from_redirect(callback))
assert linked["status"] == "authenticated" and linked["access_token"]

98
apps/users/tests/integration/test_google_flows.py Normal file
View File

@@ -0,0 +1,98 @@
import pytest
from apps.users.models import User, UserSocialAccount
from apps.users.services import google_oauth as google
pytestmark = [pytest.mark.integration, pytest.mark.django_db]
@pytest.fixture
def google_profile():
return google.GoogleProfile("provider-123", "google-person@example.com", True, "Google", "Person", "")
def _start(profile, **overrides):
payload = google.build_pending_google_flow_payload(profile)
payload.update(overrides)
return google.create_google_flow(payload)
def test_new_google_account_requires_claim_and_links_after_otp(google_profile, mocker):
send = mocker.patch("apps.users.services.google_oauth.generate_and_send_otp")
verify = mocker.patch("apps.users.services.google_oauth.verify_otp_code")
flow = _start(google_profile)
pending = google.complete_google_signup(flow=flow, mobile="09197770011", username="google_new")
assert pending["status"] == "claim_required" and pending["resolution"] == "new_account"
assert not User.objects.filter(username="google_new").exists()
send.assert_called_once_with("09197770011", "google_claim")
assert google.send_google_claim_otp(flow)["message"]
assert send.call_count == 2
authenticated = google.verify_google_claim(flow, "12345")
assert authenticated["status"] == "authenticated" and authenticated["access_token"]
verify.assert_called_once_with(mobile="09197770011", code="12345", mode="google_claim")
user = User.objects.get(username="google_new")
assert user.is_mobile_verified and not user.has_usable_password()
assert UserSocialAccount.objects.filter(user=user, provider_user_id="provider-123").exists()
assert google.get_google_flow(flow)["status"] == "authenticated"
def test_existing_email_claim_links_matching_user(google_profile, user_factory, mocker):
existing = user_factory(email=google_profile.email, is_mobile_verified=False)
mocker.patch("apps.users.services.google_oauth.generate_and_send_otp")
mocker.patch("apps.users.services.google_oauth.verify_otp_code")
flow = _start(google_profile)
pending = google.complete_google_signup(flow=flow, mobile=existing.mobile)
assert pending["resolution"] == "existing_email_claim"
authenticated = google.verify_google_claim(flow, "12345")
assert authenticated["status"] == "authenticated"
existing.refresh_from_db()
assert existing.is_mobile_verified and existing.is_email_verified
assert UserSocialAccount.objects.get(provider_user_id=google_profile.provider_user_id).user_id == existing.pk
@pytest.mark.parametrize("values,expected_code", [
({"mobile": "bad", "username": "new"}, "google_flow_error"),
({"mobile": "09197770022"}, "google_missing_username"),
({"mobile": "09197770022", "username": "taken"}, "google_username_taken"),
({"mobile": "09197770022", "username": "new", "student_id": "123"}, "google_invalid_student_id"),
({"mobile": "09197770022", "username": "new", "major": "missing"}, "google_invalid_major"),
({"mobile": "09197770022", "username": "new", "university": "missing"}, "google_invalid_university"),
])
def test_google_signup_rejects_invalid_profile_data(google_profile, user_factory, mocker, values, expected_code):
if values.get("username") == "taken":
user_factory(username="taken")
send = mocker.patch("apps.users.services.google_oauth.generate_and_send_otp")
flow = _start(google_profile)
with pytest.raises(google.GoogleOAuthFlowError) as error:
google.complete_google_signup(flow=flow, **values)
assert error.value.code == expected_code
send.assert_not_called()
assert not UserSocialAccount.objects.exists()
def test_google_mobile_conflict_is_rejected(google_profile, user_factory, mocker):
conflict = user_factory(mobile="09197770033", email="other@example.com")
send = mocker.patch("apps.users.services.google_oauth.generate_and_send_otp")
flow = _start(google_profile)
with pytest.raises(google.GoogleOAuthFlowError) as error:
google.complete_google_signup(flow=flow, mobile=conflict.mobile, username="new_google")
assert error.value.code == "google_mobile_belongs_to_other_email"
send.assert_not_called()
def test_google_claim_rejects_missing_target_and_existing_email(google_profile, user_factory, mocker):
mocker.patch("apps.users.services.google_oauth.verify_otp_code")
missing = google.create_google_flow({"status": "claim_required", "resolution": "existing_email_claim",
"target_user_id": 999999, "mobile": "09197770044",
"google_profile": google._profile_payload(google_profile)})
with pytest.raises(google.GoogleOAuthFlowError) as error:
google.verify_google_claim(missing, "12345")
assert error.value.code == "google_email_claim_failed"
user_factory(email=google_profile.email)
new_flow = google.create_google_flow({"status": "claim_required", "resolution": "new_account",
"username": "fresh", "mobile": "09197770045",
"google_profile": google._profile_payload(google_profile)})
with pytest.raises(google.GoogleOAuthFlowError) as error:
google.verify_google_claim(new_flow, "12345")
assert error.value.code == "google_email_taken"

42
apps/users/tests/integration/test_meta.py Normal file
View File

@@ -0,0 +1,42 @@
import json
import pytest
from apps.users.models import Major, University
pytestmark = [pytest.mark.integration, pytest.mark.django_db]
@pytest.mark.parametrize("name,model", [("majors", Major), ("universities", University)])
def test_meta_public_search_and_admin_lifecycle(client, user_factory, auth_headers, name, model):
staff = user_factory(is_staff=True)
admin = user_factory(is_staff=True, is_superuser=True)
member = user_factory()
path = f"/api/meta/{name}"
admin_path = f"/api/meta/admin/{name}"
assert client.get(path).status_code == 200
assert client.post(admin_path, data=json.dumps({"code": "TEST_META_CX", "name": "Unique Meta Choice"}),
content_type="application/json", **auth_headers(member)).status_code == 403
created = client.post(admin_path, data=json.dumps({"code": "TEST_META_CX", "name": "Unique Meta Choice"}),
content_type="application/json", **auth_headers(staff))
assert created.status_code == 201, created.content
item = model.objects.get(pk=created.json()["id"])
assert client.get(path, {"search": "Unique Meta Choice"}).json()["results"][0]["id"] == item.pk
assert client.get(path, {"search": "missing"}).json()["count"] == 0
assert any(row["id"] == item.pk for row in client.get(admin_path, {"search": "Unique Meta Choice"}, **auth_headers(staff)).json()["results"])
duplicate = client.post(admin_path, data=json.dumps({"code": "TEST_META_CX", "name": "Duplicate"}),
content_type="application/json", **auth_headers(staff))
assert duplicate.status_code == 400
updated = client.put(f"{admin_path}/{item.pk}", data=json.dumps({"code": "TEST_META_CX", "name": "Updated Unique Meta Choice", "is_active": False}),
content_type="application/json", **auth_headers(staff))
assert updated.status_code == 200, updated.content
item.refresh_from_db()
assert item.name == "Updated Unique Meta Choice" and not item.is_active
assert client.get(path, {"search": "TEST_META_CX"}).json()["results"] == []
assert client.get(admin_path, {"search": "TEST_META_CX"}, **auth_headers(staff)).json()["results"][0]["id"] == item.pk
assert client.delete(f"{admin_path}/{item.pk}", **auth_headers(staff)).status_code == 403
assert client.delete(f"{admin_path}/{item.pk}", **auth_headers(admin)).status_code == 200
assert model.deleted_objects.filter(pk=item.pk).exists()
assert client.post(f"{admin_path}/{item.pk}/restore", **auth_headers(admin)).status_code == 200
assert model.objects.filter(pk=item.pk).exists()

25
apps/users/tests/integration/test_user_models.py Normal file
View File

@@ -0,0 +1,25 @@
import pytest
from apps.users.models import User
from apps.users.resources import UserResource
@pytest.mark.django_db
@pytest.mark.integration
class TestUserIdentityTests:
def test_email_is_optional_and_flags_follow_identity(self):
user = User.objects.create_user(username='mobile_only', mobile='09191110000', password='password')
assert user.email is None
assert not user.is_email_verified
user.email = ' PERSON@EXAMPLE.COM '
user.save(update_fields=['email'])
user.refresh_from_db()
assert user.email == 'person@example.com'
assert user.is_email_verified
user.email = None
user.save(update_fields=['email'])
user.refresh_from_db()
assert not user.is_email_verified
def test_display_name_prefers_mobile_when_email_absent(self):
user = User.objects.create_user(username='mobile_only', mobile='09191110001', password='password')
assert user.mobile in str(user)

99
apps/users/tests/integration/test_users.py
View File

@@ -1,85 +1,72 @@
import pytest
import json
from unittest import mock
from django.test import TestCase
from django.core.files.uploadedfile import SimpleUploadedFile
from apps.users.models import User
from core.authentication import create_jwt_token
@pytest.mark.django_db
@pytest.mark.integration
class TestUsersAPIIntegrationTests:
class UsersAPIIntegrationTests(TestCase):
def setUp(self):
self.user = User.objects.create_user(
username="mobile_user",
mobile="09192000001",
email="person@example.com",
password="Sup3rSecure!123",
is_mobile_verified=True,
)
def setup_method(self):
self.user = User.objects.create_user(username='mobile_user', mobile='09192000001', email='person@example.com', password='Sup3rSecure!123', is_mobile_verified=True)
def _auth(self):
return {"HTTP_AUTHORIZATION": f"Bearer {create_jwt_token(self.user)}"}
return {'HTTP_AUTHORIZATION': f'Bearer {create_jwt_token(self.user)}'}
def test_password_login_accepts_mobile_identity(self):
response = self.client.post("/api/auth/login", json.dumps({
"identifier": self.user.mobile,
"password": "Sup3rSecure!123",
}), content_type="application/json")
self.assertEqual(response.status_code, 200)
self.assertIn("access_token", response.json())
response = self.client.post('/api/auth/login', json.dumps({'identifier': self.user.mobile, 'password': 'Sup3rSecure!123'}), content_type='application/json')
assert response.status_code == 200
assert 'access_token' in response.json()
def test_profile_update_treats_supplied_email_as_identity(self):
response = self.client.put("/api/auth/profile", json.dumps({"email": "new@example.com"}),
content_type="application/json", **self._auth())
self.assertEqual(response.status_code, 200)
response = self.client.put('/api/auth/profile', json.dumps({'email': 'new@example.com'}), content_type='application/json', **self._auth())
assert response.status_code == 200
self.user.refresh_from_db()
self.assertEqual(self.user.email, "new@example.com")
self.assertTrue(self.user.is_email_verified)
assert self.user.email == 'new@example.com'
assert self.user.is_email_verified
def test_profile_update_clears_email_identity_flag(self):
response = self.client.put("/api/auth/profile", json.dumps({"email": ""}),
content_type="application/json", **self._auth())
self.assertEqual(response.status_code, 200)
response = self.client.put('/api/auth/profile', json.dumps({'email': ''}), content_type='application/json', **self._auth())
assert response.status_code == 200
self.user.refresh_from_db()
self.assertIsNone(self.user.email)
self.assertFalse(self.user.is_email_verified)
assert self.user.email is None
assert not self.user.is_email_verified
def test_profile_requires_authentication_and_returns_account_identity(self):
self.assertEqual(self.client.get("/api/auth/profile").status_code, 401)
response = self.client.get("/api/auth/profile", **self._auth())
self.assertEqual(response.status_code, 200)
self.assertEqual(response.json()["mobile"], self.user.mobile)
assert self.client.get('/api/auth/profile').status_code == 401
response = self.client.get('/api/auth/profile', **self._auth())
assert response.status_code == 200
assert response.json()['mobile'] == self.user.mobile
def test_profile_rejects_duplicate_email(self):
User.objects.create_user(username="second", mobile="09192000002", email="taken@example.com", password="password")
response = self.client.put("/api/auth/profile", json.dumps({"email": "taken@example.com"}),
content_type="application/json", **self._auth())
self.assertEqual(response.status_code, 400)
User.objects.create_user(username='second', mobile='09192000002', email='taken@example.com', password='password')
response = self.client.put('/api/auth/profile', json.dumps({'email': 'taken@example.com'}), content_type='application/json', **self._auth())
assert response.status_code == 400
self.user.refresh_from_db()
self.assertEqual(self.user.email, "person@example.com")
assert self.user.email == 'person@example.com'
def test_profile_picture_rejects_non_image_upload(self):
image = SimpleUploadedFile("notes.txt", b"not an image", content_type="text/plain")
response = self.client.post("/api/auth/profile/picture", {"file": image}, **self._auth())
self.assertEqual(response.status_code, 400)
image = SimpleUploadedFile('notes.txt', b'not an image', content_type='text/plain')
response = self.client.post('/api/auth/profile/picture', {'file': image}, **self._auth())
assert response.status_code == 400
def test_check_username_reports_existing_user(self):
response = self.client.get("/api/auth/check-username", {"username": self.user.username})
self.assertEqual(response.status_code, 200)
self.assertTrue(response.json()["exists"])
response = self.client.get('/api/auth/check-username', {'username': self.user.username})
assert response.status_code == 200
assert response.json()['exists']
def test_legacy_email_auth_routes_are_gone(self):
self.assertEqual(self.client.get("/api/auth/verify-email/token").status_code, 404)
self.assertEqual(self.client.post("/api/auth/resend-verification").status_code, 404)
self.assertEqual(self.client.post("/api/auth/request-password-reset").status_code, 404)
assert self.client.get('/api/auth/verify-email/token').status_code == 404
assert self.client.post('/api/auth/resend-verification').status_code == 404
assert self.client.post('/api/auth/request-password-reset').status_code == 404
@mock.patch("apps.users.api.views.reset_password_with_otp")
def test_mobile_otp_password_reset_route_remains_available(self, reset):
response = self.client.post("/api/auth/reset-password", json.dumps({
"mobile": self.user.mobile,
"code": "123456",
"new_password": "AnotherSecurePass!123",
}), content_type="application/json")
self.assertEqual(response.status_code, 200)
def test_mobile_otp_password_reset_route_remains_available(self, mocker):
reset = mocker.patch('apps.users.api.views.reset_password_with_otp')
response = self.client.post('/api/auth/reset-password', json.dumps({'mobile': self.user.mobile, 'code': '123456', 'new_password': 'AnotherSecurePass!123'}), content_type='application/json')
assert response.status_code == 200
reset.assert_called_once()
@pytest.fixture(autouse=True)
def _pytest_client(self, client):
self.client = client