customerTicket/tests/test_domain.py

import pytest
import base64
from datetime import timedelta
from django.contrib.auth.models import Permission
from django.core.exceptions import PermissionDenied, ValidationError
from django.test import override_settings
from django.utils import timezone
from django.core.files.uploadedfile import SimpleUploadedFile
from apps.attachments.models import Attachment, AttachmentPolicy
from apps.accounts.models import User, normalize_mobile
from apps.messaging.models import MessageReadReceipt
from apps.notifications.models import NotificationJob
from apps.staff.models import OrganizationalUnit, StaffProfile
from apps.tickets.models import Ticket, TicketCategory
from apps.tickets.policies import TicketPolicy
from apps.tickets.services import claim_ticket, create_ticket, delete_ticket, reply_ticket

@pytest.fixture
def category(db):
    unit = OrganizationalUnit.objects.create(name="واحد تست دامنه", slug="domain-test-unit", bale_chat_id="domain-bale-group")
    unit.permissions.add(
        Permission.objects.get(codename="view_ticket", content_type__app_label="tickets"),
        Permission.objects.get(codename="add_message", content_type__app_label="messaging"),
    )
    return TicketCategory.objects.create(unit=unit, title="قیمت", slug="domain-test-price")

@pytest.fixture
def customer(db):
    return User.objects.create_user(username="customer", mobile="09121234567", account_type="CUSTOMER")

def make_staff(username):
    mobile = "09120000001" if username == "a" else "09120000002"
    user = User.objects.create_user(username=username, mobile=mobile, account_type="STAFF")
    unit = OrganizationalUnit.objects.filter(slug="domain-test-unit").first()
    StaffProfile.objects.create(user=user, primary_unit=unit, display_name=username)
    for code in ["view_ticket", "claim_ticket", "change_ticket"]:
        user.user_permissions.add(Permission.objects.get(codename=code, content_type__app_label="tickets"))
    return user

@pytest.mark.django_db
def test_mobile_normalization():
    assert normalize_mobile("0912 123 4567") == "+989121234567"

@pytest.mark.django_db
def test_customer_reply_enqueues_bale_group_notification(category, customer):
    ticket = create_ticket(customer=customer, category=category, subject="پیگیری سفارش", body="پیام نخست")
    message = reply_ticket(ticket_id=ticket.pk, user=customer, body="پاسخ تازه مشتری")
    job = NotificationJob.objects.get(event_type="CUSTOMER_TICKET_REPLY", ticket=ticket)
    assert job.channel == NotificationJob.Channel.BALE
    assert job.payload == {"message_id": message.pk,"message_count":1}
    assert job.next_attempt_at>=timezone.now()+timedelta(seconds=595)

@pytest.mark.django_db
def test_customer_ticket_and_reply_bale_events_target_owning_unit_group(category, customer, monkeypatch):
    from apps.notifications.management.commands.process_notifications import _send_bale
    sent = []
    class Provider:
        def send_message(self, chat_id, text):
            sent.append((chat_id, text))
            return type("Result", (), {"success": True})()
    monkeypatch.setattr("apps.notifications.management.commands.process_notifications.get_bale_provider", lambda: Provider())
    from apps.customers.models import CustomerProfile
    profile,_=CustomerProfile.objects.get_or_create(user=customer)
    profile.first_name="سارا"; profile.last_name="احمدی"; profile.city="تهران"; profile.save()
    ticket = create_ticket(customer=customer, category=category, subject="اعلان بله", body="پیام نخست")
    created_job = NotificationJob.objects.get(ticket=ticket, event_type="NEW_CUSTOMER_TICKET")
    _send_bale(created_job)
    reply = reply_ticket(ticket_id=ticket.pk, user=customer, body="پاسخ مشتری")
    reply_job = NotificationJob.objects.get(ticket=ticket, event_type="CUSTOMER_TICKET_REPLY")
    _send_bale(reply_job)
    assert [chat_id for chat_id, _ in sent] == ["domain-bale-group", "domain-bale-group"]
    assert "تیکت جدید مشتری" in sent[0][1]
    assert "پاسخ جدید مشتری" in sent[1][1]
    assert reply.body in sent[1][1]
    for text in (sent[0][1],sent[1][1]):
        assert "👤 سارا احمدی" in text
        assert f"📞 {customer.local_mobile}" in text
        assert f"/staff/tickets/{ticket.public_id}/" in text
        assert "🎫" in text and "📌 اعلان بله" in text and "📝" in text
    assert "📍 تهران" in sent[0][1]


@pytest.mark.django_db
def test_bale_group_notifications_are_isolated_between_units(category, customer, monkeypatch):
    from apps.notifications.management.commands.process_notifications import _send_bale

    second_unit = OrganizationalUnit.objects.create(
        name="واحد دوم اعلان بله",
        slug="second-bale-unit",
        bale_chat_id="second-bale-group",
    )
    second_category = TicketCategory.objects.create(unit=second_unit, title="حسابداری", slug="second-bale-category")
    sent = []

    class Provider:
        def send_message(self, chat_id, text):
            sent.append((chat_id, text))
            return type("Result", (), {"success": True})()

    monkeypatch.setattr("apps.notifications.management.commands.process_notifications.get_bale_provider", lambda: Provider())
    first_ticket = create_ticket(customer=customer, category=category, subject="درخواست فروش", body="پیام فروش")
    second_ticket = create_ticket(customer=customer, category=second_category, subject="درخواست مالی", body="پیام مالی")
    first_reply = reply_ticket(ticket_id=first_ticket.pk, user=customer, body="پاسخ فروش")
    second_reply = reply_ticket(ticket_id=second_ticket.pk, user=customer, body="پاسخ مالی")

    for ticket, event_type in (
        (first_ticket, "NEW_CUSTOMER_TICKET"),
        (second_ticket, "NEW_CUSTOMER_TICKET"),
        (first_ticket, "CUSTOMER_TICKET_REPLY"),
        (second_ticket, "CUSTOMER_TICKET_REPLY"),
    ):
        _send_bale(NotificationJob.objects.get(ticket=ticket, event_type=event_type))

    assert [chat_id for chat_id, _ in sent] == [
        "domain-bale-group",
        "second-bale-group",
        "domain-bale-group",
        "second-bale-group",
    ]
    assert first_reply.body in sent[2][1]
    assert second_reply.body in sent[3][1]
    assert "درخواست مالی" not in sent[0][1] and "درخواست فروش" not in sent[1][1]


@pytest.mark.django_db
def test_blank_unit_bale_chat_never_falls_back_to_global_group(category, customer, settings):
    settings.BALE_GROUP_CHAT_ID = "legacy-global-group"
    category.unit.bale_chat_id = ""
    category.unit.save(update_fields=["bale_chat_id", "updated_at"])

    ticket = create_ticket(customer=customer, category=category, subject="بدون گروه", body="پیام")
    reply_ticket(ticket_id=ticket.pk, user=customer, body="پاسخ")

    assert not NotificationJob.objects.filter(ticket=ticket, channel=NotificationJob.Channel.BALE).exists()


