102 lines
3.3 KiB
Python
102 lines
3.3 KiB
Python
from __future__ import annotations
|
|
|
|
import unittest
|
|
from datetime import UTC, datetime
|
|
|
|
from govoplan_core.core.modules import ModuleContext, ModuleManifest
|
|
from govoplan_core.core.registry import PlatformRegistry, RegistryError
|
|
from govoplan_core.core.tasks import (
|
|
WorkAssignmentRef,
|
|
WorkItem,
|
|
WorkItemPage,
|
|
WorkItemProviderRegistration,
|
|
WorkItemQuery,
|
|
WorkSourceRef,
|
|
)
|
|
|
|
|
|
class _Provider:
|
|
def list_items(self, session, principal, *, query):
|
|
del session, principal
|
|
return WorkItemPage(
|
|
items=(
|
|
WorkItem(
|
|
id="task-1",
|
|
provider_id="example.items",
|
|
owner_module="example",
|
|
tenant_id=query.tenant_id,
|
|
title="Review request",
|
|
assignments=(WorkAssignmentRef(kind="account", id="account-1"),),
|
|
sources=(
|
|
WorkSourceRef(
|
|
module_id="cases",
|
|
resource_type="case",
|
|
resource_id="case-1",
|
|
),
|
|
),
|
|
due_at=datetime(2026, 8, 7, tzinfo=UTC),
|
|
),
|
|
),
|
|
total=1,
|
|
)
|
|
|
|
|
|
class WorkItemContractTests(unittest.TestCase):
|
|
def test_provider_registration_is_lazy_and_stable(self) -> None:
|
|
registry = PlatformRegistry()
|
|
registry.register(
|
|
ModuleManifest(
|
|
id="example",
|
|
name="Example",
|
|
version="1",
|
|
work_item_providers=(
|
|
WorkItemProviderRegistration(
|
|
id="example.items",
|
|
factory=lambda _context: _Provider(),
|
|
),
|
|
),
|
|
)
|
|
)
|
|
registry.validate()
|
|
registry.configure_capability_context(
|
|
ModuleContext(registry=registry, settings=object())
|
|
)
|
|
registered, provider = registry.work_item_providers()[0]
|
|
self.assertEqual("example.items", registered.registration.id)
|
|
page = provider.list_items(
|
|
object(),
|
|
object(),
|
|
query=WorkItemQuery(tenant_id="tenant-1"),
|
|
)
|
|
self.assertEqual("Review request", page.items[0].title)
|
|
|
|
def test_provider_ids_must_be_namespaced(self) -> None:
|
|
registry = PlatformRegistry()
|
|
registry.register(
|
|
ModuleManifest(
|
|
id="example",
|
|
name="Example",
|
|
version="1",
|
|
work_item_providers=(
|
|
WorkItemProviderRegistration(
|
|
id="items",
|
|
factory=lambda _context: _Provider(),
|
|
),
|
|
),
|
|
)
|
|
)
|
|
with self.assertRaisesRegex(RegistryError, "namespaced"):
|
|
registry.validate()
|
|
|
|
def test_contract_rejects_unbounded_or_incomplete_values(self) -> None:
|
|
with self.assertRaisesRegex(ValueError, "tenant"):
|
|
WorkItemQuery(tenant_id="")
|
|
with self.assertRaisesRegex(ValueError, "assignment"):
|
|
WorkAssignmentRef(kind="account", id="")
|
|
with self.assertRaisesRegex(ValueError, "module"):
|
|
WorkSourceRef(module_id="", resource_type="case", resource_id="1")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|