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.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",
    )
    unit.full_clean()
    unit.save()
    assert unit.supervisor_mobiles == ("+989121112233", "+989352223344")

    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.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

    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_staff_created_ticket_also_notifies_owning_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="متن",
    )

    job = NotificationJob.objects.get(ticket=ticket, event_type="UNIT_SUPERVISOR_NEW_TICKET")
    assert job.payload == {"recipient_mobile": "+989124445566"}

@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="متن")
    reply_ticket(ticket_id=ticket.pk,user=a,body="پاسخ کارشناس")
    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_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_http_contract(settings,monkeypatch):
    from apps.integrations.sms.sms0098 import ZeroZeroNinetyEightSMSProvider
    settings.SMS0098_USERNAME="user"; settings.SMS0098_PASSWORD="pass"; settings.SMS0098_FROM="3000"
    calls={}
    class Response:
        text="42"
        def raise_for_status(self): return None
    def fake_get(url,**kwargs): calls.update(url=url,**kwargs); return Response()
    monkeypatch.setattr("apps.integrations.sms.sms0098.requests.get",fake_get)
    result=ZeroZeroNinetyEightSMSProvider().send_message("09202202129","پیام آزمایشی")
    assert result.success and result.provider_id=="42"
    assert calls["params"]["TO"]==9202202129
    assert calls["params"]["TEXT"]=="پیام آزمایشی"