GovOPlaN/govoplan-helpdesk
2
0
Fork 0
Code Issues 1 Pull Requests Actions Packages Projects Releases Wiki Activity

feat(helpdesk): configure ticket routing profiles

Browse Source
This commit is contained in:
zemion
2026-08-22 11:41:45 +02:00
parent 064ec6158d
commit 32bd55d5bd
24 changed files with 1584 additions and 102 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/test_helpdesk_routing.py
+155
View File
@@ -0,0 +1,155 @@
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()
tests/test_manifest.py
+21 -20
View File
@@ -2,31 +2,32 @@ from __future__ import annotations
import unittest
from govoplan_core.core.tickets import CAPABILITY_TICKET_ROUTING
from govoplan_helpdesk.backend.dsar_provider import HELPDESK_DSAR_CAPABILITY
from govoplan_helpdesk.backend.manifest import ADMIN_SCOPE, READ_SCOPE, WRITE_SCOPE, get_manifest
class ManifestSeedTests(unittest.TestCase):
def test_manifest_registers_seed_contract(self) -> None:
class ManifestTests(unittest.TestCase):
def test_manifest_registers_helpdesk_policy_overlay(self) -> None:
manifest = get_manifest()
self.assertEqual("helpdesk", manifest.id)
self.assertEqual("0.1.20", manifest.version)
self.assertEqual({READ_SCOPE, WRITE_SCOPE, ADMIN_SCOPE}, {item.scope for item in manifest.permissions})
self.assertEqual({"helpdesk_manager", "helpdesk_viewer", "helpdesk_administrator"}, {item.slug for item in manifest.role_templates})
self.assertIsNotNone(manifest.route_factory)
self.assertIsNotNone(manifest.migration_spec)
self.assertIsNotNone(manifest.frontend)
self.assertIn(CAPABILITY_TICKET_ROUTING, manifest.capability_factories)
self.assertIn(HELPDESK_DSAR_CAPABILITY, manifest.capability_factories)
self.assertEqual("vertical_slice", manifest.architecture.maturity)
self.assertEqual(manifest.id, "helpdesk")
self.assertEqual(manifest.name, "Helpdesk")
self.assertEqual(manifest.dependencies, ("access",))
self.assertEqual({permission.scope for permission in manifest.permissions}, {READ_SCOPE, WRITE_SCOPE, ADMIN_SCOPE})
self.assertEqual({role.slug for role in manifest.role_templates}, {"helpdesk_manager", "helpdesk_viewer"})
self.assertTrue(manifest.documentation)
topic = manifest.documentation[0]
self.assertEqual("reference", topic.metadata["kind"])
self.assertIn("seed_boundary", topic.metadata["consequence_classes"])
self.assertTrue(
all(
topic.translations.get("de", {}).get(field)
for field in ("title", "summary", "body")
)
)
self.assertIsNone(manifest.route_factory)
self.assertIsNone(manifest.migration_spec)
self.assertIsNone(manifest.frontend)
def test_documentation_is_complete_in_german_and_denies_duplicate_ticket_storage(self) -> None:
topics = get_manifest().documentation
for topic in topics:
self.assertTrue(all(topic.translations.get("de", {}).get(field) for field in ("title", "summary", "body")), topic.id)
boundary = next(item for item in topics if item.id == "helpdesk.module-boundary")
self.assertIn("never creates a parallel ticket store", boundary.body)
self.assertTrue(any(item.metadata.get("kind") == "workflow" for item in topics))
if __name__ == "__main__":