customerTicket/tests/test_organizational_ticketing.py

import pytest
from django.contrib.auth.models import Permission
from django.core.files.uploadedfile import SimpleUploadedFile
from django.urls import reverse

from apps.accounts.models import MagicLinkAccessLog, TicketMagicLink, User
from apps.attachments.models import AttachmentPolicy
from apps.customers.models import ContactGroup, CustomerProfile
from apps.notifications.models import NotificationJob
from apps.staff.models import OrganizationalUnit, StaffProfile
from apps.tickets.models import GroupDispatch, Ticket, TicketCategory, TicketTag
from apps.tickets.policies import TicketPolicy
from apps.tickets.selectors import staff_ticket_rows
from apps.tickets.services import create_ticket, dispatch_group_tickets, refresh_group_dispatch_status


def make_staff(username, unit):
    user = User.objects.create_user(username=username, account_type=User.AccountType.STAFF)
    StaffProfile.objects.create(user=user, display_name=username, primary_unit=unit)
    return user


def make_contact(username, mobile):
    user = User.objects.create_user(username=username, mobile=mobile, account_type=User.AccountType.CUSTOMER)
    CustomerProfile.objects.create(user=user, first_name=username)
    return user


@pytest.mark.django_db
def test_unit_permission_isolates_ticket_queries_and_object_access():
    sales = OrganizationalUnit.objects.create(name="فروش ایزوله", slug="isolated-sales")
    logistics = OrganizationalUnit.objects.create(name="بارگیری ایزوله", slug="isolated-logistics")
    view_permission = Permission.objects.get(codename="view_ticket", content_type__app_label="tickets")
    sales.permissions.add(view_permission)
    logistics.permissions.add(view_permission)
    sales_staff = make_staff("sales-agent", sales)
    logistics_staff = make_staff("logistics-agent", logistics)
    customer = make_contact("unit-customer", "09120000101")
    category = TicketCategory.objects.create(title="عمومی", slug="general")
    sales_ticket = create_ticket(customer=customer, created_by=sales_staff, unit=sales, category=category, subject="تیکت واحد فروش", body="شرح")

    assert TicketPolicy.can_view(sales_staff, sales_ticket)
    assert not TicketPolicy.can_view(logistics_staff, sales_ticket)
    assert list(staff_ticket_rows(logistics_staff)) == []


@pytest.mark.django_db
def test_group_dispatch_creates_one_isolated_ticket_and_sms_job_per_contact():
    unit = OrganizationalUnit.objects.create(name="لوجستیک ارسال گروهی", slug="dispatch-logistics")
    permission = Permission.objects.get(codename="dispatch_group_tickets", content_type__app_label="tickets")
    view_permission = Permission.objects.get(codename="view_ticket", content_type__app_label="tickets")
    unit.permissions.add(permission, view_permission)
    staff = make_staff("dispatcher", unit)
    first = make_contact("carrier-one", "09120000111")
    second = make_contact("carrier-two", "09120000112")
    group = ContactGroup.objects.create(name="باربری‌های گلپایگان", unit=unit)
    first.customer_profile.groups.add(group)
    second.customer_profile.groups.add(group)
    category = TicketCategory.objects.create(unit=unit,title="استعلام حمل", slug="test-shipping-quote")
    tag = TicketTag.objects.create(name="استعلام قیمت", slug="price-inquiry")

    dispatch, tickets = dispatch_group_tickets(user=staff, unit=unit, contact_group=group, category=category, subject="استعلام قیمت حمل", body="لطفاً قیمت پیشنهادی را ثبت کنید.", tags=[tag])

    # Creation only queues delivery; reporting promotes it after SMS succeeds.
    assert dispatch.status == GroupDispatch.Status.QUEUED
    assert dispatch.recipient_count == 2
    assert len(tickets) == 2
    assert {ticket.customer_id for ticket in tickets} == {first.pk, second.pk}
    assert all(ticket.delivery_type == Ticket.DeliveryType.GROUP for ticket in tickets)
    assert NotificationJob.objects.filter(ticket__in=tickets, channel="SMS", event_type="STAFF_TICKET_CREATED").count() == 2


