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()