feat(core): require scoped user workflows
This commit is contained in:
@@ -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
|
||||
|
||||
@@ -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:
|
||||
|
||||
85
tests/test_documentation_topic_contract.py
Normal file
85
tests/test_documentation_topic_contract.py
Normal file
@@ -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()
|
||||
Reference in New Issue
Block a user