customerTicket/tests/test_dispatch_response_opening.py

import importlib

import pytest
from django.contrib.auth.models import Permission
from django.core.exceptions import PermissionDenied, ValidationError
from django.core.files.uploadedfile import SimpleUploadedFile
from django.db import IntegrityError, transaction
from django.test import RequestFactory
from django.urls import reverse

from apps.accounts.models import User
from apps.attachments.models import Attachment, AttachmentPolicy
from apps.messaging.models import Message, MessageReadReceipt, MessageRevision
from apps.messaging.services import edit_message
from apps.notifications.models import NotificationJob, SMSTemplate
from apps.notifications.management.commands.process_notifications import _send_bale, _send_sms
from apps.staff.models import OrganizationalUnit
from apps.tickets.dispatch_responses import OPENING_EVENT, open_dispatch_responses, render_opening_sms
from apps.tickets.models import DispatchResponseControl, GroupDispatch, MessageShare, TicketCategory
from apps.tickets.policies import TicketPolicy
from apps.tickets.services import create_group_dispatch, dispatch_group_tickets, reply_ticket
from apps.tickets.sharing import share_staff_message
from tests.test_organizational_ticketing import make_contact, make_staff

pytestmark = pytest.mark.django_db


@pytest.fixture
def opening_case():
    unit = OrganizationalUnit.objects.create(name="واحد بازگشایی", slug="opening-unit", bale_chat_id="opening-room")
    unit.permissions.add(*Permission.objects.filter(content_type__app_label="tickets", codename__in=["view_ticket", "create_inquiry", "create_bulk_dispatch"]))
    staff = make_staff("opening-owner", unit)
    other = make_staff("opening-peer", unit)
    viewer = make_staff("opening-viewer", unit)
    viewer.user_permissions.add(Permission.objects.get(codename="view_all_bulk_dispatches"))
    admin = User.objects.create_superuser(username="opening-admin", password="Admin-test-91", account_type="STAFF")
    from apps.customers.models import ContactGroup
    group = ContactGroup.objects.create(name="گروه استعلام", unit=unit)
    customers = [make_contact(f"opening-contact-{i}", f"0912200010{i}") for i in range(2)]
    for customer in customers:
        customer.customer_profile.groups.add(group)
    category = TicketCategory.objects.create(unit=unit, title="استعلام", slug="opening-category")
    dispatch, tickets = dispatch_group_tickets(user=staff, unit=unit, contact_group=group, category=category,
        subject="استعلام قیمت محرمانه", body="لطفاً پیشنهاد خود را ثبت کنید.")
    message = reply_ticket(ticket_id=tickets[0].pk, user=customers[0], body="SECRET_BID_84723")
    return dict(unit=unit, staff=staff, other=other, viewer=viewer, admin=admin, group=group,
        customers=customers, category=category, dispatch=dispatch, tickets=tickets, message=message)


@pytest.mark.parametrize("actor", ["staff", "viewer", "admin"])
def test_seal_blocks_report_ticket_poll_revisions_and_receipts(client, opening_case, actor):
    c = opening_case
    client.force_login(c[actor])
    for route, public_id in [("dispatch_detail", c["dispatch"].public_id), ("detail", c["tickets"][0].public_id),
        ("messages", c["tickets"][0].public_id), ("poll", c["tickets"][0].public_id)]:
        response = client.get(reverse("staff_tickets:" + route, args=[public_id]))
        assert response.status_code == 200
        assert c["message"].body not in response.content.decode()
    report = client.get(reverse("staff_tickets:dispatch_detail", args=[c["dispatch"].public_id]))
    assert report.context["metrics"]["replied"] == 1
    assert 'data-answered="true"' in report.content.decode()
    assert 'پاسخ داده است: بله' in report.content.decode()
    assert client.get(reverse("messages:revisions", args=[c["message"].public_id])).status_code == 403
    client.post(reverse("staff_tickets:read_shares", args=[c["tickets"][0].public_id]), {"message_ids": [str(c["message"].public_id)]})
    assert not MessageReadReceipt.objects.filter(message=c["message"]).exists()


