add payment_link and banks

2026-10-06 10:03:22 +03:30
1 parent 2ae79b1ca4
commit de3135b588
@@ -0,0 +1,211 @@
+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)