import pytest from django.contrib.auth.models import Permission from django.core.management import call_command from django.urls import reverse from playwright.sync_api import expect from apps.accounts.models import User from apps.customers.models import ContactGroup, CustomerProfile from apps.staff.models import OrganizationalUnit from apps.tickets.models import GroupDispatch, TicketCategory from apps.tickets.services import create_ticket, dispatch_group_tickets from tests.e2e.pages.auth import AuthenticationPage from tests.e2e.pages.tickets import PublicTicketCreatePage pytestmark = [pytest.mark.e2e, pytest.mark.django_db(transaction=True)] PASSWORD = "E2e-org-permissions-2026" @pytest.fixture def org_data(): call_command("bootstrap") users = {name: User.objects.get(username=name) for name in ( "kimiya_ghaffari", "maryam_fotovat", "morteza_sokhaeifard", "ms_sabzali", "mehdi_azami", "fereshteh_yavari", "mostafa_sokhaeifard" )} for user in users.values(): user.set_password(PASSWORD) user.save(update_fields=["password"]) units = {slug: OrganizationalUnit.objects.get(slug=slug) for slug in ("sales", "warehouse-loading", "finance")} return users, units def make_customer(username, mobile, password=None): user = User.objects.create_user(username=username, mobile=mobile, password=password, account_type=User.AccountType.CUSTOMER) CustomerProfile.objects.create(user=user, first_name=username) return user def login_staff(page, live_server, user): AuthenticationPage(page, live_server.url).login_staff(user, PASSWORD) def test_e2e_01_sales_expert_sees_locked_unit_and_sales_categories(live_server, browser_page, org_data): users, units = org_data with browser_page() as page: login_staff(page, live_server, users["kimiya_ghaffari"]) page.goto(f"{live_server.url}{reverse('staff_tickets:create')}", wait_until="domcontentloaded") expect(page.locator("[data-staff-unit]")).to_have_text("فروش") expect(page.locator('input[type="hidden"][name="unit"]')).to_have_value(str(units["sales"].pk)) categories = page.locator('select[name="category"] option') expect(categories.filter(has_text="عمومی فروش")).to_have_count(1) expect(categories.filter(has_text="مالی و حسابداری")).to_have_count(0) groups = page.locator('select[name="groups"] option') expect(groups.filter(has_text="مخاطبان فروش")).to_have_count(1) expect(groups.filter(has_text="مخاطبان مالی")).to_have_count(0) @pytest.mark.parametrize("username", ["maryam_fotovat", "morteza_sokhaeifard"]) def test_e2e_02_sales_leadership_can_login_navigate_and_create_in_its_unit( live_server, browser_page, org_data, username ): users, units = org_data with browser_page() as page: login_staff(page, live_server, users[username]) expect(page).to_have_url(f"{live_server.url}{reverse('staff_tickets:list')}") response = page.goto(f"{live_server.url}{reverse('staff_tickets:create')}", wait_until="domcontentloaded") assert response.status == 200 expect(page.locator("[data-staff-unit]")).to_have_text("فروش") expect(page.locator('input[name="unit"]')).to_have_value(str(units["sales"].pk)) def test_e2e_05_ceo_can_login_and_open_broad_dispatch_workflow(live_server, browser_page, org_data): users, _ = org_data with browser_page() as page: login_staff(page, live_server, users["mostafa_sokhaeifard"]) expect(page).to_have_url(f"{live_server.url}{reverse('staff_tickets:list')}") response = page.goto( f"{live_server.url}{reverse('staff_tickets:dispatch_create')}", wait_until="domcontentloaded" ) assert response.status == 200 expect(page.locator("body")).to_contain_text("ارسال گروهی") def test_e2e_02_contact_group_picker_excludes_other_units(live_server, browser_page, org_data): users, units = org_data sales_group = ContactGroup.objects.get(unit=units["sales"], name__startswith="مخاطبان") with browser_page() as page: login_staff(page, live_server, users["kimiya_ghaffari"]) page.goto(f"{live_server.url}{reverse('staff_tickets:contact_create')}", wait_until="domcontentloaded") options = page.locator('[data-contact-group-selector] label') expect(options.filter(has_text="مخاطبان فروش")).to_have_count(1) expect(options.filter(has_text="بارگیری انبار")).to_have_count(0) expect(options.filter(has_text="مالی / حسابداری")).to_have_count(0) page.locator('input[name="mobile"]').fill("09120008119") page.locator('input[name="first_name"]').fill("مخاطب") page.locator('input[name="last_name"]').fill("فروش") page.locator('select[name="stakeholder_type"]').select_option(CustomerProfile.StakeholderType.CUSTOMER) page.locator(f'input[name="groups"][value="{sales_group.pk}"]').check() with page.expect_navigation(wait_until="domcontentloaded"): page.get_by_role("button", name="ذخیره مخاطب").click() expect(page.locator("body")).to_contain_text("مخاطب فروش") contact = User.objects.get(mobile="+989120008119") assert list(contact.customer_profile.groups.all()) == [sales_group] def test_e2e_02_staff_ticket_creation_finishes_with_sales_contact_group(live_server, browser_page, org_data): users, units = org_data sales_group = ContactGroup.objects.get(unit=units["sales"], name__startswith="مخاطبان") category = TicketCategory.objects.get(slug="sales-general") subject = "گردش کامل ایجاد تیکت Staff E2E" with browser_page() as page: login_staff(page, live_server, users["kimiya_ghaffari"]) page.goto(f"{live_server.url}{reverse('staff_tickets:create')}", wait_until="domcontentloaded") page.locator('input[name="customer_mobile"]').fill("09120008118") page.locator('input[name="first_name"]').fill("مشتری") page.locator('input[name="last_name"]').fill("فروش") page.locator('select[name="groups"]').select_option(str(sales_group.pk)) page.locator('select[name="category"]').select_option(str(category.pk)) page.locator('input[name="subject"]').fill(subject) page.locator('textarea[name="body"]').fill("شرح کامل ثبت توسط کارشناس فروش") with page.expect_navigation(wait_until="domcontentloaded"): page.get_by_role("button", name="ثبت تیکت و ارسال لینک امن").click() expect(page.locator("h1")).to_have_text(subject) ticket = TicketCategory.objects.get(pk=category.pk).tickets.get(subject=subject) assert ticket.owning_unit == units["sales"] assert ticket.owner == users["kimiya_ghaffari"] assert sales_group in ticket.customer.customer_profile.groups.all() def test_e2e_03_warehouse_staff_cannot_see_sales_tickets(live_server, browser_page, org_data): users, units = org_data buyer = make_customer("e2e-scope-buyer", "09120008101") create_ticket(customer=buyer, unit=units["sales"], category=TicketCategory.objects.get(slug="sales-general"), subject="تیکت محرمانه فروش E2E", body="شرح") create_ticket(customer=buyer, created_by=users["ms_sabzali"], unit=units["warehouse-loading"], category=TicketCategory.objects.get(slug="warehouse-loading-general"), subject="تیکت مجاز بارگیری E2E", body="شرح") with browser_page() as page: login_staff(page, live_server, users["ms_sabzali"]) content = page.locator("body") expect(content).to_contain_text("تیکت مجاز بارگیری E2E") expect(content).not_to_contain_text("تیکت محرمانه فروش E2E") def test_e2e_04_accounting_sees_only_finance_scope(live_server, browser_page, org_data): users, units = org_data buyer = make_customer("e2e-finance-buyer", "09120008102") create_ticket(customer=buyer, unit=units["finance"], category=TicketCategory.objects.get(slug="finance-general"), subject="تیکت مالی مجاز E2E", body="شرح") create_ticket(customer=buyer, created_by=users["ms_sabzali"], unit=units["warehouse-loading"], category=TicketCategory.objects.get(slug="warehouse-loading-general"), subject="تیکت انبار نامجاز E2E", body="شرح") with browser_page() as page: login_staff(page, live_server, users["fereshteh_yavari"]) expect(page.locator("body")).to_contain_text("تیکت مالی مجاز E2E") expect(page.locator("body")).not_to_contain_text("تیکت انبار نامجاز E2E") def prepare_operation(users, units, operation_type, suffix): warehouse = units["warehouse-loading"] contact = make_customer(f"e2e-operation-contact-{suffix}", f"091200082{suffix}") group = ContactGroup.objects.create(unit=warehouse, name=f"گروه عملیات E2E {suffix}") contact.customer_profile.groups.add(group) category = TicketCategory.objects.get(slug="warehouse-loading-general") other = users["mehdi_azami"] codename = "create_bulk_dispatch" if operation_type == GroupDispatch.OperationType.BULK else "create_inquiry" other.unit_memberships.get(unit=warehouse).permissions.add(Permission.objects.get(content_type__app_label="tickets", codename=codename)) foreign, _ = dispatch_group_tickets(user=other, unit=warehouse, contact_group=group, category=category, operation_type=operation_type, subject=f"عملیات متعلق به مهدی {suffix}", body="متن عملیات") return group, category, foreign @pytest.mark.parametrize("operation_type,suffix", [(GroupDispatch.OperationType.BULK, "03"), (GroupDispatch.OperationType.INQUIRY, "04")]) def test_e2e_05_06_limited_operation_owner_and_ceo_view_all(live_server, browser_page, org_data, operation_type, suffix): users, units = org_data group, category, foreign = prepare_operation(users, units, operation_type, suffix) own_subject = "ارسال گروهی سبزعلی E2E" if operation_type == GroupDispatch.OperationType.BULK else "استعلام سبزعلی E2E" with browser_page() as page: login_staff(page, live_server, users["ms_sabzali"]) page.goto(f"{live_server.url}{reverse('staff_tickets:dispatch_create')}", wait_until="domcontentloaded") expect(page.locator("body")).not_to_contain_text(foreign.subject) composer = page.locator(".dispatch-composer") composer.locator('select[name="operation_type"]').select_option(operation_type) composer.locator('select[name="unit"]').select_option(str(units["warehouse-loading"].pk)) composer.locator('select[name="contact_group"]').select_option(str(group.pk)) composer.locator('select[name="category"]').select_option(str(category.pk)) composer.locator('input[name="subject"]').fill(own_subject) composer.locator('textarea[name="body"]').fill("متن کامل عملیات E2E") page.get_by_role("button", name="ایجاد تیکت‌ها").click() expect(page.locator("body")).to_contain_text(own_subject) page.context.clear_cookies() login_staff(page, live_server, users["mostafa_sokhaeifard"]) page.goto(f"{live_server.url}{reverse('staff_tickets:dispatch_create')}", wait_until="domcontentloaded") expect(page.locator("body")).to_contain_text(own_subject) expect(page.locator("body")).to_contain_text(foreign.subject) def test_e2e_07_manual_unit_change_refreshes_staff_and_queue(live_server, browser_page, org_data): _, units = org_data buyer = make_customer("e2e-dynamic-manual", "09120008105", PASSWORD) with browser_page() as page: AuthenticationPage(page, live_server.url).login_customer(buyer, PASSWORD) page.goto(f"{live_server.url}{reverse('tickets:new')}", wait_until="domcontentloaded") ticket_page = PublicTicketCreatePage(page, live_server.url) sales_payload = ticket_page.select_unit(units["sales"].pk) assert {item["name"] for item in sales_payload["experts"]}.issuperset({"کیمیا غفاری", "زهرا جمالی"}) expect(page.locator("[data-general-queue]")).to_have_attribute("data-unit-id", str(units["sales"].pk)) expect(page.locator("[data-general-queue]")).to_contain_text("صف عمومی فروش") expect(page.locator("[data-expert-picker]")).to_contain_text("کیمیا غفاری") finance_payload = ticket_page.select_unit(units["finance"].pk) assert {item["name"] for item in finance_payload["experts"]}.issuperset({"خانم فرشته یاوری"}) expect(page.locator("[data-general-queue]")).to_have_attribute("data-unit-id", str(units["finance"].pk)) expect(page.locator("[data-general-queue]")).to_contain_text("صف عمومی مالی / حسابداری") expect(page.locator("[data-expert-picker]")).to_contain_text("خانم فرشته یاوری") expect(page.locator("[data-expert-picker]")).not_to_contain_text("کیمیا غفاری") def test_e2e_08_search_selection_uses_same_unit_staff_queue_pipeline(live_server, browser_page, org_data): _, units = org_data buyer = make_customer("e2e-dynamic-search", "09120008106", PASSWORD) with browser_page() as page: AuthenticationPage(page, live_server.url).login_customer(buyer, PASSWORD) page.goto(f"{live_server.url}{reverse('tickets:new')}", wait_until="domcontentloaded") ticket_page = PublicTicketCreatePage(page, live_server.url) ticket_page.select_search_result("عمومی فروش", "عمومی فروش") expect(page.locator('[data-ticket-unit]')).to_have_value(str(units["sales"].pk)) expect(page.locator("[data-general-queue]")).to_have_attribute("data-unit-id", str(units["sales"].pk)) expect(page.locator("[data-expert-picker]")).to_contain_text("کیمیا غفاری") expect(page.locator("[data-expert-picker]")).not_to_contain_text("خانم فرشته یاوری") def test_unit_change_renders_experts_when_a_legacy_cached_payload_has_no_queue(live_server, browser_page, org_data): _, units = org_data buyer = make_customer("e2e-legacy-options-payload", "09120008107", PASSWORD) with browser_page() as page: AuthenticationPage(page, live_server.url).login_customer(buyer, PASSWORD) def remove_queue_from_response(route): response = route.fetch() payload = response.json() payload.pop("queue", None) route.fulfill(response=response, json=payload) page.route("**/tickets/new/options/**", remove_queue_from_response) page.goto(f"{live_server.url}{reverse('tickets:new')}", wait_until="domcontentloaded") ticket_page = PublicTicketCreatePage(page, live_server.url) payload = ticket_page.select_unit(units["finance"].pk) assert payload.get("queue") is None expect(ticket_page.expert_picker).to_contain_text("خانم فرشته یاوری") expect(ticket_page.expert_picker).not_to_contain_text("کیمیا غفاری") expect(page.get_by_role("alert")).to_have_count(0)