@pytest.mark.django_db
def test_unit_notification_destinations_normalize_and_reject_duplicate_supervisors():
    unit = OrganizationalUnit(
        name="واحد مقصد اعلان",
        slug="notification-destinations",
        supervisor_mobile_1="0912 111 2233",
        supervisor_mobile_2="0935 222 3344",
        supervisor_mobile_4="0990 444 5566",
        supervisor_mobile_5="0921 555 6677",
    )
    unit.full_clean()
    unit.save()
    assert unit.supervisor_mobiles == ("+989121112233", "+989352223344", "+989904445566", "+989215556677")

    unit.supervisor_mobile_3 = "09121112233"
    with pytest.raises(ValidationError, match="قبلاً"):
        unit.full_clean()


@pytest.mark.django_db
def test_new_ticket_sms_is_queued_for_each_unit_supervisor_only_on_creation(category, customer, monkeypatch):
    from apps.notifications.management.commands.process_notifications import _send_sms

    unit = category.unit
    unit.supervisor_mobile_1 = "09121112233"
    unit.supervisor_mobile_2 = "09352223344"
    unit.supervisor_mobile_3 = "09903334455"
    unit.supervisor_mobile_4 = "09214445566"
    unit.supervisor_mobile_5 = "09195556677"
    unit.save()
    ticket = create_ticket(customer=customer, category=category, subject="  پیگیری   بارگیری  ", body="متن")
    jobs = list(NotificationJob.objects.filter(ticket=ticket, event_type="UNIT_SUPERVISOR_NEW_TICKET").order_by("id"))
    assert [job.payload["recipient_mobile"] for job in jobs] == list(unit.supervisor_mobiles)

    sent = []

    class Provider:
        def send_message(self, mobile, message):
            sent.append((mobile, message))
            return type("Result", (), {"success": True})()

    monkeypatch.setattr("apps.notifications.management.commands.process_notifications.get_sms_provider", lambda: Provider())
    for job in jobs:
        result, issued_link = _send_sms(job)
        assert result.success and issued_link is None

    assert [mobile for mobile, _ in sent] == list(unit.supervisor_mobiles)
    for _, message in sent:
        assert customer.local_mobile in message
        assert "موضوع: پیگیری بارگیری" in message
        assert f"دسته: {category.title}" in message
        assert "ارجاع: صف عمومی" in message

    supervisor_job_count = NotificationJob.objects.filter(event_type="UNIT_SUPERVISOR_NEW_TICKET").count()
    reply_ticket(ticket_id=ticket.pk, user=customer, body="پاسخ جدید")
    assert NotificationJob.objects.filter(event_type="UNIT_SUPERVISOR_NEW_TICKET").count() == supervisor_job_count


@pytest.mark.django_db
def test_unit_supervisor_sms_uses_editable_unit_template_and_selected_expert(category, customer, monkeypatch):
    from apps.notifications.management.commands.process_notifications import _send_sms
    from apps.notifications.models import SMSTemplate
    from apps.customers.models import CustomerProfile

    unit = category.unit
    unit.supervisor_mobile_1 = "09125556677"
    unit.save()
    expert = make_staff("selected-supervisor-sms-expert")
    expert.staff_profile.display_name = "کارشناس انتخابی پیامک"
    expert.staff_profile.primary_unit = unit
    expert.staff_profile.save()
    CustomerProfile.objects.create(user=customer, first_name="سارا", last_name="احمدی")
    template = SMSTemplate.objects.create(
        unit=unit,
        event_type=SMSTemplate.Event.UNIT_SUPERVISOR_NEW_TICKET,
        title="الگوی سفارشی سرپرستان",
        body="{first_name} {last_name} | {unit_name} | {ticket_number} | {customer_mobile} | {category_name} | {expert_name}",
    )
    template.full_clean()
    ticket = create_ticket(
        customer=customer, category=category, subject="تست الگوی قابل تنظیم",
        selected_expert=expert, body="متن",
    )
    job = NotificationJob.objects.get(ticket=ticket, event_type=SMSTemplate.Event.UNIT_SUPERVISOR_NEW_TICKET)
    sent = []

    class Provider:
        def send_message(self, mobile, message):
            sent.append((mobile, message))
            return type("Result", (), {"success": True})()

    monkeypatch.setattr("apps.notifications.management.commands.process_notifications.get_sms_provider", lambda: Provider())
    _send_sms(job)
    assert sent == [(
        "+989125556677",
        f"سارا احمدی | {unit.name} | {ticket.ticket_number} | {customer.local_mobile} | {category.title} | کارشناس انتخابی پیامک",
    )]


@pytest.mark.django_db
def test_unit_supervisor_sms_handles_missing_customer_names_and_general_queue(category, customer, monkeypatch):
    from apps.notifications.management.commands.process_notifications import _send_sms
    from apps.notifications.models import SMSTemplate

    category.unit.supervisor_mobile_1 = "09125556677"
    category.unit.save()
    SMSTemplate.objects.create(
        unit=category.unit,
        event_type=SMSTemplate.Event.UNIT_SUPERVISOR_NEW_TICKET,
        title="الگوی نام خالی سرپرستان",
        body="نام: {first_name} {last_name}\nنام جایگزین: {recipient_name}\nارجاع: {expert_name}",
    )
    ticket = create_ticket(customer=customer, category=category, subject="مشتری بدون نام", body="متن")
    job = NotificationJob.objects.get(ticket=ticket, event_type=SMSTemplate.Event.UNIT_SUPERVISOR_NEW_TICKET)
    sent = []

    class Provider:
        def send_message(self, mobile, message):
            sent.append((mobile, message))
            return type("Result", (), {"success": True})()

    monkeypatch.setattr("apps.notifications.management.commands.process_notifications.get_sms_provider", lambda: Provider())
    _send_sms(job)

    assert sent == [(
        "+989125556677",
        "نام:  \nنام جایگزین: ثبت نشده\nارجاع: صف عمومی",
    )]


@pytest.mark.django_db
def test_global_supervisor_sms_template_is_seeded_and_editable():
    from apps.notifications.models import SMSTemplate

    template = SMSTemplate.objects.get(
        unit__isnull=True, event_type=SMSTemplate.Event.UNIT_SUPERVISOR_NEW_TICKET
    )
    assert "{expert_name}" in template.body
    assert "{customer_mobile}" in template.body
    assert "{category_name}" in template.body
    assert {"expert_name", "customer_mobile", "category_name"}.issubset(template.allowed_tags)


@pytest.mark.django_db
def test_staff_created_ticket_notifies_customer_but_not_unit_supervisors(category, customer):
    category.unit.supervisor_mobile_1 = "09124445566"
    category.unit.save()
    staff = make_staff("a")

    ticket = create_ticket(
        customer=customer,
        created_by=staff,
        unit=category.unit,
        category=category,
        subject="ثبت توسط کارشناس",
        body="متن",
    )

    assert NotificationJob.objects.filter(
        ticket=ticket,
        event_type="UNIT_SUPERVISOR_NEW_TICKET",
    ).count() == 0
    customer_job = NotificationJob.objects.get(ticket=ticket, event_type="STAFF_TICKET_CREATED")
    assert customer_job.customer == customer

@pytest.mark.django_db
def test_customer_bale_replies_are_debounced_by_database_policy(category,customer):
    from apps.notifications.models import NotificationPolicy
    policy=NotificationPolicy.objects.get(key="CUSTOMER_REPLY_BALE")
    policy.quiet_period_seconds=720; policy.save()
    ticket=create_ticket(customer=customer,category=category,subject="چت مشتری",body="شروع")
    first=reply_ticket(ticket_id=ticket.pk,user=customer,body="پیام اول")
    job=NotificationJob.objects.get(ticket=ticket,event_type="CUSTOMER_TICKET_REPLY")
    first_due=job.next_attempt_at
    second=reply_ticket(ticket_id=ticket.pk,user=customer,body="پیام دوم")
    job.refresh_from_db()
    assert NotificationJob.objects.filter(ticket=ticket,event_type="CUSTOMER_TICKET_REPLY").count()==1
    assert job.payload=={"message_id":second.pk,"message_count":2}
    assert job.next_attempt_at>=first_due and job.next_attempt_at>=timezone.now()+timedelta(seconds=715)

