374 lines
12 KiB
Python
374 lines
12 KiB
Python
from __future__ import annotations
|
|
|
|
from types import SimpleNamespace
|
|
import unittest
|
|
|
|
from sqlalchemy import create_engine
|
|
from sqlalchemy.orm import Session
|
|
|
|
from govoplan_core.auth import ApiPrincipal
|
|
from govoplan_core.core.access import PrincipalRef
|
|
from govoplan_core.core.search import (
|
|
SearchAuthorizationRequest,
|
|
SearchBackfillRequest,
|
|
)
|
|
from govoplan_core.core.semantic_documentation import (
|
|
SEMANTIC_DOCUMENTATION_SUBJECT_CONTRACT_VERSION,
|
|
SemanticDocumentationSubjectDescriptor,
|
|
SemanticDocumentationSubjectPage,
|
|
SemanticDocumentationSubjectReference,
|
|
SemanticDocumentationSubjectResolution,
|
|
semantic_documentation_fingerprint,
|
|
semantic_documentation_subject_capability,
|
|
)
|
|
from govoplan_core.db.base import Base
|
|
from govoplan_core.tenancy.scope import Tenant, create_scope_tables
|
|
|
|
from govoplan_docs.backend.dsar_provider import DocsDsarProvider
|
|
from govoplan_docs.backend.search_source import SemanticDocumentationSearchSource
|
|
from govoplan_docs.backend.semantic_service import (
|
|
SemanticDocumentationAuthorizationError,
|
|
SemanticDocumentationConflictError,
|
|
create_semantic_entry,
|
|
publication_policy,
|
|
publish_semantic_entry,
|
|
select_locale_entries,
|
|
semantic_entry_payload,
|
|
set_publication_policy,
|
|
update_semantic_entry,
|
|
)
|
|
from govoplan_core.core.dsar import DsarSubjectRef
|
|
|
|
|
|
class _SubjectProvider:
|
|
provider_id = "forms.semantic_subjects"
|
|
module_id = "forms"
|
|
contract_version = SEMANTIC_DOCUMENTATION_SUBJECT_CONTRACT_VERSION
|
|
|
|
def __init__(self) -> None:
|
|
self.revision = "7"
|
|
self.denied_accounts: set[str] = set()
|
|
|
|
def reference(self, tenant_id: str) -> SemanticDocumentationSubjectReference:
|
|
fingerprint = semantic_documentation_fingerprint(
|
|
{"id": "form-1", "revision": self.revision}
|
|
)
|
|
return SemanticDocumentationSubjectReference(
|
|
module_id="forms",
|
|
tenant_id=tenant_id,
|
|
subject_kind="form",
|
|
subject_id="form-1",
|
|
observed_revision=self.revision,
|
|
observed_fingerprint=fingerprint,
|
|
)
|
|
|
|
def descriptor(self, tenant_id: str) -> SemanticDocumentationSubjectDescriptor:
|
|
return SemanticDocumentationSubjectDescriptor(
|
|
reference=self.reference(tenant_id),
|
|
labels={"de": "Anwohnerparkausweis", "en": "Resident permit"},
|
|
descriptions={"de": "Konfiguriertes Formular"},
|
|
route="/forms/form-1",
|
|
required_scopes=("forms:form:read",),
|
|
)
|
|
|
|
def list_subjects(self, session, principal, *, request):
|
|
del session
|
|
if principal.account_id in self.denied_accounts:
|
|
return SemanticDocumentationSubjectPage()
|
|
return SemanticDocumentationSubjectPage(
|
|
subjects=(self.descriptor(request.tenant_id),)
|
|
)
|
|
|
|
def resolve_subject(self, session, principal, *, reference):
|
|
del session
|
|
if principal.account_id in self.denied_accounts:
|
|
return None
|
|
descriptor = self.descriptor(reference.tenant_id)
|
|
availability = (
|
|
"changed"
|
|
if reference.observed_revision
|
|
and reference.observed_revision != descriptor.reference.observed_revision
|
|
else "available"
|
|
)
|
|
return SemanticDocumentationSubjectResolution(
|
|
requested_reference=reference,
|
|
availability=availability,
|
|
subject=descriptor,
|
|
)
|
|
|
|
|
|
class _Registry:
|
|
def __init__(self, provider: _SubjectProvider) -> None:
|
|
self.provider = provider
|
|
self.name = semantic_documentation_subject_capability("forms")
|
|
|
|
def capability_names(self):
|
|
return (self.name,)
|
|
|
|
def has_capability(self, name):
|
|
return name == self.name
|
|
|
|
def capability(self, name):
|
|
return self.provider if name == self.name else None
|
|
|
|
|
|
def _principal(
|
|
account_id: str,
|
|
*,
|
|
tenant_id: str = "tenant-1",
|
|
scopes: frozenset[str] | None = None,
|
|
) -> ApiPrincipal:
|
|
principal = PrincipalRef(
|
|
account_id=account_id,
|
|
membership_id=f"membership-{account_id}",
|
|
tenant_id=tenant_id,
|
|
scopes=scopes or frozenset({"docs:documentation:read", "forms:form:read"}),
|
|
)
|
|
return ApiPrincipal(
|
|
principal=principal,
|
|
account=SimpleNamespace(id=account_id),
|
|
user=SimpleNamespace(id=account_id),
|
|
)
|
|
|
|
|
|
def _content(title: str = "Permit form") -> dict[str, object]:
|
|
return {
|
|
"title": title,
|
|
"summary": "Tenant meaning",
|
|
"body": "Use this form for residents.",
|
|
"meaning": "Configured resident permit request.",
|
|
"intended_use": "Resident permits",
|
|
"non_intended_use": "Visitor permits",
|
|
"examples": [],
|
|
"owner_account_id": "author",
|
|
"steward_account_id": "reviewer",
|
|
"audience": [],
|
|
"classification": "internal",
|
|
"links": [],
|
|
}
|
|
|
|
|
|
class SemanticDocumentationTests(unittest.TestCase):
|
|
def setUp(self) -> None:
|
|
self.engine = create_engine("sqlite+pysqlite:///:memory:")
|
|
create_scope_tables(self.engine)
|
|
Base.metadata.create_all(self.engine)
|
|
self.session = Session(self.engine)
|
|
self.session.add(
|
|
Tenant(id="tenant-1", slug="tenant-1", name="Tenant One", settings={})
|
|
)
|
|
self.session.commit()
|
|
self.provider = _SubjectProvider()
|
|
self.registry = _Registry(self.provider)
|
|
self.author = _principal("author")
|
|
self.reviewer = _principal("reviewer")
|
|
|
|
def tearDown(self) -> None:
|
|
self.session.close()
|
|
self.engine.dispose()
|
|
|
|
def test_revision_review_policy_visibility_and_concurrency(self) -> None:
|
|
entry = create_semantic_entry(
|
|
self.session,
|
|
self.registry,
|
|
self.author,
|
|
subject=self.provider.reference("tenant-1"),
|
|
locale="de",
|
|
content=_content(),
|
|
change_reason="Initial tenant semantics",
|
|
)
|
|
self.assertEqual("reviewer_required", publication_policy(self.session, "tenant-1"))
|
|
with self.assertRaises(SemanticDocumentationAuthorizationError):
|
|
publish_semantic_entry(
|
|
self.session,
|
|
self.registry,
|
|
self.author,
|
|
entry_id=entry.id,
|
|
expected_revision=1,
|
|
change_reason="Self publish",
|
|
)
|
|
|
|
entry = publish_semantic_entry(
|
|
self.session,
|
|
self.registry,
|
|
self.reviewer,
|
|
entry_id=entry.id,
|
|
expected_revision=1,
|
|
change_reason="Independent review",
|
|
)
|
|
self.assertEqual(2, entry.current_revision)
|
|
reader_payload = semantic_entry_payload(
|
|
self.session,
|
|
self.registry,
|
|
self.author,
|
|
entry=entry,
|
|
editor=False,
|
|
)
|
|
self.assertEqual("Permit form", reader_payload["content"]["title"])
|
|
|
|
update_semantic_entry(
|
|
self.session,
|
|
self.registry,
|
|
self.author,
|
|
entry_id=entry.id,
|
|
expected_revision=2,
|
|
content=_content("Updated permit form"),
|
|
change_reason="Clarify meaning",
|
|
)
|
|
with self.assertRaises(SemanticDocumentationConflictError):
|
|
update_semantic_entry(
|
|
self.session,
|
|
self.registry,
|
|
self.author,
|
|
entry_id=entry.id,
|
|
expected_revision=2,
|
|
content=_content("Stale edit"),
|
|
change_reason="Stale",
|
|
)
|
|
reader_payload = semantic_entry_payload(
|
|
self.session,
|
|
self.registry,
|
|
self.author,
|
|
entry=entry,
|
|
editor=False,
|
|
)
|
|
self.assertEqual("Permit form", reader_payload["content"]["title"])
|
|
self.assertTrue(reader_payload["pending_draft"])
|
|
|
|
def test_subject_reauthorization_tenant_and_changed_state(self) -> None:
|
|
entry = create_semantic_entry(
|
|
self.session,
|
|
self.registry,
|
|
self.author,
|
|
subject=self.provider.reference("tenant-1"),
|
|
locale="de",
|
|
content=_content(),
|
|
change_reason="Initial",
|
|
)
|
|
publish_semantic_entry(
|
|
self.session,
|
|
self.registry,
|
|
self.reviewer,
|
|
entry_id=entry.id,
|
|
expected_revision=1,
|
|
change_reason="Review",
|
|
)
|
|
self.provider.revision = "8"
|
|
payload = semantic_entry_payload(
|
|
self.session,
|
|
self.registry,
|
|
self.author,
|
|
entry=entry,
|
|
editor=False,
|
|
)
|
|
self.assertEqual("changed", payload["subject_resolution"]["availability"])
|
|
self.provider.denied_accounts.add("author")
|
|
self.assertIsNone(
|
|
semantic_entry_payload(
|
|
self.session,
|
|
self.registry,
|
|
self.author,
|
|
entry=entry,
|
|
editor=False,
|
|
)
|
|
)
|
|
with self.assertRaises(Exception):
|
|
create_semantic_entry(
|
|
self.session,
|
|
self.registry,
|
|
_principal("other", tenant_id="tenant-2"),
|
|
subject=self.provider.reference("tenant-1"),
|
|
locale="de",
|
|
content=_content(),
|
|
change_reason="Foreign",
|
|
)
|
|
|
|
def test_locale_search_and_dsar_projection(self) -> None:
|
|
entries = []
|
|
for locale, title in (("de", "Deutsch"), ("en", "English")):
|
|
entry = create_semantic_entry(
|
|
self.session,
|
|
self.registry,
|
|
self.author,
|
|
subject=self.provider.reference("tenant-1"),
|
|
locale=locale,
|
|
content=_content(title),
|
|
change_reason="Initial",
|
|
)
|
|
publish_semantic_entry(
|
|
self.session,
|
|
self.registry,
|
|
self.reviewer,
|
|
entry_id=entry.id,
|
|
expected_revision=1,
|
|
change_reason="Review",
|
|
)
|
|
entries.append(entry)
|
|
self.assertEqual("de", select_locale_entries(entries, locale="de-AT")[0].locale)
|
|
|
|
source = SemanticDocumentationSearchSource(self.registry)
|
|
page = source.backfill(
|
|
self.session,
|
|
request=SearchBackfillRequest(
|
|
tenant_id="tenant-1",
|
|
provider_id="docs.semantic_documentation",
|
|
resource_type="semantic_documentation",
|
|
rebuild_id="rebuild-1",
|
|
),
|
|
)
|
|
self.assertEqual(2, len(page.documents))
|
|
request = SearchAuthorizationRequest(
|
|
reference=page.documents[0].reference,
|
|
source_revision=page.documents[0].source_revision,
|
|
)
|
|
self.assertTrue(source.authorize(self.session, self.author, requests=(request,))[request.reference.key])
|
|
self.provider.denied_accounts.add("author")
|
|
self.assertFalse(source.authorize(self.session, self.author, requests=(request,))[request.reference.key])
|
|
|
|
records = DocsDsarProvider().search_subject(
|
|
self.session,
|
|
tenant_id="tenant-1",
|
|
subject=DsarSubjectRef(account_id="author"),
|
|
)
|
|
self.assertGreaterEqual(len(records), 2)
|
|
self.assertNotIn("body", records[0].data)
|
|
actions = DocsDsarProvider().plan_erasure(
|
|
self.session,
|
|
tenant_id="tenant-1",
|
|
subject=DsarSubjectRef(account_id="author"),
|
|
records=records,
|
|
)
|
|
self.assertTrue(all(not action.executable for action in actions))
|
|
|
|
def test_direct_publication_policy_is_explicit(self) -> None:
|
|
self.assertEqual(
|
|
"direct",
|
|
set_publication_policy(
|
|
self.session,
|
|
tenant_id="tenant-1",
|
|
mode="direct",
|
|
),
|
|
)
|
|
entry = create_semantic_entry(
|
|
self.session,
|
|
self.registry,
|
|
self.author,
|
|
subject=self.provider.reference("tenant-1"),
|
|
locale="de",
|
|
content=_content(),
|
|
change_reason="Initial",
|
|
)
|
|
published = publish_semantic_entry(
|
|
self.session,
|
|
self.registry,
|
|
self.author,
|
|
entry_id=entry.id,
|
|
expected_revision=1,
|
|
change_reason="Direct publication",
|
|
)
|
|
self.assertEqual("published", published.lifecycle_state)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|