109 lines
4.4 KiB
Python
109 lines
4.4 KiB
Python
from __future__ import annotations
|
|
|
|
from datetime import UTC, datetime
|
|
from types import SimpleNamespace
|
|
import unittest
|
|
|
|
from sqlalchemy import create_engine
|
|
from sqlalchemy.orm import sessionmaker
|
|
|
|
from govoplan_core.core.tickets import TicketCaseEscalationCommand
|
|
from govoplan_core.db.base import Base
|
|
from govoplan_cases.backend.db import models as _case_models
|
|
from govoplan_cases.backend.service import upsert_case_status, upsert_case_type
|
|
from govoplan_cases.backend.ticket_escalation import TicketCaseEscalationProvider
|
|
|
|
|
|
NOW = datetime(2026, 8, 22, 10, 0, tzinfo=UTC)
|
|
|
|
|
|
class _Principal:
|
|
def __init__(self, tenant_id: str = "tenant-1", *, can_create: bool = True) -> None:
|
|
self.tenant_id = tenant_id
|
|
self.account_id = "case-manager-1"
|
|
self.identity_id = None
|
|
self.membership_id = "membership-1"
|
|
self.group_ids = frozenset()
|
|
self.role_ids = frozenset()
|
|
self.function_assignment_ids = frozenset()
|
|
self.acting_assignment_id = None
|
|
self.scopes = frozenset({"cases:case:create", "cases:case:read"} if can_create else {"cases:case:read"})
|
|
self.user = SimpleNamespace(id=self.account_id)
|
|
|
|
def has(self, scope: str) -> bool:
|
|
return scope in self.scopes
|
|
|
|
|
|
class TicketEscalationTests(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.principal = _Principal()
|
|
|
|
def test_escalation_creates_one_linked_case_and_replays(self) -> None:
|
|
with self.Session() as session:
|
|
upsert_case_status(
|
|
session,
|
|
self.principal,
|
|
status_key="open",
|
|
label="Open",
|
|
)
|
|
upsert_case_type(
|
|
session,
|
|
self.principal,
|
|
type_key="service-request",
|
|
label="Service request",
|
|
initial_status_key="open",
|
|
)
|
|
command = TicketCaseEscalationCommand(
|
|
tenant_id="tenant-1",
|
|
ticket_id="ticket-1",
|
|
ticket_number="TKT-2026-0001",
|
|
title="Broken streetlight",
|
|
case_type_key="service-request",
|
|
occurred_at=NOW,
|
|
idempotency_key="escalation-1",
|
|
handoff_note="Formal procedure required.",
|
|
metadata={"ticket_revision": 3},
|
|
)
|
|
provider = TicketCaseEscalationProvider()
|
|
|
|
created = provider.escalate_ticket(session, self.principal, command=command)
|
|
session.commit()
|
|
replay = provider.escalate_ticket(session, self.principal, command=command)
|
|
|
|
self.assertFalse(created.replayed)
|
|
self.assertTrue(replay.replayed)
|
|
self.assertEqual(created.case_id, replay.case_id)
|
|
self.assertEqual("service-request", created.metadata["case_type_key"])
|
|
self.assertTrue(created.case_url.startswith("/cases/"))
|
|
self.assertEqual(1, session.query(_case_models.CaseIdentity).count())
|
|
current = session.query(_case_models.CaseRecordRevision).filter(
|
|
_case_models.CaseRecordRevision.case_id == created.case_id,
|
|
_case_models.CaseRecordRevision.superseded_at.is_(None),
|
|
).one()
|
|
self.assertEqual("tickets", current.snapshot["context"]["work_item_ref"]["owner_module"])
|
|
self.assertEqual("ticket-1", current.snapshot["metadata"]["source_resource_id"])
|
|
|
|
def test_escalation_fails_closed_for_cross_tenant_or_missing_permission(self) -> None:
|
|
command = TicketCaseEscalationCommand(
|
|
tenant_id="tenant-1",
|
|
ticket_id="ticket-1",
|
|
ticket_number="TKT-1",
|
|
title="Request",
|
|
case_type_key="service-request",
|
|
occurred_at=NOW,
|
|
idempotency_key="escalation-2",
|
|
)
|
|
provider = TicketCaseEscalationProvider()
|
|
with self.Session() as session:
|
|
with self.assertRaises(PermissionError):
|
|
provider.escalate_ticket(session, _Principal("tenant-2"), command=command)
|
|
with self.assertRaises(PermissionError):
|
|
provider.escalate_ticket(session, _Principal(can_create=False), command=command)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|