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_runtime_provider_failure_is_diagnostic_only(self) -> None: def failing_provider(_context): raise RuntimeError("sensitive provider detail") registry = PlatformRegistry() registry.register(ModuleManifest( id="private-module", name="Private module", version="1.0.0", documentation_providers=(failing_provider,), )) principal = FakePrincipal({"docs:documentation:read"}) user_layers = _classify_documentation( registry, principal, settings=None, session=None, documentation_type="user", locale="en", ) admin_layers = _classify_documentation( registry, principal, settings=None, session=None, documentation_type="admin", locale="en", ) self.assertFalse(any(user_layers.values())) self.assertEqual( [topic["id"] for topic in admin_layers["evidence"]], ["private-module.runtime-documentation-unavailable"], ) 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"], []) self.assertEqual(payload["layers"]["evidence"]["sources"], []) 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()