import pytest from django.core.files.uploadedfile import SimpleUploadedFile from django.core.management import call_command from django.urls import reverse from apps.accounts.models import User from apps.customers.models import ContactGroup, CustomerProfile from apps.staff.models import OrganizationalUnit, StaffProfile, UnitMembership from apps.tickets.models import Ticket, TicketCategory from apps.tickets.policies import TicketPolicy from apps.tickets.services import create_ticket pytestmark = pytest.mark.django_db @pytest.fixture def qa_organization(): call_command("bootstrap") return { slug: OrganizationalUnit.objects.get(slug=slug) for slug in ("sales", "warehouse-loading", "finance", "executive") } REPRESENTATIVES = [ ("kimiya_ghaffari", "sales", StaffProfile.Role.SALES_EXPERT, UnitMembership.TicketScope.SELF), ("maryam_fotovat", "sales", StaffProfile.Role.SALES_SUPERVISOR, UnitMembership.TicketScope.UNIT), ("morteza_sokhaeifard", "sales", StaffProfile.Role.SALES_MANAGER, UnitMembership.TicketScope.UNIT), ("ms_sabzali", "warehouse-loading", StaffProfile.Role.WAREHOUSE_EXPERT, UnitMembership.TicketScope.UNIT), ("fereshteh_yavari", "finance", StaffProfile.Role.FINANCE_EXPERT, UnitMembership.TicketScope.UNIT), ("mostafa_sokhaeifard", "executive", StaffProfile.Role.CEO, UnitMembership.TicketScope.UNIT), ] @pytest.mark.parametrize("username,unit_slug,role,scope", REPRESENTATIVES) def test_representative_role_unit_scope_and_primary_navigation( client, qa_organization, username, unit_slug, role, scope ): user = User.objects.get(username=username) membership = user.unit_memberships.get() assert user.staff_profile.role == role assert user.staff_profile.primary_unit == qa_organization[unit_slug] assert membership.unit == qa_organization[unit_slug] assert membership.ticket_scope == scope assert membership.is_primary and membership.is_active client.force_login(user) dashboard = client.get(reverse("staff_tickets:list")) assert dashboard.status_code == 200 assert reverse("staff_tickets:create") in dashboard.content.decode() assert client.get(reverse("staff_tickets:create")).status_code == 200 def test_sales_expert_can_open_and_reply_to_public_own_and_assigned_tickets(client, qa_organization): sales = qa_organization["sales"] category = TicketCategory.objects.get(slug="sales-general") kimia = User.objects.get(username="kimiya_ghaffari") zahra = User.objects.get(username="zahra_jamali") customer = User.objects.create_user( username="qa-sales-customer", mobile="09120008401", account_type=User.AccountType.CUSTOMER ) CustomerProfile.objects.create(user=customer, first_name="مشتری") public = create_ticket(customer=customer, unit=sales, category=category, subject="عمومی QA", body="شرح") own = create_ticket( customer=customer, created_by=kimia, unit=sales, category=category, subject="ساخته کیمیا QA", body="شرح" ) assigned = create_ticket( customer=customer, unit=sales, category=category, selected_expert=kimia, subject="اختصاص کیمیا QA", body="شرح", ) unrelated = create_ticket( customer=customer, unit=sales, category=category, selected_expert=zahra, subject="اختصاص زهرا QA", body="شرح", ) client.force_login(kimia) for ticket in (public, own, assigned): assert client.get(reverse("staff_tickets:detail", args=[ticket.public_id])).status_code == 200 response = client.post( reverse("staff_tickets:reply", args=[ticket.public_id]), {"body": f"پاسخ مجاز {ticket.pk}"} ) assert response.status_code == 302 assert client.get(reverse("staff_tickets:detail", args=[unrelated.public_id])).status_code == 403 assert not TicketPolicy.can_reply(kimia, unrelated) def test_staff_ticket_creation_completes_with_unit_category_customer_group_and_message(client, qa_organization): sales = qa_organization["sales"] kimia = User.objects.get(username="kimiya_ghaffari") category = TicketCategory.objects.get(slug="sales-general") group = ContactGroup.objects.get(unit=sales, name__startswith="مخاطبان") client.force_login(kimia) response = client.post( reverse("staff_tickets:create"), { "unit": sales.pk, "customer_mobile": "09120008402", "first_name": "مشتری", "last_name": "فروش", "stakeholder_type": CustomerProfile.StakeholderType.CUSTOMER, "groups": [group.pk], "category": category.pk, "subject": "ایجاد کامل Staff QA", "body": "پیام اولیه کامل کارشناس", }, ) assert response.status_code == 302 ticket = Ticket.objects.get(subject="ایجاد کامل Staff QA") assert ticket.owning_unit == sales and ticket.assigned_unit == sales assert ticket.category == category and ticket.created_by == kimia and ticket.owner == kimia assert ticket.ticket_messages.get().body == "پیام اولیه کامل کارشناس" assert group in ticket.customer.customer_profile.groups.all() @pytest.mark.parametrize( "unit_slug,mobile", [("sales", "09120008411"), ("finance", "09120008413")] ) def test_public_options_exactly_match_each_units_staff_categories_and_queue( client, qa_organization, unit_slug, mobile ): customer = User.objects.create_user( username=f"qa-options-{unit_slug}", mobile=mobile, account_type=User.AccountType.CUSTOMER, ) CustomerProfile.objects.create(user=customer, first_name="مشتری") client.force_login(customer) unit = qa_organization[unit_slug] payload = client.get(reverse("tickets:new_options"), {"unit": unit.pk}).json() expected_experts = set( StaffProfile.objects.filter(primary_unit=unit, is_active=True, is_customer_selectable=True) .values_list("display_name", flat=True) ) expected_categories = set( TicketCategory.objects.filter(unit=unit, is_active=True, show_in_customer_form=True) .values_list("title", flat=True) ) assert {row["name"] for row in payload["experts"]} == expected_experts assert {row["label"] for row in payload["categories"]} == expected_categories assert payload["queue"] == {"unit_id": unit.pk, "label": f"صف عمومی {unit.name}"} def test_invalid_public_attachment_is_a_form_error_and_rolls_back_ticket(client, qa_organization, tmp_path, settings): settings.MEDIA_ROOT = tmp_path customer = User.objects.create_user( username="qa-invalid-attachment", mobile="09120008403", account_type=User.AccountType.CUSTOMER ) CustomerProfile.objects.create(user=customer, first_name="مشتری") client.force_login(customer) sales = qa_organization["sales"] before = Ticket.objects.count() response = client.post( reverse("tickets:new"), { "unit": sales.pk, "category": TicketCategory.objects.get(slug="sales-general").pk, "subject": "فایل غیرمجاز QA", "body": "فایل اجرایی نباید ذخیره شود", "files": SimpleUploadedFile("malware.exe", b"MZ-not-an-executable", content_type="application/octet-stream"), }, ) assert response.status_code == 200 assert "مجاز نیست" in response.content.decode() assert Ticket.objects.count() == before @pytest.mark.parametrize( "missing_field,payload_update,error_key", [ ("unit", {"unit": ""}, "unit"), ("category", {"category": ""}, "category"), ("subject", {"subject": ""}, "subject"), ("message", {"body": ""}, "__all__"), ], ) def test_public_ticket_required_field_validation_is_repeatable( client, qa_organization, missing_field, payload_update, error_key ): customer = User.objects.create_user( username=f"qa-validation-{missing_field}", mobile=f"0912000850{len(missing_field)}", account_type=User.AccountType.CUSTOMER, ) CustomerProfile.objects.create(user=customer, first_name="مشتری") client.force_login(customer) sales = qa_organization["sales"] payload = { "unit": sales.pk, "category": TicketCategory.objects.get(slug="sales-general").pk, "subject": "اعتبارسنجی عادی QA", "body": "شرح معتبر", } payload.update(payload_update) response = client.post(reverse("tickets:new"), payload) assert response.status_code == 200 assert error_key in response.context["form"].errors assert not Ticket.objects.filter(subject="اعتبارسنجی عادی QA").exists()