Amirhossein Khalili
ff99130938
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.
172 lines
10 KiB
Python
172 lines
10 KiB
Python
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
|