410 lines
16 KiB
Python
410 lines
16 KiB
Python
from __future__ import annotations
|
|
|
|
import hashlib
|
|
import unittest
|
|
from dataclasses import replace
|
|
from unittest.mock import patch
|
|
|
|
from govoplan_core.auth import ApiPrincipal
|
|
from govoplan_core.core.access import PrincipalRef
|
|
from govoplan_core.core.files import ManagedArtifactRef
|
|
from govoplan_core.core.templates import (
|
|
CAPABILITY_TEMPLATE_CATALOG,
|
|
CAPABILITY_TEMPLATE_CONTENT_LIBRARY,
|
|
CAPABILITY_TEMPLATE_RENDERER,
|
|
TemplateCompatibilityError,
|
|
TemplateContentDraftRequest,
|
|
TemplateFieldRequirement,
|
|
TemplateRenderError,
|
|
TemplateRenderRequest,
|
|
)
|
|
from govoplan_core.db.base import Base
|
|
from govoplan_core.db.session import configure_database, reset_database
|
|
from govoplan_templates.backend import rendering
|
|
from govoplan_templates.backend.capabilities import (
|
|
SqlTemplateCatalog,
|
|
SqlTemplateContentLibrary,
|
|
)
|
|
from govoplan_templates.backend.db.models import (
|
|
TemplateDefinition,
|
|
TemplateRender,
|
|
TemplateRevision,
|
|
)
|
|
from govoplan_templates.backend.rendering import (
|
|
get_render_for_principal,
|
|
list_renders,
|
|
render_template,
|
|
)
|
|
from govoplan_templates.backend.schemas import (
|
|
TemplateCreateRequest,
|
|
TemplateUpdateRequest,
|
|
)
|
|
from govoplan_templates.backend.service import (
|
|
create_template,
|
|
publish_template,
|
|
sanitize_template_html,
|
|
update_template,
|
|
)
|
|
|
|
|
|
def principal(
|
|
tenant_id: str = "tenant-1",
|
|
*,
|
|
account_id: str = "account-1",
|
|
admin: bool = True,
|
|
) -> ApiPrincipal:
|
|
scopes = {
|
|
"templates:template:read",
|
|
"templates:template:write",
|
|
"templates:template:publish",
|
|
"templates:template:render",
|
|
"files:file:upload",
|
|
}
|
|
if admin:
|
|
scopes.add("templates:template:admin")
|
|
return ApiPrincipal(
|
|
principal=PrincipalRef(
|
|
account_id=account_id,
|
|
membership_id="membership-1",
|
|
tenant_id=tenant_id,
|
|
identity_id="identity-1",
|
|
scopes=frozenset(scopes),
|
|
),
|
|
account=object(),
|
|
user=type("User", (), {"id": "user-1"})(),
|
|
)
|
|
|
|
|
|
def payload(
|
|
name: str = "Postal letter",
|
|
*,
|
|
template_type: str = "serial_letter",
|
|
body: str = "<p>Hello {{name}}</p><p>{{postal.address}}</p>",
|
|
) -> TemplateCreateRequest:
|
|
return TemplateCreateRequest.model_validate(
|
|
{
|
|
"name": name,
|
|
"template_type": template_type,
|
|
"usages": ["campaign.postal"],
|
|
"locale": "de-DE",
|
|
"required_fields": [
|
|
{"path": "name", "value_type": "string", "required": True},
|
|
{"path": "postal.address", "value_type": "string", "required": True},
|
|
],
|
|
"content_html": body,
|
|
"layout": {
|
|
"page_size": "A4",
|
|
"margin_mm": 15,
|
|
"columns": 2,
|
|
"rows": 2,
|
|
},
|
|
}
|
|
)
|
|
|
|
|
|
class _Registry:
|
|
def __init__(self, capability=None) -> None:
|
|
self._capability = capability
|
|
|
|
def has_capability(self, name: str) -> bool:
|
|
return self._capability is not None and name == "files.artifact_store"
|
|
|
|
def capability(self, name: str):
|
|
return self._capability if name == "files.artifact_store" else None
|
|
|
|
|
|
class _ArtifactStore:
|
|
def __init__(self) -> None:
|
|
self.request = None
|
|
|
|
def store_artifact(self, session, principal, *, request):
|
|
del session, principal
|
|
self.request = request
|
|
return ManagedArtifactRef(
|
|
file_asset_id="file-1",
|
|
file_version_id="version-1",
|
|
filename=request.filename,
|
|
display_path=f"Generated/Templates/{request.filename}",
|
|
content_type=request.content_type,
|
|
size_bytes=len(request.payload),
|
|
sha256=hashlib.sha256(request.payload).hexdigest(),
|
|
provenance={"module": "files", "managed": True},
|
|
)
|
|
|
|
|
|
class TemplateServiceTests(unittest.TestCase):
|
|
def setUp(self) -> None:
|
|
self.database = configure_database("sqlite:///:memory:")
|
|
Base.metadata.create_all(
|
|
self.database.engine,
|
|
tables=[
|
|
TemplateDefinition.__table__,
|
|
TemplateRevision.__table__,
|
|
TemplateRender.__table__,
|
|
],
|
|
)
|
|
|
|
def tearDown(self) -> None:
|
|
reset_database(dispose=True)
|
|
|
|
def test_catalogue_exposes_typed_contract_and_immutable_revisions(self) -> None:
|
|
with self.database.session() as session:
|
|
item, first = create_template(session, principal(), payload())
|
|
update = TemplateUpdateRequest.model_validate(
|
|
{
|
|
**payload(body="<p>Dear {{name}}</p><p>{{postal.address}}</p>").model_dump(mode="json"),
|
|
"base_revision": 1,
|
|
}
|
|
)
|
|
item, second = update_template(session, principal(), item, update)
|
|
session.commit()
|
|
|
|
refs = SqlTemplateCatalog().list_templates(
|
|
session,
|
|
principal(),
|
|
usage="campaign.postal",
|
|
)
|
|
self.assertEqual(1, len(refs))
|
|
self.assertEqual("serial_letter", refs[0].template_type)
|
|
self.assertEqual(("campaign.postal",), refs[0].revision.usages)
|
|
self.assertIn("Dear", refs[0].revision.content_html)
|
|
self.assertEqual("postal.address", refs[0].revision.required_fields[1].path)
|
|
self.assertNotEqual(first.definition_hash, second.definition_hash)
|
|
self.assertEqual(2, second.revision)
|
|
|
|
def test_content_library_creates_unpublished_provider_owned_draft(self) -> None:
|
|
with self.database.session() as session, patch(
|
|
"govoplan_templates.backend.capabilities.audit_from_principal"
|
|
), patch("govoplan_templates.backend.capabilities.emit_platform_event"):
|
|
result = SqlTemplateContentLibrary().create_content_draft(
|
|
session,
|
|
principal(),
|
|
request=TemplateContentDraftRequest(
|
|
name="Closing paragraph",
|
|
template_type="content_fragment",
|
|
usages=("campaign.content",),
|
|
locale="de",
|
|
content_text="Mit freundlichen Grüßen",
|
|
required_fields=(
|
|
TemplateFieldRequirement(
|
|
path="local.display_name",
|
|
label="Display name",
|
|
),
|
|
),
|
|
metadata={"campaign_targets": ["text"]},
|
|
),
|
|
)
|
|
session.commit()
|
|
|
|
self.assertEqual("draft", result.status)
|
|
self.assertEqual("content_fragment", result.template_type)
|
|
self.assertEqual("Mit freundlichen Grüßen", result.revision.content_text)
|
|
self.assertEqual(
|
|
"local.display_name",
|
|
result.revision.required_fields[0].path,
|
|
)
|
|
self.assertEqual(["text"], result.revision.metadata["campaign_targets"])
|
|
self.assertEqual(
|
|
"templates.content_library",
|
|
result.revision.metadata["created_through"],
|
|
)
|
|
|
|
def test_frozen_postal_snapshot_renders_deterministic_letter_bundle(self) -> None:
|
|
frozen = (
|
|
{"name": "Ada", "postal": {"address": "Street 1"}},
|
|
{"name": "Grace", "postal": {"address": "Street 2"}},
|
|
)
|
|
with self.database.session() as session:
|
|
item, _ = create_template(session, principal(), payload())
|
|
item, revision = publish_template(
|
|
session,
|
|
principal(),
|
|
item,
|
|
revision=1,
|
|
base_revision=1,
|
|
)
|
|
request = TemplateRenderRequest(
|
|
template_id=item.id,
|
|
revision=revision.revision,
|
|
usage="campaign.postal",
|
|
items=frozen,
|
|
input_snapshot={"provider": "dist_lists", "snapshot_id": "snapshot-1"},
|
|
mode="final",
|
|
idempotency_key="campaign-1:postal-output-1",
|
|
)
|
|
first = render_template(session, principal(), registry=_Registry(), request=request)
|
|
with patch.object(rendering, "run_bounded_operation", side_effect=AssertionError("idempotent render started a worker")):
|
|
second = render_template(session, principal(), registry=_Registry(), request=request)
|
|
session.commit()
|
|
|
|
self.assertEqual(first.render_id, second.render_id)
|
|
self.assertEqual(first.input_hash, second.input_hash)
|
|
self.assertEqual(first.output_sha256, second.output_sha256)
|
|
self.assertEqual(2, first.item_count)
|
|
self.assertEqual(2, first.page_count)
|
|
self.assertEqual("bounded_download", first.artifact.kind)
|
|
self.assertIn(b"Ada", first.payload)
|
|
self.assertIn(b"Grace", first.payload)
|
|
|
|
def test_real_worker_timeout_writes_neither_artifact_nor_render_evidence(self) -> None:
|
|
store = _ArtifactStore()
|
|
with self.database.session() as session:
|
|
item, _ = create_template(session, principal(), payload())
|
|
request = TemplateRenderRequest(
|
|
template_id=item.id,
|
|
usage="campaign.postal",
|
|
items=({"name": "Ada", "postal": {"address": "Street 1"}},),
|
|
persist_to_files=True,
|
|
)
|
|
limits = replace(rendering.RENDER_PROCESS_LIMITS, wall_seconds=0.001)
|
|
with patch.object(rendering, "RENDER_PROCESS_LIMITS", limits):
|
|
with self.assertRaisesRegex(TemplateRenderError, "timeout"):
|
|
render_template(session, principal(), registry=_Registry(store), request=request)
|
|
self.assertIsNone(store.request)
|
|
self.assertEqual([], list_renders(session, principal()))
|
|
|
|
def test_incompatible_input_is_rejected_before_worker_admission(self) -> None:
|
|
with self.database.session() as session:
|
|
item, _ = create_template(session, principal(), payload())
|
|
request = TemplateRenderRequest(
|
|
template_id=item.id, usage="campaign.postal", items=({"name": "Missing address"},),
|
|
)
|
|
with patch.object(rendering, "run_bounded_operation", side_effect=AssertionError("incompatible render started a worker")):
|
|
with self.assertRaises(TemplateCompatibilityError):
|
|
render_template(session, principal(), registry=_Registry(), request=request)
|
|
|
|
def test_bounded_render_history_and_payload_are_owner_scoped(self) -> None:
|
|
owner = principal(admin=False)
|
|
other = principal(account_id="account-2", admin=False)
|
|
administrator = principal(account_id="account-admin")
|
|
with self.database.session() as session:
|
|
item, _ = create_template(session, owner, payload())
|
|
result = render_template(
|
|
session,
|
|
owner,
|
|
registry=_Registry(),
|
|
request=TemplateRenderRequest(
|
|
template_id=item.id,
|
|
usage="campaign.postal",
|
|
items=(
|
|
{
|
|
"name": "Ada",
|
|
"postal": {"address": "Street 1"},
|
|
},
|
|
),
|
|
),
|
|
)
|
|
session.commit()
|
|
|
|
self.assertEqual(
|
|
result.render_id,
|
|
get_render_for_principal(session, owner, result.render_id).id,
|
|
)
|
|
with self.assertRaisesRegex(TemplateRenderError, "not found"):
|
|
get_render_for_principal(session, other, result.render_id)
|
|
self.assertEqual([], list_renders(session, other))
|
|
self.assertEqual(
|
|
result.render_id,
|
|
get_render_for_principal(
|
|
session,
|
|
administrator,
|
|
result.render_id,
|
|
).id,
|
|
)
|
|
|
|
def test_label_sheet_page_count_and_missing_fields(self) -> None:
|
|
with self.database.session() as session:
|
|
item, _ = create_template(
|
|
session,
|
|
principal(),
|
|
payload("Address labels", template_type="label_sheet"),
|
|
)
|
|
with self.assertRaises(TemplateCompatibilityError):
|
|
render_template(
|
|
session,
|
|
principal(),
|
|
registry=_Registry(),
|
|
request=TemplateRenderRequest(
|
|
template_id=item.id,
|
|
usage="campaign.postal",
|
|
items=({"name": "Missing address"},),
|
|
),
|
|
)
|
|
result = render_template(
|
|
session,
|
|
principal(),
|
|
registry=_Registry(),
|
|
request=TemplateRenderRequest(
|
|
template_id=item.id,
|
|
usage="campaign.postal",
|
|
items=tuple(
|
|
{"name": f"Person {index}", "postal": {"address": f"Street {index}"}}
|
|
for index in range(5)
|
|
),
|
|
),
|
|
)
|
|
self.assertEqual(2, result.page_count)
|
|
|
|
def test_optional_files_store_receives_hashes_without_templates_payload_copy(self) -> None:
|
|
store = _ArtifactStore()
|
|
with self.database.session() as session:
|
|
item, _ = create_template(session, principal(), payload())
|
|
item, revision = publish_template(session, principal(), item, revision=1, base_revision=1)
|
|
result = render_template(
|
|
session,
|
|
principal(),
|
|
registry=_Registry(store),
|
|
request=TemplateRenderRequest(
|
|
template_id=item.id,
|
|
revision=revision.revision,
|
|
usage="campaign.postal",
|
|
items=({"name": "Ada", "postal": {"address": "Street 1"}},),
|
|
input_snapshot={"snapshot_id": "snapshot-1"},
|
|
mode="final",
|
|
idempotency_key="managed-output-1",
|
|
persist_to_files=True,
|
|
),
|
|
)
|
|
session.commit()
|
|
row = session.get(TemplateRender, result.render_id)
|
|
|
|
self.assertEqual("managed_file", result.artifact.kind)
|
|
self.assertIsNone(row.payload)
|
|
self.assertEqual(result.output_sha256, store.request.metadata["output_sha256"])
|
|
self.assertNotIn("Ada", str(store.request.metadata))
|
|
|
|
def test_tenant_isolation_and_html_sanitization(self) -> None:
|
|
self.assertEqual("<p>Safe</p>", sanitize_template_html("<p>Safe</p><script>alert(1)</script>"))
|
|
self.assertEqual("<a>Unsafe</a>", sanitize_template_html('<a href="javascript:alert(1)">Unsafe</a>'))
|
|
with self.database.session() as session:
|
|
item, _ = create_template(session, principal("tenant-1"), payload())
|
|
session.commit()
|
|
self.assertIsNone(
|
|
SqlTemplateCatalog().get_template(
|
|
session,
|
|
principal("tenant-2"),
|
|
template_id=item.id,
|
|
)
|
|
)
|
|
|
|
|
|
class TemplateManifestTests(unittest.TestCase):
|
|
def test_manifest_announces_provider_neutral_capabilities(self) -> None:
|
|
from govoplan_templates.backend.manifest import get_manifest
|
|
|
|
manifest = get_manifest()
|
|
self.assertEqual("templates", manifest.id)
|
|
self.assertFalse(manifest.dependencies)
|
|
self.assertIn("files", manifest.optional_dependencies)
|
|
self.assertIn(CAPABILITY_TEMPLATE_CATALOG, manifest.capability_factories)
|
|
self.assertIn(
|
|
CAPABILITY_TEMPLATE_CONTENT_LIBRARY,
|
|
manifest.capability_factories,
|
|
)
|
|
self.assertIn(CAPABILITY_TEMPLATE_RENDERER, manifest.capability_factories)
|
|
self.assertEqual("@govoplan/templates-webui", manifest.frontend.package_name)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|