"""S2.1 Helpdesk / Ticketing — test suite. Coverage -------- [models] 1 Queue model fields and __str__ 2 Ticket model fields, reference_number placeholder, __str__ 3 TicketReply model fields 4 Tag model 5 SLAPolicy model 6 Category model (parent/child) 7 Ticket soft-delete (is_deleted flag) 8 Ticket manager returns only non-deleted rows [services] 9 TicketService.create — happy path 10 TicketService.create — auto-assigns default queue 11 TicketService.add_reply — public reply 12 TicketService.add_reply — internal note 13 TicketService.transition — new → open 14 TicketService.transition — open → solved sets resolved_at 15 TicketService.transition — solved → closed sets closed_at 16 TicketService.assign 17 QueueService.create 39 TicketService.escalate 40 TicketService.update — modify fields 41 TicketService.update — ignored read-only fields 42 TicketService.create — quota enforcement 43 SLA timers — start_sla_timers 44 SLA timers — mark_first_response 45 SLA timers — handle_sla_status_change (pause on on_hold) 46 SLA timers — handle_sla_status_change (resume from on_hold) 47 TicketService.transition — idempotent (same status) 48 add_reply auto-transitions new → open (public reply) [API — tenant isolation & auth] 18 GET /helpdesk/tickets/ returns 401 without auth 19 GET /helpdesk/tickets/ returns 400 without X-Tenant header 20 POST /helpdesk/tickets/ creates ticket [API — queues] 21 GET /helpdesk/queues/ lists active queues 22 POST /helpdesk/queues/ creates queue (supervisor) 23 POST /helpdesk/queues/ returns 403 without perm [API — tickets CRUD] 24 GET /helpdesk/tickets/ lists with pagination keys 25 GET /helpdesk/tickets// returns ticket detail 26 PATCH /helpdesk/tickets// updates priority 27 POST /helpdesk/tickets//transition/ changes status 28 POST /helpdesk/tickets//assign/ [API — replies] 29 GET /helpdesk/tickets//replies/ lists public replies only for portal user 30 POST /helpdesk/tickets//replies/ adds public reply 31 POST /helpdesk/tickets//replies/ internal note blocked for non-agent [API — tags & SLA & categories] 32 GET /helpdesk/tags/ lists tags 33 POST /helpdesk/tags/ creates tag (supervisor) 34 GET /helpdesk/sla-policies/ lists active policies 35 GET /helpdesk/categories/ lists categories [API — reports] 36 GET /helpdesk/reports/summary/ returns totals + per-queue data 37 GET /helpdesk/reports/summary/ counts open/solved/sla correctly [permissions] 38 Helpdesk permissions registered in the IAM registry [isolation — cross-tenant] 49 Cross-tenant ticket access blocked 50 Tenant-scoped ticket listing 51 Cross-tenant queue access blocked [events] 52 Event subscribers registered 53 ticket.created event triggers SLA timer start [automations / teams] 54 Automation model CRUD 55 AgentTeam model with members 56 AgentTeam member unique constraint """ from __future__ import annotations import pytest from django.utils import timezone from simorgh.apps.accounts.models import User from simorgh.apps.helpdesk.models import ( Category, Queue, SLAPolicy, Tag, Ticket, TicketPriority, TicketReply, TicketStatus, ) from simorgh.apps.helpdesk.services import QueueService, TicketService from simorgh.apps.iam.models import Permission, Role from simorgh.apps.iam.registry import sync_registry_to_db from simorgh.apps.memberships.models import Membership from simorgh.apps.organizations.services import create_node from simorgh.apps.tenants.models import Tenant from simorgh.core.context import RequestContext, use_request_context # --------------------------------------------------------------------------- # Fixtures # --------------------------------------------------------------------------- @pytest.fixture def tenant(db) -> Tenant: return Tenant.objects.create(slug="acme-hd", name="Acme Helpdesk") @pytest.fixture def org_node(tenant): return create_node(tenant_id=tenant.pk, name="HQ") @pytest.fixture def agent(db) -> User: return User.objects.create_user("+989010000001", password="x", email="agent@acme.com") @pytest.fixture def requester(db) -> User: return User.objects.create_user("+989010000002", password="x", email="user@acme.com") @pytest.fixture def superuser(db) -> User: return User.objects.create_superuser("+989010000099", password="x", email="root@acme.com") @pytest.fixture def queue(tenant, org_node) -> Queue: return QueueService.create( tenant_id=tenant.pk, organization_node_id=org_node.pk, name="Support", is_default=True, ) @pytest.fixture def tag(tenant) -> Tag: return Tag.objects.create( tenant=tenant, name="Bug", color="#f00", ) @pytest.fixture def sla(tenant, org_node) -> SLAPolicy: return SLAPolicy.objects.create( tenant=tenant, organization_node=org_node, name="Standard", first_response_hours=4, next_response_hours=8, resolution_hours=48, ) @pytest.fixture def category(tenant, org_node, queue) -> Category: return Category.objects.create( tenant=tenant, organization_node=org_node, name="Billing", queue=queue, ) @pytest.fixture def ticket(tenant, org_node, queue, agent) -> Ticket: return TicketService.create( tenant_id=tenant.pk, organization_node_id=org_node.pk, queue_id=queue.pk, subject="Test ticket", description="Please help.", requester_id=agent.pk, created_by_id=agent.pk, ) @pytest.fixture def all_perms(db): sync_registry_to_db() return {p.codename: p for p in Permission.objects.all()} @pytest.fixture def admin_role(tenant, all_perms) -> Role: role = Role.objects.create(tenant=tenant, code="admin", name="Admin") role.permissions.set(all_perms.values()) return role @pytest.fixture def agent_membership(agent, tenant, org_node, admin_role) -> Membership: m = Membership.objects.create( tenant=tenant, organization_node=org_node, role=admin_role, ) m.users.add(agent) return m # --------------------------------------------------------------------------- # Helper: build a RequestContext for the agent # --------------------------------------------------------------------------- def _agent_ctx(agent, tenant, org_node, all_perms) -> RequestContext: return RequestContext( actor=agent, tenant=tenant, org_node_ids=frozenset({org_node.pk}), permissions=frozenset(all_perms.keys()), ) # --------------------------------------------------------------------------- # [models] 1 — Queue model # --------------------------------------------------------------------------- @pytest.mark.django_db def test_queue_str(queue): assert str(queue) == "Support" @pytest.mark.django_db def test_queue_is_default(queue): assert queue.is_default is True assert queue.is_active is True # --------------------------------------------------------------------------- # [models] 2 — Ticket model # --------------------------------------------------------------------------- @pytest.mark.django_db def test_ticket_str(ticket): assert "Test ticket" in str(ticket) @pytest.mark.django_db def test_ticket_default_status(ticket): assert ticket.status == TicketStatus.NEW @pytest.mark.django_db def test_ticket_default_priority(ticket): assert ticket.priority == TicketPriority.NORMAL @pytest.mark.django_db def test_ticket_has_public_id(ticket): assert ticket.public_id is not None # --------------------------------------------------------------------------- # [models] 3 — TicketReply # --------------------------------------------------------------------------- @pytest.mark.django_db def test_ticket_reply_fields(ticket, agent): reply = TicketReply.objects.create( ticket=ticket, body="Hello world", is_internal=False, author_id=agent.pk, ) assert reply.body == "Hello world" assert reply.is_internal is False # --------------------------------------------------------------------------- # [models] 4 — Tag # --------------------------------------------------------------------------- @pytest.mark.django_db def test_tag_str(tag): assert str(tag) == "Bug" # --------------------------------------------------------------------------- # [models] 5 — SLAPolicy # --------------------------------------------------------------------------- @pytest.mark.django_db def test_sla_policy_hours(sla): assert sla.first_response_hours == 4 assert sla.resolution_hours == 48 # --------------------------------------------------------------------------- # [models] 6 — Category parent/child # --------------------------------------------------------------------------- @pytest.mark.django_db def test_category_parent_child(tenant, org_node, queue): parent = Category.objects.create( tenant=tenant, organization_node=org_node, name="General", queue=queue ) child = Category.objects.create( tenant=tenant, organization_node=org_node, name="Sub", queue=queue, parent=parent ) assert child.parent_id == parent.pk # --------------------------------------------------------------------------- # [models] 7-8 — Soft-delete # --------------------------------------------------------------------------- @pytest.mark.django_db def test_ticket_soft_delete_sets_flag(ticket): ticket.delete() ticket.refresh_from_db() assert ticket.is_deleted is True @pytest.mark.django_db def test_ticket_soft_delete_hidden_by_default(ticket): ticket.delete() assert Ticket.objects.filter(pk=ticket.pk).count() == 0 assert Ticket.objects.with_deleted().filter(pk=ticket.pk).count() == 1 # --------------------------------------------------------------------------- # [services] 9 — TicketService.create happy path # --------------------------------------------------------------------------- @pytest.mark.django_db def test_ticket_service_create(tenant, org_node, queue, agent): t = TicketService.create( tenant_id=tenant.pk, organization_node_id=org_node.pk, queue_id=queue.pk, subject="Need help", requester_id=agent.pk, created_by_id=agent.pk, ) assert t.pk is not None assert t.subject == "Need help" assert t.status == TicketStatus.NEW assert t.tenant_id == tenant.pk # --------------------------------------------------------------------------- # [services] 10 — auto-assigns default queue # --------------------------------------------------------------------------- @pytest.mark.django_db def test_ticket_created_in_default_queue(tenant, org_node, queue, agent): t = TicketService.create( tenant_id=tenant.pk, organization_node_id=org_node.pk, queue_id=queue.pk, subject="Auto queue", requester_id=agent.pk, created_by_id=agent.pk, ) assert t.queue_id == queue.pk # --------------------------------------------------------------------------- # [services] 11-12 — TicketService.add_reply # --------------------------------------------------------------------------- @pytest.mark.django_db def test_add_public_reply(ticket, agent): reply = TicketService.add_reply( ticket, body="Public response", is_internal=False, author_id=agent.pk, actor=agent, ) assert reply.is_internal is False assert reply.body == "Public response" @pytest.mark.django_db def test_add_internal_note(ticket, agent): reply = TicketService.add_reply( ticket, body="Internal note", is_internal=True, author_id=agent.pk, actor=agent, ) assert reply.is_internal is True # --------------------------------------------------------------------------- # [services] 13-15 — TicketService.transition # --------------------------------------------------------------------------- @pytest.mark.django_db def test_transition_new_to_open(ticket, agent): TicketService.transition(ticket, new_status="open", actor_id=agent.pk, actor=agent) ticket.refresh_from_db() assert ticket.status == TicketStatus.OPEN @pytest.mark.django_db def test_transition_to_solved_sets_resolved_at(ticket, agent): TicketService.transition(ticket, new_status="open", actor_id=agent.pk, actor=agent) TicketService.transition(ticket, new_status="solved", actor_id=agent.pk, actor=agent) ticket.refresh_from_db() assert ticket.status == TicketStatus.SOLVED assert ticket.resolved_at is not None @pytest.mark.django_db def test_transition_to_closed_sets_closed_at(ticket, agent): TicketService.transition(ticket, new_status="open", actor_id=agent.pk, actor=agent) TicketService.transition(ticket, new_status="solved", actor_id=agent.pk, actor=agent) TicketService.transition(ticket, new_status="closed", actor_id=agent.pk, actor=agent) ticket.refresh_from_db() assert ticket.status == TicketStatus.CLOSED assert ticket.closed_at is not None # --------------------------------------------------------------------------- # [services] 16 — TicketService.assign # --------------------------------------------------------------------------- @pytest.mark.django_db def test_ticket_assign(ticket, agent, requester): TicketService.assign( ticket, assigned_to_id=requester.pk, assigned_by_id=agent.pk, actor=agent, ) ticket.refresh_from_db() assert ticket.assigned_to_id == requester.pk # --------------------------------------------------------------------------- # [services] 17 — QueueService.create # --------------------------------------------------------------------------- @pytest.mark.django_db def test_queue_service_create(tenant, org_node): q = QueueService.create( tenant_id=tenant.pk, organization_node_id=org_node.pk, name="Billing Queue", description="Billing dept", is_default=False, ) assert q.pk is not None assert q.name == "Billing Queue" assert q.is_active is True # --------------------------------------------------------------------------- # API fixtures # --------------------------------------------------------------------------- @pytest.fixture def api_client_agent(agent, tenant, agent_membership, all_perms): """Authenticated API client with tenant header and all permissions.""" from rest_framework.test import APIClient client = APIClient() client.force_authenticate(user=agent) client.credentials(HTTP_X_TENANT=tenant.slug) return client @pytest.fixture def api_client_anon(): from rest_framework.test import APIClient return APIClient() @pytest.fixture def api_client_no_tenant(agent, all_perms): from rest_framework.test import APIClient client = APIClient() client.force_authenticate(user=agent) return client # --------------------------------------------------------------------------- # [API] 18-20 — auth / tenant # --------------------------------------------------------------------------- @pytest.mark.django_db def test_tickets_list_requires_auth(api_client_anon, tenant): resp = api_client_anon.get( "/api/v1/helpdesk/tickets/", HTTP_X_TENANT=tenant.slug ) assert resp.status_code == 401 @pytest.mark.django_db def test_tickets_list_requires_tenant_header(api_client_no_tenant): resp = api_client_no_tenant.get("/api/v1/helpdesk/tickets/") assert resp.status_code in (400, 401, 403) @pytest.mark.django_db def test_create_ticket_via_api(api_client_agent, queue, agent_membership, tenant): resp = api_client_agent.post( "/api/v1/helpdesk/tickets/", data={ "subject": "API Ticket", "description": "Desc", "queue_id": queue.pk, "priority": "high", }, format="json", ) assert resp.status_code == 201, resp.data assert resp.data["subject"] == "API Ticket" # --------------------------------------------------------------------------- # [API] 21-23 — queues # --------------------------------------------------------------------------- @pytest.mark.django_db def test_list_queues(api_client_agent, queue): resp = api_client_agent.get("/api/v1/helpdesk/queues/") assert resp.status_code == 200 names = [q["name"] for q in resp.data] assert "Support" in names @pytest.mark.django_db def test_create_queue_supervisor(api_client_agent, tenant, org_node, agent_membership): resp = api_client_agent.post( "/api/v1/helpdesk/queues/", data={"name": "New Queue", "description": ""}, format="json", ) assert resp.status_code == 201 assert resp.data["name"] == "New Queue" @pytest.mark.django_db def test_create_queue_forbidden_without_perm(requester, tenant, org_node, all_perms): from rest_framework.test import APIClient # Create membership with limited permissions limited_role = Role.objects.create(tenant=tenant, code="viewer-hd", name="Viewer") m = Membership.objects.create(tenant=tenant, organization_node=org_node, role=limited_role) m.users.add(requester) client = APIClient() client.force_authenticate(user=requester) client.credentials(HTTP_X_TENANT=tenant.slug) resp = client.post( "/api/v1/helpdesk/queues/", data={"name": "Forbidden", "description": ""}, format="json", ) assert resp.status_code == 403 # --------------------------------------------------------------------------- # [API] 24-28 — tickets CRUD # --------------------------------------------------------------------------- @pytest.mark.django_db def test_list_tickets_has_pagination(api_client_agent, ticket, agent_membership): resp = api_client_agent.get("/api/v1/helpdesk/tickets/") assert resp.status_code == 200 assert "total" in resp.data assert "results" in resp.data @pytest.mark.django_db def test_get_ticket_detail(api_client_agent, ticket): resp = api_client_agent.get(f"/api/v1/helpdesk/tickets/{ticket.pk}/") assert resp.status_code == 200 assert resp.data["subject"] == "Test ticket" @pytest.mark.django_db def test_patch_ticket_priority(api_client_agent, ticket): resp = api_client_agent.patch( f"/api/v1/helpdesk/tickets/{ticket.pk}/", data={"priority": "urgent"}, format="json", ) assert resp.status_code == 200 assert resp.data["priority"] == "urgent" @pytest.mark.django_db def test_ticket_transition_via_api(api_client_agent, ticket): resp = api_client_agent.post( f"/api/v1/helpdesk/tickets/{ticket.pk}/transition/", data={"status": "open"}, format="json", ) assert resp.status_code == 200 assert resp.data["status"] == "open" @pytest.mark.django_db def test_ticket_assign_via_api(api_client_agent, ticket, requester, all_perms): resp = api_client_agent.post( f"/api/v1/helpdesk/tickets/{ticket.pk}/assign/", data={"assigned_to_id": requester.pk}, format="json", ) assert resp.status_code == 200 assert resp.data["assigned_to_id"] == requester.pk # --------------------------------------------------------------------------- # [API] 29-31 — replies # --------------------------------------------------------------------------- @pytest.mark.django_db def test_list_replies_public_only_for_portal_user(ticket, requester, tenant, org_node, all_perms): """Portal user (no TICKET_CHANGE perm) should see only public replies.""" from rest_framework.test import APIClient # Create internal note TicketService.add_reply(ticket, body="secret", is_internal=True, author_id=requester.pk, actor=requester) # Create public reply TicketService.add_reply(ticket, body="public", is_internal=False, author_id=requester.pk, actor=requester) # Make ticket belong to requester ticket.requester = requester ticket.save(update_fields=["requester"]) # Portal role: no TICKET_CHANGE perm portal_role = Role.objects.create(tenant=tenant, code="portal-user", name="Portal") m = Membership.objects.create(tenant=tenant, organization_node=org_node, role=portal_role) m.users.add(requester) client = APIClient() client.force_authenticate(user=requester) client.credentials(HTTP_X_TENANT=tenant.slug) resp = client.get(f"/api/v1/helpdesk/tickets/{ticket.pk}/replies/") assert resp.status_code == 200 bodies = [r["body"] for r in resp.data] assert "public" in bodies assert "secret" not in bodies @pytest.mark.django_db def test_add_reply_via_api(api_client_agent, ticket): resp = api_client_agent.post( f"/api/v1/helpdesk/tickets/{ticket.pk}/replies/", data={"body": "Here is the answer.", "is_internal": False}, format="json", ) assert resp.status_code == 201 assert resp.data["body"] == "Here is the answer." assert resp.data["is_internal"] is False @pytest.mark.django_db def test_internal_note_blocked_for_non_agent(ticket, requester, tenant, org_node): """Non-agent cannot post internal notes.""" from rest_framework.test import APIClient # Assign ticket to requester ticket.requester = requester ticket.save(update_fields=["requester"]) portal_role = Role.objects.create(tenant=tenant, code="portal-note-test", name="Portal Note") m = Membership.objects.create(tenant=tenant, organization_node=org_node, role=portal_role) m.users.add(requester) client = APIClient() client.force_authenticate(user=requester) client.credentials(HTTP_X_TENANT=tenant.slug) resp = client.post( f"/api/v1/helpdesk/tickets/{ticket.pk}/replies/", data={"body": "Trying to post internal note.", "is_internal": True}, format="json", ) assert resp.status_code == 403 # --------------------------------------------------------------------------- # [API] 32-35 — tags, SLA, categories # --------------------------------------------------------------------------- @pytest.mark.django_db def test_list_tags(api_client_agent, tag): resp = api_client_agent.get("/api/v1/helpdesk/tags/") assert resp.status_code == 200 names = [t["name"] for t in resp.data] assert "Bug" in names @pytest.mark.django_db def test_create_tag(api_client_agent, tenant, org_node): resp = api_client_agent.post( "/api/v1/helpdesk/tags/", data={"name": "Feature", "color": "#0f0"}, format="json", ) assert resp.status_code == 201 assert resp.data["name"] == "Feature" @pytest.mark.django_db def test_list_sla_policies(api_client_agent, sla): resp = api_client_agent.get("/api/v1/helpdesk/sla-policies/") assert resp.status_code == 200 names = [p["name"] for p in resp.data] assert "Standard" in names @pytest.mark.django_db def test_list_categories(api_client_agent, category): resp = api_client_agent.get("/api/v1/helpdesk/categories/") assert resp.status_code == 200 names = [c["name"] for c in resp.data] assert "Billing" in names # --------------------------------------------------------------------------- # [API] 36-37 — reports/summary # --------------------------------------------------------------------------- @pytest.mark.django_db def test_reports_summary_structure(api_client_agent, queue): resp = api_client_agent.get("/api/v1/helpdesk/reports/summary/") assert resp.status_code == 200 assert "queues" in resp.data assert "totals" in resp.data assert "total" in resp.data["totals"] @pytest.mark.django_db def test_reports_summary_counts(api_client_agent, queue, ticket, agent): """After transitioning ticket to solved, the count should reflect correctly.""" TicketService.transition(ticket, new_status="open", actor_id=agent.pk, actor=agent) TicketService.transition(ticket, new_status="solved", actor_id=agent.pk, actor=agent) resp = api_client_agent.get("/api/v1/helpdesk/reports/summary/") assert resp.status_code == 200 totals = resp.data["totals"] assert totals["solved"] >= 1 # --------------------------------------------------------------------------- # [permissions] 38 # --------------------------------------------------------------------------- @pytest.mark.django_db def test_helpdesk_permissions_registered(all_perms): # Verify IAM registry permissions (3-part convention: module.resource.action) iam_perms = [ "helpdesk.ticket.view", "helpdesk.ticket.add", "helpdesk.ticket.change", "helpdesk.queue.view", "helpdesk.queue.change", "helpdesk.tag.view", "helpdesk.tag.change", "helpdesk.sla.view", "helpdesk.sla.change", "helpdesk.reply.view", "helpdesk.reply.add", "helpdesk.report.view", "helpdesk.automation.view", "helpdesk.aisuggestion.view", "helpdesk.aiaction.execute", "helpdesk.aiaudit.view", ] for perm_name in iam_perms: assert perm_name in all_perms, f"Permission {perm_name} not in registry" # =========================================================================== # [services] 39 — TicketService.escalate # =========================================================================== @pytest.mark.django_db def test_ticket_escalate(ticket, agent, requester): TicketService.escalate( ticket, escalated_to_id=requester.pk, reason="Needs specialist review", actor_id=agent.pk, actor=agent, ) ticket.refresh_from_db() assert ticket.escalated_to_id == requester.pk assert ticket.escalated_at is not None assert ticket.escalation_reason == "Needs specialist review" # =========================================================================== # [services] 40 — TicketService.update # =========================================================================== @pytest.mark.django_db def test_ticket_service_update(ticket, agent): TicketService.update( ticket, subject="Updated subject", priority="high", actor_id=agent.pk, actor=agent, ) ticket.refresh_from_db() assert ticket.subject == "Updated subject" assert ticket.priority == "high" # =========================================================================== # [services] 41 — TicketService.update ignores read-only fields # =========================================================================== @pytest.mark.django_db def test_ticket_update_ignores_readonly(ticket, agent): original_status = ticket.status TicketService.update( ticket, status="closed", # not in allowed set assigned_to_id=999, actor_id=agent.pk, actor=agent, ) ticket.refresh_from_db() assert ticket.status == original_status # unchanged # =========================================================================== # [services] 42 — TicketService.create quota enforcement # =========================================================================== @pytest.mark.django_db def test_ticket_create_quota_enforcement(tenant, org_node, queue, agent, monkeypatch): """When helpdesk.max_tickets is set to 1 and a ticket exists, creating another should raise ValueError.""" from simorgh.apps.helpdesk.models import Ticket # Create one ticket TicketService.create( tenant_id=tenant.pk, organization_node_id=org_node.pk, queue_id=queue.pk, subject="First ticket", requester_id=agent.pk, created_by_id=agent.pk, ) # Monkey-patch the setting resolver to return 1 def _resolve(key, tenant_id=None): if key == "helpdesk.max_tickets": return 1 return 0 import simorgh.apps.app_settings.services as svc_mod monkeypatch.setattr(svc_mod, "resolve", _resolve) with pytest.raises(ValueError, match="quota"): TicketService.create( tenant_id=tenant.pk, organization_node_id=org_node.pk, queue_id=queue.pk, subject="Second ticket", requester_id=agent.pk, created_by_id=agent.pk, ) # =========================================================================== # [services] 43 — SLA timers: start_sla_timers # =========================================================================== @pytest.mark.django_db def test_sla_start_timers(ticket, sla): ticket.sla_policy = sla ticket.save(update_fields=["sla_policy"]) TicketService.start_sla_timers(ticket_id=ticket.pk) ticket.refresh_from_db() assert ticket.first_response_due_at is not None assert ticket.resolution_due_at is not None from simorgh.apps.helpdesk.models import SLATimer timers = SLATimer.objects.filter(ticket=ticket) assert timers.count() == 2 # =========================================================================== # [services] 44 — SLA timers: mark_first_response # =========================================================================== @pytest.mark.django_db def test_sla_mark_first_response(ticket, sla): ticket.sla_policy = sla ticket.save(update_fields=["sla_policy"]) TicketService.start_sla_timers(ticket_id=ticket.pk) TicketService.mark_first_response(ticket_id=ticket.pk) ticket.refresh_from_db() assert ticket.first_responded_at is not None from simorgh.apps.helpdesk.models import SLABreachType, SLATimer timer = SLATimer.objects.get( ticket=ticket, breach_type=SLABreachType.FIRST_RESPONSE ) assert timer.is_resolved is True assert timer.resolved_at is not None # =========================================================================== # [services] 45 — SLA timers: pause on on_hold # =========================================================================== @pytest.mark.django_db def test_sla_pause_on_hold(ticket, sla): ticket.sla_policy = sla ticket.save(update_fields=["sla_policy"]) TicketService.start_sla_timers(ticket_id=ticket.pk) TicketService.handle_sla_status_change( ticket_id=ticket.pk, new_status="on_hold" ) from simorgh.apps.helpdesk.models import SLATimer timers = SLATimer.objects.filter(ticket=ticket, is_paused=True) assert timers.count() == 2 # =========================================================================== # [services] 46 — SLA timers: resume from on_hold # =========================================================================== @pytest.mark.django_db def test_sla_resume_from_hold(ticket, sla): ticket.sla_policy = sla ticket.save(update_fields=["sla_policy"]) TicketService.start_sla_timers(ticket_id=ticket.pk) # Pause TicketService.handle_sla_status_change( ticket_id=ticket.pk, new_status="on_hold" ) from simorgh.apps.helpdesk.models import SLATimer assert SLATimer.objects.filter(ticket=ticket, is_paused=True).count() == 2 # Resume TicketService.handle_sla_status_change( ticket_id=ticket.pk, new_status="open" ) assert SLATimer.objects.filter(ticket=ticket, is_paused=True).count() == 0 # =========================================================================== # [services] 47 — transition idempotent (same status) # =========================================================================== @pytest.mark.django_db def test_transition_same_status_noop(ticket, agent): result = TicketService.transition( ticket, new_status="new", actor_id=agent.pk, actor=agent ) assert result.status == "new" # unchanged, no error # =========================================================================== # [services] 48 — add_reply auto-transitions new → open # =========================================================================== @pytest.mark.django_db def test_add_public_reply_auto_opens_ticket(ticket, agent): assert ticket.status == "new" TicketService.add_reply( ticket, body="Public response from agent", is_internal=False, author_id=agent.pk, actor=agent, ) ticket.refresh_from_db() assert ticket.status == "open" # =========================================================================== # [isolation] 49 — Cross-tenant ticket access blocked (model level) # =========================================================================== @pytest.mark.django_db def test_cross_tenant_ticket_access_blocked(ticket, tenant, queue, agent): """Tickets are scoped to tenant — queryset filters by tenant.""" # Create a second tenant tenant2 = Tenant.objects.create(slug="tenant-b", name="Tenant B") org2 = create_node(tenant_id=tenant2.pk, name="Org B") queue2 = QueueService.create( tenant_id=tenant2.pk, organization_node_id=org2.pk, name="Support B", is_default=True, ) t2 = TicketService.create( tenant_id=tenant2.pk, organization_node_id=org2.pk, queue_id=queue2.pk, subject="Tenant B ticket", requester_id=agent.pk, created_by_id=agent.pk, ) # ticket belongs to tenant A, t2 belongs to tenant B # Ticket from tenant A should not appear in tenant B's queryset tickets_b = Ticket.objects.filter(tenant_id=tenant2.pk) assert t2.pk in [t.pk for t in tickets_b] assert ticket.pk not in [t.pk for t in tickets_b] # verify the ticket's own tenant is correct assert ticket.tenant_id == tenant.pk assert t2.tenant_id == tenant2.pk # =========================================================================== # [isolation] 50 — Tenant-scoped ticket listing (model level) # =========================================================================== @pytest.mark.django_db def test_tenant_scoped_ticket_listing(ticket, tenant, org_node, queue): """Tickets from other tenants must not appear in tenant-scoped queryset.""" # Create a second tenant with tickets tenant2 = Tenant.objects.create(slug="other-tenant", name="Other") org2 = create_node(tenant_id=tenant2.pk, name="Other Org") queue2 = QueueService.create( tenant_id=tenant2.pk, organization_node_id=org2.pk, name="Queue B", is_default=True, ) # Create ticket in tenant2 t2 = Ticket.objects.create( tenant=tenant2, organization_node=org2, queue=queue2, subject="Other ticket", ) # ticket belongs to tenant A, t2 to tenant B # Scoped queryset should only return tickets for the right tenant assert Ticket.objects.filter(tenant_id=tenant.pk).filter(pk=ticket.pk).exists() assert not Ticket.objects.filter(tenant_id=tenant.pk).filter(pk=t2.pk).exists() assert Ticket.objects.filter(tenant_id=tenant2.pk).filter(pk=t2.pk).exists() assert not Ticket.objects.filter(tenant_id=tenant2.pk).filter(pk=ticket.pk).exists() # =========================================================================== # [isolation] 51 — Cross-tenant queue access blocked (model level) # =========================================================================== @pytest.mark.django_db def test_cross_tenant_queue_access_blocked(queue, tenant, org_node): """Queues from tenant A should not be visible in tenant B's queryset.""" tenant2 = Tenant.objects.create(slug="tenant-c", name="Tenant C") org2 = create_node(tenant_id=tenant2.pk, name="Org C") queue2 = QueueService.create( tenant_id=tenant2.pk, organization_node_id=org2.pk, name="Queue C", is_default=True, ) # queue belongs to tenant A, queue2 belongs to tenant C assert Queue.objects.filter(tenant_id=tenant.pk).filter(pk=queue.pk).exists() assert not Queue.objects.filter(tenant_id=tenant.pk).filter(pk=queue2.pk).exists() assert Queue.objects.filter(tenant_id=tenant2.pk).filter(pk=queue2.pk).exists() assert not Queue.objects.filter(tenant_id=tenant2.pk).filter(pk=queue.pk).exists() # =========================================================================== # [events] 52 — Event subscribers registered # =========================================================================== @pytest.mark.django_db def test_helpdesk_events_registered(): """Verify all expected helpdesk events are registered in the event bus.""" from simorgh.apps.helpdesk.events import ( TICKET_CREATED, TICKET_UPDATED, TICKET_STATUS_CHANGED, TICKET_ASSIGNED, TICKET_ESCALATED, TICKET_SOLVED, TICKET_CLOSED, TICKET_REOPENED, REPLY_ADDED, SLA_BREACHED, AUTOMATION_FIRED, CSAT_SURVEY_DUE, CSAT_RECEIVED, ) # EventSpec objects have a 'name' attribute with the event string events = [ TICKET_CREATED, TICKET_UPDATED, TICKET_STATUS_CHANGED, TICKET_ASSIGNED, TICKET_ESCALATED, TICKET_SOLVED, TICKET_CLOSED, TICKET_REOPENED, REPLY_ADDED, SLA_BREACHED, AUTOMATION_FIRED, CSAT_SURVEY_DUE, CSAT_RECEIVED, ] for event_spec in events: name = getattr(event_spec, "name", event_spec) assert isinstance(name, str), f"Event {event_spec!r} has no string name" assert name.startswith("helpdesk."), f"Event {name} missing helpdesk prefix" # =========================================================================== # [events] 53 — ticket.created triggers SLA timer # =========================================================================== @pytest.mark.django_db def test_ticket_created_dispatches_sla_timer(ticket, sla): """Creating a ticket with an SLA policy should automatically start timers.""" ticket.sla_policy = sla ticket.save(update_fields=["sla_policy"]) # Simulate what the event subscriber would do from simorgh.apps.helpdesk.services import TicketService TicketService.start_sla_timers(ticket_id=ticket.pk) from simorgh.apps.helpdesk.models import SLATimer assert SLATimer.objects.filter(ticket=ticket).count() == 2 # =========================================================================== # [automations] 54 — Automation model CRUD # =========================================================================== @pytest.mark.django_db def test_automation_model_create(tenant, org_node): from simorgh.apps.helpdesk.models import Automation, AutomationTrigger a = Automation.objects.create( tenant=tenant, organization_node=org_node, name="Auto-escalate critical", trigger=AutomationTrigger.TICKET_CREATED, conditions=[{"field": "priority", "op": "eq", "value": "critical"}], actions=[{"type": "escalate", "user_id": 1}], is_active=True, ) assert a.pk is not None assert a.trigger == AutomationTrigger.TICKET_CREATED assert len(a.conditions) == 1 assert a.is_active is True @pytest.mark.django_db def test_automation_soft_delete(tenant, org_node): from simorgh.apps.helpdesk.models import Automation, AutomationTrigger a = Automation.objects.create( tenant=tenant, organization_node=org_node, name="Test automation", trigger=AutomationTrigger.TIME_ELAPSED, ) a.delete() a.refresh_from_db() assert a.is_deleted is True # =========================================================================== # [teams] 55 — AgentTeam model with members # =========================================================================== @pytest.mark.django_db def test_agent_team_with_members(tenant, org_node, agent, requester): from simorgh.apps.helpdesk.models import AgentTeam, AgentTeamMember, AgentTeamRole team = AgentTeam.objects.create( tenant=tenant, organization_node=org_node, name="Technical Support", description="Handles tech issues", is_active=True, ) assert team.pk is not None m1 = AgentTeamMember.objects.create( team=team, user=agent, role=AgentTeamRole.LEAD, added_by=agent, ) m2 = AgentTeamMember.objects.create( team=team, user=requester, role=AgentTeamRole.MEMBER, added_by=agent, ) assert team.memberships.count() == 2 assert str(m1) == f"{agent} → {team} (lead)" assert m2.role == AgentTeamRole.MEMBER # =========================================================================== # [teams] 56 — AgentTeam member unique constraint # =========================================================================== @pytest.mark.django_db def test_agent_team_member_unique(tenant, org_node, agent): from simorgh.apps.helpdesk.models import AgentTeam, AgentTeamMember team = AgentTeam.objects.create( tenant=tenant, organization_node=org_node, name="Unique Test Team", ) AgentTeamMember.objects.create(team=team, user=agent, added_by=agent) with pytest.raises(Exception): AgentTeamMember.objects.create(team=team, user=agent, added_by=agent)