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", status=Event.StatusChoices.DRAFT)
|
|
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
|