@pytest.mark.django_db
def test_reopened_ticket_bale_notification_is_immediate(category,customer):
    ticket=create_ticket(customer=customer,category=category,subject="بازگشایی",body="شروع")
    Ticket.objects.filter(pk=ticket.pk).update(status=Ticket.Status.CLOSED)
    message=reply_ticket(ticket_id=ticket.pk,user=customer,body="لطفاً دوباره بررسی شود")
    job=NotificationJob.objects.get(ticket=ticket,event_type="CUSTOMER_TICKET_REOPENED")
    assert job.payload=={"message_id":message.pk,"message_count":1}
    assert job.next_attempt_at<=timezone.now()

@pytest.mark.django_db
def test_bale_worker_defers_active_customer_chat(category,customer):
    from apps.notifications.management.commands.process_notifications import _prepare_customer_reply_bale
    ticket=create_ticket(customer=customer,category=category,subject="گفتگوی فعال",body="شروع")
    reply_ticket(ticket_id=ticket.pk,user=customer,body="هنوز در حال گفتگو")
    job=NotificationJob.objects.get(ticket=ticket,event_type="CUSTOMER_TICKET_REPLY")
    NotificationJob.objects.filter(pk=job.pk).update(status="PROCESSING",next_attempt_at=timezone.now())
    assert _prepare_customer_reply_bale(job) is False
    job.refresh_from_db()
    assert job.status=="PENDING" and job.last_error=="customer_chat_active"

@pytest.mark.django_db
def test_bale_ticket_excerpt_is_compact_and_whitespace_normalized(category,customer,settings):
    from apps.notifications.management.commands.process_notifications import _bale_ticket_text
    settings.PUBLIC_BASE_URL="https://support.example.com"
    long_body=("متن طولانی   با فاصله\n"*30)
    ticket=create_ticket(customer=customer,category=category,subject="خلاصه کوتاه",body=long_body)
    text=_bale_ticket_text(ticket)
    excerpt=text.split("📝 ",1)[1].split("\n",1)[0]
    assert len(excerpt)<=280 and excerpt.endswith("…")
    assert "  " not in excerpt and "https://support.example.com/staff/tickets/" in text

@pytest.mark.django_db
def test_ticket_full_delete_requires_manager_permission_and_denies_sales_expert(category, customer):
    ticket = create_ticket(customer=customer, category=category, subject="حذف پرونده", body="متن اولیه")
    expert = User.objects.create_user(username="delete-expert", mobile="09120000003", account_type="STAFF")
    StaffProfile.objects.create(user=expert, primary_unit=category.unit, display_name="کارشناس", role=StaffProfile.Role.SALES_EXPERT)
    expert.user_permissions.add(Permission.objects.get(codename="delete_ticket", content_type__app_label="tickets"))
    with pytest.raises(PermissionDenied):
        delete_ticket(ticket_id=ticket.pk, user=expert)
    manager = User.objects.create_user(username="delete-manager", mobile="09120000004", account_type="STAFF")
    StaffProfile.objects.create(user=manager, primary_unit=category.unit, display_name="مدیر", role=StaffProfile.Role.SALES_MANAGER)
    for codename in ("delete_ticket", "view_all_tickets"):
        manager.user_permissions.add(Permission.objects.get(codename=codename, content_type__app_label="tickets"))
    manager.unit_memberships.get(unit=category.unit).permissions.add(
        Permission.objects.get(codename="delete_ticket", content_type__app_label="tickets")
    )
    delete_ticket(ticket_id=ticket.pk, user=manager)
    assert not Ticket.objects.filter(pk=ticket.pk).exists()
    assert not NotificationJob.objects.filter(ticket_id=ticket.pk).exists()

@pytest.mark.django_db
@override_settings(DEBUG=True, DEV_OTP_MOBILE="+989202202129", DEV_OTP_CODE="12345")
def test_fixed_otp_is_available_for_development_only():
    from apps.accounts.models import OTPChallenge
    challenge, code = OTPChallenge.issue("09202202129")
    assert code == "12345"
    assert challenge.verify("12345")

@pytest.mark.django_db
@override_settings(DEBUG=False, DEV_OTP_MOBILE="+989202202129", DEV_OTP_CODE="12345")
def test_fixed_otp_is_disabled_outside_development():
    from apps.accounts.models import OTPChallenge
    _, code = OTPChallenge.issue("09202202129")
    assert code != "12345"

@pytest.mark.django_db
def test_customer_cannot_view_others_ticket(category, customer):
    other = User.objects.create_user(username="other", mobile="09121111111", account_type="CUSTOMER")
    ticket = create_ticket(customer=customer, category=category, subject="موضوع معتبر", body="متن")
    assert TicketPolicy.can_view(customer, ticket)
    assert not TicketPolicy.can_view(other, ticket)

@pytest.mark.django_db
def test_selected_expert_is_private(category, customer):
    a, b = make_staff("a"), make_staff("b")
    ticket = create_ticket(customer=customer, category=category, subject="موضوع خصوصی", selected_expert=a, body="متن")
    assert TicketPolicy.can_view(a, ticket)
    assert not TicketPolicy.can_view(b, ticket)

@pytest.mark.django_db
def test_claim_locks_ticket(category, customer):
    a, b = make_staff("a"), make_staff("b")
    ticket = create_ticket(customer=customer, category=category, subject="موضوع عمومی", body="متن")
    claim_ticket(ticket.pk, a)
    ticket.refresh_from_db()
    assert ticket.owner == a and ticket.status == Ticket.Status.OPEN
    with pytest.raises(PermissionDenied):
        claim_ticket(ticket.pk, b)
    assert not TicketPolicy.can_view(b, ticket)

@pytest.mark.django_db
def test_closed_customer_reply_reopens(category, customer):
    ticket = create_ticket(customer=customer, category=category, subject="موضوع بسته", body="متن")
    ticket.status = Ticket.Status.CLOSED
    ticket.save()
    reply_ticket(ticket_id=ticket.pk, user=customer, body="دوباره نیاز دارم")
    ticket.refresh_from_db()
    assert ticket.status == Ticket.Status.REOPENED

@pytest.mark.django_db
def test_staff_reply_waits_customer(category, customer):
    a = make_staff("a")
    ticket = create_ticket(customer=customer, category=category, subject="پاسخ کارشناس", body="متن")
    claim_ticket(ticket.pk, a)
    reply_ticket(ticket_id=ticket.pk, user=a, body="پاسخ")
    ticket.refresh_from_db()
    assert ticket.status == Ticket.Status.WAITING_CUSTOMER
    assert ticket.first_response_at

