from datetime import timedelta from unittest.mock import Mock, patch import pytest import requests from django.core.management import call_command from django.urls import reverse from django.utils import timezone from apps.accounts.models import User from apps.core.admin import VoipConfigurationAdmin from apps.core.models import VoipConfiguration from apps.messaging.models import MessageReadReceipt from apps.notifications.models import AnnouncementJob, NotificationJob from apps.staff.models import OrganizationalUnit from apps.tickets.models import TicketAccessGrant, TicketCategory from apps.tickets.services import create_ticket, reply_ticket @pytest.fixture def announcement_catalog(db): unit = OrganizationalUnit.objects.create(name="واحد تماس", slug="announcement-unit") category = TicketCategory.objects.create(unit=unit, title="پیگیری", slug="announcement-category") return category @pytest.fixture def announcement_people(db): staff = User.objects.create_superuser(username="announcement-staff", password="safe-password", account_type="STAFF") customer = User.objects.create_user(username="announcement-customer", mobile="09120000000", account_type="CUSTOMER") return staff, customer @pytest.fixture def announcement_data(announcement_catalog, announcement_people): configuration = VoipConfiguration.objects.create( announcement_url="http://37.255.224.143:8009/v1/announcements", announcement_id="custom/fereshte1_G711", announcement_delay_minutes=90, ) return announcement_catalog, *announcement_people, configuration def due(job): AnnouncementJob.objects.filter(pk=job.pk).update(next_attempt_at=timezone.now() - timedelta(seconds=1)) def test_staff_creation_and_each_public_reply_queue_independent_calls(announcement_data): category, staff, customer, _ = announcement_data ticket = create_ticket(customer=customer, created_by=staff, category=category, subject="پیگیری مشتری", body="پیام نخست") first = ticket.ticket_messages.get() first_job = AnnouncementJob.objects.get(message=first) assert timedelta(minutes=89) <= first_job.next_attempt_at - first.created_at <= timedelta(minutes=91) reply = reply_ticket(ticket_id=ticket.pk, user=staff, body="پاسخ تازه") assert AnnouncementJob.objects.filter(message__ticket=ticket).count() == 2 assert AnnouncementJob.objects.get(message=reply).idempotency_key != first_job.idempotency_key reply_ticket(ticket_id=ticket.pk, user=staff, body="یادداشت داخلی", internal=True) reply_ticket(ticket_id=ticket.pk, user=customer, body="پاسخ مشتری") assert AnnouncementJob.objects.filter(message__ticket=ticket).count() == 2 customer_ticket = create_ticket(customer=customer, category=category, subject="درخواست مشتری", body="پیام مشتری") assert not AnnouncementJob.objects.filter(message__ticket=customer_ticket).exists() @pytest.mark.parametrize("api_key", ["", "secret-key"]) def test_due_unread_message_calls_once_with_optional_api_key(announcement_data, api_key): category, staff, customer, configuration = announcement_data configuration.announcement_api_key = api_key configuration.save() ticket = create_ticket(customer=customer, created_by=staff, category=category, subject="پیگیری مشتری", body="پیام نخست") job = AnnouncementJob.objects.get(message__ticket=ticket) response = Mock(content=b'{}') response.json.return_value = {} with patch("apps.integrations.announcements.requests.post", return_value=response) as post: call_command("process_notifications", once=True) assert not post.called due(job) call_command("process_notifications", once=True) call_command("process_notifications", once=True) job.refresh_from_db() assert job.status == AnnouncementJob.Status.SENT and job.attempts == 1 post.assert_called_once() assert post.call_args.args == (configuration.announcement_url,) assert post.call_args.kwargs["json"] == {"customer_number": "09120000000", "announcement_id": configuration.announcement_id} assert post.call_args.kwargs["headers"] == ({"Idempotency-Key": job.idempotency_key, "X-API-Key": api_key} if api_key else {"Idempotency-Key": job.idempotency_key}) def test_customer_read_cancels_only_that_message(announcement_data): category, staff, customer, _ = announcement_data ticket = create_ticket(customer=customer, created_by=staff, category=category, subject="پیگیری مشتری", body="پیام نخست") second = reply_ticket(ticket_id=ticket.pk, user=staff, body="پاسخ دوم") first_job = AnnouncementJob.objects.get(message=ticket.ticket_messages.exclude(pk=second.pk).get()) second_job = AnnouncementJob.objects.get(message=second) MessageReadReceipt.objects.create(message=first_job.message, user=customer) due(first_job) due(second_job) with patch("apps.integrations.announcements.requests.post", return_value=Mock(content=b"", success=True)) as post: call_command("process_notifications", once=True) first_job.refresh_from_db() second_job.refresh_from_db() assert first_job.status == AnnouncementJob.Status.CANCELLED and first_job.last_error == "customer_read" assert second_job.status == AnnouncementJob.Status.SENT post.assert_called_once() def test_provider_failure_retries_with_same_idempotency_key(announcement_data): category, staff, customer, _ = announcement_data ticket = create_ticket(customer=customer, created_by=staff, category=category, subject="پیگیری مشتری", body="پیام نخست") job = AnnouncementJob.objects.get(message__ticket=ticket) keys = [] def fail(*args, **kwargs): keys.append(kwargs["headers"]["Idempotency-Key"]) raise requests.Timeout("timeout") with patch("apps.integrations.announcements.requests.post", side_effect=fail): for attempt in range(3): due(job) call_command("process_notifications", once=True) job.refresh_from_db() assert job.attempts == attempt + 1 assert job.status == AnnouncementJob.Status.FAILED assert keys == [job.idempotency_key] * 3 assert job.last_error == "service_timeout" @pytest.mark.parametrize( ("provider_result", "expected_error"), [ (requests.ConnectionError("secret-in-provider-error"), "service_unreachable"), (requests.HTTPError("503 Service Unavailable"), "service_http_error"), (Mock(content=b"not-json", json=Mock(side_effect=ValueError("bad JSON"))), "invalid_service_response"), (Mock(content=b"[]", json=Mock(return_value=[])), "invalid_service_response"), (Mock(content=b'{"success": false}', json=Mock(return_value={"success": False})), "service_rejected"), ], ) def test_provider_errors_are_sanitized_and_do_not_stop_worker(announcement_data, provider_result, expected_error): category, staff, customer, _ = announcement_data ticket = create_ticket(customer=customer, created_by=staff, category=category, subject="پیگیری مشتری", body="پیام نخست") second = reply_ticket(ticket_id=ticket.pk, user=staff, body="پیام دوم") first_job = AnnouncementJob.objects.get(message=ticket.ticket_messages.exclude(pk=second.pk).get()) second_job = AnnouncementJob.objects.get(message=second) due(first_job) due(second_job) success = Mock(content=b"") with patch("apps.integrations.announcements.requests.post", side_effect=[provider_result, success]) as post: call_command("process_notifications", once=True) first_job.refresh_from_db() second_job.refresh_from_db() assert post.call_count == 2 assert first_job.status == AnnouncementJob.Status.PENDING assert first_job.attempts == 1 and first_job.last_error == expected_error assert second_job.status == AnnouncementJob.Status.SENT assert first_job.next_attempt_at > timezone.now() def test_customer_reply_prevents_unnecessary_call(announcement_data): category, staff, customer, _ = announcement_data ticket = create_ticket(customer=customer, created_by=staff, category=category, subject="پیگیری مشتری", body="پیام نخست") job = AnnouncementJob.objects.get(message__ticket=ticket) reply_ticket(ticket_id=ticket.pk, user=customer, body="پاسخ مشتری") due(job) with patch("apps.integrations.announcements.requests.post") as post: call_command("process_notifications", once=True) job.refresh_from_db() assert job.status == AnnouncementJob.Status.CANCELLED and job.last_error == "customer_replied" post.assert_not_called() def test_missing_configuration_cancels_without_call(announcement_data): category, staff, customer, configuration = announcement_data ticket = create_ticket(customer=customer, created_by=staff, category=category, subject="پیگیری مشتری", body="پیام نخست") job = AnnouncementJob.objects.get(message__ticket=ticket) configuration.announcement_url = "" configuration.save() due(job) with patch("apps.integrations.announcements.requests.post") as post: call_command("process_notifications", once=True) job.refresh_from_db() assert job.status == AnnouncementJob.Status.CANCELLED and job.last_error == "announcement_not_configured" post.assert_not_called() def test_database_delay_and_admin_api_key_controls(announcement_data): category, staff, customer, configuration = announcement_data configuration.announcement_delay_minutes = 15 configuration.announcement_api_key = "stored-secret" configuration.save() ticket = create_ticket(customer=customer, created_by=staff, category=category, subject="پیگیری مشتری", body="پیام نخست") job = AnnouncementJob.objects.get(message__ticket=ticket) assert timedelta(minutes=14) <= job.next_attempt_at - job.message.created_at <= timedelta(minutes=16) form_type = VoipConfigurationAdmin.ConfigurationForm data = { "announcement_url": configuration.announcement_url, "announcement_id": configuration.announcement_id, "announcement_delay_minutes": 15, "voip_call_url": "", } unchanged = form_type(data=data, instance=configuration) assert "stored-secret" not in unchanged.as_p() assert unchanged.is_valid(), unchanged.errors assert unchanged.save().announcement_api_key == "stored-secret" cleared = form_type(data={**data, "clear_announcement_api_key": "on"}, instance=configuration) assert cleared.is_valid(), cleared.errors assert cleared.save().announcement_api_key == "" def test_staff_ticket_shows_call_history_without_claiming_connection(client, announcement_data): category, staff, customer, configuration = announcement_data configuration.announcement_api_key = "private-provider-key" configuration.save() ticket = create_ticket(customer=customer, created_by=staff, category=category, subject="پیگیری مشتری", body="پیام نخست") reply = reply_ticket(ticket_id=ticket.pk, user=staff, body="پاسخ تازه") first_job = AnnouncementJob.objects.get(message=ticket.ticket_messages.exclude(pk=reply.pk).get()) second_job = AnnouncementJob.objects.get(message=reply) AnnouncementJob.objects.filter(pk=first_job.pk).update(status=AnnouncementJob.Status.FAILED, attempts=3, last_error="service_timeout") AnnouncementJob.objects.filter(pk=second_job.pk).update(status=AnnouncementJob.Status.SENT, attempts=1, sent_at=timezone.now()) client.force_login(staff) content = client.get(f"/staff/tickets/{ticket.public_id}/").content.decode() assert "وضعیت پیامک‌های مشتری" in content assert "وضعیت اطلاع‌رسانی‌های تلفنی" in content assert content.count('aria-label="نمایش جزئیات اطلاع‌رسانی تلفنی"') == 2 assert "سرویس تماس در مهلت مقرر پاسخ نداد." in content assert "درخواست تماس پذیرفته شد" in content assert "پذیرش درخواست، به‌تنهایی برقرار شدن" in content assert configuration.announcement_url not in content assert configuration.announcement_api_key not in content def test_staff_can_queue_immediate_call_and_see_it_in_ticket(client, announcement_data): category, staff, customer, configuration = announcement_data ticket = create_ticket(customer=customer, created_by=staff, category=category, subject="تماس فوری", body="پیام نخست") automatic = AnnouncementJob.objects.get(message__ticket=ticket) MessageReadReceipt.objects.create(message=automatic.message, user=customer) detail_url = reverse("staff_tickets:detail", args=[ticket.public_id]) call_url = reverse("staff_tickets:announcement_now", args=[ticket.public_id]) client.force_login(staff) detail = client.get(detail_url).content.decode() assert call_url in detail assert "اطلاع‌رسانی تلفنی فوری" in detail assert "پیام صوتی از پیش ضبط‌شده" in detail assert detail.index("وضعیت پیامک‌های مشتری") < detail.index('aria-label="اطلاع‌رسانی تلفنی فوری"') assert client.get(call_url).status_code == 405 provider_response = Mock(content=b"{}") provider_response.json.return_value = {} with patch("apps.integrations.announcements.requests.post", return_value=provider_response) as post: response = client.post(call_url, follow=True) assert response.status_code == 200 assert "درخواست تماس توسط سرویس پذیرفته شد." in response.content.decode() job = AnnouncementJob.objects.get(ticket=ticket, source=AnnouncementJob.Source.MANUAL) assert job.message is None and job.requested_by == staff assert job.status == AnnouncementJob.Status.SENT assert "درخواست فوری" in response.content.decode() assert AnnouncementJob.objects.get(pk=automatic.pk).status == AnnouncementJob.Status.PENDING post.assert_called_once() assert post.call_args.args == (configuration.announcement_url,) assert post.call_args.kwargs["json"] == {"customer_number": customer.local_mobile, "announcement_id": configuration.announcement_id} assert post.call_args.kwargs["headers"] == {"Idempotency-Key": job.idempotency_key} def test_immediate_call_coalesces_repeated_clicks(client, announcement_data): category, staff, customer, _ = announcement_data ticket = create_ticket(customer=customer, category=category, subject="تماس فوری", body="پیام مشتری") url = reverse("staff_tickets:announcement_now", args=[ticket.public_id]) client.force_login(staff) with patch("apps.integrations.announcements.requests.post", return_value=Mock(content=b"")) as post: assert client.post(url).status_code == 302 second = client.post(url, follow=True) assert "در دقیقهٔ گذشته ثبت شده است" in second.content.decode() assert AnnouncementJob.objects.filter(ticket=ticket, source=AnnouncementJob.Source.MANUAL).count() == 1 post.assert_called_once() def test_immediate_call_requires_staff_reply_access_and_configuration(client, announcement_data): category, staff, customer, configuration = announcement_data ticket = create_ticket(customer=customer, category=category, subject="تماس فوری", body="پیام مشتری") url = reverse("staff_tickets:announcement_now", args=[ticket.public_id]) outsider = User.objects.create_user(username="announcement-viewer", account_type="STAFF") TicketAccessGrant.objects.create(ticket=ticket, user=outsider, granted_by=staff, permission_level=TicketAccessGrant.Level.VIEW_ONLY) client.force_login(outsider) assert url not in client.get(reverse("staff_tickets:detail", args=[ticket.public_id])).content.decode() assert client.post(url).status_code == 403 client.force_login(customer) assert client.post(url).status_code == 403 assert not AnnouncementJob.objects.filter(ticket=ticket).exists() client.force_login(staff) configuration.announcement_url = "" configuration.save() page = client.get(reverse("staff_tickets:detail", args=[ticket.public_id])).content.decode() assert "اطلاع‌رسانی تلفنی فوری" in page and "disabled" in page result = client.post(url, follow=True) assert "تنظیمات اطلاع‌رسانی تلفنی کامل نیست." in result.content.decode() assert not AnnouncementJob.objects.filter(ticket=ticket).exists() def test_immediate_call_is_processed_before_older_scheduled_calls(client, announcement_data): category, staff, customer, _ = announcement_data ticket = create_ticket(customer=customer, created_by=staff, category=category, subject="تماس فوری", body="پیام نخست") for message in ticket.ticket_messages.all(): due(AnnouncementJob.objects.get(message=message)) client.force_login(staff) with patch("apps.integrations.announcements.requests.post", return_value=Mock(content=b"")) as immediate_post: client.post(reverse("staff_tickets:announcement_now", args=[ticket.public_id])) immediate = AnnouncementJob.objects.get(ticket=ticket, source=AnnouncementJob.Source.MANUAL) assert immediate.status == AnnouncementJob.Status.SENT immediate_post.assert_called_once() with patch("apps.integrations.announcements.requests.post", return_value=Mock(content=b"")) as post: call_command("process_notifications", once=True) assert all(call.kwargs["headers"]["Idempotency-Key"] != immediate.idempotency_key for call in post.call_args_list) def test_immediate_call_retries_provider_outage_without_breaking_ticket(client, announcement_data): category, staff, customer, _ = announcement_data ticket = create_ticket(customer=customer, category=category, subject="تماس فوری", body="پیام مشتری") url = reverse("staff_tickets:announcement_now", args=[ticket.public_id]) client.force_login(staff) with patch("apps.integrations.announcements.requests.post", side_effect=requests.ConnectionError("provider secret")) as post: result = client.post(url, follow=True) job = AnnouncementJob.objects.get(ticket=ticket, source=AnnouncementJob.Source.MANUAL) assert "تلاش مجدد در صف قرار گرفت" in result.content.decode() assert job.status == AnnouncementJob.Status.PENDING assert job.last_error == "service_unreachable" and job.attempts == 1 assert post.call_args.kwargs["headers"]["Idempotency-Key"] == job.idempotency_key assert client.get(reverse("staff_tickets:detail", args=[ticket.public_id])).status_code == 200 due(job) with patch("apps.integrations.announcements.requests.post", return_value=Mock(content=b"")) as retry: call_command("process_notifications", once=True) job.refresh_from_db() assert job.status == AnnouncementJob.Status.SENT assert retry.call_args.kwargs["headers"]["Idempotency-Key"] == job.idempotency_key @pytest.mark.django_db def test_worker_skips_sms_provider_when_no_delivery_check_is_due(): with patch("apps.notifications.management.commands.process_notifications.get_sms_provider", side_effect=RuntimeError("SMS provider unavailable")) as sms_provider: call_command("process_notifications", once=True) sms_provider.assert_not_called() def test_sms_delivery_provider_error_does_not_stop_worker(announcement_data): category, staff, customer, _ = announcement_data ticket = create_ticket(customer=customer, created_by=staff, category=category, subject="تماس فوری", body="پیام") sms_job = NotificationJob.objects.create( channel=NotificationJob.Channel.SMS, event_type="CUSTOMER_TICKET_CREATED", ticket=ticket, customer=customer, status=NotificationJob.Status.SENT, next_attempt_at=timezone.now(), delivery_status=NotificationJob.DeliveryStatus.PENDING, delivery_next_check_at=timezone.now() - timedelta(seconds=1), provider_message_id="provider-id", dedupe_key="sms-delivery-failure-does-not-stop-worker", ) with patch("apps.notifications.management.commands.process_notifications.get_sms_provider", side_effect=RuntimeError("provider unavailable")): call_command("process_notifications", once=True) sms_job.refresh_from_db() assert sms_job.delivery_status == NotificationJob.DeliveryStatus.PENDING assert sms_job.delivery_next_check_at > timezone.now()