def test_unit_wide_access_does_not_grant_unrelated_managed_dispatch_tickets(client, opening_case):
    c = opening_case
    assert not TicketPolicy.can_view(c["other"], c["tickets"][0])
    assert not c["tickets"][0].__class__.objects.filter(TicketPolicy.viewable_q(c["other"]), pk=c["tickets"][0].pk).exists()
    client.force_login(c["other"])
    assert client.get(reverse("staff_tickets:detail", args=[c["tickets"][0].public_id])).status_code == 403
    c["tickets"][0].owner = c["other"]
    c["tickets"][0].save()
    assert TicketPolicy.can_view(c["other"], c["tickets"][0])
    response = client.get(reverse("staff_tickets:detail", args=[c["tickets"][0].public_id]))
    assert response.status_code == 200 and c["message"].body not in response.content.decode()


@pytest.mark.parametrize("mode", ["SEALED", "INDIVIDUAL", "LEGACY"])
def test_report_shows_full_response_history_and_files_without_ticket_navigation(client, opening_case, mode):
    from apps.core.templatetags.jalali import jalali_datetime_precise
    c = opening_case
    control = c["dispatch"].response_control
    if mode == "LEGACY":
        control.delete()
    elif mode == "INDIVIDUAL":
        control.mode, control.notify_sms = mode, False
        control.save()
    body = "متن کامل پیشنهاد " * 30 + "پایان متن بلند"
    answer = reply_ticket(ticket_id=c["tickets"][0].pk, user=c["customers"][0], body=body)
    AttachmentPolicy.objects.create(attachment_type="PDF", max_file_size=1024 * 1024,
        allowed_extensions=[".pdf"], allowed_mime_types=["application/pdf"])
    from apps.attachments.services import save_attachments
    save_attachments(answer, [SimpleUploadedFile("offer.pdf", b"%PDF-1.4\n%%EOF", content_type="application/pdf")], c["customers"][0])
    file = answer.attachments.get()
    deleted = reply_ticket(ticket_id=c["tickets"][0].pk, user=c["customers"][0], body="DELETED_OFFER")
    from django.utils import timezone
    deleted.deleted_at = timezone.now()
    deleted.save(update_fields=["deleted_at"])
    client.force_login(c["staff"])
    url = reverse("staff_tickets:dispatch_detail", args=[c["dispatch"].public_id])
    ticket_url = reverse("staff_tickets:detail", args=[c["tickets"][0].public_id])
    attachment_url = reverse("attachments:view", args=[file.public_id])
    if mode == "SEALED":
        locked = client.get(url)
        assert body not in locked.content.decode() and attachment_url not in locked.content.decode()
        assert not locked.context["tickets"][0].report_customer_responses
        open_dispatch_responses(c["dispatch"], c["staff"], render_opening_sms(c["dispatch"]))
    report = client.get(url)
    html = report.content.decode()
    assert body in html and c["message"].body in html and deleted.body not in html
    assert attachment_url in html and jalali_datetime_precise(answer.created_at) in html
    assert (f'href="{ticket_url}"' in html) == (mode != "SEALED")


def test_customer_can_read_and_edit_own_sealed_answer(client, opening_case):
    c = opening_case
    client.force_login(c["customers"][0])
    response = client.get(reverse("tickets:detail", args=[c["tickets"][0].public_id]))
    assert c["message"].body in response.content.decode()
    edit_message(c["message"].pk, c["customers"][0], "SECRET_EDITED_BID")
    client.force_login(c["admin"])
    assert client.get(reverse("messages:revisions", args=[c["message"].public_id])).status_code == 403


def test_opening_is_once_and_uses_confirmed_template_snapshot(opening_case, monkeypatch):
    c = opening_case
    text = render_opening_sms(c["dispatch"])
    control, opened = open_dispatch_responses(c["dispatch"], c["staff"], text)
    assert opened and control.opened_by == c["staff"] and control.opened_at
    assert control.respondents_snapshot == [str(c["customers"][0].public_id)]
    job = NotificationJob.objects.get(event_type=OPENING_EVENT)
    SMSTemplate.objects.filter(event_type=OPENING_EVENT).update(body="متن تغییر یافته")
    assert open_dispatch_responses(c["dispatch"], c["admin"], text)[1] is False
    assert NotificationJob.objects.filter(event_type=OPENING_EVENT).count() == 1
    sent = []
    class Provider:
        def send_message(self, mobile, body):
            sent.append((mobile, body))
            return "test-provider-id"
    monkeypatch.setattr("apps.notifications.management.commands.process_notifications.get_sms_provider", lambda: Provider())
    assert _send_sms(job) == ("test-provider-id", None)
    assert sent == [(c["customers"][0].mobile, text)]
    reply_ticket(ticket_id=c["tickets"][1].pk, user=c["customers"][1], body="پاسخ پس از بازگشایی")
    assert NotificationJob.objects.filter(event_type=OPENING_EVENT).count() == 1