@pytest.mark.django_db
def test_attachment_uuid_path_fits_storage_field(tmp_path, category, customer):
    AttachmentPolicy.objects.create(attachment_type="IMAGE", max_file_size=10 * 1024 * 1024, allowed_extensions=[".png"])
    png=base64.b64decode("iVBORw0KGgoAAAANSUhEUgAAAAEAAAABCAQAAAC1HAwCAAAAC0lEQVR42mNk+A8AAQUBAScY42YAAAAASUVORK5CYII=")
    upload = SimpleUploadedFile("customer-image.png", png, content_type="image/png")
    with override_settings(MEDIA_ROOT=tmp_path):
        ticket = create_ticket(customer=customer, category=category, subject="درخواست تصویری", body="تصویر پیوست شد", files=[upload])
        attachment = ticket.ticket_messages.get().attachments.get()
        assert attachment.file.name.startswith("tickets/")
        assert len(attachment.file.name) <= 500
        assert attachment.file.storage.exists(attachment.file.name)

@pytest.mark.django_db
def test_spoofed_image_is_rejected(category, customer):
    from django.core.exceptions import ValidationError
    AttachmentPolicy.objects.create(attachment_type="IMAGE",max_file_size=1024,allowed_extensions=[".png"])
    upload=SimpleUploadedFile("fake.png",b"not an image",content_type="image/png")
    with pytest.raises(ValidationError): create_ticket(customer=customer,category=category,subject="تصویر نامعتبر",body="فایل",files=[upload])

@pytest.mark.django_db
def test_selected_expert_can_transfer_and_old_access_is_removed(category,customer):
    from apps.tickets.services import transfer_ticket
    a,b=make_staff("a"),make_staff("b")
    for user in (a,b): user.user_permissions.add(Permission.objects.get(codename="transfer_ticket",content_type__app_label="tickets"))
    ticket=create_ticket(customer=customer,category=category,subject="انتقال انتخابی",selected_expert=a,body="متن")
    transfer_ticket(ticket.pk,a,b); ticket.refresh_from_db()
    assert ticket.owner==b and ticket.selected_expert is None
    assert not TicketPolicy.can_view(a,ticket) and TicketPolicy.can_view(b,ticket)

@pytest.mark.django_db
def test_internal_note_is_hidden_and_does_not_enqueue_sms(category,customer):
    from apps.notifications.models import NotificationJob
    a=make_staff("a"); a.user_permissions.add(Permission.objects.get(codename="add_internal_note",content_type__app_label="messaging"))
    ticket=create_ticket(customer=customer,category=category,subject="یادداشت داخلی",body="متن"); claim_ticket(ticket.pk,a)
    sms_count=NotificationJob.objects.filter(channel="SMS").count()
    msg=reply_ticket(ticket_id=ticket.pk,user=a,body="محرمانه",internal=True)
    assert msg.is_internal and NotificationJob.objects.filter(channel="SMS").count()==sms_count

@pytest.mark.django_db
def test_customer_can_soft_delete_own_message_but_not_another_authors(category,customer):
    from apps.messaging.services import soft_delete_message
    staff=make_staff("a")
    ticket=create_ticket(customer=customer,category=category,subject="حذف پیام مشتری",body="پیام خود مشتری")
    own=ticket.ticket_messages.get()
    deleted=soft_delete_message(own.pk,customer)
    assert deleted.deleted_at is not None and deleted.deleted_by==customer
    staff_message=reply_ticket(ticket_id=ticket.pk,user=staff,body="پاسخ کارشناس")
    with pytest.raises(PermissionDenied): soft_delete_message(staff_message.pk,customer)

@pytest.mark.django_db
def test_staff_reply_sms_is_debounced_using_database_policy(category,customer):
    from apps.notifications.models import NotificationJob,NotificationPolicy
    from django.utils import timezone
    policy=NotificationPolicy.objects.get(key="STAFF_REPLY_SMS")
    policy.quiet_period_seconds=420; policy.save()
    a=make_staff("a"); ticket=create_ticket(customer=customer,category=category,subject="کنترل پیامک پاسخ",selected_expert=a,body="متن")
    first=reply_ticket(ticket_id=ticket.pk,user=a,body="پاسخ اول")
    job=NotificationJob.objects.get(ticket=ticket,event_type="STAFF_REPLY")
    first_due=job.next_attempt_at
    second=reply_ticket(ticket_id=ticket.pk,user=a,body="پاسخ دوم")
    job.refresh_from_db()
    assert NotificationJob.objects.filter(ticket=ticket,event_type="STAFF_REPLY").count()==1
    assert job.payload=={"latest_message_id":second.pk} and job.next_attempt_at>=first_due
    assert job.next_attempt_at>=timezone.now()+timedelta(seconds=415)

@pytest.mark.django_db
def test_customer_reply_cancels_pending_staff_sms(category,customer):
    a=make_staff("a"); ticket=create_ticket(customer=customer,category=category,subject="گفت‌وگوی فعال",selected_expert=a,body="متن")
    reply_ticket(ticket_id=ticket.pk,user=a,body="پاسخ کارشناس")
    reply_ticket(ticket_id=ticket.pk,user=customer,body="پاسخ سریع مشتری")
    job=NotificationJob.objects.get(ticket=ticket,event_type="STAFF_REPLY")
    assert job.status==NotificationJob.Status.CANCELLED and job.last_error=="customer_replied"

@pytest.mark.django_db
def test_due_unread_staff_reply_is_sent_once(category,customer,monkeypatch):
    from django.core.management import call_command
    from apps.integrations.sms.base import SMSResult
    from apps.notifications.management.commands import process_notifications
    a=make_staff("a"); ticket=create_ticket(customer=customer,category=category,subject="پاسخ خوانده نشده",selected_expert=a,body="متن")
    message=reply_ticket(ticket_id=ticket.pk,user=a,body="پاسخ کارشناس")
    observer=make_staff("observer")
    other_customer=User.objects.create_user(username="other-sms-customer",mobile="09123334444",account_type="CUSTOMER")
    MessageReadReceipt.objects.create(message=message,user=observer)
    MessageReadReceipt.objects.create(message=message,user=other_customer)
    job=NotificationJob.objects.get(ticket=ticket,event_type="STAFF_REPLY")
    NotificationJob.objects.filter(pk=job.pk).update(next_attempt_at=timezone.now()-timedelta(seconds=1))
    ticket.last_activity_at=timezone.now()-timedelta(hours=1); ticket.save(update_fields=["last_activity_at"])
    sent=[]
    class Provider:
        def send_message(self,mobile,message): sent.append((mobile,message)); return SMSResult(True,"ok")
    monkeypatch.setattr(process_notifications,"get_sms_provider",lambda:Provider())
    call_command("process_notifications",once=True,job_id=job.pk)
    job.refresh_from_db()
    assert job.status==NotificationJob.Status.SENT and len(sent)==1
    call_command("process_notifications",once=True,job_id=job.pk)
    assert len(sent)==1


@pytest.mark.django_db
def test_due_staff_reply_is_cancelled_only_when_ticket_customer_has_seen_it(category,customer,monkeypatch):
    from django.core.management import call_command
    from apps.notifications.management.commands import process_notifications

    staff=make_staff("a")
    ticket=create_ticket(customer=customer,category=category,subject="پاسخ دیده شده",selected_expert=staff,body="متن")
    message=reply_ticket(ticket_id=ticket.pk,user=staff,body="پاسخ کارشناس")
    MessageReadReceipt.objects.create(message=message,user=customer)
    job=NotificationJob.objects.get(ticket=ticket,event_type="STAFF_REPLY")
    NotificationJob.objects.filter(pk=job.pk).update(next_attempt_at=timezone.now()-timedelta(seconds=1))
    ticket.last_activity_at=timezone.now()-timedelta(hours=1)
    ticket.save(update_fields=["last_activity_at"])

    class Provider:
        def send_message(self,mobile,message):
            raise AssertionError("پیام دیده‌شده توسط خود مشتری نباید به درگاه ارسال شود")

    monkeypatch.setattr(process_notifications,"get_sms_provider",lambda:Provider())
    call_command("process_notifications",once=True,job_id=job.pk)
    job.refresh_from_db()
    assert job.status==NotificationJob.Status.CANCELLED
    assert job.last_error=="customer_read"