@pytest.mark.django_db
def test_staff_created_ticket_sms_uses_its_actual_organizational_unit(monkeypatch, settings):
    from apps.integrations.sms.base import SMSResult
    from apps.notifications.management.commands import process_notifications

    settings.PUBLIC_BASE_URL = "https://support.example.com"
    finance = OrganizationalUnit.objects.create(name="حسابداری", slug="sms-finance")
    staff = make_staff("finance-sms-agent", finance)
    customer = make_contact("finance-sms-customer", "09120000501")
    category = TicketCategory.objects.create(unit=finance, title="اسناد مالی", slug="sms-financial-documents")
    ticket = create_ticket(
        customer=customer,
        created_by=staff,
        unit=finance,
        category=category,
        subject="ارسال سند مالی",
        body="لطفاً سند را بررسی کنید.",
    )
    captured = {}

    class Provider:
        def send_message(self, mobile, message):
            captured.update(mobile=mobile, message=message)
            return SMSResult(True, "sent")

    monkeypatch.setattr(process_notifications, "get_sms_provider", lambda: Provider())
    process_notifications._send_sms(NotificationJob.objects.get(ticket=ticket))

    assert "توسط واحد حسابداری برای شما ثبت شد" in captured["message"]
    assert "توسط واحد فروش" not in captured["message"]


@pytest.mark.django_db
def test_group_dispatch_sms_uses_dispatch_organizational_unit(monkeypatch, settings):
    from apps.integrations.sms.base import SMSResult
    from apps.notifications.management.commands import process_notifications

    settings.PUBLIC_BASE_URL = "https://support.example.com"
    warehouse = OrganizationalUnit.objects.get(name="بارگیری انبار")
    warehouse.permissions.add(Permission.objects.get(codename="dispatch_group_tickets", content_type__app_label="tickets"))
    staff = make_staff("warehouse-sms-agent", warehouse)
    customer = make_contact("warehouse-sms-customer", "09120000502")
    group = ContactGroup.objects.create(name="رانندگان", unit=warehouse)
    customer.customer_profile.groups.add(group)
    category = TicketCategory.objects.create(unit=warehouse, title="هماهنگی بار", slug="sms-loading")
    _, tickets = dispatch_group_tickets(
        user=staff,
        unit=warehouse,
        contact_group=group,
        category=category,
        subject="هماهنگی بارگیری",
        body="زمان بارگیری را اعلام کنید.",
    )
    captured = []

    class Provider:
        def send_message(self, mobile, message):
            captured.append(message)
            return SMSResult(True, "sent")

    monkeypatch.setattr(process_notifications, "get_sms_provider", lambda: Provider())
    process_notifications._send_sms(NotificationJob.objects.get(ticket=tickets[0]))

    assert "توسط واحد بارگیری انبار برای شما ثبت شد" in captured[0]
    assert "توسط واحد فروش" not in captured[0]


@pytest.mark.django_db
def test_group_dispatch_copies_shared_document_into_each_private_ticket():
    unit=OrganizationalUnit.objects.create(name="مالی گروهی",slug="bulk-finance")
    unit.permissions.add(Permission.objects.get(codename="dispatch_group_tickets",content_type__app_label="tickets"))
    staff=make_staff("bulk-finance-user",unit)
    first=make_contact("bulk-supplier-one","09120000311"); second=make_contact("bulk-supplier-two","09120000312")
    group=ContactGroup.objects.create(name="تامین‌کنندگان اسناد",unit=unit)
    first.customer_profile.groups.add(group); second.customer_profile.groups.add(group)
    category=TicketCategory.objects.create(title="سند گروهی",slug="bulk-document")
    AttachmentPolicy.objects.create(attachment_type="PDF",max_file_size=1024*1024,allowed_extensions=[".pdf"])
    upload=SimpleUploadedFile("receipt.pdf",b"%PDF-1.4\n%%EOF",content_type="application/pdf")

    _,tickets=dispatch_group_tickets(user=staff,unit=unit,contact_group=group,category=category,subject="ارسال سند مالی مشترک",body="سند را بررسی کنید.",files=[upload])

    assert len(tickets)==2
    assert all(ticket.ticket_messages.get().attachments.filter(original_name="receipt.pdf").count()==1 for ticket in tickets)
    paths=[ticket.ticket_messages.get().attachments.get().file.name for ticket in tickets]
    assert len(set(paths))==2


