156 lines
5.4 KiB
Python
156 lines
5.4 KiB
Python
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.tickets import TicketRoutingRequest
|
|
from govoplan_core.db.base import Base
|
|
from govoplan_helpdesk.backend.domain import ServiceProfile
|
|
from govoplan_helpdesk.backend.service import (
|
|
HelpdeskConflictError,
|
|
SqlHelpdeskRoutingProvider,
|
|
list_service_profiles,
|
|
upsert_service_profile,
|
|
)
|
|
|
|
|
|
NOW = datetime(2026, 8, 22, 11, 0, tzinfo=UTC)
|
|
|
|
|
|
class _Principal:
|
|
tenant_id = "tenant-1"
|
|
account_id = "admin-1"
|
|
identity_id = None
|
|
membership_id = "membership-1"
|
|
user = SimpleNamespace(id="admin-1")
|
|
scopes = frozenset({"helpdesk:workspace:admin"})
|
|
|
|
def has(self, scope: str) -> bool:
|
|
return scope in self.scopes
|
|
|
|
|
|
class HelpdeskRoutingTests(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_profile_is_replay_safe_and_routes_ticket_without_storing_it(self) -> None:
|
|
with self.Session() as session:
|
|
profile = _profile()
|
|
created = upsert_service_profile(
|
|
session,
|
|
self.principal,
|
|
profile=profile,
|
|
expected_revision=None,
|
|
idempotency_key="profile-create-1",
|
|
)
|
|
session.commit()
|
|
replay = upsert_service_profile(
|
|
session,
|
|
self.principal,
|
|
profile=profile,
|
|
expected_revision=None,
|
|
idempotency_key="profile-create-1",
|
|
)
|
|
self.assertEqual(created.to_dict(), replay.to_dict())
|
|
self.assertEqual(1, len(list_service_profiles(session, self.principal)))
|
|
|
|
plan = SqlHelpdeskRoutingProvider().route_ticket(
|
|
session,
|
|
self.principal,
|
|
request=TicketRoutingRequest(
|
|
tenant_id="tenant-1",
|
|
ticket_id="ticket-1",
|
|
ticket_type="incident",
|
|
priority="urgent",
|
|
title="Network outage",
|
|
received_at=NOW,
|
|
),
|
|
)
|
|
self.assertEqual("it-operations", plan.queue_ref)
|
|
self.assertEqual(NOW + timedelta(minutes=30), plan.service_target_at)
|
|
self.assertEqual("it-support", plan.metadata["profile_key"])
|
|
self.assertEqual(0, len([table for table in Base.metadata.tables if table == "tickets"]))
|
|
|
|
def test_ordered_matching_no_match_and_revision_conflicts_are_explicit(self) -> None:
|
|
with self.Session() as session:
|
|
upsert_service_profile(
|
|
session,
|
|
self.principal,
|
|
profile=_profile(),
|
|
expected_revision=None,
|
|
idempotency_key="profile-create-2",
|
|
)
|
|
session.commit()
|
|
provider = SqlHelpdeskRoutingProvider()
|
|
unmatched = provider.route_ticket(
|
|
session,
|
|
self.principal,
|
|
request=TicketRoutingRequest(
|
|
tenant_id="tenant-1",
|
|
ticket_id="ticket-2",
|
|
ticket_type="report",
|
|
priority="low",
|
|
title="General feedback",
|
|
received_at=NOW,
|
|
queue_hint="citizen-service",
|
|
),
|
|
)
|
|
self.assertEqual("citizen-service", unmatched.queue_ref)
|
|
self.assertIsNone(unmatched.service_target_at)
|
|
self.assertFalse(unmatched.metadata["matched"])
|
|
|
|
with self.assertRaises(HelpdeskConflictError):
|
|
upsert_service_profile(
|
|
session,
|
|
self.principal,
|
|
profile=ServiceProfile.from_mapping({**_profile().to_dict(), "revision": 2, "label": "Changed"}),
|
|
expected_revision=9,
|
|
idempotency_key="profile-stale",
|
|
)
|
|
|
|
def test_profile_writes_recheck_administration_permission(self) -> None:
|
|
principal = SimpleNamespace(
|
|
tenant_id="tenant-1",
|
|
account_id="viewer-1",
|
|
scopes=frozenset({"helpdesk:workspace:read"}),
|
|
)
|
|
with self.Session() as session, self.assertRaises(PermissionError):
|
|
upsert_service_profile(
|
|
session,
|
|
principal,
|
|
profile=_profile(),
|
|
expected_revision=None,
|
|
idempotency_key="unauthorized-profile-write",
|
|
)
|
|
|
|
|
|
def _profile() -> ServiceProfile:
|
|
return ServiceProfile(
|
|
tenant_id="tenant-1",
|
|
profile_id="profile-1",
|
|
profile_key="it-support",
|
|
revision=1,
|
|
label="IT support",
|
|
description="IT incidents and requests.",
|
|
queue_ref="it-operations",
|
|
ticket_types=("incident", "problem", "request"),
|
|
priorities=("low", "normal", "high", "urgent"),
|
|
target_minutes={"urgent": 30, "high": 240},
|
|
default_target_minutes=1_440,
|
|
active=True,
|
|
sort_order=10,
|
|
recorded_at=NOW,
|
|
change_reason="Created IT support profile.",
|
|
)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|