@pytest.mark.django_db
def test_every_failed_sms_gets_exactly_three_retries_three_minutes_apart(category,customer,monkeypatch,settings):
    from django.core.management import call_command
    from apps.integrations.sms.base import SMSResult
    from apps.notifications.management.commands import process_notifications
    settings.SMS_MAX_RETRIES=3
    settings.SMS_RETRY_DELAY_SECONDS=180
    ticket=create_ticket(customer=customer,category=category,subject="آزمون retry پیامک",body="متن")
    job=NotificationJob.objects.get(ticket=ticket,event_type="CUSTOMER_TICKET_CREATED")
    calls=[]
    class Provider:
        def send_message(self,mobile,message):
            calls.append((mobile,message))
            return SMSResult(False,error="0098SMS [کد -19]: شارژ پنل برای ارسال کافی نیست.",provider_code="-19")
    monkeypatch.setattr(process_notifications,"get_sms_provider",lambda:Provider())

    for attempt in range(1,5):
        before=timezone.now()
        call_command("process_notifications",once=True,job_id=job.pk)
        job.refresh_from_db()
        assert job.attempts==attempt
        assert job.provider_status_code=="-19"
        assert "کد -19" in job.last_error
        if attempt<4:
            assert job.status==NotificationJob.Status.RETRYING
            assert timedelta(seconds=179) <= job.next_attempt_at-before <= timedelta(seconds=181)
            NotificationJob.objects.filter(pk=job.pk).update(next_attempt_at=timezone.now()-timedelta(seconds=1))
        else:
            assert job.status==NotificationJob.Status.FAILED

    call_command("process_notifications",once=True,job_id=job.pk)
    assert len(calls)==4


@pytest.mark.django_db
def test_worker_persists_0098_delivery_result(category,customer,monkeypatch,settings):
    from django.core.management import call_command
    from apps.integrations.sms.base import SMSDeliveryResult
    from apps.notifications.management.commands import process_notifications
    ticket=create_ticket(customer=customer,category=category,subject="استعلام دلیوری",body="متن")
    job=NotificationJob.objects.get(ticket=ticket,event_type="CUSTOMER_TICKET_CREATED")
    NotificationJob.objects.filter(pk=job.pk).update(
        status=NotificationJob.Status.SENT,
        provider_message_id="123456789",
        delivery_status=NotificationJob.DeliveryStatus.PENDING,
        delivery_next_check_at=timezone.now()-timedelta(seconds=1),
    )
    class Provider:
        def get_delivery_status(self,provider_id):
            assert provider_id=="123456789"
            return SMSDeliveryResult("DELIVERED","5","0098SMS [کد 5]: پیامک به گوشی گیرنده رسیده است.")
    monkeypatch.setattr(process_notifications,"get_sms_provider",lambda:Provider())

    call_command("process_notifications",once=True,job_id=job.pk)
    job.refresh_from_db()
    assert job.delivery_status==NotificationJob.DeliveryStatus.DELIVERED
    assert job.delivery_status_code=="5"
    assert job.delivery_next_check_at is None


@pytest.mark.django_db
def test_worker_rechecks_provisional_delivery_until_definitive_result(category,customer,monkeypatch,settings):
    from django.core.management import call_command
    from apps.integrations.sms.base import SMSDeliveryResult
    from apps.notifications.management.commands import process_notifications
    settings.SMS_DELIVERY_CHECK_DELAY_SECONDS=180
    settings.SMS_DELIVERY_MAX_CHECKS=15
    ticket=create_ticket(customer=customer,category=category,subject="پیگیری طولانی دلیوری",body="متن")
    job=NotificationJob.objects.get(ticket=ticket,event_type="CUSTOMER_TICKET_CREATED")
    NotificationJob.objects.filter(pk=job.pk).update(
        status=NotificationJob.Status.SENT, sent_at=timezone.now(), provider_message_id="123456789",
        delivery_status=NotificationJob.DeliveryStatus.PENDING,
        delivery_next_check_at=timezone.now()-timedelta(seconds=1),
    )
    results=[
        SMSDeliveryResult("UNKNOWN","8","وضعیت هنوز مشخص نشده است.",is_final=False),
        SMSDeliveryResult("UNDELIVERED","6","تحویل هنوز تأیید نشده است.",is_final=False),
        SMSDeliveryResult("DELIVERED","5","پیامک به گوشی رسیده است."),
    ]
    class Provider:
        def get_delivery_status(self,provider_id): return results.pop(0)
    provider=Provider()
    monkeypatch.setattr(process_notifications,"get_sms_provider",lambda:provider)

    for expected_status,expected_attempts in ((NotificationJob.DeliveryStatus.UNKNOWN,1),(NotificationJob.DeliveryStatus.UNDELIVERED,2)):
        before=timezone.now()
        call_command("process_notifications",once=True,job_id=job.pk)
        job.refresh_from_db()
        assert job.delivery_status==expected_status
        assert job.delivery_check_attempts==expected_attempts
        assert timedelta(seconds=179) <= job.delivery_next_check_at-before <= timedelta(seconds=181)
        NotificationJob.objects.filter(pk=job.pk).update(delivery_next_check_at=timezone.now()-timedelta(seconds=1))

    call_command("process_notifications",once=True,job_id=job.pk)
    job.refresh_from_db()
    assert job.delivery_status==NotificationJob.DeliveryStatus.DELIVERED
    assert job.delivery_check_attempts==3 and job.delivery_next_check_at is None