@pytest.mark.django_db
def test_magic_link_records_every_open_with_ip_and_user_agent(client):
    unit = OrganizationalUnit.objects.create(name="حسابداری ممیزی", slug="audit-accounting")
    staff = make_staff("accountant", unit)
    customer = make_contact("supplier", "09120000121")
    category = TicketCategory.objects.create(title="سند مالی", slug="financial-document")
    ticket = create_ticket(customer=customer, created_by=staff, unit=unit, category=category, subject="ارسال رسید مالی", body="رسید پیوست شد.")
    token, link = TicketMagicLink.issue_with_record(ticket, created_by=staff)

    response = client.get(reverse("accounts:magic", args=[token]), HTTP_USER_AGENT="Audit Browser/1.0", REMOTE_ADDR="203.0.113.7")

    assert response.status_code == 302
    link.refresh_from_db()
    assert link.view_count == 1
    assert link.last_used_ip == "203.0.113.7"
    assert link.last_user_agent == "Audit Browser/1.0"
    assert link.created_by == staff
    assert link.created_unit == unit
    assert MagicLinkAccessLog.objects.filter(magic_link=link, ip_address="203.0.113.7").count() == 1


@pytest.mark.django_db
def test_dispatch_pages_are_available_only_to_the_dispatch_unit(client):
    unit = OrganizationalUnit.objects.create(name="انبار دسترسی", slug="access-warehouse")
    other_unit = OrganizationalUnit.objects.create(name="فروش دسترسی تست", slug="sales-2")
    for codename in ("view_ticket", "dispatch_group_tickets"):
        permission = Permission.objects.get(codename=codename, content_type__app_label="tickets")
        unit.permissions.add(permission)
    staff = make_staff("warehouse-user", unit)
    outsider = make_staff("sales-user", other_unit)
    customer = make_contact("warehouse-contact", "09120000131")
    group = ContactGroup.objects.create(name="تامین‌کنندگان", unit=unit)
    customer.customer_profile.groups.add(group)
    category = TicketCategory.objects.create(unit=unit, title="موجودی", slug="test-inventory-access")

    client.force_login(staff)
    assert client.get(reverse("staff_tickets:dispatch_create")).status_code == 200
    response = client.post(reverse("staff_tickets:dispatch_create"), {"unit": unit.pk, "contact_group": group.pk, "category": category.pk, "subject": "استعلام موجودی کالا", "body": "موجودی را اعلام کنید."})
    assert response.status_code == 302
    dispatch = GroupDispatch.objects.get()
    assert response.url == reverse("staff_tickets:dispatch_detail", args=[dispatch.public_id])

    client.force_login(outsider)
    assert client.get(reverse("staff_tickets:dispatch_detail", args=[dispatch.public_id])).status_code == 403


@pytest.mark.django_db
def test_staff_operations_pages_respect_unit_scoped_permissions(client):
    unit = OrganizationalUnit.objects.create(name="عملیات", slug="operations")
    other = OrganizationalUnit.objects.create(name="واحد محرمانه", slug="private-unit")
    codenames = [
        ("customers", "view_customerprofile"), ("customers", "add_customerprofile"),
        ("customers", "view_contactgroup"), ("customers", "add_contactgroup"),
        ("accounts", "view_ticketmagiclink"), ("notifications", "view_smstemplate"),
        ("notifications", "add_smstemplate"),
    ]
    unit.permissions.add(*[Permission.objects.get(content_type__app_label=app, codename=code) for app, code in codenames])
    staff = make_staff("operations-agent", unit)
    visible = make_contact("visible-contact", "09120000201")
    hidden = make_contact("hidden-contact", "09120000202")
    visible_group = ContactGroup.objects.create(name="گروه مجاز", unit=unit)
    hidden_group = ContactGroup.objects.create(name="گروه محرمانه", unit=other)
    visible.customer_profile.groups.add(visible_group)
    hidden.customer_profile.groups.add(hidden_group)

    client.force_login(staff)
    response = client.get(reverse("staff_tickets:contacts"))
    assert response.status_code == 200
    assert "visible-contact" in response.content.decode()
    assert "hidden-contact" not in response.content.decode()
    assert client.get(reverse("staff_tickets:groups")).status_code == 200
    assert client.get(reverse("staff_tickets:magic_audit")).status_code == 200
    assert client.get(reverse("staff_tickets:sms_templates")).status_code == 200


@pytest.mark.django_db
def test_superuser_without_staff_profile_can_open_all_staff_contacts(client):
    unit = OrganizationalUnit.objects.create(name="واحد مخاطبان مدیر", slug="admin-contacts-unit")
    contact = make_contact("admin-visible-contact", "09120000503")
    group = ContactGroup.objects.create(name="گروه قابل مشاهده مدیر", unit=unit)
    contact.customer_profile.groups.add(group)
    User.objects.create_user(
        username="customer-without-profile",
        mobile="09120000504",
        account_type=User.AccountType.CUSTOMER,
    )
    admin = User.objects.create_superuser(
        username="contacts-admin",
        password="safe-password",
        account_type=User.AccountType.STAFF,
    )

    client.force_login(admin)
    response = client.get(reverse("staff_tickets:contacts"))

    assert response.status_code == 200
    assert "admin-visible-contact" in response.content.decode()
    assert "customer-without-profile" not in response.content.decode()


