from __future__ import annotations from datetime import UTC, datetime, timedelta from types import SimpleNamespace import unittest from sqlalchemy import create_engine from sqlalchemy.orm import sessionmaker from govoplan_core.core.events import EventObjectRef, EventTenantRef, PlatformEvent from govoplan_core.core.search import SearchBackfillRequest from govoplan_core.core.tickets import ( CAPABILITY_TICKET_CASE_ESCALATION, CAPABILITY_TICKET_ROUTING, TicketCaseEscalationResult, TicketRoutingPlan, ) from govoplan_core.db.base import Base from govoplan_tickets.backend.db.models import TicketComment, TicketEscalation, TicketHistory from govoplan_tickets.backend.domain import TicketLink, TicketRecord, TicketSubjectRef from govoplan_tickets.backend.search_source import PROVIDER_ID, RESOURCE_TYPE, TicketsSearchSource from govoplan_tickets.backend.service import ( ADMIN_SCOPE, ASSIGN_SCOPE, READ_SCOPE, REPORT_SCOPE, RESOLVE_SCOPE, TRIAGE_SCOPE, TicketConflictError, add_ticket_comment, add_ticket_link, assign_ticket, create_ticket, escalate_ticket_to_case, get_ticket, list_ticket_comments, list_tickets, resolve_ticket, ticket_history, triage_ticket, ) NOW = datetime(2026, 8, 22, 9, 0, tzinfo=UTC) class _Principal: def __init__(self, account_id: str, scopes: set[str], tenant_id: str = "tenant-1") -> None: self.account_id = account_id self.identity_id = None self.membership_id = f"membership-{account_id}" self.tenant_id = tenant_id self.scopes = frozenset(scopes) self.group_ids = frozenset() self.role_ids = frozenset() self.function_assignment_ids = frozenset() self.acting_assignment_id = None self.user = SimpleNamespace(id=account_id) def has(self, scope: str) -> bool: return scope in self.scopes class _Integrations: def route_ticket(self, session, principal, *, request): del session, principal return TicketRoutingPlan( provider_id="helpdesk", queue_ref=request.queue_hint or "citizen-service", service_target_at=request.received_at + timedelta(hours=24), explanation="Matched the default service profile.", ) def escalate_ticket(self, session, principal, *, command): del session, principal return TicketCaseEscalationResult( provider_id="cases", case_id=f"case-{command.ticket_id}", case_number=f"CASE-{command.ticket_number}", case_url=f"/cases/case-{command.ticket_id}", ) class _Registry: def __init__(self, integrations: _Integrations | None = None) -> None: self.capabilities = ( { CAPABILITY_TICKET_ROUTING: integrations, CAPABILITY_TICKET_CASE_ESCALATION: integrations, } if integrations else {} ) def has_capability(self, name): return name in self.capabilities def capability(self, name): return self.capabilities[name] def active_module_ids(self): return ("tickets", "cases", "helpdesk", "search") if self.capabilities else ("tickets",) class TicketServiceTests(unittest.TestCase): def setUp(self) -> None: engine = create_engine("sqlite+pysqlite:///:memory:", future=True) Base.metadata.create_all(engine) self.Session = sessionmaker(bind=engine, expire_on_commit=False) self.manager = _Principal( "manager-1", {READ_SCOPE, REPORT_SCOPE, TRIAGE_SCOPE, ASSIGN_SCOPE, RESOLVE_SCOPE, ADMIN_SCOPE}, ) self.reporter = _Principal("reporter-1", {READ_SCOPE, REPORT_SCOPE}) def test_lifecycle_is_tenant_safe_replay_safe_and_revisioned(self) -> None: registry = _Registry(_Integrations()) with self.Session() as session: created = create_ticket( session, self.reporter, record=_record(reporter="reporter-1"), idempotency_key="report-1", registry=registry, ) session.commit() self.assertEqual("citizen-service", created.queue_ref) self.assertEqual(NOW + timedelta(hours=24), created.service_target_at) replay = create_ticket( session, self.reporter, record=_record(reporter="reporter-1"), idempotency_key="report-1", registry=registry, ) self.assertEqual(created.to_dict(), replay.to_dict()) triaged = triage_ticket( session, self.manager, ticket_id=created.ticket_id, expected_revision=1, changes={"status": "triaged", "priority": "high", "visibility": "restricted"}, recorded_at=NOW + timedelta(minutes=5), change_reason="Classified by the service desk.", idempotency_key="triage-1", registry=registry, ) assigned = assign_ticket( session, self.manager, ticket_id=created.ticket_id, expected_revision=triaged.revision, assignee=TicketSubjectRef(kind="account", id="manager-1", label="Manager"), recorded_at=NOW + timedelta(minutes=10), change_reason="Assigned to the duty manager.", idempotency_key="assign-1", registry=registry, ) resolved = resolve_ticket( session, self.manager, ticket_id=created.ticket_id, expected_revision=assigned.revision, target_status="resolved", resolution_summary="Streetlight repair was commissioned.", recorded_at=NOW + timedelta(hours=2), change_reason="Resolution confirmed.", idempotency_key="resolve-1", registry=registry, ) session.commit() self.assertEqual(4, resolved.revision) self.assertEqual("resolved", resolved.status) self.assertEqual(4, session.query(TicketHistory).count()) outsider = _Principal("outsider", {READ_SCOPE}) self.assertIsNone(get_ticket(session, outsider, ticket_id=created.ticket_id)) self.assertIsNotNone(get_ticket(session, self.reporter, ticket_id=created.ticket_id)) other_tenant = _Principal("manager-1", self.manager.scopes, tenant_id="tenant-2") self.assertIsNone(get_ticket(session, other_tenant, ticket_id=created.ticket_id)) def test_links_comments_history_and_case_escalation_are_auditable(self) -> None: registry = _Registry(_Integrations()) with self.Session() as session: created = create_ticket( session, self.manager, record=_record(reporter="manager-1"), idempotency_key="report-2", registry=registry, ) linked = add_ticket_link( session, self.manager, ticket_id=created.ticket_id, expected_revision=1, link=TicketLink( link_id="file-1", kind="attachment", owner_module="files", resource_type="file", resource_id="file-1", label="Photo", url="/files/file-1", ), recorded_at=NOW + timedelta(minutes=1), change_reason="Attached inspection photo.", idempotency_key="link-1", registry=registry, ) commented, comment = add_ticket_comment( session, self.manager, ticket_id=created.ticket_id, expected_revision=linked.revision, comment_id="comment-1", body="Forwarded to maintenance.", visibility="internal", recorded_at=NOW + timedelta(minutes=2), idempotency_key="comment-op-1", registry=registry, ) escalated, outcome = escalate_ticket_to_case( session, self.manager, ticket_id=created.ticket_id, expected_revision=commented.revision, case_type_key="service-request", occurred_at=NOW + timedelta(minutes=3), handoff_note="Formal procedure required.", idempotency_key="escalate-1", registry=registry, ) session.commit() self.assertEqual("comment-1", comment["comment_id"]) self.assertEqual("case-ticket-1", outcome["case_id"]) self.assertEqual(1, session.query(TicketComment).count()) self.assertEqual(1, session.query(TicketEscalation).count()) self.assertTrue(any(item.kind == "case" for item in escalated.links)) self.assertEqual(4, len(ticket_history(session, self.manager, ticket_id=created.ticket_id))) self.assertEqual(1, len(list_ticket_comments(session, self.manager, ticket_id=created.ticket_id))) replayed, replay_outcome = escalate_ticket_to_case( session, self.manager, ticket_id=created.ticket_id, expected_revision=commented.revision, case_type_key="service-request", occurred_at=NOW + timedelta(minutes=3), handoff_note="Formal procedure required.", idempotency_key="escalate-1", registry=registry, ) self.assertEqual(escalated.revision, replayed.revision) self.assertTrue(replay_outcome["replayed"]) def test_revision_conflicts_and_search_rechecks_fail_closed(self) -> None: with self.Session() as session: created = create_ticket( session, self.manager, record=_record(reporter="manager-1"), idempotency_key="report-3", ) triage_ticket( session, self.manager, ticket_id=created.ticket_id, expected_revision=1, changes={"status": "triaged"}, recorded_at=NOW + timedelta(minutes=1), change_reason="Triaged.", idempotency_key="triage-3", ) session.commit() with self.assertRaises(TicketConflictError): triage_ticket( session, self.manager, ticket_id=created.ticket_id, expected_revision=1, changes={"priority": "urgent"}, recorded_at=NOW + timedelta(minutes=2), change_reason="Stale update.", idempotency_key="triage-stale", ) session.rollback() source = TicketsSearchSource() page = source.backfill( session, request=SearchBackfillRequest( tenant_id="tenant-1", provider_id=PROVIDER_ID, resource_type=RESOURCE_TYPE, rebuild_id="rebuild-1", ), ) self.assertEqual(1, len(page.documents)) event = PlatformEvent( type="tickets.ticket.triaged", module_id="tickets", tenant=EventTenantRef(id="tenant-1"), resource=EventObjectRef(type="ticket", id=created.ticket_id), ) changes = source.index_changes_for_event(session, event=event, delivery_key="delivery-1") self.assertEqual("upsert", changes[0].kind) items, total = list_tickets(session, self.manager) self.assertEqual(1, total) self.assertEqual(created.ticket_id, items[0].ticket_id) def test_report_and_comment_permissions_are_enforced_below_the_router(self) -> None: viewer = _Principal("viewer-1", {READ_SCOPE}) with self.Session() as session: with self.assertRaises(PermissionError): create_ticket( session, viewer, record=_record(reporter="viewer-1"), idempotency_key="viewer-report", ) created = create_ticket( session, self.reporter, record=_record(reporter="reporter-1"), idempotency_key="report-comment-test", ) with self.assertRaises(PermissionError): add_ticket_comment( session, viewer, ticket_id=created.ticket_id, expected_revision=created.revision, comment_id="viewer-comment", body="A read-only viewer must not append a comment.", visibility="external", recorded_at=NOW + timedelta(minutes=1), idempotency_key="viewer-comment-op", ) def _record(*, reporter: str) -> TicketRecord: return TicketRecord( tenant_id="tenant-1", ticket_id="ticket-1", ticket_number="TKT-2026-0001", revision=1, ticket_type="request", priority="normal", status="new", title="Broken streetlight", description="The lamp at the town square is not working.", visibility="restricted", reporter=TicketSubjectRef(kind="account", id=reporter), requester=TicketSubjectRef(kind="account", id=reporter), received_at=NOW, recorded_at=NOW, change_reason="Reported through the authenticated portal.", ) if __name__ == "__main__": unittest.main()