from __future__ import annotations from types import SimpleNamespace from unittest.mock import Mock, call, patch import pytest from fastapi import HTTPException from sqlalchemy import Column, String, Table, create_engine from sqlalchemy.orm import Session from govoplan_campaign.backend import route_support from govoplan_campaign.backend.routes import attachments as attachment_routes from govoplan_campaign.backend.routes import versions as 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.db.models import Campaign, CampaignIssue, CampaignVersion from govoplan_campaign.backend.persistence.campaigns import CampaignPersistenceError, load_campaign_config_from_json from govoplan_campaign.backend.persistence.versions import _updated_runtime_json, update_campaign_version from govoplan_campaign.backend.integrations import MailCampaignIntegration, MailProfileError from govoplan_campaign.backend.sending.execution import ExecutionSnapshotError, create_execution_snapshot, ensure_execution_snapshot from govoplan_campaign.backend.schemas import CampaignVersionUpdateRequest from govoplan_core.core.change_sequence import ChangeSequenceEntry from govoplan_core.db.base import Base 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_attachment_preview_reports_legacy_mail_boundary_as_validation_error() -> None: campaign = SimpleNamespace(id="campaign-1") version = SimpleNamespace( id="version-1", campaign_id=campaign.id, raw_json=_campaign_json({"smtp": {"host": "legacy.example.test"}}), ) principal = SimpleNamespace( tenant_id="tenant-1", user=SimpleNamespace(id="user-1"), ) with ( patch.object(attachment_routes, "_get_campaign_for_principal"), patch.object(attachment_routes, "_require_permission"), patch.object(attachment_routes, "_get_campaign_for_tenant", return_value=campaign), patch.object(attachment_routes, "_get_version_for_tenant", return_value=version), patch.object(attachment_routes, "_require_mail_profile_use_if_needed"), patch.object( attachment_routes, "_attachment_preview_for_version", side_effect=CampaignMailProfileBoundaryError("Select an authorized Mail profile."), ), pytest.raises(HTTPException) as captured, ): attachment_routes.preview_campaign_attachments( campaign.id, version.id, session=object(), # type: ignore[arg-type] principal=principal, # type: ignore[arg-type] ) assert captured.value.status_code == 422 assert captured.value.detail == "Select an authorized Mail profile." 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"] == "9" 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_explicit_legacy_migration_accepts_an_unchanged_authorized_profile_reference() -> None: legacy = _campaign_json({ "mail_profile_id": "profile-1", "smtp": {"host": "smtp.example.test", "password": "legacy-secret"}, "imap": None, "credentials": {}, "inherit_smtp_credentials": True, "inherit_imap_credentials": True, }) submitted = _campaign_json({"mail_profile_id": "profile-1"}) integration = SimpleNamespace(assert_campaign_mail_policy_allows_json=Mock()) session = object() with patch("govoplan_campaign.backend.persistence.versions.mail_integration", return_value=integration): result = _updated_runtime_json( session, # type: ignore[arg-type] tenant_id="tenant-1", campaign=SimpleNamespace(id="campaign-1"), version=SimpleNamespace(raw_json=legacy), raw_json=submitted, source_base_path=None, migrate_legacy_mail_settings=True, ) assert result == submitted assert result is not submitted assert campaign_mail_profile_boundary_violations(result) == () assert "legacy-secret" not in repr(result) assert legacy["server"]["smtp"]["password"] == "legacy-secret" # type: ignore[index] integration.assert_campaign_mail_policy_allows_json.assert_called_once_with( session, tenant_id="tenant-1", raw_json=result, campaign_id="campaign-1" ) def test_explicit_legacy_migration_requires_a_profile_and_preserves_mail_authorization() -> None: legacy = _campaign_json({"smtp": {"password": "legacy-secret"}}) integration = SimpleNamespace(assert_campaign_mail_policy_allows_json=Mock( side_effect=MailProfileError("Mail profile is not authorized for this campaign") )) with patch("govoplan_campaign.backend.persistence.versions.mail_integration", return_value=integration): for submitted, expected_error in ( (_campaign_json(), CampaignPersistenceError), (_campaign_json({"mail_profile_id": "unauthorized-profile"}), MailProfileError), (_campaign_json({"mail_profile_id": "profile-1", "smtp": {}}), CampaignMailProfileBoundaryError), ): with pytest.raises(expected_error): _updated_runtime_json( object(), # type: ignore[arg-type] tenant_id="tenant-1", campaign=SimpleNamespace(id="campaign-1"), version=SimpleNamespace(raw_json=legacy), raw_json=submitted, source_base_path=None, migrate_legacy_mail_settings=True, ) assert integration.assert_campaign_mail_policy_allows_json.call_count == 1 assert legacy["server"]["smtp"]["password"] == "legacy-secret" # type: ignore[index] def test_persisted_legacy_draft_can_migrate_then_save_other_content_without_changing_history() -> None: engine = create_engine("sqlite+pysqlite:///:memory:") for table_name in ("access_users", "access_groups"): if table_name not in Base.metadata.tables: Table(table_name, Base.metadata, Column("id", String(36), primary_key=True)) Base.metadata.create_all(engine, tables=[ Base.metadata.tables["access_users"], Base.metadata.tables["access_groups"], ChangeSequenceEntry.__table__, Campaign.__table__, CampaignVersion.__table__, CampaignIssue.__table__, ]) legacy = _campaign_json({ "mail_profile_id": "profile-1", "smtp": {"host": "old.example.test", "password": "historical-secret"}, "imap": None, "credentials": {}, "inherit_smtp_credentials": True, "inherit_imap_credentials": True, }) integration = SimpleNamespace(assert_campaign_mail_policy_allows_json=Mock()) try: with Session(engine) as session, patch( "govoplan_campaign.backend.persistence.versions.mail_integration", return_value=integration ): campaign = Campaign(id="campaign-1", tenant_id="tenant-1", external_id="campaign-1", name="Campaign", current_version_id="draft-1") historical = CampaignVersion(id="history-1", campaign_id=campaign.id, version_number=1, raw_json=legacy, workflow_state="final") draft = CampaignVersion(id="draft-1", campaign_id=campaign.id, version_number=2, raw_json=legacy, editor_state={"created_from": "minimal_campaign"}) session.add_all((campaign, historical, draft)) session.commit() original_revision = draft.edit_revision assert draft.mail_profile_migration_required migrated = update_campaign_version( session, tenant_id="tenant-1", campaign_id=campaign.id, version_id=draft.id, raw_json=_campaign_json({"mail_profile_id": "profile-1"}), editor_state={"created_from": "minimal_campaign"}, expected_revision=original_revision, migrate_legacy_mail_settings=True, autosave=True, ) session.expire_all() persisted = session.get(CampaignVersion, migrated.id) assert persisted is not None assert not persisted.mail_profile_migration_required assert persisted.raw_json["server"] == {"mail_profile_id": "profile-1"} assert persisted.edit_revision > original_revision assert persisted.autosaved_at is not None assert "historical-secret" not in repr(persisted.raw_json) assert session.get(CampaignVersion, historical.id).raw_json == legacy edited = _campaign_json({"mail_profile_id": "profile-1"}) edited["template"]["subject"] = "Edited after migration" # type: ignore[index] updated = update_campaign_version( session, tenant_id="tenant-1", campaign_id=campaign.id, version_id=persisted.id, raw_json=edited, expected_revision=persisted.edit_revision, ) session.expire_all() assert session.get(CampaignVersion, updated.id).raw_json["template"]["subject"] == "Edited after migration" assert session.get(CampaignVersion, historical.id).raw_json == legacy assert integration.assert_campaign_mail_policy_allows_json.call_count == 1 finally: engine.dispose() 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(route_support, "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(route_support, "_require_campaign_profile_use_if_needed") as require_profile: route_support._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"), ]