from __future__ import annotations from datetime import UTC, datetime from govoplan_campaign.backend.integrations import TemplatesCampaignIntegration from govoplan_campaign.backend.routes.campaigns import ( _campaign_content_library_item, _campaign_content_required_fields, ) from govoplan_campaign.backend.schemas import CampaignContentLibrarySaveRequest from govoplan_core.core.templates import ( TemplateContentDraftRequest, TemplateRef, TemplateRevisionRef, ) class _Catalog: def __init__(self, template: TemplateRef) -> None: self.template = template self.calls: list[dict[str, object]] = [] def list_templates(self, session, principal, **kwargs): del session, principal self.calls.append(kwargs) return (self.template,) def get_template(self, session, principal, **kwargs): del session, principal, kwargs return self.template def check_compatibility(self, session, principal, **kwargs): del session, principal, kwargs raise AssertionError("Compatibility is not needed for raw reusable content.") class _ContentLibrary: def __init__(self, template: TemplateRef) -> None: self.template = template self.request: TemplateContentDraftRequest | None = None def create_content_draft(self, session, principal, *, request): del session, principal self.request = request return self.template def _template_ref() -> TemplateRef: revision = TemplateRevisionRef( id="revision-1", template_id="template-1", revision=3, definition_hash="a" * 64, template_type="content_fragment", usages=("campaign.content",), locale="de", required_fields=(), output_profiles=(), content_text="Mit freundlichen Grussen", metadata={ "campaign_kind": "fragment", "campaign_targets": ["subject"], }, published_at=datetime.now(tz=UTC), ) return TemplateRef( id="template-1", tenant_id="tenant-1", name="Closing", slug="closing", template_type="content_fragment", status="published", current_revision=3, current_revision_id=revision.id, published_revision_id=revision.id, revision=revision, ) def test_campaign_content_integration_filters_usage_and_delegates_draft_creation() -> None: template = _template_ref() catalog = _Catalog(template) writer = _ContentLibrary(template) integration = TemplatesCampaignIntegration(catalog, None, writer) listed = integration.list_content_templates(object(), object(), query="close") request = TemplateContentDraftRequest( name="Closing", template_type="content_fragment", usages=("campaign.content",), content_text="Regards", ) created = integration.create_content_draft( object(), object(), request=request, ) assert listed == (template,) assert catalog.calls == [ {"query": "close", "usage": "campaign.content", "limit": 100} ] assert writer.request is request assert created is template def test_campaign_content_payload_keeps_revision_and_declared_target() -> None: payload = _campaign_content_library_item(_template_ref()) assert payload["kind"] == "fragment" assert payload["targets"] == ["subject"] assert payload["text"] == "Mit freundlichen Grussen" assert payload["published"] is True assert payload["revision"] == 3 assert payload["required_fields"] == [] def test_campaign_content_required_fields_are_normalized_and_deduplicated() -> None: payload = CampaignContentLibrarySaveRequest( name="Greeting", kind="campaign_part", subject="Hello {{ local:display_name }}", text="Reference ${global.case_reference}; hello {{local::display_name}}", html="
{{fields.salutation}} {{local:to[2].email}} {{unsupported namespace}}
", ) requirements = _campaign_content_required_fields(payload) assert [item.path for item in requirements] == [ "global.case_reference", "local.display_name", "local.to.2.email", "salutation", ] def test_content_save_request_rejects_empty_selected_fragment() -> None: try: CampaignContentLibrarySaveRequest( name="Empty", kind="fragment", target="html", text="Only text", ) except ValueError as exc: assert "selected fragment field is empty" in str(exc) else: raise AssertionError("Expected empty target validation to fail")