feat: expose form semantic documentation subjects

tests/test_semantic_subjects.py

@@ -0,0 +1,263 @@
+from __future__ import annotations
+
+from dataclasses import dataclass, replace
+from datetime import UTC, datetime, timedelta
+import unittest
+
+from sqlalchemy import create_engine
+from sqlalchemy.orm import Session
+
+from govoplan_core.core.institutional import (
+    FormDefinition,
+    FormFieldDefinition,
+    FormLocalization,
+    FormPageDefinition,
+    FormSectionDefinition,
+    InstitutionalReference,
+    TemporalRevision,
+)
+from govoplan_core.core.semantic_documentation import (
+    SemanticDocumentationSubjectQuery,
+)
+from govoplan_forms.backend.db.models import FormDefinitionRevision
+from govoplan_forms.backend.semantic_subjects import (
+    FormsSemanticDocumentationSubjectProvider,
+)
+from govoplan_forms.backend.service import record_form_definition
+
+
+NOW = datetime(2026, 8, 21, 8, 0, tzinfo=UTC)
+
+
+@dataclass
+class Principal:
+    tenant_id: str = "tenant-1"
+    account_id: str = "author-1"
+    scopes: frozenset[str] = frozenset({"forms:definition:read"})
+
+    def has(self, scope: str) -> bool:
+        return scope in self.scopes
+
+
+def definition(
+    revision: int,
+    *,
+    fields: tuple[FormFieldDefinition, ...] | None = None,
+) -> FormDefinition:
+    resolved_fields = fields or (
+        FormFieldDefinition(key="name", label="Name", required=True),
+        FormFieldDefinition(key="delivery", label="Delivery method"),
+    )
+    return FormDefinition(
+        reference=InstitutionalReference(
+            kind="form",
+            owner_module="forms",
+            object_id="resident-permit",
+            tenant_id="tenant-1",
+            version=str(revision),
+        ),
+        key="resident-permit",
+        temporal=TemporalRevision(
+            revision=str(revision),
+            recorded_at=NOW + timedelta(minutes=revision),
+            change_reason=f"Revision {revision}",
+        ),
+        title="Resident permit",
+        fields=resolved_fields,
+        publication_state="published",
+        pages=(
+            FormPageDefinition(
+                key="application",
+                title="Application",
+                sections=(
+                    FormSectionDefinition(
+                        key="details",
+                        title="Applicant details",
+                        field_keys=tuple(field.key for field in resolved_fields),
+                    ),
+                ),
+            ),
+        ),
+        fallback_locale="de",
+        localizations=(
+            FormLocalization(
+                locale="de",
+                title="Anwohnerparkausweis",
+                field_labels={field.key: field.label for field in resolved_fields},
+                page_titles={"application": "Antrag"},
+                section_titles={"details": "Angaben"},
+            ),
+        ),
+    )
+
+
+class FormsSemanticSubjectTests(unittest.TestCase):
+    def setUp(self) -> None:
+        self.engine = create_engine("sqlite+pysqlite:///:memory:")
+        FormDefinitionRevision.__table__.create(self.engine)
+        self.session = Session(self.engine)
+        self.principal = Principal()
+        self.provider = FormsSemanticDocumentationSubjectProvider()
+
+    def tearDown(self) -> None:
+        self.session.close()
+        self.engine.dispose()
+
+    def subjects(self):
+        return self.provider.list_subjects(
+            self.session,
+            self.principal,
+            request=SemanticDocumentationSubjectQuery(tenant_id="tenant-1", limit=200),
+        ).subjects
+
+    def test_exposes_safe_form_field_and_section_descriptors(self) -> None:
+        record_form_definition(
+            self.session,
+            self.principal,
+            definition=definition(1),
+        )
+        subjects = self.subjects()
+        self.assertEqual(4, len(subjects))
+        form = next(item for item in subjects if item.reference.anchor is None)
+        field = next(
+            item
+            for item in subjects
+            if item.reference.anchor and item.reference.anchor.kind == "field"
+        )
+        self.assertEqual("Anwohnerparkausweis", form.labels["de"])
+        self.assertEqual("forms:definition:read", field.required_scopes[0])
+        self.assertTrue(field.route.startswith("/forms?formId="))
+        self.assertTrue(field.route_anchor.startswith("field-"))
+        self.assertNotIn("constraints", field.to_dict())
+
+    def test_identity_survives_reorder_and_label_change_but_fingerprint_changes(self) -> None:
+        first = definition(1)
+        record_form_definition(self.session, self.principal, definition=first)
+        before = {
+            item.route_anchor: item.reference
+            for item in self.subjects()
+            if item.reference.anchor and item.reference.anchor.kind == "field"
+        }
+        revised = replace(
+            definition(2),
+            fields=(
+                first.fields[1],
+                replace(first.fields[0], label="Full legal name"),
+            ),
+            pages=(
+                FormPageDefinition(
+                    key="application",
+                    title="Application",
+                    sections=(
+                        FormSectionDefinition(
+                            key="details",
+                            title="Applicant details",
+                            field_keys=("delivery", "name"),
+                        ),
+                    ),
+                ),
+            ),
+        )
+        record_form_definition(
+            self.session,
+            self.principal,
+            definition=revised,
+            expected_revision="1",
+        )
+        after = {
+            item.route_anchor: item.reference
+            for item in self.subjects()
+            if item.reference.anchor and item.reference.anchor.kind == "field"
+        }
+        self.assertEqual(
+            before["field-name"].stable_key,
+            after["field-name"].stable_key,
+        )
+        self.assertNotEqual(
+            before["field-name"].observed_fingerprint,
+            after["field-name"].observed_fingerprint,
+        )
+        resolution = self.provider.resolve_subject(
+            self.session,
+            self.principal,
+            reference=before["field-name"],
+        )
+        self.assertEqual("changed", resolution.availability)
+
+    def test_deleted_and_recreated_key_gets_new_lineage(self) -> None:
+        first = definition(1)
+        record_form_definition(self.session, self.principal, definition=first)
+        old = next(
+            item.reference
+            for item in self.subjects()
+            if item.route_anchor == "field-delivery"
+        )
+        record_form_definition(
+            self.session,
+            self.principal,
+            definition=definition(2, fields=(first.fields[0],)),
+            expected_revision="1",
+        )
+        deleted = self.provider.resolve_subject(
+            self.session,
+            self.principal,
+            reference=old,
+        )
+        self.assertEqual("missing", deleted.availability)
+        self.assertEqual("field_deleted", deleted.reason_code)
+
+        recreated_field = replace(first.fields[1], label="Recreated delivery")
+        record_form_definition(
+            self.session,
+            self.principal,
+            definition=definition(3, fields=(first.fields[0], recreated_field)),
+            expected_revision="2",
+        )
+        recreated = next(
+            item.reference
+            for item in self.subjects()
+            if item.route_anchor == "field-delivery"
+        )
+        self.assertNotEqual(old.stable_key, recreated.stable_key)
+        still_deleted = self.provider.resolve_subject(
+            self.session,
+            self.principal,
+            reference=old,
+        )
+        self.assertEqual("field_deleted", still_deleted.reason_code)
+
+    def test_resolution_denies_cross_tenant_and_missing_scope(self) -> None:
+        record_form_definition(
+            self.session,
+            self.principal,
+            definition=definition(1),
+        )
+        reference = self.subjects()[0].reference
+        denied = replace(self.principal, scopes=frozenset())
+        self.assertIsNone(
+            self.provider.resolve_subject(
+                self.session,
+                denied,
+                reference=reference,
+            )
+        )
+        foreign = replace(self.principal, tenant_id="tenant-2")
+        self.assertIsNone(
+            self.provider.resolve_subject(
+                self.session,
+                foreign,
+                reference=reference,
+            )
+        )
+        self.assertEqual(
+            (),
+            self.provider.list_subjects(
+                self.session,
+                denied,
+                request=SemanticDocumentationSubjectQuery(tenant_id="tenant-1"),
+            ).subjects,
+        )
+
+
+if __name__ == "__main__":
+    unittest.main()