Release govoplan-mail v0.1.27: stabilize credentials, folder encoding and transport progress

tests/test_campaign_imap_batch.py

@@ -0,0 +1,233 @@
+"""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()

tests/test_campaign_protocol_authorization.py

@@ -0,0 +1,187 @@
+"""Real stored hierarchy and policy; provider calls are replaced, never live."""
+from __future__ import annotations
+
+import json
+from types import SimpleNamespace
+from unittest.mock import Mock, patch
+
+import pytest
+from sqlalchemy import Column, String, Table, create_engine
+from sqlalchemy.orm import Session
+
+from govoplan_core.admin.models import SystemSettings
+from govoplan_core.core.campaigns import CampaignMailPolicyContext
+from govoplan_core.db.base import Base
+from govoplan_core.security.credential_envelopes import CredentialEnvelope
+from govoplan_core.security.secrets import encrypt_secret
+from govoplan_core.tenancy.scope import Tenant
+from govoplan_mail.backend import capabilities, mail_profiles, server_hierarchy
+from govoplan_mail.backend.db.models import MailProfilePolicy, MailServerCredentialBinding, MailServerEndpoint, MailServerProfile
+from govoplan_mail.backend.sending import smtp as smtp_module
+
+
+@pytest.fixture
+def hierarchy(tmp_path):
+    engine = create_engine(f"sqlite+pysqlite:///{tmp_path / 'smtp-policy.db'}")
+    if "access_users" not in Base.metadata.tables:
+        Table("access_users", Base.metadata, Column("id", String(36), primary_key=True))
+    for table in (Base.metadata.tables["access_users"], SystemSettings.__table__, Tenant.__table__,
+                  CredentialEnvelope.__table__, MailServerProfile.__table__, MailServerEndpoint.__table__,
+                  MailServerCredentialBinding.__table__, MailProfilePolicy.__table__):
+        table.create(engine)
+    with Session(engine) as session:
+        session.add(SystemSettings(id="global", settings={}))
+        session.add(Tenant(id="tenant-1", slug="test", name="Test", settings={}))
+        profile = MailServerProfile(id="profile-1", tenant_id="tenant-1", scope_type="tenant", scope_id="tenant-1", name="Test mail", slug="test",
+            smtp_config={"host": "smtp.example.test", "port": 587, "security": "starttls"},
+            imap_config={"host": "imap.example.test", "port": 993, "security": "tls"},
+            smtp_transport_revision="smtp-current", imap_transport_revision="imap-current", inherit_to_lower_scopes=True)
+        session.add(profile)
+        for protocol, port in (("smtp", 587), ("imap", 993)):
+            server = MailServerEndpoint(id=f"{protocol}-server", profile_id=profile.id, tenant_id="tenant-1", protocol=protocol,
+                name=protocol, scope_type="tenant", scope_id="tenant-1", inherit_to_lower_scopes=True,
+                is_default=True, is_active=True, transport_revision=f"{protocol}-current",
+                config={"host": f"{protocol}.example.test", "port": port, "security": "starttls" if protocol == "smtp" else "tls"})
+            credential = CredentialEnvelope(id=f"{protocol}-credential", tenant_id="tenant-1", scope_type="tenant", scope_id="tenant-1",
+                name=protocol, credential_kind="username_password", public_data={"username": f"{protocol}-user"},
+                secret_data_encrypted=encrypt_secret(json.dumps({"password": f"fake-{protocol}-password"})), secret_keys=["password"],
+                allowed_modules=["mail"], allowed_server_refs=[f"mail:{protocol}-server"], inherit_to_lower_scopes=True, is_active=True)
+            session.add_all([server, credential, MailServerCredentialBinding(id=f"{protocol}-binding", server_id=server.id,
+                credential_id=credential.id, is_default=True)])
+        for scope in ("system", "tenant"):
+            session.add(MailProfilePolicy(id=f"{scope}-policy", tenant_id=None if scope == "system" else "tenant-1",
+                scope_type=scope, scope_id=None if scope == "system" else "tenant-1",
+                policy={"smtp_credentials": {"inherit": False}, "imap_credentials": {"inherit": False}}))
+        session.commit()
+        context = CampaignMailPolicyContext(id="campaign-1", tenant_id="tenant-1")
+        provider = SimpleNamespace(get_campaign_mail_policy_context=lambda *_args, **_kwargs: context)
+        # Only the optional Campaign context provider and external I/O are mocked.
+        # Profiles, policy inheritance, endpoint/credential ACLs and decryption are real.
+        with patch.object(mail_profiles, "_campaign_policy_provider", return_value=provider), \
+             patch("socket.create_connection", side_effect=AssertionError("No live network in regression tests")):
+            yield SimpleNamespace(session=session, profile=profile, context=context)
+    engine.dispose()
+
+
+def smtp_args(**overrides):
+    return {"tenant_id": "tenant-1", "campaign_id": "campaign-1", "profile_id": "profile-1",
+            "envelope_from": "sender@example.test", "envelope_recipients": ["recipient@example.test"], "from_header": "sender@example.test",
+            "expected_smtp_transport_revision": "smtp-current", "smtp_server_id": "smtp-server", "smtp_credential_id": "smtp-credential", **overrides}
+
+
+def selection():
+    return {"smtp_server_id": "smtp-server", "smtp_credential_id": "smtp-credential", "imap_server_id": "imap-server", "imap_credential_id": "imap-credential"}
+
+
+def test_smtp_batch_accepts_explicit_smtp_when_other_protocol_requires_explicit_selection(hierarchy):
+    # The full frozen selection is valid; runtime SMTP deliberately carries only SMTP.
+    summary = capabilities.campaign_profile_delivery_summary(hierarchy.session, tenant_id="tenant-1", campaign_id="campaign-1", profile_id="profile-1", **selection())
+    assert summary["smtp_available"] and summary["imap_available"]
+    fake_connection = Mock()
+    with patch.object(smtp_module, "_open_smtp", return_value=fake_connection) as opener, \
+         patch.object(server_hierarchy, "resolve_credential_envelope", wraps=server_hierarchy.resolve_credential_envelope) as decrypt:
+        with capabilities.campaign_smtp_batch(hierarchy.session, **smtp_args()) as batch:
+            assert batch.status == "ready"
+            assert batch.connection_count == 1
+    assert [call.kwargs["credential_id"] for call in decrypt.call_args_list] == ["smtp-credential"]
+    assert opener.call_args.args[0].username == "smtp-user"
+    fake_connection.sendmail.assert_not_called()
+    fake_connection.send_message.assert_not_called()
+
+
+def test_smtp_single_uses_same_selected_protocol_authorization(hierarchy):
+    result = SimpleNamespace(envelope_recipients=["recipient@example.test"], refused_recipients={})
+    with patch.object(capabilities, "send_email_bytes", return_value=result) as send, \
+         patch.object(server_hierarchy, "resolve_credential_envelope", wraps=server_hierarchy.resolve_credential_envelope) as decrypt:
+        sent = capabilities.send_campaign_email_bytes(hierarchy.session, message_bytes=b"frozen test message", **smtp_args())
+    assert sent.accepted_count == 1
+    assert [call.kwargs["credential_id"] for call in decrypt.call_args_list] == ["smtp-credential"]
+    assert send.call_args.kwargs["smtp_config"].username == "smtp-user"
+
+
+def imap_args(**overrides):
+    return {"tenant_id": "tenant-1", "campaign_id": "campaign-1", "profile_id": "profile-1",
+            "message_bytes": b"frozen test message", "folder": "Sent",
+            "expected_smtp_transport_revision": "smtp-current", "expected_imap_transport_revision": "imap-current",
+            # An IMAP-only call need not submit an unrelated SMTP credential.
+            "smtp_server_id": "smtp-server", "imap_server_id": "imap-server", "imap_credential_id": "imap-credential", **overrides}
+
+
+def test_imap_append_checks_and_decrypts_only_selected_protocol(hierarchy):
+    with patch.object(capabilities, "append_message_to_sent", return_value=SimpleNamespace(folder="Sent")) as append, \
+         patch.object(server_hierarchy, "resolve_credential_envelope", wraps=server_hierarchy.resolve_credential_envelope) as decrypt:
+        result = capabilities.append_campaign_message_to_sent(hierarchy.session, **imap_args())
+    assert result.folder == "Sent"
+    assert [call.kwargs["credential_id"] for call in decrypt.call_args_list] == ["imap-credential"]
+    assert append.call_args.kwargs["imap_config"].username == "imap-user"
+
+
+@pytest.mark.parametrize("operation", ["smtp_batch", "smtp_single", "imap_append", "imap_endpoint_only"])
+def test_selected_protocol_still_requires_explicit_credentials_before_decryption_or_provider(hierarchy, operation):
+    if operation == "imap_endpoint_only":
+        hierarchy.profile.imap_config = None  # Current endpoint exists without the legacy mirror.
+        hierarchy.session.commit()
+    with patch.object(server_hierarchy, "resolve_credential_envelope", side_effect=AssertionError("Policy must reject before decryption")) as decrypt, \
+         patch.object(smtp_module, "_open_smtp", side_effect=AssertionError("No network")) as opener, \
+         patch.object(capabilities, "send_email_bytes", side_effect=AssertionError("No SMTP")) as send, \
+         patch.object(capabilities, "append_message_to_sent", side_effect=AssertionError("No IMAP")) as append:
+        with pytest.raises(mail_profiles.MailProfileError, match=f"effective {'SMTP' if operation.startswith('smtp') else 'IMAP'}"):
+            if operation == "smtp_batch":
+                with capabilities.campaign_smtp_batch(hierarchy.session, **smtp_args(smtp_credential_id=None)): pass
+            elif operation == "smtp_single":
+                capabilities.send_campaign_email_bytes(hierarchy.session, message_bytes=b"test", **smtp_args(smtp_credential_id=None))
+            else:
+                capabilities.append_campaign_message_to_sent(hierarchy.session, **imap_args(imap_credential_id=None))
+    decrypt.assert_not_called(); opener.assert_not_called(); send.assert_not_called(); append.assert_not_called()
+
+
+@pytest.mark.parametrize("operation", ["smtp_batch", "smtp_single", "imap_append"])
+@pytest.mark.parametrize("mutation", ["stale_revision", "inactive_credential", "wrong_server", "wrong_tenant"])
+def test_selected_transport_revision_and_credential_authority_remain_fail_closed(hierarchy, operation, mutation):
+    protocol = "smtp" if operation.startswith("smtp") else "imap"
+    overrides = {}
+    if mutation == "stale_revision":
+        overrides[f"expected_{protocol}_transport_revision"] = "stale-build-revision"
+    elif mutation == "wrong_server":
+        overrides[f"{protocol}_credential_id"] = "imap-credential" if protocol == "smtp" else "smtp-credential"
+    else:
+        credential = hierarchy.session.get(CredentialEnvelope, f"{protocol}-credential")
+        if mutation == "inactive_credential": credential.is_active = False
+        else: credential.tenant_id = "other-tenant"
+        hierarchy.session.commit()
+    with patch.object(server_hierarchy, "resolve_credential_envelope", side_effect=AssertionError("Reject stale/unauthorized before decrypt")) as decrypt, \
+         patch.object(smtp_module, "_open_smtp", side_effect=AssertionError("No network")) as opener, \
+         patch.object(capabilities, "send_email_bytes", side_effect=AssertionError("No SMTP")) as send, \
+         patch.object(capabilities, "append_message_to_sent", side_effect=AssertionError("No IMAP")) as append:
+        with pytest.raises(mail_profiles.MailProfileError):
+            if operation == "smtp_batch":
+                with capabilities.campaign_smtp_batch(hierarchy.session, **smtp_args(**overrides)): pass
+            elif operation == "smtp_single":
+                capabilities.send_campaign_email_bytes(hierarchy.session, message_bytes=b"test", **smtp_args(**overrides))
+            else:
+                capabilities.append_campaign_message_to_sent(hierarchy.session, **imap_args(**overrides))
+    decrypt.assert_not_called(); opener.assert_not_called(); send.assert_not_called(); append.assert_not_called()
+
+
+@pytest.mark.parametrize("missing", ["smtp_credential_id", "imap_credential_id"])
+def test_full_authoring_and_summary_still_check_both_protocols_without_decryption(hierarchy, missing):
+    complete = selection()
+    raw = {"server": {"mail_profile_id": "profile-1", **complete}}
+    mail_profiles.assert_campaign_mail_policy_allows_json(hierarchy.session, tenant_id="tenant-1", campaign_id="campaign-1", raw_json=raw)
+    incomplete = {**complete, missing: None}
+    with patch.object(server_hierarchy, "resolve_credential_envelope", side_effect=AssertionError("Summary must not decrypt")) as decrypt:
+        with pytest.raises(mail_profiles.MailProfileError, match="explicit credential selection"):
+            capabilities.campaign_profile_delivery_summary(hierarchy.session, tenant_id="tenant-1", campaign_id="campaign-1", profile_id="profile-1", **incomplete)
+        with pytest.raises(mail_profiles.MailProfileError, match="explicit credential selection"):
+            mail_profiles.assert_campaign_mail_policy_allows_json(hierarchy.session, tenant_id="tenant-1", campaign_id="campaign-1", raw_json={"server": {"mail_profile_id": "profile-1", **incomplete}})
+    decrypt.assert_not_called()
+
+
+def test_batch_still_enforces_all_recipient_domains_before_connection(hierarchy):
+    row = hierarchy.session.get(MailProfilePolicy, "system-policy")
+    row.policy = {**row.policy, "blacklist": {"recipient_domains": ["blocked.example"]}}
+    hierarchy.session.commit()
+    with patch.object(smtp_module, "_open_smtp", side_effect=AssertionError("Forbidden recipient must never connect")) as opener:
+        with pytest.raises(mail_profiles.MailProfileError, match="effective Mail policy"):
+            with capabilities.campaign_smtp_batch(hierarchy.session, **smtp_args(envelope_recipients=["ok@example.test", "no@blocked.example"])): pass
+    opener.assert_not_called()

tests/test_documentation.py

@@ -8,6 +8,7 @@ from sqlalchemy.orm import Session
 
 from govoplan_core.core.modules import DocumentationContext
 from govoplan_mail.backend.documentation import (
+    _credential_line,
     documentation_configuration_states,
     documentation_topics,
 )
@@ -26,6 +27,46 @@ class _Principal:
 
 
 class MailRuntimeDocumentationTests(unittest.TestCase):
+    def test_mailbox_toolbar_contract_documents_context_refresh_and_read_only_bounds_in_both_languages(self) -> None:
+        from govoplan_mail.backend.manifest import manifest
+        topic = next(item for item in manifest.documentation if item.id == "mail.workflow.read-mailbox")
+        self.assertEqual(set(topic.documentation_types), {"user", "admin"})
+        self.assertTrue({"mail.mailbox.reload", "mail.mailbox.tools"}.issubset(topic.metadata["help_contexts"]))
+        for phrase in ("one right-aligned Reload", "Mailbox tools", "IMAP retains the current page", "JMAP starts a fresh cursor chain", "Failed refreshes preserve usable loaded data", "ignore late reads", "do not grant profile administration rights"):
+            self.assertIn(phrase, topic.body)
+        for phrase in ("genau einmal Neuladen rechts", "Postfachwerkzeuge", "IMAP behält die Seite", "JMAP beginnt", "Verspätete Antworten", "Fehlgeschlagene Aktualisierungen", "keine Profilverwaltungsrechte"):
+            self.assertIn(phrase, topic.translations["de"]["body"])
+
+    def test_imap_batch_contract_documents_bounds_and_per_message_safety_in_both_languages(self) -> None:
+        from govoplan_mail.backend.manifest import manifest
+        topic = next(item for item in manifest.documentation if item.id == "mail.reference.campaign-delivery-contract")
+        for body in (topic.body, topic.translations["de"]["body"]):
+            self.assertIn("campaign_imap_batch", body)
+            self.assertIn("GOVOPLAN_IMAP_BATCH_REUSE", body)
+            self.assertIn("MULTIAPPEND", body)
+            self.assertIn("100", body)
+            self.assertIn("300", body)
+        self.assertIn("permissions and recovery evidence are never cached", topic.body)
+        self.assertIn("no automatic APPEND replay", topic.body)
+        self.assertIn("Berechtigungen und Nachweise werden nicht zwischengespeichert", topic.translations["de"]["body"])
+
+    def test_campaign_contract_documents_protocol_scoped_runtime_and_complete_validation(self) -> None:
+        from govoplan_mail.backend.manifest import manifest
+        topic = next(item for item in manifest.documentation if item.id == "mail.reference.campaign-delivery-contract")
+        self.assertIn("Runtime credential-selection checks are protocol-scoped", topic.body)
+        self.assertIn("still check both configured protocols", topic.body)
+        self.assertIn("before credential decryption or provider contact", topic.body)
+        self.assertIn("protokollbezogen", topic.translations["de"]["body"])
+        self.assertIn("weiterhin beide konfigurierten Protokolle", topic.translations["de"]["body"])
+
+    def test_credential_policy_guidance_describes_explicit_mail_references_not_local_secrets(self) -> None:
+        text = _credential_line({"smtp_credentials": {"inherit": False}, "imap_credentials": {"inherit": True}})
+        self.assertIn("SMTP requires an explicit Mail credential", text)
+        self.assertIn("IMAP allows a profile default or explicit Mail credential", text)
+        self.assertIn("Secrets remain in Mail", text)
+        self.assertNotIn("local credentials", text)
+        self.assertNotIn("only for protocols that inherit", text)
+
     def setUp(self) -> None:
         self.session = Session()
 
@@ -182,7 +223,7 @@ class MailRuntimeDocumentationTests(unittest.TestCase):
 
         topics = {topic.id: topic for topic in get_manifest().documentation}
         self.assertEqual(topics["mail.workflow.choose-and-test-profile"].metadata["help_contexts"], ["mail.profiles", "app.settings"])
-        self.assertEqual(topics["mail.workflow.read-mailbox"].metadata["help_contexts"], ["mail.list", "mail.mailbox"])
+        self.assertEqual(topics["mail.workflow.read-mailbox"].metadata["help_contexts"], ["mail.list", "mail.mailbox", "mail.mailbox.reload", "mail.mailbox.tools"])
         self.assertIn("mail.admin.profiles", topics["mail.profiles-and-policy"].metadata["help_contexts"])
         self.assertIn("mail.bounce-processing", topics["mail.bounce-processing"].metadata["help_contexts"])
 

tests/test_imap_batch.py

@@ -0,0 +1,231 @@
+from __future__ import annotations
+
+import imaplib
+import unittest
+from dataclasses import replace
+from unittest.mock import Mock, patch
+
+from govoplan_mail.backend.config import ImapConfig
+from govoplan_mail.backend.sending.imap import (
+    ImapAppendError,
+    ImapBatchPolicy,
+    ImapBatchSession,
+    ImapConfigurationError,
+    append_message_to_sent,
+)
+
+
+class BatchClient:
+    utf8_enabled = False
+
+    def __init__(self, *, wire_folder=b"Gesendete &APw-bermittlung"):
+        self.login = Mock(return_value=("OK", []))
+        self.list = Mock(return_value=("OK", [b'(\\Sent) "/" "' + wire_folder + b'"']))
+        self.noop = Mock(return_value=("OK", []))
+        self.append = Mock(return_value=("OK", [b"APPEND complete"]))
+        self.logout = Mock(return_value=("BYE", []))
+        self.shutdown = Mock()
+
+
+def config(**changes):
+    return ImapConfig(
+        host="imap.example.test", port=993, security="tls",
+        username="service", password="secret", sent_folder="auto",
+    ).model_copy(update=changes)
+
+
+class ImapBatchTests(unittest.TestCase):
+    def test_one_login_discovery_and_logout_for_many_sequential_appends(self):
+        client = BatchClient()
+        with (
+            patch("govoplan_mail.backend.sending.imap.validate_outbound_host"),
+            patch("govoplan_mail.backend.sending.imap._OutboundPolicyIMAP4SSL", return_value=client) as connect,
+            ImapBatchSession(config()) as batch,
+        ):
+            results = [append_message_to_sent(bytes([i]), imap_config=config(), batch_session=batch) for i in range(50)]
+            self.assertEqual(batch.connection_count, 1)
+            self.assertEqual(batch.reconnect_count, 0)
+            client.logout.assert_not_called()
+        connect.assert_called_once()
+        client.login.assert_called_once_with("service", "secret")
+        client.list.assert_called_once()
+        client.noop.assert_not_called()
+        client.logout.assert_called_once()
+        self.assertEqual(client.append.call_count, 50)
+        self.assertEqual([call.args[3] for call in client.append.call_args_list], [bytes([i]) for i in range(50)])
+        self.assertEqual({call.args[0] for call in client.append.call_args_list}, {'"Gesendete &APw-bermittlung"'})
+        self.assertEqual({item.folder for item in results}, {"Gesendete übermittlung"})
+        self.assertEqual([item.session_reused for item in results], [False] + [True] * 49)
+        self.assertEqual({item.connection_sequence for item in results}, {1})
+
+    def test_count_rotation_rediscovers_original_wire_names_on_new_connection(self):
+        first = BatchClient(wire_folder=b"&U,BTFw-&ZeVnLIqe-")
+        second = BatchClient(wire_folder=b"&U,BTF2XlZyyKng-")
+        with (
+            patch("govoplan_mail.backend.sending.imap._open_imap", side_effect=[first, second]) as connect,
+            ImapBatchSession(config(), policy=replace(ImapBatchPolicy(), max_messages_per_connection=2)) as batch,
+        ):
+            results = [batch.append(b"message") for _ in range(3)]
+        self.assertEqual(connect.call_count, 2)
+        first.list.assert_called_once()
+        second.list.assert_called_once()
+        first.logout.assert_called_once()
+        second.logout.assert_called_once()
+        self.assertEqual([item.connection_sequence for item in results], [1, 1, 2])
+        self.assertEqual([item.session_reused for item in results], [False, True, False])
+        self.assertEqual(results[-1].reconnect_count, 1)
+        self.assertEqual(first.append.call_args.args[0], '"&U,BTFw-&ZeVnLIqe-"')
+        self.assertEqual(second.append.call_args.args[0], '"&U,BTF2XlZyyKng-"')
+
+    def test_age_rotation_is_checked_before_next_append(self):
+        clock = [0.0]
+        first, second = BatchClient(), BatchClient()
+        with (
+            patch("govoplan_mail.backend.sending.imap.time.monotonic", side_effect=lambda: clock[0]),
+            patch("govoplan_mail.backend.sending.imap._open_imap", side_effect=[first, second]),
+            ImapBatchSession(config(), policy=replace(ImapBatchPolicy(), max_connection_age_seconds=20)) as batch,
+        ):
+            batch.append(b"one")
+            clock[0] = 20.0
+            self.assertFalse(batch.append(b"two").session_reused)
+        first.noop.assert_not_called()
+        self.assertEqual(first.append.call_count + second.append.call_count, 2)
+
+    def test_idle_probe_can_reconnect_before_append_without_replaying_prior_message(self):
+        clock = [0.0]
+        first, second = BatchClient(), BatchClient()
+        first.noop.side_effect = imaplib.IMAP4.abort("gone")
+        with (
+            patch("govoplan_mail.backend.sending.imap.time.monotonic", side_effect=lambda: clock[0]),
+            patch("govoplan_mail.backend.sending.imap._open_imap", side_effect=[first, second]),
+            ImapBatchSession(config()) as batch,
+        ):
+            batch.append(b"one")
+            clock[0] = 31.0
+            result = batch.append(b"two")
+        first.noop.assert_called_once()
+        self.assertEqual(first.append.call_args.args[3], b"one")
+        self.assertEqual(second.append.call_args.args[3], b"two")
+        self.assertEqual(result.reconnect_count, 1)
+
+    def test_healthy_idle_probe_reuses_connection(self):
+        client = BatchClient()
+        with (
+            patch("govoplan_mail.backend.sending.imap._open_imap", return_value=client),
+            ImapBatchSession(config(), policy=replace(ImapBatchPolicy(), idle_health_check_seconds=0)) as batch,
+        ):
+            batch.append(b"one")
+            self.assertTrue(batch.append(b"two").session_reused)
+        client.noop.assert_called_once()
+
+    def test_only_pre_effect_connection_failures_are_retried(self):
+        client = BatchClient()
+        with (
+            patch("govoplan_mail.backend.sending.imap._open_imap", side_effect=[OSError("offline"), client]) as connect,
+            ImapBatchSession(config()) as batch,
+        ):
+            result = batch.append(b"one")
+        self.assertEqual(connect.call_count, 2)
+        client.append.assert_called_once()
+        self.assertEqual(result.reconnect_count, 1)
+
+    def test_pre_effect_reconnects_are_bounded_and_not_unknown(self):
+        with (
+            patch("govoplan_mail.backend.sending.imap._open_imap", side_effect=OSError("offline")) as connect,
+            ImapBatchSession(config()) as batch,
+            self.assertRaises(ImapAppendError) as caught,
+        ):
+            batch.append(b"one")
+        self.assertEqual(connect.call_count, 2)
+        self.assertTrue(caught.exception.temporary)
+        self.assertFalse(caught.exception.outcome_unknown)
+
+    def test_authentication_rejection_is_not_retried(self):
+        with (
+            patch("govoplan_mail.backend.sending.imap._open_imap", side_effect=imaplib.IMAP4.error("bad login")) as connect,
+            ImapBatchSession(config()) as batch,
+            self.assertRaises(ImapAppendError) as caught,
+        ):
+            batch.append(b"one")
+        connect.assert_called_once()
+        self.assertFalse(caught.exception.temporary)
+        self.assertFalse(caught.exception.outcome_unknown)
+
+    def test_ambiguous_append_is_never_replayed_and_connection_is_discarded(self):
+        first, second = BatchClient(), BatchClient()
+        first.append.side_effect = imaplib.IMAP4.abort("accepted but reply lost")
+        with (
+            patch("govoplan_mail.backend.sending.imap._open_imap", side_effect=[first, second]) as connect,
+            ImapBatchSession(config()) as batch,
+        ):
+            with self.assertRaises(ImapAppendError) as caught:
+                batch.append(b"uncertain")
+            self.assertTrue(caught.exception.outcome_unknown)
+            self.assertFalse(caught.exception.temporary)
+            self.assertEqual(connect.call_count, 1)
+            first.logout.assert_called_once()
+            batch.append(b"different independently claimed message")
+        first.append.assert_called_once()
+        second.append.assert_called_once()
+        self.assertNotEqual(first.append.call_args.args[3], second.append.call_args.args[3])
+
+    def test_definitive_append_rejection_is_not_retried_or_unknown(self):
+        client = BatchClient()
+        client.append.return_value = ("NO", [b"quota"])
+        with (
+            patch("govoplan_mail.backend.sending.imap._open_imap", return_value=client) as connect,
+            ImapBatchSession(config()) as batch,
+            self.assertRaises(ImapAppendError) as caught,
+        ):
+            batch.append(b"one")
+        self.assertFalse(caught.exception.outcome_unknown)
+        connect.assert_called_once()
+        client.append.assert_called_once()
+
+    def test_logout_failure_does_not_reverse_accepted_message(self):
+        client = BatchClient()
+        client.logout.side_effect = OSError("gone")
+        with patch("govoplan_mail.backend.sending.imap._open_imap", return_value=client):
+            result = append_message_to_sent(b"one", imap_config=config())
+        self.assertEqual(result.bytes_appended, 3)
+        client.shutdown.assert_called_once()
+
+    def test_disabling_reuse_keeps_individual_appends_and_closes_every_connection(self):
+        clients = [BatchClient() for _ in range(3)]
+        with (
+            patch("govoplan_mail.backend.sending.imap._open_imap", side_effect=clients),
+            ImapBatchSession(config(), policy=replace(ImapBatchPolicy(), reuse_connections=False)) as batch,
+        ):
+            results = [batch.append(b"one") for _ in clients]
+        self.assertEqual([item.connection_sequence for item in results], [1, 2, 3])
+        self.assertFalse(any(item.session_reused for item in results))
+        for client in clients:
+            client.append.assert_called_once()
+            client.logout.assert_called_once()
+
+    def test_config_mismatch_closed_and_overlapping_calls_fail_before_provider(self):
+        with patch("govoplan_mail.backend.sending.imap._open_imap") as connect:
+            batch = ImapBatchSession(config())
+            with self.assertRaises(ImapConfigurationError):
+                append_message_to_sent(b"one", imap_config=config(password="different"), batch_session=batch)
+            with batch._exclusive_append(), self.assertRaises(ImapConfigurationError):
+                batch.append(b"overlap")
+            batch.close()
+            with self.assertRaises(ImapConfigurationError):
+                batch.append(b"closed")
+        connect.assert_not_called()
+
+    def test_environment_values_are_bounded(self):
+        with patch.dict("os.environ", {
+            "GOVOPLAN_IMAP_BATCH_REUSE": "false",
+            "GOVOPLAN_IMAP_BATCH_MAX_MESSAGES": "0",
+            "GOVOPLAN_IMAP_BATCH_MAX_AGE_SECONDS": "invalid",
+            "GOVOPLAN_IMAP_BATCH_IDLE_HEALTH_CHECK_SECONDS": "999999",
+            "GOVOPLAN_IMAP_BATCH_RECONNECT_ATTEMPTS": "999999",
+        }):
+            policy = ImapBatchPolicy.from_environment()
+        self.assertFalse(policy.reuse_connections)
+        self.assertEqual(policy.max_messages_per_connection, 1)
+        self.assertEqual(policy.max_connection_age_seconds, 300)
+        self.assertEqual(policy.idle_health_check_seconds, 3600)
+        self.assertEqual(policy.reconnect_attempts, 5)

tests/test_imap_mailbox_encoding.py

@@ -0,0 +1,225 @@
+from __future__ import annotations
+
+import unittest
+from unittest.mock import patch
+
+from govoplan_mail.backend.config import ImapConfig
+from govoplan_mail.backend.sending.imap import (
+    ImapAppendError,
+    ImapConfigurationError,
+    _decode_mailbox_name,
+    _encode_mailbox_name,
+    _extract_mailbox_name,
+    _list_imap_folders_on_client,
+    _quote_mailbox_name,
+    _select_readonly,
+    append_message_to_sent,
+)
+
+
+class MailboxClient:
+    utf8_enabled = False
+    capabilities = ("IMAP4REV1", "UTF8=ACCEPT")
+    untagged_responses = {"EXISTS": [b"2"], "UIDVALIDITY": [b"1"]}
+
+    def __init__(self, listing=None):
+        self.listing = listing or []
+        self.list_calls = 0
+        self.status_calls = []
+        self.select_calls = []
+        self.append_calls = []
+        self.logged_out = False
+
+    def list(self):
+        self.list_calls += 1
+        return "OK", self.listing
+
+    def status(self, mailbox, items):
+        self.status_calls.append((mailbox, items))
+        return "OK", [b'"mailbox" (MESSAGES 2 UNSEEN 1)']
+
+    def select(self, mailbox, readonly=False):
+        self.select_calls.append((mailbox, readonly))
+        return "OK", [b"2"]
+
+    def response(self, code):
+        return "OK", [b"1"] if code == "UIDVALIDITY" else []
+
+    def append(self, mailbox, flags, date_time, message):
+        self.append_calls.append((mailbox, flags, message))
+        return "OK", [b"APPEND complete"]
+
+    def logout(self):
+        self.logged_out = True
+        return "BYE", []
+
+
+class ImapMailboxEncodingTests(unittest.TestCase):
+    def test_rfc_and_real_provider_vectors_round_trip(self):
+        # RFC 3501 section 5.1.3 plus the reported German folder and UTF-16
+        # surrogate pairs. A plain '+' is not a shift in modified UTF-7.
+        for display, wire in (
+            ("INBOX", "INBOX"),
+            ("Entwürfe", "Entw&APw-rfe"),
+            ("R&D", "R&-D"),
+            ("+Plus & Sons", "+Plus &- Sons"),
+            ("~peter/mail/台北/日本語", "~peter/mail/&U,BTFw-/&ZeVnLIqe-"),
+            ("📨", "&2D3c6A-"),
+            ('A \\ "B"', 'A \\ "B"'),
+        ):
+            with self.subTest(display=display):
+                self.assertEqual(_encode_mailbox_name(display), wire)
+                self.assertEqual(_decode_mailbox_name(wire), display)
+
+    def test_list_decodes_names_before_standard_folder_detection_and_status(self):
+        client = MailboxClient([
+            b'(\\HasNoChildren) "/" "Entw&APw-rfe"',
+            b'(\\HasNoChildren) "/" "Gel&APY-scht"',
+            b'(\\HasNoChildren) "/" "R&-D"',
+        ])
+        result = _list_imap_folders_on_client(
+            client, host="imap.example.org", port=993, security="tls", include_status=True,
+        )
+        self.assertEqual([folder.name for folder in result.folders], ["Entwürfe", "Gelöscht", "R&D"])
+        self.assertEqual(result.detected_folder_mappings, {"drafts": "Entwürfe", "trash": "Gelöscht"})
+        self.assertTrue(all(folder.message_count == 2 and folder.unseen_count == 1 for folder in result.folders))
+        self.assertEqual([call[0] for call in client.status_calls], [
+            '"Entw&APw-rfe"', '"Gel&APY-scht"', '"R&-D"',
+        ])
+        self.assertEqual(client.list_calls, 1)
+
+    def test_literal_names_are_decoded_without_stripping_or_unquoting_content(self):
+        for wire, expected in (
+            (b"Entw&APw-rfe", "Entwürfe"),
+            (b'"R&-D" ', '"R&D" '),
+        ):
+            with self.subTest(wire=wire):
+                line = b'(\\Drafts) "/" {' + str(len(wire)).encode("ascii") + b"}"
+                self.assertEqual(_extract_mailbox_name((line, wire)), (expected, {"\\drafts"}))
+        self.assertIsNone(_extract_mailbox_name(b""))
+        self.assertIsNone(_extract_mailbox_name(None))
+
+    def test_list_rejects_wrong_literal_lengths_and_invalid_provider_encoding(self):
+        with self.assertRaisesRegex(ImapAppendError, "literal"):
+            _extract_mailbox_name((b'() "/" {99}', b"INBOX"))
+        for wire in (b"&APw", b"&!bad-", b"&AGE-", b"&AA-", b"&APx-", b"&2AA-", b"Entw\xffrfe"):
+            with self.subTest(wire=wire), self.assertRaisesRegex(ImapAppendError, "encoding"):
+                _extract_mailbox_name(b'() "/" "' + wire + b'"')
+
+    def test_select_encodes_unicode_and_quotes_protocol_metacharacters(self):
+        client = MailboxClient()
+        folder = 'Entwürfe / R&D / "Q" \\'
+        self.assertEqual(_select_readonly(client, folder), (2, "1"))
+        self.assertEqual(client.select_calls, [('"Entw&APw-rfe / R&-D / \\"Q\\" \\\\"', True)])
+        self.assertEqual(client.list_calls, 0)
+
+    def test_quoted_name_escaping_round_trips_independently_of_charset_encoding(self):
+        for folder in ('Entwürfe "R&D"', ' \\"quoted"\\ ', 'R&D', '📨/日本語', 'back\\slash'):
+            with self.subTest(folder=folder):
+                quoted = _quote_mailbox_name(folder)
+                self.assertEqual(_extract_mailbox_name('() "/" ' + quoted), (folder, set()))
+
+    def test_saved_wire_names_resolve_without_double_encoding(self):
+        client = MailboxClient([b'() "/" "Entw&APw-rfe"'])
+        _select_readonly(client, "Entw&APw-rfe")
+        _select_readonly(client, "Entwürfe")
+        self.assertEqual(client.select_calls, [('"Entw&APw-rfe"', True)] * 2)
+        self.assertEqual(client.list_calls, 1)
+
+    def test_literal_name_wins_when_legacy_alias_is_ambiguous(self):
+        client = MailboxClient([
+            b'() "/" "Entw&APw-rfe"',
+            b'() "/" "Entw&-APw-rfe"',
+        ])
+        _select_readonly(client, "Entw&APw-rfe")
+        self.assertEqual(client.select_calls, [('"Entw&-APw-rfe"', True)])
+
+    def test_saved_ampersand_and_literal_ampersand_remain_distinct(self):
+        client = MailboxClient([b'() "/" "R&-D"'])
+        _select_readonly(client, "R&-D")
+        _select_readonly(client, "R&D")
+        self.assertEqual(client.select_calls, [('"R&-D"', True)] * 2)
+
+    def test_provider_wire_form_is_preserved_on_the_listed_connection(self):
+        client = MailboxClient([b'() "/" "&U,BTFw-&ZeVnLIqe-"'])
+        result = _list_imap_folders_on_client(
+            client, host="imap.example.org", port=993, security="tls", include_status=True,
+        )
+        self.assertEqual(result.folders[0].name, "台北日本語")
+        self.assertEqual(client.status_calls[0][0], '"&U,BTFw-&ZeVnLIqe-"')
+
+    def test_utf8_mode_preserves_literal_ampersands_and_does_not_decode_again(self):
+        client = MailboxClient(['() "/" "Entwürfe &APw-"'.encode("utf-8")])
+        client.utf8_enabled = True
+        result = _list_imap_folders_on_client(
+            client, host="imap.example.org", port=993, security="tls", include_status=True,
+        )
+        self.assertEqual(result.folders[0].name, "Entwürfe &APw-")
+        _select_readonly(client, result.folders[0].name)
+        self.assertEqual(client.select_calls, [('"Entwürfe &APw-"', True)])
+        self.assertEqual(client.status_calls[0][0], '"Entwürfe &APw-"')
+
+    def test_utf8_capability_alone_does_not_change_the_encoding(self):
+        client = MailboxClient()
+        self.assertEqual(_quote_mailbox_name("Entwürfe", client=client), '"Entw&APw-rfe"')
+        with self.assertRaisesRegex(ImapAppendError, "encoding"):
+            _extract_mailbox_name('() "/" "Entwürfe"'.encode("utf-8"))
+
+    def test_changing_utf8_mode_invalidates_cached_wire_names(self):
+        client = MailboxClient([b'() "/" "Entw&APw-rfe"'])
+        _list_imap_folders_on_client(
+            client, host="imap.example.org", port=993, security="tls", include_status=False,
+        )
+        client.utf8_enabled = True
+        self.assertEqual(_quote_mailbox_name("Entwürfe", client=client), '"Entwürfe"')
+
+    def test_select_rejects_control_characters_before_sending_any_command(self):
+        for folder in ("INBOX\r\nLOGOUT", "bad\x00folder", "bad\x7ffolder", "bad\u2028folder"):
+            client = MailboxClient()
+            with self.subTest(folder=folder), self.assertRaisesRegex(ImapConfigurationError, "control"):
+                _select_readonly(client, folder)
+            self.assertEqual(client.select_calls, [])
+
+    def test_append_auto_detects_unicode_sent_folder_and_uses_original_wire_name(self):
+        client = MailboxClient([b'(\\Sent) "/" "Gesendet &APw-"'])
+        config = ImapConfig(host="imap.example.org", sent_folder="auto")
+        with patch("govoplan_mail.backend.sending.imap._open_imap", return_value=client):
+            result = append_message_to_sent(b"Subject: test\r\n\r\nBody", imap_config=config)
+        self.assertEqual(result.folder, "Gesendet ü")
+        self.assertEqual(client.append_calls[0][0], '"Gesendet &APw-"')
+        self.assertTrue(client.logged_out)
+
+    def test_append_explicit_unicode_and_saved_wire_names_target_the_same_mailbox(self):
+        for folder in ("Entwürfe", "Entw&APw-rfe"):
+            with self.subTest(folder=folder):
+                client = MailboxClient([b'() "/" "Entw&APw-rfe"'])
+                config = ImapConfig(host="imap.example.org", sent_folder=folder)
+                with patch("govoplan_mail.backend.sending.imap._open_imap", return_value=client):
+                    append_message_to_sent(b"Subject: test\r\n\r\nBody", imap_config=config)
+                self.assertEqual(client.append_calls[0][0], '"Entw&APw-rfe"')
+
+    def test_failed_legacy_lookup_never_selects_or_appends_to_a_guessed_mailbox(self):
+        client = MailboxClient()
+        with patch.object(client, "list", return_value=("NO", [b"Denied"])), self.assertRaisesRegex(
+            ImapAppendError, "resolving a saved folder name",
+        ):
+            _select_readonly(client, "Entw&APw-rfe")
+        self.assertEqual(client.select_calls, [])
+
+    def test_append_saved_wire_name_failure_is_not_an_unknown_append_outcome(self):
+        client = MailboxClient()
+        config = ImapConfig(host="imap.example.org", sent_folder="Entw&APw-rfe")
+        with (
+            patch("govoplan_mail.backend.sending.imap._open_imap", return_value=client),
+            patch.object(client, "list", side_effect=OSError("connection lost")),
+            self.assertRaises(ImapAppendError) as caught,
+        ):
+            append_message_to_sent(b"Subject: test\r\n\r\nBody", imap_config=config)
+        self.assertTrue(caught.exception.temporary)
+        self.assertFalse(caught.exception.outcome_unknown)
+        self.assertEqual(client.append_calls, [])
+        self.assertTrue(client.logged_out)
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_mail_profile_helpers.py

@@ -563,7 +563,7 @@ class MailProfilePolicyHelperTests(unittest.TestCase):
             "credential-1",
         )
 
-    def test_campaign_delivery_fails_when_policy_requires_local_credentials(self):
+    def test_campaign_delivery_fails_when_policy_requires_missing_explicit_credentials(self):
         profile = SimpleNamespace(imap_config=None)
         policy = EffectiveMailProfilePolicy(
             smtp_credentials=EffectiveCredentialPolicy(inherit=False),
@@ -572,6 +572,44 @@ class MailProfilePolicyHelperTests(unittest.TestCase):
         with self.assertRaisesRegex(MailProfileError, "explicit credential selection"):
             _assert_campaign_inherits_profile_credentials(profile, policy)
 
+    def test_explicit_mail_credentials_satisfy_independent_protocol_selection_policy(self):
+        profile = SimpleNamespace(imap_config={"host": "imap.example.test"})
+        policy = EffectiveMailProfilePolicy(
+            smtp_credentials=EffectiveCredentialPolicy(inherit=False),
+            imap_credentials=EffectiveCredentialPolicy(inherit=False),
+        )
+        with self.assertRaisesRegex(MailProfileError, "effective IMAP"):
+            _assert_campaign_inherits_profile_credentials(profile, policy, {"smtp_credential_id": "smtp-credential"})
+        _assert_campaign_inherits_profile_credentials(profile, policy, {
+            "smtp_credential_id": "smtp-credential", "imap_credential_id": "imap-credential",
+        })
+        # Allowing a default never forbids an explicitly selected Mail credential.
+        _assert_campaign_inherits_profile_credentials(profile, EffectiveMailProfilePolicy(), {
+            "smtp_credential_id": "smtp-credential", "imap_credential_id": "imap-credential",
+        })
+
+    def test_credential_selection_false_is_overridable_only_without_an_ancestor_lock(self):
+        for locked in (False, True):
+            with self.subTest(locked=locked):
+                policy = EffectiveMailProfilePolicy()
+                _merge_policy(policy, {
+                    "smtp_credentials": {"inherit": False},
+                    "allow_lower_level_limits": {"smtp_credentials.inherit": not locked},
+                }, source="system")
+                _merge_policy(policy, {
+                    "smtp_credentials": {"inherit": True},
+                    "allow_lower_level_limits": {"smtp_credentials.inherit": True},
+                }, source="tenant", source_id="tenant-1")
+                self.assertEqual(policy.smtp_credentials.inherit, not locked)
+                self.assertEqual(policy.allow_lower_level_limits["smtp_credentials.inherit"], not locked)
+
+    def test_null_credential_selection_inherits_the_parent_choice(self):
+        policy = EffectiveMailProfilePolicy()
+        _merge_policy(policy, {"smtp_credentials": {"inherit": False}}, source="system")
+        _merge_policy(policy, {"smtp_credentials": {"inherit": None}}, source="tenant", source_id="tenant-1")
+        self.assertFalse(policy.smtp_credentials.inherit)
+        self.assertEqual(policy.smtp_credentials.inherit_source, "system")
+
     def test_merge_policy_respects_locked_lower_level_limits(self):
         policy = EffectiveMailProfilePolicy()
         _merge_policy(