def test_template_change_requires_new_confirmation_and_view_permission_is_not_open_permission(client, opening_case):
    c = opening_case
    with pytest.raises(PermissionDenied):
        open_dispatch_responses(c["dispatch"], c["viewer"], render_opening_sms(c["dispatch"]))
    c["viewer"].user_permissions.add(Permission.objects.get(codename="open_all_dispatch_responses"))
    viewer = User.objects.get(pk=c["viewer"].pk)
    client.force_login(viewer)
    url = reverse("staff_tickets:dispatch_open_responses", args=[c["dispatch"].public_id])
    assert client.post(url).status_code == 200
    proof = client.get(url).context["confirmation"]
    SMSTemplate.objects.filter(event_type=OPENING_EVENT).update(body="متن جدید بازگشایی")
    client.post(url, {"confirmation": proof})
    assert not DispatchResponseControl.objects.get(dispatch=c["dispatch"]).opened_at
    proof = client.get(url).context["confirmation"]
    assert client.post(url, {"confirmation": proof}).status_code == 302
    assert DispatchResponseControl.objects.get(dispatch=c["dispatch"]).opened_by == viewer


@pytest.mark.parametrize("count,mode,allowed", [(15, "SEALED", True), (16, "SEALED", False), (2, "INDIVIDUAL", False)])
def test_sms_recipient_limit_is_frozen_and_individual_mode_disables_it(opening_case, count, mode, allowed):
    c = opening_case
    for i in range(2, count):
        make_contact(f"limit-{i}", f"09122000{i:03}").customer_profile.groups.add(c["group"])
    dispatch, tickets = create_group_dispatch(user=c["staff"], unit=c["unit"], contact_group=c["group"], category=c["category"],
        subject="آزمون سقف مخاطبان", body="متن استعلام", response_mode=mode, notify_opening_sms=True)
    control = DispatchResponseControl.objects.get(dispatch=dispatch)
    assert control.recipient_count_snapshot == count and control.notify_sms is allowed
    if count == 16:
        c["group"].contacts.clear()
        with transaction.atomic(), pytest.raises(IntegrityError):
            DispatchResponseControl.objects.filter(pk=control.pk).update(notify_sms=True)
    if mode == "INDIVIDUAL":
        message = reply_ticket(ticket_id=tickets[0].pk, user=c["customers"][0], body="پاسخ جداگانه")
        from apps.tickets.staff_conversation import visible_messages
        assert visible_messages(tickets[0], c["staff"]).filter(pk=message.pk).exists()


def test_sealed_attachment_shared_routes_admin_and_direct_media_are_blocked(client, opening_case, settings, tmp_path):
    c = opening_case
    settings.MEDIA_ROOT = tmp_path
    settings.USE_NGINX_MEDIA = True
    from apps.attachments.services import save_attachments
    AttachmentPolicy.objects.create(attachment_type="PDF", max_file_size=1024 * 1024,
        allowed_extensions=[".pdf"], allowed_mime_types=["application/pdf"])
    save_attachments(c["message"], [SimpleUploadedFile("secret.pdf", b"%PDF-1.4\n%%EOF", content_type="application/pdf")], c["customers"][0])
    attachment = Attachment.objects.get(message=c["message"])
    assert attachment.relative_path.startswith("dispatch-responses/")
    MessageRevision.objects.create(message=c["message"], body="OLD_SECRET_BID", edited_by=c["customers"][0])
    with pytest.raises(PermissionDenied):
        share_staff_message(c["message"], c["staff"], c["other"])
    grant = MessageShare.objects.create(message=c["message"], user=c["other"], granted_by=c["staff"])
    client.force_login(c["other"])
    assert client.get(reverse("staff_tickets:shared_message_attachment", args=[grant.public_id, attachment.public_id])).status_code == 403
    client.force_login(c["admin"])
    for route in ("view", "download"):
        assert client.get(reverse("attachments:" + route, args=[attachment.public_id])).status_code == 403
    for name, obj in [("messaging_message", c["message"]), ("messaging_messagerevision", c["message"].revisions.get()), ("attachments_attachment", attachment)]:
        response = client.get(reverse("admin:" + name + "_change", args=[obj.pk]), follow=True)
        assert response.redirect_chain and c["message"].body not in response.content.decode()
    open_dispatch_responses(c["dispatch"], c["admin"], render_opening_sms(c["dispatch"]))
    assert client.get(reverse("admin:messaging_message_change", args=[c["message"].pk])).status_code == 200
    client.force_login(c["customers"][0])
    response = client.get(reverse("attachments:view", args=[attachment.public_id]))
    assert response.status_code == 200 and "no-store" in response["Cache-Control"]
    assert b"".join(response.streaming_content).startswith(b"%PDF")
    from config.urls import debug_media
    from django.http import Http404
    with pytest.raises(Http404):
        debug_media(RequestFactory().get("/media/"), attachment.relative_path)