@pytest.mark.django_db
def test_definitive_staff_created_delivery_failure_schedules_exactly_one_delayed_resend(category,customer,monkeypatch,settings):
    from django.core.management import call_command
    from apps.integrations.sms.base import SMSDeliveryResult,SMSResult
    from apps.notifications.management.commands import process_notifications

    settings.SMS_DEFINITIVE_DELIVERY_RETRY_AFTER_SECONDS=2700
    staff=make_staff("delivery-retry-staff")
    ticket=create_ticket(customer=customer,created_by=staff,unit=category.unit,category=category,subject="ارسال قطعی ناموفق",body="متن")
    source=NotificationJob.objects.get(ticket=ticket,event_type="STAFF_TICKET_CREATED")
    sent_at=timezone.now()
    NotificationJob.objects.filter(pk=source.pk).update(
        status=NotificationJob.Status.SENT,attempts=1,sent_at=sent_at,
        provider_message_id="123456789",delivery_status=NotificationJob.DeliveryStatus.PENDING,
        delivery_next_check_at=timezone.now()-timedelta(seconds=1),
    )

    class Provider:
        def get_delivery_status(self,provider_id):
            return SMSDeliveryResult("EXPIRED","7","مهلت دریافت پیامک منقضی شده است.",is_final=True)
        def send_message(self,mobile,message):
            return SMSResult(True,provider_id="223456789",provider_code="2")
    monkeypatch.setattr(process_notifications,"get_sms_provider",lambda:Provider())

    call_command("process_notifications",once=True,job_id=source.pk)
    source.refresh_from_db()
    retry=NotificationJob.objects.get(delivery_retry_source=source)
    assert source.delivery_is_final is True
    assert source.delivery_status==NotificationJob.DeliveryStatus.EXPIRED
    assert retry.event_type=="STAFF_TICKET_CREATED" and retry.status==NotificationJob.Status.PENDING
    assert retry.attempts==0 and retry.ticket_id==ticket.pk and retry.customer_id==customer.pk
    assert sent_at+timedelta(seconds=2699) <= retry.next_attempt_at <= sent_at+timedelta(seconds=2701)

    process_notifications._schedule_definitive_delivery_retry(source)
    assert NotificationJob.objects.filter(delivery_retry_source=source).count()==1

    NotificationJob.objects.filter(pk=retry.pk).update(next_attempt_at=timezone.now()-timedelta(seconds=1))
    call_command("process_notifications",once=True,job_id=retry.pk)
    retry.refresh_from_db()
    assert retry.status==NotificationJob.Status.SENT and retry.attempts==1
    assert retry.provider_message_id=="223456789"

    retry.sent_at=timezone.now()-timedelta(hours=1)
    retry.delivery_status=NotificationJob.DeliveryStatus.EXPIRED
    retry.delivery_is_final=True
    retry.save()
    assert process_notifications._schedule_definitive_delivery_retry(retry) is None
    assert NotificationJob.objects.filter(ticket=ticket,event_type="STAFF_TICKET_CREATED").count()==2


@pytest.mark.django_db
def test_provisional_or_unrelated_delivery_states_never_schedule_automatic_resend(category,customer,monkeypatch,settings):
    from django.core.management import call_command
    from apps.integrations.sms.base import SMSDeliveryResult
    from apps.notifications.management.commands import process_notifications

    settings.SMS_DELIVERY_MAX_CHECKS=2
    staff=make_staff("provisional-retry-staff")
    staff_ticket=create_ticket(customer=customer,created_by=staff,unit=category.unit,category=category,subject="دلیوری موقت",body="متن")
    staff_job=NotificationJob.objects.get(ticket=staff_ticket,event_type="STAFF_TICKET_CREATED")
    NotificationJob.objects.filter(pk=staff_job.pk).update(
        status=NotificationJob.Status.SENT,sent_at=timezone.now()-timedelta(hours=1),
        provider_message_id="123456789",delivery_status=NotificationJob.DeliveryStatus.PENDING,
        delivery_next_check_at=timezone.now()-timedelta(seconds=1),
    )
    class Provider:
        def get_delivery_status(self,provider_id):
            return SMSDeliveryResult("UNDELIVERED","6","تحویل هنوز تأیید نشده است.",is_final=False)
    monkeypatch.setattr(process_notifications,"get_sms_provider",lambda:Provider())
    call_command("process_notifications",once=True,job_id=staff_job.pk)
    staff_job.refresh_from_db()
    assert staff_job.delivery_is_final is False
    assert not NotificationJob.objects.filter(delivery_retry_source=staff_job).exists()

    customer_ticket=create_ticket(customer=customer,category=category,subject="رویداد خارج از دامنه",body="متن")
    unrelated=NotificationJob.objects.get(ticket=customer_ticket,event_type="CUSTOMER_TICKET_CREATED")
    unrelated.status=NotificationJob.Status.SENT
    unrelated.sent_at=timezone.now()-timedelta(hours=1)
    unrelated.delivery_status=NotificationJob.DeliveryStatus.EXPIRED
    unrelated.delivery_is_final=True
    unrelated.save()
    assert process_notifications._schedule_definitive_delivery_retry(unrelated) is None


@pytest.mark.django_db
def test_0098_code_6_schedules_one_resend_only_after_monitoring_is_exhausted(category,customer,monkeypatch,settings):
    from django.core.management import call_command
    from apps.integrations.sms.base import SMSDeliveryResult
    from apps.notifications.management.commands import process_notifications

    settings.SMS_DELIVERY_MAX_CHECKS=1
    settings.SMS_DEFINITIVE_DELIVERY_RETRY_AFTER_SECONDS=2700
    staff=make_staff("terminal-code-6-staff")
    ticket=create_ticket(customer=customer,created_by=staff,unit=category.unit,category=category,subject="پایان پایش کد ۶",body="متن")
    source=NotificationJob.objects.get(ticket=ticket,event_type="STAFF_TICKET_CREATED")
    NotificationJob.objects.filter(pk=source.pk).update(
        status=NotificationJob.Status.SENT,sent_at=timezone.now()-timedelta(minutes=45),
        provider_message_id="123456789",delivery_status=NotificationJob.DeliveryStatus.PENDING,
        delivery_next_check_at=timezone.now()-timedelta(seconds=1),
    )
    class Provider:
        def get_delivery_status(self,provider_id):
            return SMSDeliveryResult("UNDELIVERED","6","تحویل هنوز تأیید نشده است.",is_final=False)
    monkeypatch.setattr(process_notifications,"get_sms_provider",lambda:Provider())

    call_command("process_notifications",once=True,job_id=source.pk)
    source.refresh_from_db()
    retry=NotificationJob.objects.get(delivery_retry_source=source)
    assert source.delivery_status==NotificationJob.DeliveryStatus.UNDELIVERED
    assert source.delivery_is_final is True and source.delivery_next_check_at is None
    assert retry.next_attempt_at<=timezone.now()+timedelta(seconds=2)
    process_notifications._schedule_definitive_delivery_retry(source)
    assert NotificationJob.objects.filter(delivery_retry_source=source).count()==1


@pytest.mark.django_db
def test_0098_minus_31_schedules_one_resend_at_45_minute_floor(category,customer,monkeypatch,settings):
    from django.core.management import call_command
    from apps.integrations.sms.base import SMSDeliveryResult
    from apps.notifications.management.commands import process_notifications

    settings.SMS_DEFINITIVE_DELIVERY_RETRY_AFTER_SECONDS=2700
    staff=make_staff("safe-minus-31-staff")
    ticket=create_ticket(customer=customer,created_by=staff,unit=category.unit,category=category,subject="شناسه یافت نشد",body="متن")
    source=NotificationJob.objects.get(ticket=ticket,event_type="STAFF_TICKET_CREATED")
    sent_at=timezone.now()
    NotificationJob.objects.filter(pk=source.pk).update(
        status=NotificationJob.Status.SENT,sent_at=sent_at,
        provider_message_id="123456789",delivery_status=NotificationJob.DeliveryStatus.PENDING,
        delivery_next_check_at=timezone.now()-timedelta(seconds=1),
    )
    class Provider:
        def get_delivery_status(self,provider_id):
            return SMSDeliveryResult("ERROR","-31","شناسه پیامک یافت نشد.",is_final=True)
    monkeypatch.setattr(process_notifications,"get_sms_provider",lambda:Provider())

    call_command("process_notifications",once=True,job_id=source.pk)
    source.refresh_from_db()
    retry=NotificationJob.objects.get(delivery_retry_source=source)
    assert source.delivery_is_final is True and source.delivery_status_code=="-31"
    assert sent_at+timedelta(seconds=2699)<=retry.next_attempt_at<=sent_at+timedelta(seconds=2701)
    process_notifications._schedule_definitive_delivery_retry(source)
    assert NotificationJob.objects.filter(delivery_retry_source=source).count()==1


