from __future__ import annotations from types import SimpleNamespace from unittest.mock import call, 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 authorized Mail resources"): 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"] == "7" 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() def test_campaign_wide_effect_authorizes_every_affected_version() -> None: session = object() principal = SimpleNamespace(tenant_id="tenant-1") with patch.object(router, "_require_campaign_profile_use_if_needed") as require_profile: router._require_campaign_versions_profile_use( # noqa: SLF001 - security boundary regression test session, # type: ignore[arg-type] principal, # type: ignore[arg-type] "campaign-1", {"version-2", "version-1"}, ) assert require_profile.call_args_list == [ call(session, principal, "campaign-1", "version-1"), call(session, principal, "campaign-1", "version-2"), ]