From a18499cbb5cdd4c2d6322fa2dcf86a58ef5e100c Mon Sep 17 00:00:00 2001 From: Albrecht Degering Date: Wed, 22 Jul 2026 01:55:24 +0200 Subject: [PATCH] feat(core): require scoped user workflows --- src/govoplan_core/core/modules.py | 29 ++++++++ src/govoplan_core/core/registry.py | 6 ++ tests/test_documentation_topic_contract.py | 85 ++++++++++++++++++++++ 3 files changed, 120 insertions(+) create mode 100644 tests/test_documentation_topic_contract.py diff --git a/src/govoplan_core/core/modules.py b/src/govoplan_core/core/modules.py index d9c8ba1..bc777b9 100644 --- a/src/govoplan_core/core/modules.py +++ b/src/govoplan_core/core/modules.py @@ -237,6 +237,35 @@ class DocumentationTopic: metadata: Mapping[str, Any] = field(default_factory=dict) +def user_workflow_scope_condition_issues(topic: DocumentationTopic) -> tuple[str, ...]: + """Return fail-closed authoring issues for a user-facing workflow topic. + + Topic conditions are alternatives. Every alternative therefore needs an + explicit scope constraint; otherwise one unscoped alternative would make + the workflow visible regardless of the actor's task authority. + """ + + raw_kind = topic.metadata.get("kind") + kind = raw_kind.strip().lower().replace("_", "-") if isinstance(raw_kind, str) else "" + if kind != "workflow" or "user" not in topic.documentation_types: + return () + if not topic.conditions: + return ("user workflow topics must declare at least one scope-conditioned alternative",) + + unscoped_alternatives = tuple( + index + for index, condition in enumerate(topic.conditions, start=1) + if not any(scope.strip() for scope in (*condition.required_scopes, *condition.any_scopes)) + ) + if not unscoped_alternatives: + return () + alternatives = ", ".join(str(index) for index in unscoped_alternatives) + return ( + "every user workflow condition alternative must declare required_scopes or any_scopes; " + f"unscoped alternative(s): {alternatives}", + ) + + @dataclass(frozen=True, slots=True) class DocumentationContext: registry: object diff --git a/src/govoplan_core/core/registry.py b/src/govoplan_core/core/registry.py index c160727..6a81587 100644 --- a/src/govoplan_core/core/registry.py +++ b/src/govoplan_core/core/registry.py @@ -21,6 +21,7 @@ from govoplan_core.core.modules import ( SUPPORTED_FRONTEND_ASSET_MANIFEST_CONTRACT_VERSION, SUPPORTED_MANIFEST_CONTRACT_VERSION, TenantSummaryProvider, + user_workflow_scope_condition_issues, ) from govoplan_core.core.versioning import format_version_range, version_range_is_valid, version_satisfies_range @@ -369,6 +370,11 @@ def _validate_manifest_shape(manifest: ModuleManifest) -> None: _validate_manifest_frontend(manifest) for item in manifest.nav_items: _validate_nav_item(manifest.id, item) + for topic in manifest.documentation: + for issue in user_workflow_scope_condition_issues(topic): + raise RegistryError( + f"Module {manifest.id!r} documentation topic {topic.id!r}: {issue}" + ) def _validate_manifest_identity(manifest: ModuleManifest) -> None: diff --git a/tests/test_documentation_topic_contract.py b/tests/test_documentation_topic_contract.py new file mode 100644 index 0000000..4c8656e --- /dev/null +++ b/tests/test_documentation_topic_contract.py @@ -0,0 +1,85 @@ +from __future__ import annotations + +import unittest + +from govoplan_core.core.modules import ( + DocumentationCondition, + DocumentationTopic, + ModuleManifest, + user_workflow_scope_condition_issues, +) +from govoplan_core.core.registry import PlatformRegistry, RegistryError + + +def workflow_topic( + *, + conditions: tuple[DocumentationCondition, ...], + documentation_types: tuple[str, ...] = ("user",), + kind: str = "workflow", +) -> DocumentationTopic: + return DocumentationTopic( + id="example.workflow.task", + title="Complete a task", + summary="Complete the example task.", + documentation_types=documentation_types, # type: ignore[arg-type] + conditions=conditions, + metadata={"kind": kind}, + ) + + +class DocumentationTopicContractTests(unittest.TestCase): + def test_user_workflow_requires_scope_conditions(self) -> None: + topic = workflow_topic(conditions=()) + + self.assertEqual( + user_workflow_scope_condition_issues(topic), + ("user workflow topics must declare at least one scope-conditioned alternative",), + ) + with self.assertRaisesRegex(RegistryError, "scope-conditioned alternative"): + registry_for(topic).validate() + + def test_every_condition_alternative_must_be_scope_conditioned(self) -> None: + topic = workflow_topic( + conditions=( + DocumentationCondition(required_scopes=("example:item:read",)), + DocumentationCondition(required_modules=("example",)), + ) + ) + + with self.assertRaisesRegex(RegistryError, r"unscoped alternative\(s\): 2"): + registry_for(topic).validate() + + def test_scoped_user_workflow_and_non_user_topics_are_accepted(self) -> None: + scoped = workflow_topic( + conditions=( + DocumentationCondition(required_scopes=("example:item:read",)), + DocumentationCondition(any_scopes=("example:item:write", "example:item:admin")), + ) + ) + admin_workflow = workflow_topic( + conditions=(), + documentation_types=("admin",), + ) + user_reference = workflow_topic(conditions=(), kind="reference") + + self.assertEqual(user_workflow_scope_condition_issues(scoped), ()) + self.assertEqual(user_workflow_scope_condition_issues(admin_workflow), ()) + self.assertEqual(user_workflow_scope_condition_issues(user_reference), ()) + registry_for(scoped, admin_workflow, user_reference).validate() + + +def registry_for(*topics: DocumentationTopic) -> PlatformRegistry: + registry = PlatformRegistry() + registry.register( + ModuleManifest( + id="example", + name="Example", + version="1.0.0", + documentation=topics, + ) + ) + return registry + + +if __name__ == "__main__": + unittest.main()