@pytest.mark.django_db
def test_blocked_and_delivery_error_are_not_definitive_resend_candidates(category,customer):
    from apps.notifications.management.commands import process_notifications

    staff=make_staff("nonretryable-delivery-staff")
    ticket=create_ticket(customer=customer,created_by=staff,unit=category.unit,category=category,subject="عدم ارسال مجدد خطا",body="متن")
    source=NotificationJob.objects.get(ticket=ticket,event_type="STAFF_TICKET_CREATED")
    source.status=NotificationJob.Status.SENT
    source.sent_at=timezone.now()-timedelta(hours=1)
    source.delivery_is_final=True
    for status,code in (
        (NotificationJob.DeliveryStatus.BLOCKED,"4"),
        (NotificationJob.DeliveryStatus.ERROR,"9"),
    ):
        source.delivery_status=status
        source.delivery_status_code=code
        source.save(update_fields=["status","sent_at","delivery_is_final","delivery_status","delivery_status_code"])
        assert process_notifications._schedule_definitive_delivery_retry(source) is None
    assert not NotificationJob.objects.filter(delivery_retry_source=source).exists()


@pytest.mark.django_db
def test_historical_expired_staff_ticket_delivery_is_queued_once_by_data_migration(category,customer,settings):
    import importlib
    from django.apps import apps as django_apps

    settings.SMS_DEFINITIVE_DELIVERY_RETRY_AFTER_SECONDS=2700
    settings.SMS_HISTORICAL_DELIVERY_RETRY_STAGGER_SECONDS=30
    staff=make_staff("historical-delivery-staff")
    ticket=create_ticket(customer=customer,created_by=staff,unit=category.unit,category=category,subject="خطای قدیمی دلیوری",body="متن")
    source=NotificationJob.objects.get(ticket=ticket,event_type="STAFF_TICKET_CREATED")
    NotificationJob.objects.filter(pk=source.pk).update(
        status=NotificationJob.Status.SENT,sent_at=timezone.now()-timedelta(days=1),
        provider_message_id="987654321",delivery_status=NotificationJob.DeliveryStatus.EXPIRED,
        delivery_next_check_at=None,delivery_is_final=False,
    )
    migration=importlib.import_module("apps.notifications.migrations.0013_notificationjob_definitive_delivery_retry")

    before=timezone.now()
    migration.queue_historical_definitive_delivery_retries(django_apps,None)
    retry=NotificationJob.objects.get(delivery_retry_source=source)
    assert retry.status==NotificationJob.Status.PENDING
    assert retry.next_attempt_at>=before+timedelta(seconds=29)
    migration.queue_historical_definitive_delivery_retries(django_apps,None)
    assert NotificationJob.objects.filter(delivery_retry_source=source).count()==1


@pytest.mark.django_db
def test_historical_safe_0098_results_are_queued_once_by_followup_migration(category,customer,settings):
    import importlib
    from django.apps import apps as django_apps

    settings.SMS_DEFINITIVE_DELIVERY_RETRY_AFTER_SECONDS=2700
    settings.SMS_HISTORICAL_DELIVERY_RETRY_STAGGER_SECONDS=30
    settings.SMS_DELIVERY_MAX_CHECKS=15
    staff=make_staff("historical-safe-0098-staff")
    sources=[]
    for index,(status,code) in enumerate((("UNDELIVERED","6"),("EXPIRED","7"),("ERROR","-31"))):
        ticket=create_ticket(
            customer=customer,created_by=staff,unit=category.unit,category=category,
            subject=f"خطای تاریخی امن {index}",body="متن",
        )
        source=NotificationJob.objects.get(ticket=ticket,event_type="STAFF_TICKET_CREATED")
        NotificationJob.objects.filter(pk=source.pk).update(
            status=NotificationJob.Status.SENT,sent_at=timezone.now()-timedelta(days=1),
            provider_message_id=f"98765432{index}",delivery_status=status,
            delivery_status_code=code,delivery_next_check_at=None,delivery_is_final=False,
        )
        sources.append(source)
    unsafe_ticket=create_ticket(customer=customer,created_by=staff,unit=category.unit,category=category,subject="خطای تاریخی ناامن",body="متن")
    unsafe=NotificationJob.objects.get(ticket=unsafe_ticket,event_type="STAFF_TICKET_CREATED")
    NotificationJob.objects.filter(pk=unsafe.pk).update(
        status=NotificationJob.Status.SENT,sent_at=timezone.now()-timedelta(days=1),
        provider_message_id="987654329",delivery_status=NotificationJob.DeliveryStatus.ERROR,
        delivery_status_code="9",delivery_next_check_at=None,delivery_is_final=True,
    )
    migration=importlib.import_module("apps.notifications.migrations.0014_backfill_safe_0098_delivery_retries")

    before=timezone.now()
    migration.queue_historical_safe_0098_delivery_retries(django_apps,None)
    for position,source in enumerate(sources,start=1):
        source.refresh_from_db()
        retry=NotificationJob.objects.get(delivery_retry_source=source)
        assert source.delivery_is_final is True
        assert retry.next_attempt_at>=before+timedelta(seconds=position*30-1)
    assert not NotificationJob.objects.filter(delivery_retry_source=unsafe).exists()
    migration.queue_historical_safe_0098_delivery_retries(django_apps,None)
    assert NotificationJob.objects.filter(delivery_retry_source_id__in=[source.pk for source in sources]).count()==3

@pytest.mark.django_db
def test_customer_ticket_created_sms_has_seven_day_magic_link(category,customer,settings,monkeypatch):
    from datetime import timedelta
    from django.utils import timezone
    from apps.accounts.models import TicketMagicLink
    from apps.integrations.sms.base import SMSResult
    from apps.notifications.management.commands import process_notifications
    from apps.notifications.models import NotificationJob
    settings.PUBLIC_BASE_URL="https://support.example.com"
    settings.MAGIC_LINK_TTL_DAYS=7
    ticket=create_ticket(customer=customer,category=category,subject="پیام تشکر",body="متن")
    job=NotificationJob.objects.get(ticket=ticket,event_type="CUSTOMER_TICKET_CREATED")
    captured={}
    class Provider:
        def send_message(self,mobile,message): captured.update(mobile=mobile,message=message); return SMSResult(True,"ok")
    monkeypatch.setattr(process_notifications,"get_sms_provider",lambda:Provider())
    before=timezone.now()
    result,link=process_notifications._send_sms(job)
    assert result.success and link.created_reason=="customer_created_ticket"
    assert before+timedelta(days=7)-timedelta(seconds=1) <= link.expires_at <= before+timedelta(days=7)+timedelta(seconds=1)
    assert "از ثبت درخواست شما متشکریم" in captured["message"]
    assert "پس از بررسی و پاسخ کارشناس، از طریق پیامک" in captured["message"]
    assert "https://support.example.com/t/" in captured["message"]

@pytest.mark.django_db
def test_sms_templates_are_database_managed_and_distinguish_ticket_creator():
    from apps.notifications.models import SMSTemplate
    customer_text=SMSTemplate.objects.get(event_type="CUSTOMER_TICKET_CREATED").body
    staff_text=SMSTemplate.objects.get(event_type="STAFF_TICKET_CREATED").body
    assert customer_text!=staff_text
    assert "از ثبت درخواست شما متشکریم" in customer_text
    assert "توسط واحد {unit_name} برای شما ثبت شد" in staff_text
    assert "{magic_link}" in customer_text and "{first_name}" in customer_text

