379 lines
13 KiB
Python
379 lines
13 KiB
Python
from __future__ import annotations
|
|
|
|
import unittest
|
|
from types import SimpleNamespace
|
|
|
|
from govoplan_core.core.modules import (
|
|
CapabilityDocumentation,
|
|
DocumentationTopic,
|
|
ModuleContext,
|
|
ModuleManifest,
|
|
)
|
|
from govoplan_core.core.registry import PlatformRegistry, RegistryError
|
|
from govoplan_core.core.semantic_documentation import (
|
|
SEMANTIC_DOCUMENTATION_SUBJECT_CONTRACT_VERSION,
|
|
SemanticDocumentationBreadcrumb,
|
|
SemanticDocumentationContractError,
|
|
SemanticDocumentationSubjectAnchor,
|
|
SemanticDocumentationSubjectDescriptor,
|
|
SemanticDocumentationSubjectPage,
|
|
SemanticDocumentationSubjectQuery,
|
|
SemanticDocumentationSubjectReference,
|
|
SemanticDocumentationSubjectResolution,
|
|
list_semantic_documentation_subjects,
|
|
resolve_semantic_documentation_subject,
|
|
semantic_documentation_fingerprint,
|
|
semantic_documentation_subject_capability,
|
|
semantic_documentation_subject_provider_names,
|
|
semantic_documentation_subject_providers,
|
|
)
|
|
|
|
|
|
def reference(
|
|
*,
|
|
tenant_id: str = "tenant-1",
|
|
revision: str | None = "4",
|
|
fingerprint: str | None = None,
|
|
anchor: SemanticDocumentationSubjectAnchor | None = None,
|
|
) -> SemanticDocumentationSubjectReference:
|
|
return SemanticDocumentationSubjectReference(
|
|
module_id="forms",
|
|
tenant_id=tenant_id,
|
|
subject_kind="form",
|
|
subject_id="permit-application",
|
|
anchor=anchor,
|
|
observed_revision=revision,
|
|
observed_fingerprint=fingerprint
|
|
or semantic_documentation_fingerprint({"revision": revision}),
|
|
)
|
|
|
|
|
|
def descriptor(
|
|
*,
|
|
tenant_id: str = "tenant-1",
|
|
revision: str = "4",
|
|
fingerprint: str | None = None,
|
|
label: str = "Permit application",
|
|
anchor: SemanticDocumentationSubjectAnchor | None = None,
|
|
) -> SemanticDocumentationSubjectDescriptor:
|
|
return SemanticDocumentationSubjectDescriptor(
|
|
reference=reference(
|
|
tenant_id=tenant_id,
|
|
revision=revision,
|
|
fingerprint=fingerprint,
|
|
anchor=anchor,
|
|
),
|
|
labels={"de": "Antrag auf Parkerlaubnis", "en": label},
|
|
descriptions={"de": "Konfiguriertes Antragsformular."},
|
|
breadcrumbs=(
|
|
SemanticDocumentationBreadcrumb(
|
|
label="Forms",
|
|
subject_kind="form",
|
|
subject_id="permit-application",
|
|
),
|
|
),
|
|
route="/forms/permit-application",
|
|
route_anchor=anchor.id if anchor else None,
|
|
audience=("case_worker",),
|
|
classification="internal",
|
|
required_scopes=("forms:definition:read",),
|
|
)
|
|
|
|
|
|
class Provider:
|
|
provider_id = "forms.semantic_subjects"
|
|
module_id = "forms"
|
|
contract_version = SEMANTIC_DOCUMENTATION_SUBJECT_CONTRACT_VERSION
|
|
|
|
def __init__(self) -> None:
|
|
self.list_calls = 0
|
|
self.resolve_calls = 0
|
|
|
|
def list_subjects(self, _session, principal, *, request):
|
|
self.list_calls += 1
|
|
if getattr(principal, "account_id", "") == "denied":
|
|
return SemanticDocumentationSubjectPage()
|
|
return SemanticDocumentationSubjectPage(subjects=(descriptor(),))
|
|
|
|
def resolve_subject(self, _session, principal, *, reference):
|
|
self.resolve_calls += 1
|
|
if getattr(principal, "account_id", "") == "denied":
|
|
return None
|
|
current = descriptor()
|
|
availability = (
|
|
"changed"
|
|
if reference.observed_fingerprint
|
|
!= current.reference.observed_fingerprint
|
|
else "available"
|
|
)
|
|
return SemanticDocumentationSubjectResolution(
|
|
requested_reference=reference,
|
|
availability=availability,
|
|
subject=current,
|
|
)
|
|
|
|
|
|
class SemanticDocumentationContractTests(unittest.TestCase):
|
|
def test_reference_identity_is_stable_across_labels_and_revisions(self) -> None:
|
|
original = descriptor(label="Old label")
|
|
renamed = descriptor(revision="5", label="New label")
|
|
|
|
self.assertEqual(
|
|
original.reference.stable_key,
|
|
renamed.reference.stable_key,
|
|
)
|
|
self.assertNotEqual(
|
|
original.reference.observed_fingerprint,
|
|
renamed.reference.observed_fingerprint,
|
|
)
|
|
|
|
def test_nested_anchor_is_typed_and_changes_subject_identity(self) -> None:
|
|
form = reference()
|
|
field = reference(
|
|
anchor=SemanticDocumentationSubjectAnchor(
|
|
kind="field",
|
|
id="vehicle-registration",
|
|
)
|
|
)
|
|
|
|
self.assertNotEqual(form.stable_key, field.stable_key)
|
|
restored = SemanticDocumentationSubjectReference.from_mapping(
|
|
field.to_dict()
|
|
)
|
|
self.assertEqual(restored, field)
|
|
|
|
def test_reference_mapping_rejects_unbounded_provider_metadata(self) -> None:
|
|
payload = reference().to_dict()
|
|
payload["credentials"] = {"token": "must-not-cross"}
|
|
|
|
with self.assertRaisesRegex(
|
|
SemanticDocumentationContractError,
|
|
"unsupported fields: credentials",
|
|
):
|
|
SemanticDocumentationSubjectReference.from_mapping(payload)
|
|
|
|
def test_descriptors_reject_remote_routes_and_missing_localization(self) -> None:
|
|
with self.assertRaisesRegex(
|
|
SemanticDocumentationContractError,
|
|
"local absolute paths",
|
|
):
|
|
SemanticDocumentationSubjectDescriptor(
|
|
reference=reference(),
|
|
labels={"de": "Formular"},
|
|
route="https://provider.example/form",
|
|
)
|
|
with self.assertRaisesRegex(
|
|
SemanticDocumentationContractError,
|
|
"labels are required",
|
|
):
|
|
SemanticDocumentationSubjectDescriptor(
|
|
reference=reference(),
|
|
labels={},
|
|
)
|
|
|
|
def test_resolution_distinguishes_change_supersession_and_absence(self) -> None:
|
|
old = reference(
|
|
revision="3",
|
|
fingerprint=semantic_documentation_fingerprint({"revision": "3"}),
|
|
)
|
|
current = descriptor()
|
|
changed = SemanticDocumentationSubjectResolution(
|
|
requested_reference=old,
|
|
availability="changed",
|
|
subject=current,
|
|
)
|
|
superseded = SemanticDocumentationSubjectResolution(
|
|
requested_reference=old,
|
|
availability="superseded",
|
|
superseded_by=SemanticDocumentationSubjectReference(
|
|
module_id="forms",
|
|
tenant_id="tenant-1",
|
|
subject_kind="form",
|
|
subject_id="permit-application-v2",
|
|
),
|
|
)
|
|
missing = SemanticDocumentationSubjectResolution(
|
|
requested_reference=old,
|
|
availability="missing",
|
|
reason_code="subject_removed",
|
|
)
|
|
|
|
self.assertEqual(changed.availability, "changed")
|
|
self.assertEqual(superseded.superseded_by.subject_id, "permit-application-v2")
|
|
self.assertEqual(missing.reason_code, "subject_removed")
|
|
with self.assertRaisesRegex(
|
|
SemanticDocumentationContractError,
|
|
"require a revision or fingerprint change",
|
|
):
|
|
SemanticDocumentationSubjectResolution(
|
|
requested_reference=current.reference,
|
|
availability="changed",
|
|
subject=current,
|
|
)
|
|
|
|
def test_subject_pages_reject_duplicate_stable_identities(self) -> None:
|
|
with self.assertRaisesRegex(
|
|
SemanticDocumentationContractError,
|
|
"duplicate identities",
|
|
):
|
|
SemanticDocumentationSubjectPage(
|
|
subjects=(descriptor(), descriptor(label="Renamed")),
|
|
)
|
|
|
|
def test_registry_discovers_optional_providers_without_direct_imports(self) -> None:
|
|
provider = Provider()
|
|
registry = provider_registry(provider)
|
|
|
|
self.assertEqual(
|
|
semantic_documentation_subject_provider_names(registry),
|
|
("documentation.semantic_subjects.forms",),
|
|
)
|
|
self.assertEqual(
|
|
semantic_documentation_subject_providers(registry),
|
|
(("forms", provider),),
|
|
)
|
|
|
|
def test_listing_and_resolution_are_tenant_and_principal_aware(self) -> None:
|
|
provider = Provider()
|
|
registry = provider_registry(provider)
|
|
allowed = SimpleNamespace(tenant_id="tenant-1", account_id="account-1")
|
|
wrong_tenant = SimpleNamespace(
|
|
tenant_id="tenant-2", account_id="account-1"
|
|
)
|
|
denied = SimpleNamespace(tenant_id="tenant-1", account_id="denied")
|
|
|
|
pages = list_semantic_documentation_subjects(
|
|
registry,
|
|
object(),
|
|
allowed,
|
|
request=SemanticDocumentationSubjectQuery(tenant_id="tenant-1"),
|
|
)
|
|
self.assertEqual(len(pages), 1)
|
|
self.assertEqual(pages[0][1].subjects[0].reference.tenant_id, "tenant-1")
|
|
self.assertEqual(
|
|
list_semantic_documentation_subjects(
|
|
registry,
|
|
object(),
|
|
wrong_tenant,
|
|
request=SemanticDocumentationSubjectQuery(tenant_id="tenant-1"),
|
|
),
|
|
(),
|
|
)
|
|
self.assertIsNone(
|
|
resolve_semantic_documentation_subject(
|
|
registry,
|
|
object(),
|
|
denied,
|
|
reference=reference(),
|
|
)
|
|
)
|
|
self.assertEqual(provider.list_calls, 1)
|
|
self.assertEqual(provider.resolve_calls, 1)
|
|
|
|
def test_missing_optional_module_resolves_without_loading_feature_code(self) -> None:
|
|
principal = SimpleNamespace(tenant_id="tenant-1", account_id="account-1")
|
|
result = resolve_semantic_documentation_subject(
|
|
PlatformRegistry(),
|
|
object(),
|
|
principal,
|
|
reference=reference(),
|
|
)
|
|
|
|
self.assertEqual(result.availability, "temporarily_unavailable")
|
|
self.assertEqual(result.reason_code, "provider_unavailable")
|
|
|
|
def test_manifest_requires_exact_capability_docs_and_static_baselines(self) -> None:
|
|
provider = Provider()
|
|
capability = semantic_documentation_subject_capability("forms")
|
|
missing_baseline = PlatformRegistry()
|
|
missing_baseline.register(
|
|
ModuleManifest(
|
|
id="forms",
|
|
name="Forms",
|
|
version="1.0.0",
|
|
capability_factories={capability: lambda _context: provider},
|
|
capability_documentation={
|
|
capability: CapabilityDocumentation(
|
|
label="Form semantics",
|
|
summary="Lists documentation-safe configured form subjects.",
|
|
contract_version=(
|
|
SEMANTIC_DOCUMENTATION_SUBJECT_CONTRACT_VERSION
|
|
),
|
|
)
|
|
},
|
|
documentation=(static_topic("admin"),),
|
|
)
|
|
)
|
|
|
|
with self.assertRaisesRegex(RegistryError, "static user and administrator"):
|
|
missing_baseline.validate()
|
|
|
|
wrong_capability = PlatformRegistry()
|
|
wrong_capability.register(
|
|
ModuleManifest(
|
|
id="forms",
|
|
name="Forms",
|
|
version="1.0.0",
|
|
capability_factories={
|
|
semantic_documentation_subject_capability("workflow"): (
|
|
lambda _context: provider
|
|
)
|
|
},
|
|
)
|
|
)
|
|
with self.assertRaisesRegex(RegistryError, "must be"):
|
|
wrong_capability.validate()
|
|
|
|
def test_fingerprint_is_canonical_and_rejects_non_json_values(self) -> None:
|
|
self.assertEqual(
|
|
semantic_documentation_fingerprint({"b": 2, "a": 1}),
|
|
semantic_documentation_fingerprint({"a": 1, "b": 2}),
|
|
)
|
|
with self.assertRaisesRegex(
|
|
SemanticDocumentationContractError,
|
|
"canonical JSON",
|
|
):
|
|
semantic_documentation_fingerprint({"unsafe": object()})
|
|
|
|
|
|
def static_topic(documentation_type: str) -> DocumentationTopic:
|
|
return DocumentationTopic(
|
|
id=f"forms.semantic.{documentation_type}",
|
|
title="Form semantics",
|
|
summary="Static form semantics baseline.",
|
|
documentation_types=(documentation_type,), # type: ignore[arg-type]
|
|
)
|
|
|
|
|
|
def provider_registry(provider: Provider) -> PlatformRegistry:
|
|
capability = semantic_documentation_subject_capability("forms")
|
|
registry = PlatformRegistry()
|
|
registry.register(
|
|
ModuleManifest(
|
|
id="forms",
|
|
name="Forms",
|
|
version="1.0.0",
|
|
capability_factories={capability: lambda _context: provider},
|
|
capability_documentation={
|
|
capability: CapabilityDocumentation(
|
|
label="Form semantic subjects",
|
|
summary="Lists safe configured forms and fields for Docs.",
|
|
contract_version=(
|
|
SEMANTIC_DOCUMENTATION_SUBJECT_CONTRACT_VERSION
|
|
),
|
|
documentation_types=("admin", "user"),
|
|
)
|
|
},
|
|
documentation=(static_topic("admin"), static_topic("user")),
|
|
)
|
|
)
|
|
registry.validate()
|
|
registry.configure_capability_context(
|
|
ModuleContext(registry=registry, settings=object())
|
|
)
|
|
return registry
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|