Amirhossein Khalili
6c5fadef2f
Replace persisted event statuses with publication and cancellation flags, and compute public lifecycle labels from event timing and registration capacity. Keep filters, registration checks, and profile responses aligned with the derived status.
328 lines
16 KiB
Python
328 lines
16 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_public_profile_hides_personal_fields_until_the_user_opts_in(client, user_factory, auth_headers, event_factory):
|
|
from apps.events.models import Event, Registration
|
|
|
|
member = user_factory(
|
|
username="public_member",
|
|
first_name="Public",
|
|
last_name="Member",
|
|
email="public.member@example.org",
|
|
mobile="09195550123",
|
|
student_id="402123456",
|
|
year_of_study=1402,
|
|
bio="A public introduction",
|
|
)
|
|
event = event_factory(slug="public-profile-event")
|
|
Registration.objects.create(event=event, user=member, status=Registration.StatusChoices.CONFIRMED)
|
|
draft_event = event_factory(slug="private-profile-event", is_published=False)
|
|
Registration.objects.create(event=draft_event, user=member, status=Registration.StatusChoices.ATTENDED)
|
|
|
|
public_path = f"/api/auth/public-profile/{member.username}"
|
|
hidden_response = client.get(public_path)
|
|
assert hidden_response.status_code == 200
|
|
hidden = hidden_response.json()
|
|
assert hidden["full_name"] == "Public Member"
|
|
assert hidden["details_visible"] is False
|
|
assert hidden["mobile"] is None
|
|
assert hidden["student_id"] is None
|
|
assert hidden["year_of_study"] is None
|
|
assert "date_joined" not in hidden
|
|
assert "blog_activity" not in hidden
|
|
assert [row["slug"] for row in hidden["events"]] == [event.slug]
|
|
assert hidden.get("email") is None
|
|
|
|
updated = client.put(
|
|
"/api/auth/profile",
|
|
data=json.dumps({"show_public_profile_details": True}),
|
|
content_type="application/json",
|
|
**auth_headers(member),
|
|
)
|
|
assert updated.status_code == 200, updated.content
|
|
visible = client.get(public_path).json()
|
|
assert visible["details_visible"] is True
|
|
assert visible["mobile"] == member.mobile
|
|
assert visible["student_id"] == member.student_id
|
|
assert visible["year_of_study"] == member.year_of_study
|
|
assert visible["email"] == member.email
|
|
assert "date_joined" not in visible
|
|
assert "blog_activity" not in visible
|
|
|
|
|
|
def test_public_profile_does_not_expose_placeholder_email(client, user_factory):
|
|
member = user_factory(
|
|
username="public_placeholder_email",
|
|
email="pending-public-placeholder@noemail.local",
|
|
show_public_profile_details=True,
|
|
)
|
|
|
|
response = client.get(f"/api/auth/public-profile/{member.username}")
|
|
|
|
assert response.status_code == 200
|
|
assert response.json().get("email") is None
|
|
|
|
|
|
def test_public_profile_does_not_expose_inactive_users(client, user_factory):
|
|
member = user_factory(username="inactive_public_member", is_active=False)
|
|
response = client.get(f"/api/auth/public-profile/{member.username}")
|
|
assert response.status_code == 404
|
|
|
|
|
|
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
|
|
assert changed.json()["major_code"] == major.code
|
|
assert changed.json()["university_code"] == university.code
|
|
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_profile_update_allows_username_changes_and_limits_plain_text_bio(client, user_factory, auth_headers):
|
|
member = user_factory(username="before_edit")
|
|
another = user_factory(username="already_taken")
|
|
headers = auth_headers(member)
|
|
path = "/api/auth/profile"
|
|
|
|
changed = client.put(
|
|
path,
|
|
data=json.dumps({"username": "after_edit", "bio": "A short plain text introduction"}),
|
|
content_type="application/json",
|
|
**headers,
|
|
)
|
|
assert changed.status_code == 200, changed.content
|
|
assert changed.json()["username"] == "after_edit"
|
|
|
|
duplicate = client.put(
|
|
path,
|
|
data=json.dumps({"username": another.username}),
|
|
content_type="application/json",
|
|
**headers,
|
|
)
|
|
assert duplicate.status_code == 400
|
|
|
|
too_long = client.put(
|
|
path,
|
|
data=json.dumps({"bio": "a" * 501}),
|
|
content_type="application/json",
|
|
**headers,
|
|
)
|
|
assert too_long.status_code == 422
|
|
at_limit = client.put(
|
|
path,
|
|
data=json.dumps({"bio": "a" * 500}),
|
|
content_type="application/json",
|
|
**headers,
|
|
)
|
|
assert at_limit.status_code == 200, at_limit.content
|
|
|
|
|
|
def test_profile_username_changes_follow_public_username_rules(client, user_factory, auth_headers):
|
|
member = user_factory(username="amir")
|
|
user_factory(username="already_taken")
|
|
headers = auth_headers(member)
|
|
path = "/api/auth/profile"
|
|
|
|
unchanged_legacy_name = client.put(
|
|
path,
|
|
data=json.dumps({"username": "amir"}),
|
|
content_type="application/json",
|
|
**headers,
|
|
)
|
|
too_short = client.put(
|
|
path,
|
|
data=json.dumps({"username": "abc1"}),
|
|
content_type="application/json",
|
|
**headers,
|
|
)
|
|
invalid_characters = client.put(
|
|
path,
|
|
data=json.dumps({"username": "valid-name"}),
|
|
content_type="application/json",
|
|
**headers,
|
|
)
|
|
duplicate = client.put(
|
|
path,
|
|
data=json.dumps({"username": "already_taken"}),
|
|
content_type="application/json",
|
|
**headers,
|
|
)
|
|
available = client.put(
|
|
path,
|
|
data=json.dumps({"username": "valid_name42"}),
|
|
content_type="application/json",
|
|
**headers,
|
|
)
|
|
case_insensitive_availability = client.get("/api/auth/check-username?username=ALREADY_TAKEN")
|
|
|
|
assert unchanged_legacy_name.status_code == 200
|
|
assert too_short.status_code == 400
|
|
assert invalid_characters.status_code == 400
|
|
assert duplicate.status_code == 400
|
|
assert available.status_code == 200, available.content
|
|
assert available.json()["username"] == "valid_name42"
|
|
assert case_insensitive_availability.status_code == 200
|
|
assert case_insensitive_availability.json()["exists"] is True
|
|
|
|
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
|