@pytest.mark.django_db(transaction=True)
def test_magic_link_is_ten_character_base36_and_retries_collision_without_select(category,customer,monkeypatch):
    import re
    from django.test.utils import CaptureQueriesContext
    from django.db import connection
    from django.utils import timezone
    from apps.accounts.models import TicketMagicLink
    ticket=create_ticket(customer=customer,category=category,subject="توکن کوتاه",body="متن")
    TicketMagicLink.objects.create(ticket=ticket,customer=customer,token_hash=TicketMagicLink.digest("aaaaaaaaaa"),expires_at=timezone.now()+__import__('datetime').timedelta(days=7))
    values=iter(["aaaaaaaaaa","b2c3d4e5f6"])
    monkeypatch.setattr(TicketMagicLink,"generate_secure_token",staticmethod(lambda length=10:next(values)))
    with CaptureQueriesContext(connection) as queries: token=TicketMagicLink.issue(ticket)
    assert token=="b2c3d4e5f6" and re.fullmatch(r"[a-z0-9]{10}",token)
    assert not any(query["sql"].lstrip().upper().startswith("SELECT") for query in queries)

@pytest.mark.django_db
def test_audio_only_webm_is_saved_as_voice(category,customer,monkeypatch,tmp_path,settings):
    from apps.attachments.services import save_attachments
    settings.MEDIA_ROOT=tmp_path
    AttachmentPolicy.objects.update_or_create(attachment_type="VOICE",defaults={"max_file_size":1024*1024,"allowed_extensions":[".webm"],"allowed_mime_types":[]})
    ticket=create_ticket(customer=customer,category=category,subject="ضبط صدا",body="متن")
    message=ticket.ticket_messages.first()
    monkeypatch.setattr("apps.attachments.services.magic.from_buffer",lambda *args,**kwargs:"video/webm")
    upload=SimpleUploadedFile("voice.webm",b"webm-audio-content",content_type="audio/webm")
    attachment=save_attachments(message,[upload],customer)[0]
    assert attachment.attachment_type=="VOICE" and attachment.extension==".webm"

@pytest.mark.django_db
def test_webotp_sms_format_is_origin_bound(settings):
    from apps.integrations.sms.providers import webotp_message
    settings.WEBOTP_HOST="example.com"
    assert webotp_message("123456")=="کد شما: 123456\n@example.com #123456"

@override_settings(SMS_PROVIDER_CLASS="apps.integrations.sms.sms0098.ZeroZeroNinetyEightSMSProvider")
def test_sms_provider_contract_and_default(settings):
    from apps.integrations.sms.base import SMSProvider
    from apps.integrations.sms.sms0098 import ZeroZeroNinetyEightSMSProvider
    from apps.integrations.sms.factory import get_sms_provider,reset_sms_provider
    assert issubclass(ZeroZeroNinetyEightSMSProvider,SMSProvider)
    reset_sms_provider()
    assert isinstance(get_sms_provider(),ZeroZeroNinetyEightSMSProvider)
    reset_sms_provider()

def test_behin_provider_maps_send_array_contract(settings):
    from apps.integrations.sms.behin import BehinSMSProvider
    settings.BEHIN_SMS_USERNAME="user"; settings.BEHIN_SMS_PASSWORD="pass"; settings.BEHIN_SMS_SPECIAL_NUMBER="3000"
    calls={}
    class Service:
        def SendArray(self,**kwargs): calls.update(kwargs); return [987654]
    class Client: service=Service()
    provider=BehinSMSProvider(); provider._client=Client()
    result=provider.send_message("+989202202129","پیام آزمایشی")
    assert result.success and "987654" in result.provider_id
    assert calls["RecipientNumber"]=={"string":["+989202202129"]}
    assert calls["CheckingMessageID"]=={"long":[0]}
    assert calls["SpecialNumber"]=="3000" and calls["IsFlashMessage"] is False

def test_0098_provider_maps_soap_contract(settings):
    from apps.integrations.sms.sms0098 import ZeroZeroNinetyEightSMSProvider
    settings.SMS0098_USERNAME="user"; settings.SMS0098_PASSWORD="pass"; settings.SMS0098_FROM="3000"
    calls={}
    class Service:
        def SendSMSWithID(self,**parameters): calls.update(parameters); return "123456789"
    class Client: service=Service()
    provider=ZeroZeroNinetyEightSMSProvider(); provider._client=Client()
    result=provider.send_message("09202202129","پیام آزمایشی")
    assert result.success and result.provider_id=="123456789"
    assert calls["mobileno"]=="09202202129"
    assert calls["text"]=="پیام آزمایشی"
    assert calls["pnlno"]=="3000"
    assert set(calls)=={"username","password","text","mobileno","pnlno"}


@pytest.mark.parametrize("code,expected", [
    ("-3", "تطابق ندارند"), ("10", "اشتباه است"), ("11", "کاراکتر غیرمجاز"),
    ("-17", "خالی است"), ("-18", "خطای شارژ"), ("-19", "شارژ پنل"),
    ("-22", "شماره موبایل"), ("66", "تطابق ندارند"), ("1111", "کاراکتر غیرمجاز"),
    ("Hang", "مسدود است"), ("No Doc", "مرحله دوم"), ("-23", "مجوز ارسال"),
    ("-24", "شماره پنل"), ("-25", "اعتبار شماره پنل"), ("-26", "تنظیمات کد OTP"),
    ("-27", "فرمت کد OTP"), ("-28", "تأیید نشده"), ("Wrong timestamp", "timestamp"),
    ("Wrong time", "خارج از بازه"), ("-29", "فرمت کد OTP"), ("-31", "یافت نشد"),
    ("-32", "مطابقت ندارد"), ("13", "IP غیرمجاز"),
])
def test_0098_provider_maps_every_documented_send_error(settings,code,expected):
    from apps.integrations.sms.sms0098 import ZeroZeroNinetyEightSMSProvider
    settings.SMS0098_USERNAME="user"; settings.SMS0098_PASSWORD="pass"; settings.SMS0098_FROM="3000"
    provider=ZeroZeroNinetyEightSMSProvider()
    result=provider._parse_send_result(code)
    assert not result.success
    assert result.provider_code==code
    assert expected in result.error


@pytest.mark.parametrize("code,status,is_final", [("4","BLOCKED",True),("5","DELIVERED",True),("6","UNDELIVERED",False),("7","EXPIRED",True),("8","UNKNOWN",False),("9","ERROR",True),("12","ERROR",True),("13","ERROR",True),("-31","ERROR",True)])
def test_0098_provider_maps_documented_delivery_states(settings,code,status,is_final):
    from apps.integrations.sms.sms0098 import ZeroZeroNinetyEightSMSProvider
    settings.SMS0098_USERNAME="user"; settings.SMS0098_PASSWORD="pass"; settings.SMS0098_FROM="3000"
    class Service:
        def smsdeliveryState(self,**parameters): return code
    class Client: service=Service()
    provider=ZeroZeroNinetyEightSMSProvider(); provider._client=Client()
    result=provider.get_delivery_status("123456789")
    assert result.status==status and result.provider_code==code and result.is_final is is_final
    assert f"کد {code}" in result.description