264 lines
8.6 KiB
Python
264 lines
8.6 KiB
Python
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()
|