287 lines
17 KiB
Python
287 lines
17 KiB
Python
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_each_public_queue_and_staff_message_endpoint_is_strictly_unit_scoped(client, organization):
|
|
sales, finance = organization["sales"], organization["finance"]
|
|
sales_staff = User.objects.get(username="kimiya_ghaffari")
|
|
other_sales_staff = User.objects.get(username="zahra_jamali")
|
|
finance_staff = User.objects.get(username="fereshteh_yavari")
|
|
buyer = customer("isolated-queue-buyer", "09120007111")
|
|
sales_ticket = create_ticket(
|
|
customer=buyer,
|
|
unit=sales,
|
|
category=TicketCategory.objects.get(slug="sales-general"),
|
|
subject="صف عمومی فقط فروش",
|
|
body="پیام فروش",
|
|
)
|
|
finance_ticket = create_ticket(
|
|
customer=buyer,
|
|
unit=finance,
|
|
category=TicketCategory.objects.get(slug="finance-general"),
|
|
subject="صف عمومی فقط مالی",
|
|
body="پیام مالی",
|
|
)
|
|
|
|
assert set(staff_ticket_rows(sales_staff)) == {sales_ticket}
|
|
assert finance_ticket not in staff_ticket_rows(sales_staff)
|
|
assert set(staff_ticket_rows(finance_staff)) == {finance_ticket}
|
|
assert sales_ticket not in staff_ticket_rows(finance_staff)
|
|
|
|
client.force_login(sales_staff)
|
|
assert client.get(reverse("staff_tickets:detail", args=[finance_ticket.public_id])).status_code == 403
|
|
assert client.get(reverse("staff_tickets:messages", args=[finance_ticket.public_id])).status_code == 403
|
|
assert client.get(reverse("staff_tickets:poll", args=[finance_ticket.public_id])).status_code == 403
|
|
|
|
routed_ticket = create_ticket(
|
|
customer=buyer,
|
|
unit=sales,
|
|
category=TicketCategory.objects.get(slug="sales-general"),
|
|
subject="صف منتقلشده از فروش به مالی",
|
|
body="پیام منتقلشده",
|
|
)
|
|
routed_ticket.assigned_unit = finance
|
|
routed_ticket.save(update_fields=["assigned_unit", "updated_at"])
|
|
assert routed_ticket not in staff_ticket_rows(sales_staff)
|
|
assert not TicketPolicy.can_view(sales_staff, routed_ticket)
|
|
assert routed_ticket in staff_ticket_rows(finance_staff)
|
|
assert TicketPolicy.can_view(finance_staff, routed_ticket)
|
|
|
|
# Once a SELF-scope queue item is claimed, it is no longer visible to a
|
|
# different expert merely because both experts belong to the same unit.
|
|
sales_ticket.owner = other_sales_staff
|
|
sales_ticket.save(update_fields=["owner", "updated_at"])
|
|
assert sales_ticket not in staff_ticket_rows(sales_staff)
|
|
assert not TicketPolicy.can_view(sales_staff, sales_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()
|