@pytest.mark.django_db
def test_staff_can_visually_add_and_remove_contact_group_memberships(client):
    unit = OrganizationalUnit.objects.create(name="واحد ویرایش مخاطب", slug="contact-membership-editor")
    unit.permissions.add(*Permission.objects.filter(
        content_type__app_label="customers",
        codename__in=("view_customerprofile", "change_customerprofile"),
    ))
    staff = make_staff("contact-membership-agent", unit)
    contact = make_contact("editable-contact", "09120000505")
    first = ContactGroup.objects.create(name="گروه اولیه", unit=unit)
    second = ContactGroup.objects.create(name="گروه جایگزین", unit=unit)
    contact.customer_profile.groups.add(first)
    client.force_login(staff)
    url = reverse("staff_tickets:contact_edit", args=[contact.public_id])

    page = client.get(url)
    content = page.content.decode()
    assert page.status_code == 200
    assert "data-contact-group-selector" in content
    assert "data-group-search" in content
    assert str(first.pk) in {str(value) for value in page.context["form"]["groups"].value()}
    assert f'value="{second.pk}"' in content

    response = client.post(url, {
        "mobile": contact.local_mobile,
        "first_name": "مخاطب",
        "last_name": "ویرایش‌شده",
        "organization_name": "",
        "stakeholder_type": CustomerProfile.StakeholderType.CUSTOMER,
        "groups": [second.pk],
        "province": "",
        "city": "",
    })

    assert response.status_code == 302
    assert set(contact.customer_profile.groups.values_list("pk", flat=True)) == {second.pk}


@pytest.mark.django_db
def test_staff_ticket_create_supports_multiple_previewable_attachments(client, monkeypatch, settings, tmp_path):
    settings.MEDIA_ROOT = tmp_path
    unit = OrganizationalUnit.objects.create(name="واحد پیوست کارشناس", slug="staff-attachment-unit")
    unit.permissions.add(Permission.objects.get(content_type__app_label="tickets", codename="add_ticket"))
    staff = make_staff("staff-attachment-agent", unit)
    category = TicketCategory.objects.create(unit=unit, title="مدارک", slug="staff-attachment-category")
    AttachmentPolicy.objects.create(attachment_type="IMAGE", max_file_size=1024 * 1024, allowed_extensions=[".jpg"])
    AttachmentPolicy.objects.create(attachment_type="PDF", max_file_size=1024 * 1024, allowed_extensions=[".pdf"])
    monkeypatch.setattr(
        "apps.attachments.services.magic.from_buffer",
        lambda content, mime=True: "image/jpeg" if content.startswith(b"\xff\xd8") else "application/pdf",
    )
    client.force_login(staff)
    url = reverse("staff_tickets:create")

    page = client.get(url)
    content = page.content.decode()
    assert page.status_code == 200
    assert "data-ticket-composer" in content
    assert "data-gallery-input" in content
    assert "data-document-input" in content
    assert "data-file-preview" in content
    assert "attachment-uploader.js" in content
    assert "ایجاد پرونده از طرف واحد پیوست کارشناس" in content

    response = client.post(url, {
        "unit": unit.pk,
        "customer_mobile": "09120000506",
        "first_name": "مشتری",
        "last_name": "پیوست",
        "stakeholder_type": CustomerProfile.StakeholderType.CUSTOMER,
        "category": category.pk,
        "subject": "ارسال همزمان تصویر و سند",
        "body": "دو فایل برای این پرونده ثبت می‌شود.",
        "files": [
            SimpleUploadedFile("photo.jpg", b"\xff\xd8\xff\xe0image", content_type="image/jpeg"),
            SimpleUploadedFile("document.pdf", b"%PDF-1.4\n%%EOF", content_type="application/pdf"),
        ],
    })

    assert response.status_code == 302
    ticket = Ticket.objects.get(subject="ارسال همزمان تصویر و سند")
    assert set(ticket.ticket_messages.get().attachments.values_list("original_name", flat=True)) == {
        "photo.jpg", "document.pdf",
    }


