from datetime import timedelta from uuid import uuid4 import pytest from django.core.files.uploadedfile import SimpleUploadedFile from django.test import Client from django.urls import reverse from django.utils import timezone from apps.accounts.magic_access import MAGIC_LINK_SESSION_KEY from apps.accounts.models import MagicLinkAccessLog, TicketMagicLink, User from apps.attachments.models import Attachment, AttachmentPolicy from apps.customers.models import CustomerProfile from apps.messaging.models import Message from apps.tickets.models import TicketCategory from apps.tickets.services import create_ticket, reply_ticket pytestmark = pytest.mark.django_db CONFIRMATION_MESSAGE = ( "شما با حساب دیگری وارد سامانه شده‌اید. برای مشاهده تیکت مرتبط با این لینک، " "از حساب فعلی خارج شوید. پس از خروج، تیکت به‌صورت خودکار باز می‌شود و نیازی به ورود مجدد ندارید." ) @pytest.fixture def magic_case(settings, tmp_path): settings.MEDIA_ROOT = tmp_path category = TicketCategory.objects.create(title="لینک امن", slug="magic-confirmation") owner = User.objects.create_user(username="link-owner", mobile="09121000101", account_type="CUSTOMER") other = User.objects.create_user(username="link-other", mobile="09121000102", account_type="CUSTOMER") CustomerProfile.objects.create(user=owner) CustomerProfile.objects.create(user=other) ticket = create_ticket(customer=owner, category=category, subject="موضوع محرمانه", body="پیام محرمانه") sibling = create_ticket(customer=owner, category=category, subject="تیکت خارج از لینک", body="پیام دیگر") token, link = TicketMagicLink.issue_with_record(ticket) return owner, other, ticket, sibling, link, reverse("accounts:magic", args=[token]) def confirmation_data(user): return {"action": "logout_and_open", "account_id": str(user.pk)} @pytest.mark.parametrize("account_type", ["CUSTOMER", "STAFF"]) def test_mismatched_account_get_requires_confirmation_without_leaking_ticket(client, magic_case, account_type): owner, other, ticket, sibling, link, url = magic_case other.account_type = account_type other.save(update_fields=["account_type"]) client.force_login(other) response = client.get(url) content = response.content.decode() assert response.status_code == 200 assert CONFIRMATION_MESSAGE in content assert "خروج از حساب و مشاهده تیکت" in content assert "انصراف از مشاهده این تیکت/پیام" in content assert ticket.subject not in content and "پیام محرمانه" not in content assert owner.mobile not in content assert client.session["_auth_user_id"] == str(other.pk) assert MAGIC_LINK_SESSION_KEY not in client.session assert MagicLinkAccessLog.objects.count() == 0 link.refresh_from_db() assert link.view_count == 0 and link.used_at is None assert "no-store" in response["Cache-Control"] assert response["Referrer-Policy"] == "same-origin" @pytest.mark.parametrize("account_type", ["CUSTOMER", "STAFF"]) def test_confirmed_logout_opens_only_linked_ticket_and_uses_owner_for_reply(client, magic_case, account_type): owner, other, ticket, sibling, link, url = magic_case other.account_type = account_type other.save(update_fields=["account_type"]) client.force_login(other) session = client.session session["unrelated_account_data"] = "must disappear" session.save() old_session_key = session.session_key response = client.post(url, confirmation_data(other), follow=True) assert response.status_code == 200 assert response.redirect_chain == [(reverse("tickets:detail", args=[ticket.public_id]), 302)] assert "پیام محرمانه" in response.content.decode() assert "فقط به همین تیکت دسترسی دارید" in response.content.decode() assert "_auth_user_id" not in client.session assert "unrelated_account_data" not in client.session assert client.session.session_key != old_session_key assert client.session[MAGIC_LINK_SESSION_KEY] == link.pk for destination in ( reverse("tickets:detail", args=[sibling.public_id]), reverse("tickets:list"), reverse("tickets:new"), reverse("accounts:profile"), reverse("payments:home"), ): assert client.get(destination).status_code == 302 for name in ("messages", "poll"): assert client.get(reverse(f"tickets:{name}", args=[ticket.public_id])).status_code == 200 reply = client.post(reverse("tickets:reply", args=[ticket.public_id]), {"body": "پاسخ پس از خروج"}) assert reply.status_code == 302 assert Message.objects.get(ticket=ticket, body="پاسخ پس از خروج").author_id == owner.pk link.refresh_from_db() assert link.view_count == 1 assert MagicLinkAccessLog.objects.filter(magic_link=link).count() == 1 def test_cancel_preserves_current_account_and_does_not_consume_link(client, magic_case): owner, other, ticket, sibling, link, url = magic_case client.force_login(other) client.get(url) response = client.get(reverse("accounts:home"), follow=True) assert response.status_code == 200 assert client.session["_auth_user_id"] == str(other.pk) assert MAGIC_LINK_SESSION_KEY not in client.session assert MagicLinkAccessLog.objects.count() == 0 @pytest.mark.parametrize("invalid_state", ["expired", "revoked", "inactive", "unknown"]) def test_link_revalidated_before_logout(client, magic_case, invalid_state): owner, other, ticket, sibling, link, url = magic_case client.force_login(other) assert client.get(url).status_code == 200 if invalid_state == "expired": link.expires_at = timezone.now() - timedelta(seconds=1) link.save(update_fields=["expires_at"]) elif invalid_state == "revoked": link.revoked_at = timezone.now() link.save(update_fields=["revoked_at"]) elif invalid_state == "inactive": owner.is_active = False owner.save(update_fields=["is_active"]) else: url = reverse("accounts:magic", args=["unknown-token"]) response = client.post(url, confirmation_data(other)) assert response.status_code == 302 and response.url == reverse("accounts:login") assert client.session["_auth_user_id"] == str(other.pk) assert MAGIC_LINK_SESSION_KEY not in client.session assert MagicLinkAccessLog.objects.count() == 0 @pytest.mark.parametrize("payload", [{}, {"action": "other"}, {"action": "logout_and_open", "account_id": "wrong"}]) def test_unconfirmed_post_does_not_logout(client, magic_case, payload): owner, other, ticket, sibling, link, url = magic_case client.force_login(other) assert client.post(url, payload).status_code == 200 assert client.session["_auth_user_id"] == str(other.pk) assert MagicLinkAccessLog.objects.count() == 0 def test_confirmation_from_previous_account_cannot_logout_new_account(client, magic_case): owner, other, ticket, sibling, link, url = magic_case client.force_login(other) client.get(url) replacement = User.objects.create_user(username="replacement", mobile="09121000103", account_type="CUSTOMER") client.force_login(replacement) response = client.post(url, confirmation_data(other)) assert response.status_code == 200 assert client.session["_auth_user_id"] == str(replacement.pk) assert MagicLinkAccessLog.objects.count() == 0 def test_logout_confirmation_requires_csrf_and_valid_post_succeeds(magic_case): owner, other, ticket, sibling, link, url = magic_case client = Client(enforce_csrf_checks=True) client.force_login(other) client.get(url) assert client.post(url, confirmation_data(other)).status_code == 403 assert client.session["_auth_user_id"] == str(other.pk) response = client.post(url, {**confirmation_data(other), "csrfmiddlewaretoken": client.cookies["csrftoken"].value}) assert response.status_code == 302 and response.url == reverse("tickets:detail", args=[ticket.public_id]) assert "_auth_user_id" not in client.session @pytest.mark.parametrize("authenticated", [False, True]) def test_existing_anonymous_and_matching_account_flow_remains_direct(client, magic_case, authenticated): owner, other, ticket, sibling, link, url = magic_case if authenticated: client.force_login(owner) response = client.get(url) assert response.status_code == 302 and response.url == reverse("tickets:detail", args=[ticket.public_id]) assert client.get(response.url).status_code == 200 assert "no-store" in response["Cache-Control"] if authenticated: assert client.session["_auth_user_id"] == str(owner.pk) assert MAGIC_LINK_SESSION_KEY not in client.session else: assert "_auth_user_id" not in client.session assert client.session[MAGIC_LINK_SESSION_KEY] == link.pk def test_link_attachment_permissions_remain_scoped_after_logout(client, magic_case): owner, other, ticket, sibling, link, url = magic_case AttachmentPolicy.objects.create(attachment_type="PDF", allowed_extensions=[".pdf"], max_file_size=1024 * 1024) file = SimpleUploadedFile("linked.pdf", b"%PDF-1.4 linked", content_type="application/pdf") reply_ticket(ticket_id=ticket.pk, user=owner, body="پیوست مجاز", files=[file]) attachment = ticket.ticket_messages.get(body="پیوست مجاز").attachments.get() internal = Message.objects.create(ticket=ticket, author=other, body="یادداشت داخلی", is_internal=True) internal_attachment = Attachment.objects.get(pk=attachment.pk) internal_attachment.pk = None internal_attachment.public_id = uuid4() internal_attachment.message = internal internal_attachment.save() client.force_login(other) client.post(url, confirmation_data(other)) assert client.get(reverse("attachments:download", args=[attachment.public_id])).status_code == 200 assert client.get(reverse("attachments:download", args=[internal_attachment.public_id])).status_code == 403 attachment.message = sibling.ticket_messages.first() attachment.save(update_fields=["message"]) assert client.get(reverse("attachments:download", args=[attachment.public_id])).status_code == 403 @pytest.mark.parametrize("method", ["put", "delete"]) def test_unsupported_method_never_logs_out(client, magic_case, method): owner, other, ticket, sibling, link, url = magic_case client.force_login(other) assert getattr(client, method)(url).status_code == 405 assert client.session["_auth_user_id"] == str(other.pk)