import pytest from django.contrib.auth.models import Permission 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.forms import StaffCreateTicketForm from apps.tickets.models import GroupDispatch, Ticket, TicketCategory from apps.tickets.policies import DispatchPolicy, TicketPolicy from apps.tickets.selectors import staff_ticket_rows from apps.tickets.services import create_ticket, dispatch_group_tickets pytestmark = pytest.mark.django_db @pytest.fixture def organization(): call_command("bootstrap") return { "sales": OrganizationalUnit.objects.get(slug="sales"), "warehouse": OrganizationalUnit.objects.get(slug="warehouse-loading"), "finance": OrganizationalUnit.objects.get(slug="finance"), "executive": OrganizationalUnit.objects.get(slug="executive"), } def customer(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 def add_contact(unit, username, mobile, group_name): user = customer(username, mobile) group = ContactGroup.objects.create(unit=unit, name=group_name) user.customer_profile.groups.add(group) return user, group def test_bootstrap_creates_requested_users_roles_units_and_explicit_permissions(organization): assert not OrganizationalUnit.objects.filter(slug="accounting").exists() assert not OrganizationalUnit.objects.filter(slug__in=("logistics", "warehouse")).exists() expected = { "kimiya_ghaffari": ("sales", StaffProfile.Role.SALES_EXPERT, UnitMembership.TicketScope.SELF), "zahra_jamali": ("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), "mostafa_sokhaeifard": ("executive", StaffProfile.Role.CEO, UnitMembership.TicketScope.UNIT), "ms_sabzali": ("warehouse-loading", StaffProfile.Role.WAREHOUSE_EXPERT, UnitMembership.TicketScope.UNIT), "mehdi_azami": ("warehouse-loading", StaffProfile.Role.WAREHOUSE_EXPERT, UnitMembership.TicketScope.UNIT), "fereshteh_yavari": ("finance", StaffProfile.Role.FINANCE_EXPERT, UnitMembership.TicketScope.UNIT), "fatemeh_sokhaei": ("finance", StaffProfile.Role.FINANCE_EXPERT, UnitMembership.TicketScope.UNIT), "rana_nobakht": ("finance", StaffProfile.Role.FINANCE_EXPERT, UnitMembership.TicketScope.UNIT), "mehdi_liaqat": ("finance", StaffProfile.Role.FINANCE_EXPERT, UnitMembership.TicketScope.UNIT), "aliasghar_sokhaeifard": ("finance", StaffProfile.Role.FINANCE_EXPERT, UnitMembership.TicketScope.UNIT), } for username, (unit_slug, role, scope) in expected.items(): user = User.objects.get(username=username) assert user.staff_profile.role == role assert user.staff_profile.primary_unit.slug == unit_slug assert list(user.unit_memberships.values_list("unit__slug", flat=True)) == [unit_slug] assert user.unit_memberships.get().ticket_scope == scope sabzali = User.objects.get(username="ms_sabzali") assert DispatchPolicy.can_create(sabzali, organization["warehouse"], GroupDispatch.OperationType.BULK) assert DispatchPolicy.can_create(sabzali, organization["warehouse"], GroupDispatch.OperationType.INQUIRY) assert not DispatchPolicy.can_view_all(sabzali, GroupDispatch.OperationType.BULK) ceo = User.objects.get(username="mostafa_sokhaeifard") assert ceo.has_perm("tickets.view_all_tickets") assert DispatchPolicy.can_view_all(ceo, GroupDispatch.OperationType.BULK) assert DispatchPolicy.can_view_all(ceo, GroupDispatch.OperationType.INQUIRY) def test_bootstrap_merges_legacy_accounting_into_canonical_finance(client, organization): finance = organization["finance"] legacy = OrganizationalUnit.objects.create( slug="accounting", name="حسابداری", is_active=True, accepts_external_tickets=True, show_in_customer_form=True, ) legacy_category = TicketCategory.objects.create( unit=legacy, title="اسناد حسابداری قدیمی", slug="legacy-accounting-documents", ) legacy_group = ContactGroup.objects.create(unit=legacy, name="مخاطبان حسابداری قدیمی") legacy_user = User.objects.create_user( username="legacy-accountant", account_type=User.AccountType.STAFF, ) StaffProfile.objects.create( user=legacy_user, primary_unit=legacy, role=StaffProfile.Role.FINANCE_EXPERT, display_name="کارشناس حسابداری قدیمی", is_customer_selectable=True, ) call_command("bootstrap") assert not OrganizationalUnit.objects.filter(slug="accounting").exists() legacy_category.refresh_from_db() legacy_group.refresh_from_db() legacy_user.staff_profile.refresh_from_db() assert legacy_category.unit == finance assert legacy_group.unit == finance assert legacy_user.staff_profile.primary_unit == finance assert list(legacy_user.unit_memberships.values_list("unit", flat=True)) == [finance.pk] buyer = customer("legacy-merge-buyer", "09120007109") client.force_login(buyer) response = client.get(reverse("tickets:new_options"), {"unit": finance.pk}) assert response["Cache-Control"] == "private, no-cache, no-store, must-revalidate" assert "کارشناس حسابداری قدیمی" in {item["name"] for item in response.json()["experts"]} def test_ticket_visibility_obeys_self_unit_and_global_scopes(organization): sales, warehouse, finance = organization["sales"], organization["warehouse"], organization["finance"] kimia = User.objects.get(username="kimiya_ghaffari") zahra = User.objects.get(username="zahra_jamali") supervisor = User.objects.get(username="maryam_fotovat") loader = User.objects.get(username="ms_sabzali") accountant = User.objects.get(username="fereshteh_yavari") ceo = User.objects.get(username="mostafa_sokhaeifard") buyer = customer("permission-buyer", "09120007101") sales_category = TicketCategory.objects.get(slug="sales-general") warehouse_category = TicketCategory.objects.get(slug="warehouse-loading-general") finance_category = TicketCategory.objects.get(slug="finance-general") public_sales = create_ticket(customer=buyer, unit=sales, category=sales_category, subject="درخواست عمومی فروش", body="شرح عمومی") selected_sales = create_ticket(customer=buyer, unit=sales, category=sales_category, subject="درخواست انتخابی زهرا", body="شرح", selected_expert=zahra) kimia_owned = create_ticket(customer=buyer, created_by=kimia, unit=sales, category=sales_category, subject="ایجادشده توسط کیمیا", body="شرح") warehouse_ticket = create_ticket(customer=buyer, created_by=loader, unit=warehouse, category=warehouse_category, subject="درخواست داخلی بارگیری", body="شرح") finance_ticket = create_ticket(customer=buyer, unit=finance, category=finance_category, subject="درخواست مالی", body="شرح") assert set(staff_ticket_rows(kimia)) == {public_sales, kimia_owned} assert TicketPolicy.can_reply(kimia, public_sales) assert not TicketPolicy.can_view(kimia, selected_sales) assert {public_sales, selected_sales, kimia_owned}.issubset(set(staff_ticket_rows(supervisor))) assert set(staff_ticket_rows(loader)) == {warehouse_ticket} assert set(staff_ticket_rows(accountant)) == {finance_ticket} assert set(staff_ticket_rows(ceo)) == {public_sales, selected_sales, kimia_owned, warehouse_ticket, finance_ticket} def test_staff_ticket_form_locks_primary_unit_and_rejects_tampered_unit_category_and_group(client, organization): kimia = User.objects.get(username="kimiya_ghaffari") sales, finance = organization["sales"], organization["finance"] sales_category = TicketCategory.objects.get(slug="sales-general") finance_category = TicketCategory.objects.get(slug="finance-general") sales_group = ContactGroup.objects.get(unit=sales, name__startswith="مخاطبان") finance_group = ContactGroup.objects.get(unit=finance, name__startswith="مخاطبان") form = StaffCreateTicketForm(user=kimia) assert list(form.fields["unit"].queryset) == [sales] assert sales_category in form.fields["category"].queryset assert not form.fields["category"].queryset.exclude(unit=sales).exists() assert sales_group in form.fields["groups"].queryset and finance_group not in form.fields["groups"].queryset client.force_login(kimia) response = client.get(reverse("staff_tickets:create")) assert response.status_code == 200 assert b'type="hidden" name="unit"' in response.content before = Ticket.objects.count() response = client.post(reverse("staff_tickets:create"), { "unit": finance.pk, "customer_mobile": "09120007102", "first_name": "مشتری", "last_name": "آزمایشی", "groups": [finance_group.pk], "category": finance_category.pk, "subject": "درخواست دستکاری شده", "body": "نباید ثبت شود", }) assert response.status_code == 200 and Ticket.objects.count() == before def test_contact_create_only_accepts_groups_from_users_unit(client, organization): kimia = User.objects.get(username="kimiya_ghaffari") sales_group = ContactGroup.objects.get(unit=organization["sales"], name__startswith="مخاطبان") finance_group = ContactGroup.objects.get(unit=organization["finance"], name__startswith="مخاطبان") client.force_login(kimia) page = client.get(reverse("staff_tickets:contact_create")) assert f'value="{sales_group.pk}"'.encode() in page.content assert f'value="{finance_group.pk}"'.encode() not in page.content response = client.post(reverse("staff_tickets:contact_create"), { "mobile": "09120007103", "first_name": "مخاطب", "stakeholder_type": CustomerProfile.StakeholderType.CUSTOMER, "groups": [finance_group.pk], }) assert response.status_code == 200 assert not User.objects.filter(mobile="+989120007103").exists() @pytest.mark.parametrize("operation_type", [GroupDispatch.OperationType.BULK, GroupDispatch.OperationType.INQUIRY]) def test_dispatch_and_inquiry_limited_owner_and_global_view_all(client, organization, operation_type): warehouse = organization["warehouse"] category = TicketCategory.objects.get(slug="warehouse-loading-general") _, group = add_contact(warehouse, f"recipient-{operation_type.lower()}", "09120007104" if operation_type == "BULK" else "09120007105", f"گروه {operation_type}") sabzali = User.objects.get(username="ms_sabzali") other = User.objects.get(username="mehdi_azami") permission = Permission.objects.get(content_type__app_label="tickets", codename="create_bulk_dispatch" if operation_type == "BULK" else "create_inquiry") other.unit_memberships.get(unit=warehouse).permissions.add(permission) own, _ = dispatch_group_tickets(user=sabzali, unit=warehouse, contact_group=group, category=category, operation_type=operation_type, subject=f"عملیات خود {operation_type}", body="متن عملیات") other_item, _ = dispatch_group_tickets(user=other, unit=warehouse, contact_group=group, category=category, operation_type=operation_type, subject=f"عملیات دیگر {operation_type}", body="متن عملیات") client.force_login(sabzali) listing = client.get(reverse("staff_tickets:dispatch_create")) assert own.subject in listing.content.decode() and other_item.subject not in listing.content.decode() assert client.get(reverse("staff_tickets:dispatch_detail", args=[other_item.public_id])).status_code == 403 client.force_login(User.objects.get(username="mostafa_sokhaeifard")) listing = client.get(reverse("staff_tickets:dispatch_create")) assert own.subject in listing.content.decode() and other_item.subject in listing.content.decode() assert client.get(reverse("staff_tickets:dispatch_detail", args=[other_item.public_id])).status_code == 200 def test_user_without_operation_permission_is_denied(client, organization): user = User.objects.get(username="mehdi_azami") client.force_login(user) assert client.get(reverse("staff_tickets:dispatch_create")).status_code == 403 def test_customer_options_and_search_are_unit_scoped_and_backend_rejects_cross_unit_expert(client, organization): sales, finance = organization["sales"], organization["finance"] buyer = customer("dynamic-customer", "09120007106") client.force_login(buyer) sales_payload = client.get(reverse("tickets:new_options"), {"unit": sales.pk}).json() finance_payload = client.get(reverse("tickets:new_options"), {"unit": finance.pk}).json() assert sales_payload["queue"] == {"unit_id": sales.pk, "label": "صف عمومی فروش"} assert finance_payload["queue"]["unit_id"] == finance.pk assert {item["name"] for item in sales_payload["experts"]}.issuperset({"کیمیا غفاری", "زهرا جمالی"}) assert not ({item["name"] for item in sales_payload["experts"]} & {item["name"] for item in finance_payload["experts"]}) search = client.get(reverse("tickets:new_search"), {"q": "مالی و حسابداری"}).json()["results"] assert search and search[0]["unit_id"] == finance.pk finance_expert = User.objects.get(username="fereshteh_yavari").staff_profile response = client.post(reverse("tickets:new"), { "unit": sales.pk, "category": TicketCategory.objects.get(slug="sales-general").pk, "selected_expert": finance_expert.pk, "subject": "انتخاب کارشناس نامرتبط", "body": "این درخواست باید رد شود", }) assert response.status_code == 200 assert not Ticket.objects.filter(subject="انتخاب کارشناس نامرتبط").exists() def test_customer_new_initial_queue_uses_the_default_unit(client, organization): buyer = customer("default-unit-customer", "09120007110") client.force_login(buyer) response = client.get(reverse("tickets:new")) assert response.status_code == 200 assert response.context["selected_unit"] == organization["sales"] assert f'data-unit-id="{organization["sales"].pk}"'.encode() in response.content assert "صف عمومی فروش" in response.content.decode()