def test_bale_does_not_disclose_managed_response_text(opening_case, monkeypatch):
    c = opening_case
    profile = c["staff"].staff_profile
    profile.bale_chat_id = "private-dispatch-owner"
    profile.save(update_fields=["bale_chat_id"])
    sent = []
    class Provider:
        def send_message(self, chat_id, body):
            sent.append((chat_id, body))
    monkeypatch.setattr("apps.notifications.management.commands.process_notifications.get_bale_provider", lambda: Provider())
    job = NotificationJob.objects.get(event_type="CUSTOMER_TICKET_REPLY", ticket=c["tickets"][0])
    _send_bale(job)
    assert sent and sent[0][0] == "private-dispatch-owner" and c["message"].body not in sent[0][1]


def test_seed_only_adds_missing_opening_template_and_preserves_customizations():
    from django.apps import apps
    from django.db import connection
    migration = importlib.import_module("apps.notifications.migrations.0021_dispatch_response_opening")
    original = list(SMSTemplate.objects.exclude(event_type=OPENING_EVENT).values())
    SMSTemplate.objects.filter(event_type=OPENING_EVENT).update(body="قالب اختصاصی حفظ شود")
    from types import SimpleNamespace
    schema_editor = SimpleNamespace(connection=connection)
    migration.seed_opening_template(apps, schema_editor)
    migration.seed_opening_template(apps, schema_editor)
    assert SMSTemplate.objects.get(event_type=OPENING_EVENT).body == "قالب اختصاصی حفظ شود"
    assert list(SMSTemplate.objects.exclude(event_type=OPENING_EVENT).values()) == original


def test_view_only_staff_can_list_reports_without_creation_permission(client, opening_case):
    c = opening_case
    unit = OrganizationalUnit.objects.create(name="واحد مشاهده", slug="opening-observer")
    viewer = make_staff("only-view-dispatches", unit)
    viewer.user_permissions.add(Permission.objects.get(codename="view_all_bulk_dispatches"))
    client.force_login(viewer)
    response = client.get(reverse("staff_tickets:dispatch_create"))
    assert response.status_code == 200 and c["dispatch"].subject in response.content.decode()
    assert response.context["staff_operations_access"]["dispatches"] is True
    assert 'id="dispatch-compose-form"' not in response.content.decode()
    assert client.post(reverse("staff_tickets:dispatch_create"), {}).status_code == 403
    assert client.get(reverse("staff_tickets:dispatch_open_responses", args=[c["dispatch"].public_id])).status_code == 403


def test_opening_rolls_back_if_outbox_cannot_be_saved(opening_case, monkeypatch):
    c = opening_case
    def fail(*args, **kwargs):
        raise RuntimeError("outbox unavailable")
    monkeypatch.setattr("apps.notifications.services.enqueue", fail)
    with pytest.raises(RuntimeError):
        open_dispatch_responses(c["dispatch"], c["staff"], render_opening_sms(c["dispatch"]))
    assert not DispatchResponseControl.objects.get(dispatch=c["dispatch"]).opened_at
    assert not NotificationJob.objects.filter(event_type=OPENING_EVENT).exists()


def test_existing_worker_sends_opening_job_once_without_issuing_magic_link(opening_case, monkeypatch):
    from django.core.management import call_command
    from apps.integrations.sms.base import SMSResult
    from apps.accounts.models import TicketMagicLink
    c = opening_case
    text = render_opening_sms(c["dispatch"])
    open_dispatch_responses(c["dispatch"], c["staff"], text)
    job = NotificationJob.objects.get(event_type=OPENING_EVENT)
    sent = []
    class Provider:
        def send_message(self, mobile, body):
            sent.append((mobile, body))
            return SMSResult(True, "opening-test-provider")
    monkeypatch.setattr("apps.notifications.management.commands.process_notifications.get_sms_provider", lambda: Provider())
    links_before = TicketMagicLink.objects.count()
    call_command("process_notifications", once=True, job_id=job.pk)
    call_command("process_notifications", once=True, job_id=job.pk)
    job.refresh_from_db()
    assert job.status == "SENT" and sent == [(c["customers"][0].mobile, text)]
    assert TicketMagicLink.objects.count() == links_before