@pytest.mark.django_db
def test_unit_scoped_add_ticket_preserves_contact_ownership_and_stakeholder(client):
    unit=OrganizationalUnit.objects.create(name="حسابداری عملیاتی",slug="accounting-ops")
    unit.permissions.add(Permission.objects.get(content_type__app_label="tickets",codename="add_ticket"))
    staff=make_staff("unit-accountant",unit)
    group=ContactGroup.objects.create(name="تامین‌کنندگان مالی",unit=unit)
    category=TicketCategory.objects.create(title="رسید مالی",slug="receipt-ops")
    client.force_login(staff)

    response=client.post(reverse("staff_tickets:create"),{
        "unit":unit.pk,"customer_mobile":"09120000301","first_name":"تامین‌کننده","last_name":"نمونه",
        "organization_name":"شرکت نمونه","stakeholder_type":CustomerProfile.StakeholderType.SUPPLIER,
        "groups":[group.pk],"category":category.pk,"subject":"ارسال رسید پرداخت","body":"رسید مالی پیوست خواهد شد.",
    })

    assert response.status_code==302
    ticket=Ticket.objects.get(customer__mobile="+989120000301")
    assert ticket.owning_unit==unit and ticket.created_by==staff
    assert ticket.customer.customer_profile.stakeholder_type==CustomerProfile.StakeholderType.SUPPLIER
    assert ticket.customer.customer_profile.groups.filter(pk=group.pk).exists()


@pytest.mark.django_db
def test_dispatch_retry_is_unit_scoped_and_refreshes_persistent_status(client):
    unit=OrganizationalUnit.objects.create(name="لوجستیک retry",slug="logistics-retry")
    other=OrganizationalUnit.objects.create(name="فروش retry",slug="sales-retry")
    permission=Permission.objects.get(content_type__app_label="tickets",codename="dispatch_group_tickets")
    unit.permissions.add(permission)
    staff=make_staff("retry-dispatcher",unit); outsider=make_staff("retry-outsider",other)
    contact=make_contact("retry-carrier","09120000302")
    group=ContactGroup.objects.create(name="باربری retry",unit=unit); contact.customer_profile.groups.add(group)
    category=TicketCategory.objects.create(title="استعلام retry",slug="quote-retry")
    dispatch,tickets=dispatch_group_tickets(user=staff,unit=unit,contact_group=group,category=category,subject="استعلام هزینه جدید",body="قیمت را ثبت کنید.")
    job=NotificationJob.objects.get(ticket=tickets[0],channel="SMS")
    job.status=NotificationJob.Status.FAILED; job.attempts=3; job.last_error="provider"; job.save()
    refresh_group_dispatch_status(dispatch.pk); dispatch.refresh_from_db()
    assert dispatch.status==GroupDispatch.Status.PARTIAL

    client.force_login(outsider)
    assert client.post(reverse("staff_tickets:dispatch_retry",args=[dispatch.public_id])).status_code==403
    client.force_login(staff)
    assert client.post(reverse("staff_tickets:dispatch_retry",args=[dispatch.public_id])).status_code==302
    job.refresh_from_db(); dispatch.refresh_from_db()
    assert (job.status,job.attempts,job.last_error)==(NotificationJob.Status.PENDING,0,"")
    assert dispatch.status==GroupDispatch.Status.QUEUED


@pytest.mark.django_db
def test_magic_link_can_be_revoked_only_with_unit_permission(client):
    unit=OrganizationalUnit.objects.create(name="حسابرسی",slug="audit-unit")
    other=OrganizationalUnit.objects.create(name="واحد دیگر audit",slug="audit-other")
    view=Permission.objects.get(content_type__app_label="accounts",codename="view_ticketmagiclink")
    change=Permission.objects.get(content_type__app_label="accounts",codename="change_ticketmagiclink")
    unit.permissions.add(view,change)
    auditor=make_staff("unit-auditor",unit); outsider=make_staff("audit-outsider",other)
    customer=make_contact("audit-contact","09120000303")
    category=TicketCategory.objects.create(title="ممیزی",slug="audit-revoke")
    ticket=create_ticket(customer=customer,created_by=auditor,unit=unit,category=category,subject="سند قابل ممیزی",body="سند")
    _,link=TicketMagicLink.issue_with_record(ticket,created_by=auditor)

    client.force_login(outsider)
    assert client.post(reverse("staff_tickets:magic_revoke",args=[link.pk])).status_code==403
    client.force_login(auditor)
    assert client.post(reverse("staff_tickets:magic_revoke",args=[link.pk])).status_code==302
    link.refresh_from_db(); assert link.revoked_at is not None