"""Batch transport reuse must never become an authorization or evidence cache.""" from __future__ import annotations import imaplib from dataclasses import asdict from types import SimpleNamespace from unittest.mock import Mock, patch import pytest from govoplan_core.security.credential_envelopes import CredentialEnvelope from govoplan_mail.backend import capabilities, server_hierarchy from govoplan_mail.backend.db.models import MailProfilePolicy, MailServerEndpoint from govoplan_mail.backend.mail_profiles import MailProfileError from govoplan_mail.backend.sending import imap as transport from govoplan_mail.backend.sending.imap import ImapAppendError, ImapConfigurationError from test_campaign_protocol_authorization import hierarchy, imap_args # noqa: F401 def client(): return SimpleNamespace( utf8_enabled=False, append=Mock(return_value=("OK", [b"provider internal secret response"])), logout=Mock(return_value=("BYE", [])), noop=Mock(return_value=("OK", [])), ) def recovery(): return SimpleNamespace(replayed=False, succeed_imap=Mock(), reject=Mock(), unknown=Mock()) def test_reuses_transport_but_decrypts_and_authorizes_each_message_and_records_each_effect(hierarchy): connection = client() effects = [recovery(), recovery()] with ( patch.object(transport, "_open_imap", return_value=connection) as open_connection, patch.object(capabilities, "begin_provider_effect_recovery", side_effect=effects) as begin, patch.object(capabilities, "_authorized_campaign_profile", wraps=capabilities._authorized_campaign_profile) as authorize, patch.object(capabilities, "assert_mail_policy_allows_send", wraps=capabilities.assert_mail_policy_allows_send) as policy, patch.object(server_hierarchy, "resolve_credential_envelope", wraps=server_hierarchy.resolve_credential_envelope) as decrypt, capabilities.campaign_imap_batch(tenant_id="tenant-1", campaign_id="campaign-1") as batch, ): assert batch.connection_count == 0 open_connection.assert_not_called() results = [capabilities.append_campaign_message_to_sent( hierarchy.session, **imap_args(message_bytes=f"message-{index}".encode(), recovery_effect_id=f"effect-{index}"), ) for index in range(2)] assert batch.connection_count == 1 assert batch.reconnect_count == 0 assert authorize.call_count == policy.call_count == 2 assert [call.kwargs["credential_id"] for call in decrypt.call_args_list] == ["imap-credential"] * 2 assert [call.kwargs["effect_id"] for call in begin.call_args_list] == ["effect-0", "effect-1"] assert [item.session_reused for item in results] == [False, True] assert [item.connection_sequence for item in results] == [1, 1] assert "secret" not in repr([asdict(item) for item in results]) assert "imap.example" not in repr(results) open_connection.assert_called_once() connection.logout.assert_called_once() for effect in effects: effect.succeed_imap.assert_called_once_with(folder="Sent") effect.reject.assert_not_called() effect.unknown.assert_not_called() @pytest.mark.parametrize("change", ["credential_revoked", "smtp_revision_changed", "imap_revision_changed"]) def test_next_message_rechecks_current_authority_and_frozen_revisions_before_decryption(hierarchy, change): connection = client() with ( patch.object(transport, "_open_imap", return_value=connection), patch.object(server_hierarchy, "resolve_credential_envelope", wraps=server_hierarchy.resolve_credential_envelope) as decrypt, capabilities.campaign_imap_batch(tenant_id="tenant-1", campaign_id="campaign-1"), ): capabilities.append_campaign_message_to_sent(hierarchy.session, **imap_args()) if change == "credential_revoked": hierarchy.session.get(CredentialEnvelope, "imap-credential").is_active = False else: protocol = change.split("_")[0] hierarchy.session.get(MailServerEndpoint, f"{protocol}-server").transport_revision = "changed" hierarchy.session.commit() with pytest.raises(MailProfileError): capabilities.append_campaign_message_to_sent(hierarchy.session, **imap_args(message_bytes=b"blocked")) assert decrypt.call_count == 1 connection.append.assert_called_once() def test_effective_credential_policy_is_not_cached_by_warm_batch(hierarchy): for scope in ("system", "tenant"): hierarchy.session.get(MailProfilePolicy, f"{scope}-policy").policy = { "smtp_credentials": {"inherit": False}, "imap_credentials": {"inherit": True}, } hierarchy.session.commit() connection = client() with ( patch.object(transport, "_open_imap", return_value=connection), patch.object(server_hierarchy, "resolve_credential_envelope", wraps=server_hierarchy.resolve_credential_envelope) as decrypt, capabilities.campaign_imap_batch(tenant_id="tenant-1", campaign_id="campaign-1"), ): capabilities.append_campaign_message_to_sent(hierarchy.session, **imap_args(imap_credential_id=None)) hierarchy.session.get(MailProfilePolicy, "tenant-policy").policy = { "imap_credentials": {"inherit": False}, } hierarchy.session.commit() with pytest.raises(MailProfileError, match="explicit credential"): capabilities.append_campaign_message_to_sent(hierarchy.session, **imap_args(imap_credential_id=None)) assert decrypt.call_count == 1 connection.append.assert_called_once() def test_copying_context_to_a_worker_cannot_share_the_batch_connection(hierarchy): from concurrent.futures import ThreadPoolExecutor from contextvars import copy_context with ( patch.object(capabilities, "_authorized_campaign_profile") as authorize, capabilities.campaign_imap_batch(tenant_id="tenant-1", campaign_id="campaign-1"), ThreadPoolExecutor(max_workers=1) as executor, ): future = executor.submit(copy_context().run, capabilities.append_campaign_message_to_sent, hierarchy.session, **imap_args()) with pytest.raises(ImapConfigurationError, match="scope"): future.result(timeout=5) authorize.assert_not_called() @pytest.mark.parametrize("scope", [{"tenant_id": "other"}, {"campaign_id": "other"}]) def test_cross_scope_batch_is_rejected_before_authorization_decryption_evidence_or_network(hierarchy, scope): with ( patch.object(capabilities, "_authorized_campaign_profile") as authorize, patch.object(capabilities, "begin_provider_effect_recovery") as begin, patch.object(transport, "_open_imap") as open_connection, capabilities.campaign_imap_batch(tenant_id="tenant-1", campaign_id="campaign-1"), pytest.raises(ImapConfigurationError, match="scope"), ): capabilities.append_campaign_message_to_sent(hierarchy.session, **imap_args(**scope)) authorize.assert_not_called() begin.assert_not_called() open_connection.assert_not_called() def test_changed_resolved_secret_does_not_reuse_previous_authenticated_connection(hierarchy): from govoplan_core.security.secrets import encrypt_secret connections = [client(), client()] with ( patch.object(transport, "_open_imap", side_effect=connections) as open_connection, capabilities.campaign_imap_batch(tenant_id="tenant-1", campaign_id="campaign-1") as batch, ): capabilities.append_campaign_message_to_sent(hierarchy.session, **imap_args()) hierarchy.session.get(CredentialEnvelope, "imap-credential").secret_data_encrypted = encrypt_secret('{"password":"new fake password"}') hierarchy.session.commit() result = capabilities.append_campaign_message_to_sent(hierarchy.session, **imap_args(message_bytes=b"new credential")) assert batch.connection_count == result.connection_sequence == 2 assert not result.session_reused assert open_connection.call_count == 2 assert open_connection.call_args_list[0].args[0].password != open_connection.call_args_list[1].args[0].password for connection in connections: connection.logout.assert_called_once() connection.append.assert_called_once() def test_nested_context_restores_outer_connection_and_always_cleans_up(hierarchy): outer_client, inner_client = client(), client() with patch.object(transport, "_open_imap", side_effect=[outer_client, inner_client]) as open_connection: with capabilities.campaign_imap_batch(tenant_id="tenant-1", campaign_id="campaign-1") as outer: capabilities.append_campaign_message_to_sent(hierarchy.session, **imap_args()) with pytest.raises(ValueError, match="caller"): with capabilities.campaign_imap_batch(tenant_id="tenant-1", campaign_id="campaign-1"): capabilities.append_campaign_message_to_sent(hierarchy.session, **imap_args()) raise ValueError("caller failure") inner_client.logout.assert_called_once() assert capabilities._ACTIVE_IMAP_BATCH.get() is outer result = capabilities.append_campaign_message_to_sent(hierarchy.session, **imap_args()) assert result.session_reused assert capabilities._ACTIVE_IMAP_BATCH.get() is None assert open_connection.call_count == 2 assert outer_client.append.call_count == 2 outer_client.logout.assert_called_once() def test_known_success_evidence_does_not_replay_even_with_warm_connection(hierarchy): connection = client() first, replayed = recovery(), recovery() replayed.replayed = True with ( patch.object(transport, "_open_imap", return_value=connection), patch.object(capabilities, "begin_provider_effect_recovery", side_effect=[first, replayed]), capabilities.campaign_imap_batch(tenant_id="tenant-1", campaign_id="campaign-1"), ): capabilities.append_campaign_message_to_sent(hierarchy.session, **imap_args(recovery_effect_id="same-effect")) with pytest.raises(ImapAppendError, match="already succeeded") as caught: capabilities.append_campaign_message_to_sent(hierarchy.session, **imap_args(recovery_effect_id="same-effect")) assert caught.value.outcome_unknown connection.append.assert_called_once() replayed.succeed_imap.assert_not_called() def test_ambiguous_effect_is_recorded_once_sanitized_and_never_replayed(hierarchy): connection = client() connection.append.side_effect = imaplib.IMAP4.abort("provider host and secret") effect = recovery() with ( patch.object(transport, "_open_imap", return_value=connection) as open_connection, patch.object(capabilities, "begin_provider_effect_recovery", return_value=effect), capabilities.campaign_imap_batch(tenant_id="tenant-1", campaign_id="campaign-1"), pytest.raises(ImapAppendError) as caught, ): capabilities.append_campaign_message_to_sent(hierarchy.session, **imap_args(recovery_effect_id="effect-1")) assert caught.value.outcome_unknown assert "secret" not in str(caught.value) assert "secret" not in repr(effect.unknown.call_args) effect.unknown.assert_called_once() effect.succeed_imap.assert_not_called() effect.reject.assert_not_called() open_connection.assert_called_once() connection.append.assert_called_once() connection.logout.assert_called_once() def test_evidence_finalization_failure_closes_batch_and_prevents_further_effects(hierarchy): connection = client() effect = recovery() effect.succeed_imap.side_effect = OSError("evidence store unavailable") with ( patch.object(transport, "_open_imap", return_value=connection), patch.object(capabilities, "begin_provider_effect_recovery", return_value=effect), capabilities.campaign_imap_batch(tenant_id="tenant-1", campaign_id="campaign-1"), ): with pytest.raises(ImapAppendError) as caught: capabilities.append_campaign_message_to_sent(hierarchy.session, **imap_args(recovery_effect_id="effect-1")) assert caught.value.outcome_unknown with pytest.raises(ImapConfigurationError): capabilities.append_campaign_message_to_sent(hierarchy.session, **imap_args(message_bytes=b"must not run")) connection.append.assert_called_once() connection.logout.assert_called_once()