@pytest.mark.django_db(transaction=True)
def test_seed_round_trip_preserves_opening_with_user_primary_key_gaps(opening_case, settings, tmp_path, monkeypatch):
    from django.core.management import call_command
    from django.db import connection
    if connection.vendor != "sqlite":
        pytest.skip("Seed export is deliberately restricted to SQLite sources")
    c = opening_case
    settings.MEDIA_ROOT = tmp_path / "source-media"
    User.objects.create_user(username="discarded-seed-contact").delete()
    customer = make_contact("stable-seed-respondent", "09122000888")
    customer.customer_profile.groups.add(c["group"])
    dispatch, tickets = create_group_dispatch(user=c["staff"], unit=c["unit"], contact_group=c["group"],
        category=c["category"], subject="بازگشایی با انتقال داده", body="متن", notify_opening_sms=False)
    ticket = next(ticket for ticket in tickets if ticket.customer_id == customer.pk)
    reply_ticket(ticket_id=ticket.pk, user=customer, body="پاسخ منتقل‌شونده")
    control, _ = open_dispatch_responses(dispatch, c["staff"])
    snapshot = control.respondents_snapshot
    opened_at = control.opened_at
    assert snapshot == [str(customer.public_id)]
    bundle = tmp_path / "bundle"
    call_command("export_production_seed", output=str(bundle))
    call_command("flush", interactive=False, verbosity=0)
    settings.MEDIA_ROOT = tmp_path / "destination-media"
    monkeypatch.setattr(connection, "vendor", "postgresql")
    call_command("import_production_seed", bundle_dir=str(bundle))
    restored = DispatchResponseControl.objects.get(dispatch_id=dispatch.pk)
    # Django's existing JSON fixture format retains millisecond precision.
    assert 0 <= (opened_at - restored.opened_at).total_seconds() < 0.001
    assert restored.respondents_snapshot == snapshot
    assert restored.opened_by.username == c["staff"].username
    assert str(User.objects.get(username=customer.username).public_id) in restored.respondents_snapshot


def test_managed_reply_bale_never_falls_back_to_a_broad_unit_room(opening_case, monkeypatch):
    c = opening_case
    sent = []
    class Provider:
        def send_message(self, chat_id, text):
            sent.append((chat_id, text))
    monkeypatch.setattr("apps.notifications.management.commands.process_notifications.get_bale_provider", lambda: Provider())
    job = NotificationJob.objects.get(event_type="CUSTOMER_TICKET_REPLY", ticket=c["tickets"][0])
    assert _send_bale(job) is None
    assert not sent


def test_acknowledging_question_does_not_mark_a_sealed_response_ticket_read(client, opening_case):
    from apps.tickets.models import TicketReadState
    c = opening_case
    ticket = c["tickets"][0]
    question = ticket.ticket_messages.get(author=c["staff"])
    client.force_login(c["admin"])
    url = reverse("staff_tickets:read_shares", args=[ticket.public_id])
    client.post(url, {"message_ids": [str(question.public_id)]})
    assert MessageReadReceipt.objects.filter(message=question, user=c["admin"]).exists()
    assert not TicketReadState.objects.filter(ticket=ticket, user=c["admin"]).exists()
    open_dispatch_responses(c["dispatch"], c["admin"], render_opening_sms(c["dispatch"]))
    assert not MessageReadReceipt.objects.filter(message=c["message"], user=c["admin"]).exists()
    client.post(url, {"message_ids": [str(c["message"].public_id)]})
    assert TicketReadState.objects.filter(ticket=ticket, user=c["admin"]).exists()


def test_existing_share_audit_cannot_disclose_sealed_body_to_creator_or_admin(client, opening_case):
    c = opening_case
    MessageShare.objects.create(message=c["message"], user=c["other"], granted_by=c["staff"])
    for actor in ("staff", "admin"):
        client.force_login(c[actor])
        response = client.get(reverse("staff_tickets:detail", args=[c["tickets"][0].public_id]))
        assert response.status_code == 200 and c["message"].body not in response.content.decode()
        assert not response.context["message_shares"].filter(message=c["message"]).exists()
    open_dispatch_responses(c["dispatch"], c["staff"], render_opening_sms(c["dispatch"]))
    response = client.get(reverse("staff_tickets:detail", args=[c["tickets"][0].public_id]))
    assert c["message"].body in response.content.decode()