231 lines
10 KiB
Python
231 lines
10 KiB
Python
from __future__ import annotations
|
|
|
|
from types import SimpleNamespace
|
|
from unittest.mock import patch
|
|
|
|
import pytest
|
|
from fastapi import HTTPException
|
|
|
|
from govoplan_campaign.backend import router
|
|
from govoplan_campaign.backend.campaign.loader import CampaignSchemaError, validate_against_schema
|
|
from govoplan_campaign.backend.campaign.mail_profile_boundary import (
|
|
CampaignMailProfileBoundaryError,
|
|
assert_campaign_uses_mail_profile_reference,
|
|
campaign_mail_profile_boundary_violations,
|
|
campaign_mail_profile_id,
|
|
)
|
|
from govoplan_campaign.backend.campaign.models import DeliveryConfig
|
|
from govoplan_campaign.backend.persistence.campaigns import CampaignPersistenceError, load_campaign_config_from_json
|
|
from govoplan_campaign.backend.persistence.versions import update_campaign_version
|
|
from govoplan_campaign.backend.integrations import MailCampaignIntegration
|
|
from govoplan_campaign.backend.sending.execution import ExecutionSnapshotError, create_execution_snapshot, ensure_execution_snapshot
|
|
from govoplan_campaign.backend.schemas import CampaignVersionUpdateRequest
|
|
|
|
|
|
def _campaign_json(server: dict[str, object] | None = None) -> dict[str, object]:
|
|
return {
|
|
"version": "1.0",
|
|
"campaign": {"id": "campaign-1", "name": "Campaign", "mode": "send"},
|
|
"server": server or {},
|
|
"recipients": {"from": [{"email": "sender@example.test"}]},
|
|
"template": {"subject": "Subject", "text": "Body", "body_mode": "text"},
|
|
"entries": {"inline": []},
|
|
}
|
|
|
|
|
|
def test_campaign_mail_contract_accepts_only_a_stable_profile_reference() -> None:
|
|
raw = _campaign_json({"mail_profile_id": " profile-1 "})
|
|
|
|
assert_campaign_uses_mail_profile_reference(raw)
|
|
|
|
assert campaign_mail_profile_id(raw) == "profile-1"
|
|
assert campaign_mail_profile_boundary_violations(raw) == ()
|
|
|
|
|
|
def test_mail_profile_documentation_is_classified_for_adaptive_views() -> None:
|
|
from govoplan_campaign.backend.manifest import get_manifest
|
|
|
|
manifest = get_manifest()
|
|
topics = {topic.id: topic for topic in manifest.documentation}
|
|
|
|
workflow = topics["campaigns.mail-profile-user-journey"]
|
|
assert workflow.metadata["kind"] == "workflow"
|
|
assert workflow.metadata["route"] == "/campaigns/{campaign_id}/mail-settings"
|
|
assert workflow.conditions[0].required_scopes == (
|
|
"campaigns:campaign:update",
|
|
"mail:profile:use",
|
|
)
|
|
assert workflow.metadata["prerequisites"]
|
|
assert workflow.metadata["steps"]
|
|
assert workflow.metadata["outcome"]
|
|
assert workflow.metadata["verification"]
|
|
assert "campaigns.mail-profile-governance" in workflow.metadata["related_topic_ids"]
|
|
|
|
assert topics["campaigns.mail-profile-governance"].metadata["kind"] == "reference"
|
|
assert topics["campaigns.mail-profile-operations"].metadata["kind"] == "reference"
|
|
assert topics["campaigns.workflow.prepare-validate-and-build"].metadata["kind"] == "workflow"
|
|
assert topics["campaigns.workflow.complete-review"].metadata["kind"] == "workflow"
|
|
assert topics["campaigns.workflow.retry-and-reconcile"].metadata["kind"] == "workflow"
|
|
assert topics["campaigns.reference.composition-assurance"].metadata["kind"] == "reference"
|
|
|
|
|
|
@pytest.mark.parametrize("legacy_key", ["smtp", "imap", "credentials", "inherit_smtp_credentials", "profile_id"])
|
|
def test_campaign_mail_contract_rejects_every_legacy_server_field(legacy_key: str) -> None:
|
|
raw = _campaign_json({"mail_profile_id": "profile-1", legacy_key: {}})
|
|
|
|
with pytest.raises(CampaignMailProfileBoundaryError, match="select an authorized Mail profile"):
|
|
assert_campaign_uses_mail_profile_reference(raw)
|
|
|
|
|
|
def test_persisted_schema_rejects_inline_transport_even_without_a_secret() -> None:
|
|
with pytest.raises(CampaignSchemaError, match="Additional properties are not allowed"):
|
|
validate_against_schema(_campaign_json({"smtp": {"host": "smtp.example.test"}}))
|
|
|
|
|
|
def test_loader_rejects_inline_transport_before_optional_mail_summary() -> None:
|
|
integration = SimpleNamespace(campaign_profile_delivery_summary=lambda *_args, **_kwargs: pytest.fail("must not resolve"))
|
|
with patch("govoplan_campaign.backend.persistence.campaigns.mail_integration", return_value=integration):
|
|
with pytest.raises(CampaignMailProfileBoundaryError, match="remove campaign-local SMTP/IMAP settings"):
|
|
load_campaign_config_from_json(
|
|
object(), # type: ignore[arg-type]
|
|
tenant_id="tenant-1",
|
|
raw_json=_campaign_json({"smtp": {"password": "secret"}}),
|
|
)
|
|
|
|
|
|
def test_loader_uses_only_non_secret_mail_profile_capabilities() -> None:
|
|
raw = _campaign_json({"mail_profile_id": "profile-1"})
|
|
|
|
def summary(_session, **kwargs):
|
|
assert kwargs["profile_id"] == "profile-1"
|
|
return {
|
|
"mail_profile_id": "profile-1",
|
|
"smtp_available": True,
|
|
"imap_available": False,
|
|
"smtp_transport_revision": "opaque-smtp",
|
|
"imap_transport_revision": None,
|
|
# Even a broken/malicious provider cannot inject extra material into
|
|
# Campaign's strict in-memory ServerConfig.
|
|
"host": "smtp.example.test",
|
|
"password": "secret",
|
|
}
|
|
|
|
integration = SimpleNamespace(campaign_profile_delivery_summary=summary)
|
|
with patch("govoplan_campaign.backend.persistence.campaigns.mail_integration", return_value=integration):
|
|
config = load_campaign_config_from_json(
|
|
object(), # type: ignore[arg-type]
|
|
tenant_id="tenant-1",
|
|
raw_json=raw,
|
|
)
|
|
|
|
assert raw["server"] == {"mail_profile_id": "profile-1"}
|
|
assert config.server.mail_profile_id == "profile-1"
|
|
assert config.server.profile_capabilities.smtp_available is True
|
|
assert config.server.profile_capabilities.imap_available is False
|
|
assert "smtp.example.test" not in repr(config.server)
|
|
assert "secret" not in repr(config.server)
|
|
|
|
|
|
def test_new_execution_snapshot_stores_reference_and_evidence_not_transport_material() -> None:
|
|
raw = _campaign_json({"mail_profile_id": "profile-1"})
|
|
version = SimpleNamespace(id="version-1", raw_json=raw)
|
|
|
|
payload, _digest = create_execution_snapshot(
|
|
version, # type: ignore[arg-type]
|
|
mail_profile_id="profile-1",
|
|
smtp_transport_revision="opaque-smtp-evidence",
|
|
imap_transport_revision="opaque-imap-evidence",
|
|
delivery=DeliveryConfig(),
|
|
)
|
|
|
|
assert payload["snapshot_version"] == "5"
|
|
assert payload["mail_profile_id"] == "profile-1"
|
|
assert "smtp" not in payload
|
|
assert "imap" not in payload
|
|
assert payload["smtp_transport_revision"] == "opaque-smtp-evidence"
|
|
assert payload["imap_transport_revision"] == "opaque-imap-evidence"
|
|
|
|
|
|
def test_legacy_execution_snapshot_is_preserved_but_fails_closed() -> None:
|
|
version = SimpleNamespace(
|
|
raw_json=_campaign_json({"mail_profile_id": "profile-1"}),
|
|
execution_snapshot={"snapshot_version": "3", "smtp": {"host": "legacy.example.test"}},
|
|
execution_snapshot_hash=None,
|
|
)
|
|
with patch(
|
|
"govoplan_campaign.backend.sending.execution.files_integration",
|
|
return_value=SimpleNamespace(available=False),
|
|
):
|
|
with pytest.raises(ExecutionSnapshotError, match="preserved for audit only"):
|
|
ensure_execution_snapshot(object(), version) # type: ignore[arg-type]
|
|
|
|
assert version.execution_snapshot["smtp"]["host"] == "legacy.example.test"
|
|
|
|
|
|
def test_campaign_mail_adapter_does_not_expose_raw_transport_helpers() -> None:
|
|
integration = MailCampaignIntegration(SimpleNamespace())
|
|
|
|
for name in (
|
|
"smtp_config_from_profile",
|
|
"imap_config_from_profile",
|
|
"send_email_bytes",
|
|
"send_email_message",
|
|
"materialize_campaign_mail_profile_config",
|
|
):
|
|
assert not hasattr(integration, name)
|
|
|
|
|
|
def test_editing_a_legacy_record_requires_an_explicit_profile_migration() -> None:
|
|
legacy_raw = _campaign_json({"smtp": {"host": "smtp.example.test", "password": "secret"}})
|
|
version = SimpleNamespace(id="version-1", campaign_id="campaign-1", raw_json=legacy_raw)
|
|
campaign = SimpleNamespace(id="campaign-1", current_version_id="version-1")
|
|
|
|
with (
|
|
patch("govoplan_campaign.backend.persistence.versions.get_campaign_version_for_tenant", return_value=version),
|
|
patch("govoplan_campaign.backend.persistence.versions._require_campaign", return_value=campaign),
|
|
patch("govoplan_campaign.backend.persistence.versions.ensure_current_working_version"),
|
|
patch("govoplan_campaign.backend.persistence.versions.is_version_locked", return_value=False),
|
|
):
|
|
with pytest.raises(CampaignPersistenceError, match="explicitly save the migration"):
|
|
update_campaign_version(
|
|
object(), # type: ignore[arg-type]
|
|
tenant_id="tenant-1",
|
|
campaign_id="campaign-1",
|
|
version_id="version-1",
|
|
raw_json=_campaign_json({"mail_profile_id": "profile-1"}),
|
|
)
|
|
|
|
assert version.raw_json is legacy_raw
|
|
assert legacy_raw["server"]["smtp"]["password"] == "secret" # type: ignore[index]
|
|
|
|
|
|
def test_fork_inherited_profile_requires_mail_profile_use_scope() -> None:
|
|
principal = SimpleNamespace(
|
|
tenant_id="tenant-1",
|
|
user=SimpleNamespace(id="user-1"),
|
|
)
|
|
campaign = SimpleNamespace(id="campaign-1")
|
|
source = SimpleNamespace(
|
|
id="version-1",
|
|
campaign_id="campaign-1",
|
|
raw_json={"server": {"mail_profile_id": "profile-1"}},
|
|
)
|
|
with (
|
|
patch.object(router, "_get_campaign_for_principal", return_value=campaign),
|
|
patch.object(router, "_require_permission"),
|
|
patch.object(router, "_get_version_for_tenant", return_value=source),
|
|
patch.object(router, "has_scope", return_value=False),
|
|
patch.object(router, "fork_campaign_version_for_edit") as fork,
|
|
):
|
|
with pytest.raises(HTTPException) as captured:
|
|
router.fork_version_for_edit(
|
|
"campaign-1",
|
|
"version-1",
|
|
CampaignVersionUpdateRequest(),
|
|
session=object(), # type: ignore[arg-type]
|
|
principal=principal, # type: ignore[arg-type]
|
|
)
|
|
|
|
assert captured.value.status_code == 403
|
|
fork.assert_not_called()
|