"""Exercise real version routes/persistence without transport or filesystem effects.""" from __future__ import annotations import copy from types import SimpleNamespace from unittest.mock import Mock, 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.archive_encryption import ( CampaignArchiveEncryptionError, assert_archive_encryption_allowed, stamp_legacy_zipcrypto_acknowledgements, ) from govoplan_campaign.backend.campaign.mail_profile_boundary import ( CampaignMailProfileBoundaryError, assert_campaign_uses_mail_profile_reference, ) from govoplan_campaign.backend.db.models import Campaign, CampaignIssue, CampaignVersion from govoplan_campaign.backend.integrations import MailProfileError from govoplan_campaign.backend.persistence.campaigns import ( CampaignPersistenceError, build_campaign_version, load_campaign_config_from_json, validate_campaign_version, ) from govoplan_campaign.backend.schemas import CampaignVersionDetailResponse, CampaignVersionUpdateRequest from govoplan_core.auth import ApiPrincipal from govoplan_core.core.access import PrincipalRef from govoplan_core.core.change_sequence import ChangeSequenceEntry from govoplan_core.db.base import Base def _principal(*scopes: str) -> ApiPrincipal: return ApiPrincipal( principal=PrincipalRef( account_id="account-1", membership_id="user-1", tenant_id="tenant-1", scopes=frozenset(scopes), ), account=SimpleNamespace(id="account-1"), user=SimpleNamespace(id="user-1"), ) def _legacy() -> dict: return { "version": "1.0", "campaign": {"id": "campaign-1", "name": "Campaign", "mode": "send"}, "server": { "mail_profile_id": "profile-1", "smtp": {"host": "old.example.test", "password": "stored-secret"}, "imap": None, "credentials": {}, "inherit_smtp_credentials": True, }, "recipients": {"from": [{"email": "sender@example.test"}]}, "template": {"subject": "Subject", "text": "Body", "body_mode": "text"}, "entries": {"inline": []}, "attachments": {"zip": {"enabled": True, "archives": [{ "id": "archive-1", "method": "zip_standard", "password_enabled": True, "password_delivery_channel": "separate_mail", }]}}, } @pytest.fixture def repair(): engine = create_engine("sqlite+pysqlite:///:memory:") for name in ("access_users", "access_groups"): if name not in Base.metadata.tables: 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__, ]) with Session(engine) as session: campaign = Campaign( id="campaign-1", tenant_id="tenant-1", external_id="campaign-1", name="Campaign", current_version_id="draft-1", ) draft = CampaignVersion(id="draft-1", campaign_id=campaign.id, version_number=2, raw_json=_legacy()) history = CampaignVersion(id="history-1", campaign_id=campaign.id, version_number=1, raw_json=_legacy(), workflow_state="final") session.add_all((campaign, draft, history)) session.commit() audits: list[dict] = [] def audit(_session, _principal, **kwargs): audits.append(kwargs) if kwargs.get("commit"): _session.commit() integration = SimpleNamespace(assert_campaign_mail_policy_allows_json=Mock()) with ( patch.object(route_support, "_get_campaign_for_principal", return_value=campaign), patch.object(route_support, "audit_from_principal", side_effect=audit), patch("govoplan_campaign.backend.persistence.versions.mail_integration", return_value=integration), patch("govoplan_campaign.backend.archive_encryption.get_registry", return_value=None), ): yield SimpleNamespace(session=session, campaign=campaign, draft=draft, history=history, integration=integration, audits=audits) engine.dispose() def _public(repair) -> dict: return CampaignVersionDetailResponse.model_validate(repair.draft).raw_json def _save(repair, raw: dict, *, migrate: bool = False, principal=None): return route_support._update_campaign_version_detail_response( repair.session, principal or _principal("mail:profile:use"), repair.campaign.id, repair.draft.id, CampaignVersionUpdateRequest( campaign_json=raw, base_revision=repair.draft.edit_revision, migrate_legacy_mail_settings=migrate, ), if_match=repair.draft.strong_etag, autosave=True, audit_action="campaign.version_autosaved", ) @pytest.mark.parametrize("mail_first", [True, False]) def test_independent_repairs_persist_in_either_order_without_altering_history(repair, mail_first): original_zip = copy.deepcopy(repair.draft.raw_json["attachments"]["zip"]) original_server = copy.deepcopy(repair.draft.raw_json["server"]) first = _public(repair) if not mail_first: first["attachments"]["zip"]["archives"][0]["method"] = "aes" result = _save(repair, first, migrate=mail_first) repair.session.expire_all() assert repair.draft.edit_revision == 2 assert result.mail_profile_migration_required is not mail_first assert "stored-secret" not in repr(result) assert repair.draft.build_summary is None assert repair.draft.execution_snapshot is None if mail_first: assert repair.draft.raw_json["attachments"]["zip"] == original_zip assert repair.draft.raw_json["server"] == {"mail_profile_id": "profile-1"} # Real validation/build gates still reject the unresolved ZIP policy. for action in (validate_campaign_version, build_campaign_version): with patch( "govoplan_campaign.backend.persistence.campaigns.load_version_config", return_value=(repair.draft, None, object()), ), pytest.raises(CampaignPersistenceError, match="blocked"): action(repair.session, tenant_id="tenant-1", version_id=repair.draft.id) else: assert repair.draft.raw_json["server"] == original_server repair.integration.assert_campaign_mail_policy_allows_json.assert_not_called() # The authoritative config loader still blocks legacy transport. with pytest.raises(CampaignMailProfileBoundaryError, match="remove campaign-local"): load_campaign_config_from_json(repair.session, tenant_id="tenant-1", raw_json=repair.draft.raw_json) second = _public(repair) if mail_first: second["attachments"]["zip"]["archives"][0]["method"] = "aes" result = _save(repair, second, migrate=not mail_first) repair.session.expire_all() assert repair.draft.edit_revision == 3 assert not result.mail_profile_migration_required assert_campaign_uses_mail_profile_reference(repair.draft.raw_json) assert_archive_encryption_allowed(repair.session, repair.campaign, repair.draft.raw_json) assert repair.history.raw_json == _legacy() assert len(repair.audits) == 2 assert repair.audits[0]["details"]["legacy_mail_settings_preserved"] is not mail_first assert repair.audits[0]["details"]["legacy_mail_settings_migrated"] is mail_first assert not any(item["details"]["legacy_zipcrypto_acknowledgements"] for item in repair.audits) def test_zip_repair_does_not_require_use_of_unchanged_revoked_mail_profile(repair): repair.integration.assert_campaign_mail_policy_allows_json.side_effect = MailProfileError("Profile revoked") raw = _public(repair) raw["attachments"]["zip"]["archives"][0]["method"] = "aes" result = _save(repair, raw, principal=_principal()) assert result.mail_profile_migration_required repair.integration.assert_campaign_mail_policy_allows_json.assert_not_called() with pytest.raises(HTTPException) as denied: _save(repair, _public(repair), migrate=True, principal=_principal()) assert denied.value.status_code == 403 assert "mail:profile:use" in denied.value.detail with pytest.raises(HTTPException, match="Profile revoked"): _save(repair, _public(repair), migrate=True) assert repair.draft.mail_profile_migration_required def test_archive_repair_is_saveable_after_mail_migration_and_a_stricter_credential_policy(repair): _save(repair, _public(repair), migrate=True) repair.integration.assert_campaign_mail_policy_allows_json.reset_mock() repair.integration.assert_campaign_mail_policy_allows_json.side_effect = MailProfileError( "SMTP credential policy requires an explicit credential selection" ) raw = _public(repair) raw["attachments"]["zip"]["archives"][0]["method"] = "aes" result = _save(repair, raw, principal=_principal()) assert not result.mail_profile_migration_required assert result.raw_json["server"] == {"mail_profile_id": "profile-1"} assert result.raw_json["attachments"]["zip"]["archives"][0]["method"] == "aes" repair.integration.assert_campaign_mail_policy_allows_json.assert_not_called() changed_mail = _public(repair) changed_mail["server"]["smtp_server_id"] = "smtp-1" with pytest.raises(HTTPException) as missing_permission: _save(repair, changed_mail, principal=_principal()) assert missing_permission.value.status_code == 403 with pytest.raises(HTTPException, match="explicit credential selection"): _save(repair, changed_mail) assert repair.draft.raw_json["server"] == {"mail_profile_id": "profile-1"} @pytest.mark.parametrize("mutation", ["new_profile", "remove_profile", "credential", "inline", "echo_inline"]) def test_archive_save_cannot_modify_or_introduce_mail_transport_without_migration(repair, mutation): raw = _public(repair) raw["attachments"]["zip"]["archives"][0]["method"] = "aes" if mutation == "new_profile": raw["server"]["mail_profile_id"] = "other-profile" elif mutation == "remove_profile": raw["server"] = {} elif mutation == "credential": raw["server"].update(smtp_server_id="smtp-1", smtp_credential_id="credential-1") elif mutation == "inline": raw["server"]["smtp"] = {"password": "injected-secret"} else: raw["server"] = copy.deepcopy(repair.draft.raw_json["server"]) with pytest.raises(HTTPException): _save(repair, raw) assert repair.draft.raw_json == _legacy() assert repair.draft.edit_revision == 1 @pytest.mark.parametrize("mutation", ["acknowledgement", "reason", "actor", "method", "channel", "typed_boolean"]) def test_mail_migration_cannot_grandfather_modified_zip_settings(repair, mutation): raw = _public(repair) archive = raw["attachments"]["zip"]["archives"][0] if mutation == "typed_boolean": archive["password_enabled"] = 1 # Python True == 1 is not exact JSON equality. elif mutation == "method": archive["method"] = "unknown" elif mutation == "channel": archive["method"] = "aes" archive["password_delivery_channel"] = "same_mail" else: archive[{ "acknowledgement": "legacy_zipcrypto_acknowledged", "reason": "legacy_zipcrypto_reason", "actor": "legacy_zipcrypto_acknowledged_by", }[mutation]] = True if mutation == "acknowledgement" else "forged evidence" with pytest.raises(HTTPException): _save(repair, raw, migrate=True) assert repair.draft.raw_json == _legacy() assert repair.draft.edit_revision == 1 def test_unchanged_zip_is_not_restamped_even_after_legacy_permission_revocation(repair): raw = _public(repair) archive = raw["attachments"]["zip"]["archives"][0] archive.update( legacy_zipcrypto_acknowledged=True, legacy_zipcrypto_reason="Existing recipient compatibility requirement", legacy_zipcrypto_acknowledged_by="original-actor", legacy_zipcrypto_acknowledged_at="2026-08-01T10:00:00+00:00", ) repair.draft.raw_json = {**copy.deepcopy(repair.draft.raw_json), "attachments": copy.deepcopy(raw["attachments"])} repair.session.commit() result = _save(repair, raw, migrate=True, principal=_principal("mail:profile:use")) assert result.raw_json["attachments"]["zip"]["archives"][0] == archive assert repair.audits[-1]["details"]["legacy_zipcrypto_acknowledgements"] == [] with pytest.raises(CampaignArchiveEncryptionError, match="blocked"): assert_archive_encryption_allowed(repair.session, repair.campaign, repair.draft.raw_json) with patch( "govoplan_campaign.backend.archive_encryption.effective_archive_encryption_policy", return_value=SimpleNamespace( available=True, allowed_password_encryption_methods={"aes", "zip_standard"}, allowed_password_delivery_channels={"separate_mail"}, reason="Policy allows legacy", ), ), pytest.raises(CampaignArchiveEncryptionError, match="Missing scope"): assert_archive_encryption_allowed( repair.session, repair.campaign, repair.draft.raw_json, principal=_principal("mail:profile:use"), ) def test_unchanged_missing_zip_remains_noop_without_policy_lookup(repair): with patch("govoplan_campaign.backend.archive_encryption.effective_archive_encryption_policy") as policy: candidate, evidence = stamp_legacy_zipcrypto_acknowledgements( repair.session, repair.campaign, {}, {"template": {"text": "New"}}, principal=_principal(), ) assert candidate == {"template": {"text": "New"}} assert evidence == [] policy.assert_not_called() @pytest.mark.parametrize("target", ["individual", "global"]) def test_recipient_address_order_round_trips_through_the_real_save_route(repair, target): addresses = [ {"name": "Zulu", "email": "zulu@example.test"}, {"name": "Alpha", "email": "alpha@example.test"}, {"name": "Beta", "email": "beta@example.test"}, ] raw = _public(repair) if target == "individual": raw["entries"]["inline"] = [{"id": "entry-1", "to": addresses}] else: raw["recipients"]["to"] = addresses result = _save(repair, raw, principal=_principal("campaigns:recipient:write")) repair.session.expire_all() persisted = repair.session.get(CampaignVersion, "draft-1") assert persisted is not None if target == "individual": assert persisted.raw_json["entries"]["inline"][0]["to"] == addresses assert result.raw_json["entries"]["inline"][0]["to"] == addresses else: assert persisted.raw_json["recipients"]["to"] == addresses assert result.raw_json["recipients"]["to"] == addresses assert persisted.mail_profile_migration_required assert persisted.edit_revision == 2