95 lines
2.9 KiB
Python
95 lines
2.9 KiB
Python
from __future__ import annotations
|
|
|
|
from datetime import UTC, datetime
|
|
import unittest
|
|
|
|
from govoplan_core.core.tickets import (
|
|
CAPABILITY_TICKET_CASE_ESCALATION,
|
|
CAPABILITY_TICKET_ROUTING,
|
|
TicketCaseEscalationCommand,
|
|
TicketCaseEscalationResult,
|
|
TicketRoutingPlan,
|
|
TicketRoutingRequest,
|
|
ticket_case_escalation_provider,
|
|
ticket_routing_provider,
|
|
)
|
|
|
|
|
|
class _Provider:
|
|
def route_ticket(self, session, principal, *, request):
|
|
del session, principal, request
|
|
return TicketRoutingPlan(provider_id="helpdesk", queue_ref="citizen-service")
|
|
|
|
def escalate_ticket(self, session, principal, *, command):
|
|
del session, principal, command
|
|
return TicketCaseEscalationResult(
|
|
provider_id="cases",
|
|
case_id="case-1",
|
|
case_number="CASE-1",
|
|
case_url="/cases/case-1",
|
|
)
|
|
|
|
|
|
class _Registry:
|
|
def __init__(self, capabilities):
|
|
self.capabilities = capabilities
|
|
|
|
def has_capability(self, name):
|
|
return name in self.capabilities
|
|
|
|
def capability(self, name):
|
|
return self.capabilities[name]
|
|
|
|
|
|
class TicketContractTests(unittest.TestCase):
|
|
def test_optional_providers_fail_open_when_absent(self) -> None:
|
|
registry = _Registry({})
|
|
self.assertIsNone(ticket_routing_provider(registry))
|
|
self.assertIsNone(ticket_case_escalation_provider(registry))
|
|
|
|
def test_optional_providers_resolve_structurally(self) -> None:
|
|
provider = _Provider()
|
|
registry = _Registry(
|
|
{
|
|
CAPABILITY_TICKET_ROUTING: provider,
|
|
CAPABILITY_TICKET_CASE_ESCALATION: provider,
|
|
}
|
|
)
|
|
self.assertIs(provider, ticket_routing_provider(registry))
|
|
self.assertIs(provider, ticket_case_escalation_provider(registry))
|
|
|
|
def test_commands_validate_tenant_time_and_relative_case_link(self) -> None:
|
|
instant = datetime(2026, 8, 22, 9, 0, tzinfo=UTC)
|
|
request = TicketRoutingRequest(
|
|
tenant_id="tenant-1",
|
|
ticket_id="ticket-1",
|
|
ticket_type="request",
|
|
priority="normal",
|
|
title="Broken streetlight",
|
|
received_at=instant,
|
|
)
|
|
self.assertEqual("ticket-1", request.ticket_id)
|
|
|
|
command = TicketCaseEscalationCommand(
|
|
tenant_id="tenant-1",
|
|
ticket_id="ticket-1",
|
|
ticket_number="TKT-1",
|
|
title="Broken streetlight",
|
|
case_type_key="service-request",
|
|
occurred_at=instant,
|
|
idempotency_key="escalation-1",
|
|
)
|
|
self.assertEqual("service-request", command.case_type_key)
|
|
|
|
with self.assertRaises(ValueError):
|
|
TicketCaseEscalationResult(
|
|
provider_id="cases",
|
|
case_id="case-1",
|
|
case_number="CASE-1",
|
|
case_url="https://other.example/cases/1",
|
|
)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|