Add shared work-item provider contracts

2026-08-06 16:06:17 +02:00
parent b5f5be15f6
commit d65d7a8e5f
8 changed files with 844 additions and 140 deletions
@@ -0,0 +1,101 @@
+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()