240 lines
10 KiB
Python
240 lines
10 KiB
Python
from __future__ import annotations
|
|
|
|
import unittest
|
|
from inspect import signature
|
|
from types import SimpleNamespace
|
|
from unittest.mock import patch
|
|
|
|
from fastapi import HTTPException
|
|
|
|
from govoplan_access.backend.manifest import get_manifest as get_access_manifest
|
|
from govoplan_core.core.modules import DocumentationCondition, DocumentationLink, DocumentationTopic, ModuleManifest
|
|
from govoplan_core.core.registry import PlatformRegistry
|
|
from govoplan_tenancy.backend.manifest import get_manifest as get_tenancy_manifest
|
|
from govoplan_docs.backend.api.v1.routes import (
|
|
_classify_documentation,
|
|
_condition_visibility,
|
|
_documentation_topic_anchor,
|
|
_documentation_topic_groups,
|
|
docs_context,
|
|
)
|
|
from govoplan_docs.backend.manifest import get_manifest as get_docs_manifest
|
|
|
|
|
|
class FakePrincipal:
|
|
tenant_id = "tenant-1"
|
|
user = SimpleNamespace(id="user-1")
|
|
|
|
def __init__(self, scopes: set[str]) -> None:
|
|
self.scopes = frozenset(scopes)
|
|
|
|
def has(self, required_scope: str) -> bool:
|
|
return required_scope in self.scopes
|
|
|
|
|
|
class DocsContextTests(unittest.TestCase):
|
|
def test_topic_groups_preserve_layer_classification(self) -> None:
|
|
registry = PlatformRegistry()
|
|
registry.register(get_tenancy_manifest())
|
|
registry.register(get_access_manifest())
|
|
registry.register(get_docs_manifest())
|
|
principal = FakePrincipal({"admin:users:read", "admin:groups:read", "docs:documentation:read"})
|
|
|
|
layers = _classify_documentation(
|
|
registry,
|
|
principal,
|
|
settings=None,
|
|
session=None,
|
|
documentation_type="admin",
|
|
locale="en",
|
|
)
|
|
groups = _documentation_topic_groups(layers)
|
|
|
|
workflow_ids = {topic["id"] for topic in groups["workflow"]}
|
|
reference_ids = {topic["id"] for topic in groups["reference"]}
|
|
pattern_ids = {topic["id"] for topic in groups["pattern"]}
|
|
system_ids = {topic["id"] for topic in groups["system"]}
|
|
|
|
self.assertIn("access.workflow.grant-user-access", workflow_ids)
|
|
self.assertIn("access.reference.admin-access-fields", reference_ids)
|
|
self.assertIn("docs.pattern.field-help", pattern_ids)
|
|
self.assertIn("docs.configured-system-documentation", system_ids)
|
|
|
|
configured_ids = {topic["id"] for topic in layers["configured"]}
|
|
always_ids = {topic["id"] for topic in layers["always"]}
|
|
self.assertIn("access.workflow.grant-user-access", configured_ids)
|
|
self.assertIn("docs.pattern.field-help", always_ids)
|
|
|
|
def test_user_projection_omits_topics_without_required_access(self) -> None:
|
|
registry = PlatformRegistry()
|
|
registry.register(get_tenancy_manifest())
|
|
registry.register(get_access_manifest())
|
|
registry.register(get_docs_manifest())
|
|
principal = FakePrincipal({"docs:documentation:read"})
|
|
|
|
layers = _classify_documentation(
|
|
registry,
|
|
principal,
|
|
settings=None,
|
|
session=None,
|
|
documentation_type="user",
|
|
locale="en",
|
|
)
|
|
groups = _documentation_topic_groups(layers)
|
|
workflow_ids = {topic["id"] for topic in groups["workflow"]}
|
|
|
|
self.assertNotIn("access.workflow.grant-user-access", workflow_ids)
|
|
self.assertNotIn("access.workflow.grant-user-access", {topic["id"] for topic in layers["available"]})
|
|
|
|
def test_user_projection_is_a_bounded_safe_whitelist(self) -> None:
|
|
registry = PlatformRegistry()
|
|
registry.register(ModuleManifest(
|
|
id="example",
|
|
name="Example",
|
|
version="9.9.9",
|
|
documentation=(
|
|
DocumentationTopic(
|
|
id="example.workflow.safe",
|
|
title="Safe task",
|
|
summary="A task the actor may perform.",
|
|
documentation_types=("user",),
|
|
conditions=(DocumentationCondition(required_scopes=("docs:documentation:read",)),),
|
|
links=(
|
|
DocumentationLink(label="Open task", href="/example", kind="runtime"),
|
|
DocumentationLink(label="Unsafe runtime", href="javascript:alert(1)", kind="runtime"),
|
|
DocumentationLink(label="Backslash runtime", href="/\\evil.invalid", kind="runtime"),
|
|
DocumentationLink(label="Public help", href="https://example.invalid/help", kind="public"),
|
|
DocumentationLink(label="Repository", href="example/docs/SECRET.md", kind="repository"),
|
|
DocumentationLink(label="API", href="/api/v1/example", kind="api"),
|
|
),
|
|
metadata={
|
|
"kind": "workflow",
|
|
"screen": "Example",
|
|
"steps": ["Do the safe thing."],
|
|
"help_contexts": ["example.list"],
|
|
"current_configuration": ["The bounded option is enabled."],
|
|
"constraints": [{
|
|
"id": "domain",
|
|
"label": "Domain",
|
|
"description": "Use an approved domain.",
|
|
"values": ["*.example.invalid"],
|
|
"secret": "must-not-pass",
|
|
}],
|
|
"raw_policy": {"secret": "must-not-pass"},
|
|
"api_path": "/api/v1/internal",
|
|
},
|
|
),
|
|
),
|
|
))
|
|
|
|
layers = _classify_documentation(
|
|
registry,
|
|
FakePrincipal({"docs:documentation:read"}),
|
|
settings=None,
|
|
session=None,
|
|
documentation_type="user",
|
|
locale="en",
|
|
)
|
|
topic = layers["configured"][0]
|
|
|
|
self.assertEqual([link["href"] for link in topic["links"]], ["/example", "https://example.invalid/help"])
|
|
self.assertEqual(topic["conditions"], [])
|
|
self.assertEqual(topic["configuration_keys"], [])
|
|
self.assertEqual(topic["blockers"], {"modules": [], "capabilities": [], "scopes": []})
|
|
self.assertNotIn("raw_policy", topic["metadata"])
|
|
self.assertNotIn("api_path", topic["metadata"])
|
|
self.assertEqual(topic["metadata"]["help_contexts"], ["example.list"])
|
|
self.assertEqual(topic["metadata"]["constraints"][0], {
|
|
"id": "domain",
|
|
"label": "Domain",
|
|
"description": "Use an approved domain.",
|
|
"values": ["*.example.invalid"],
|
|
})
|
|
|
|
def test_docs_default_to_user_and_admin_projection_requires_admin_authority(self) -> None:
|
|
query_default = signature(docs_context).parameters["documentation_type"].default
|
|
self.assertEqual(query_default.default, "user")
|
|
|
|
with self.assertRaises(HTTPException) as raised:
|
|
docs_context(
|
|
SimpleNamespace(),
|
|
documentation_type="admin",
|
|
locale="en",
|
|
principal=FakePrincipal({"docs:documentation:read"}),
|
|
)
|
|
self.assertEqual(raised.exception.status_code, 403)
|
|
|
|
def test_user_actor_projection_does_not_disclose_identity_or_scope_count(self) -> None:
|
|
registry = PlatformRegistry()
|
|
registry.register(get_docs_manifest())
|
|
request = SimpleNamespace(app=SimpleNamespace(state=SimpleNamespace(govoplan_registry=registry)))
|
|
empty_layers = {"always": [], "configured": [], "available": [], "evidence": []}
|
|
|
|
with patch("govoplan_docs.backend.api.v1.routes._documentation_layers", return_value=empty_layers):
|
|
payload = docs_context(
|
|
request,
|
|
documentation_type="user",
|
|
locale="en",
|
|
principal=FakePrincipal({"docs:documentation:read"}),
|
|
)
|
|
|
|
self.assertEqual(payload["actor"]["available_documentation_types"], ["user"])
|
|
self.assertNotIn("tenant_id", payload["actor"])
|
|
self.assertNotIn("user_id", payload["actor"])
|
|
self.assertNotIn("scope_count", payload["actor"])
|
|
self.assertEqual(payload["layers"]["configured"]["permissions"], [])
|
|
self.assertEqual(payload["layers"]["available"]["routes"], [])
|
|
|
|
def test_topic_anchor_is_stable(self) -> None:
|
|
self.assertEqual(
|
|
_documentation_topic_anchor("access", "access.workflow.grant-user-access"),
|
|
"docs-topic-access-access-workflow-grant-user-access",
|
|
)
|
|
|
|
def test_condition_visibility_reports_module_capability_and_scope_blockers(self) -> None:
|
|
registry = PlatformRegistry()
|
|
condition = DocumentationCondition(
|
|
required_modules=("mail",),
|
|
any_modules=("files", "campaigns"),
|
|
missing_modules=("legacy",),
|
|
required_capabilities=("mail.delivery",),
|
|
required_scopes=("docs:documentation:read",),
|
|
any_scopes=("mail:profile:read", "admin:policies:read"),
|
|
)
|
|
principal = FakePrincipal({"docs:documentation:read"})
|
|
|
|
active, reason, blockers = _condition_visibility(condition, {"legacy"}, registry, principal)
|
|
|
|
self.assertFalse(active)
|
|
self.assertEqual(
|
|
reason,
|
|
"missing modules: mail; "
|
|
"requires one installed module from: files, campaigns; "
|
|
"not active when installed: legacy; "
|
|
"missing capabilities: mail.delivery; "
|
|
"requires one scope from: mail:profile:read, admin:policies:read",
|
|
)
|
|
self.assertEqual(blockers["modules"], ["mail", "files", "campaigns", "legacy"])
|
|
self.assertEqual(blockers["capabilities"], ["mail.delivery"])
|
|
self.assertEqual(blockers["scopes"], ["mail:profile:read", "admin:policies:read"])
|
|
|
|
def test_condition_visibility_accepts_any_module_scope_and_capability(self) -> None:
|
|
registry = PlatformRegistry()
|
|
registry.register_capability_factory("mail", "mail.delivery", lambda context: object())
|
|
condition = DocumentationCondition(
|
|
any_modules=("files", "campaigns"),
|
|
required_capabilities=("mail.delivery",),
|
|
any_scopes=("mail:profile:read", "admin:policies:read"),
|
|
)
|
|
principal = FakePrincipal({"admin:policies:read"})
|
|
|
|
active, reason, blockers = _condition_visibility(condition, {"campaigns"}, registry, principal)
|
|
|
|
self.assertTrue(active)
|
|
self.assertEqual(reason, "conditions satisfied")
|
|
self.assertEqual(blockers, {"modules": [], "capabilities": [], "scopes": []})
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|