Files
customerTicket/tests/test_role_workflow_regression.py
2026-09-10 11:34:19 +03:30

195 lines
8.6 KiB
Python

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()