from __future__ import annotations import importlib.util import json import smtplib import sys from pathlib import Path from typing import Any import pytest from govoplan_campaign.backend.campaign import load_campaign_config REPOSITORY_ROOT = Path(__file__).resolve().parents[1] RUNNER_PATH = REPOSITORY_ROOT / "dev" / "mail-testbed" / "run_campaign_acceptance.py" FIXTURE_PATH = REPOSITORY_ROOT / "examples" / "greenmail-delivery" / "campaign.json" def _load_runner(): spec = importlib.util.spec_from_file_location("govoplan_campaign_greenmail_acceptance", RUNNER_PATH) assert spec is not None and spec.loader is not None module = importlib.util.module_from_spec(spec) sys.modules[spec.name] = module spec.loader.exec_module(module) return module runner = _load_runner() class _Response: def __init__(self, status_code: int, payload: dict[str, Any]) -> None: self.status_code = status_code self._payload = payload def json(self) -> dict[str, Any]: return self._payload class _AcceptanceClient: def __init__(self) -> None: self.campaign_json: dict[str, Any] | None = None self.send_calls = 0 def post(self, path: str, **kwargs: Any) -> _Response: if path == "/api/v1/campaigns": self.campaign_json = kwargs["json"]["config"] return _Response( 200, { "campaign": {"id": "campaign-internal"}, "version": {"id": "version-internal"}, }, ) if path.endswith("/validate"): return _Response(200, {"ok": True, "error_count": 0, "warning_count": 0}) if path.endswith("/build"): return _Response( 200, { "built_count": 1, "build_failed_count": 0, "queueable_count": 1, }, ) if path.endswith("/send-now"): self.send_calls += 1 if self.send_calls == 2: return _Response(422, {"detail": "Already accepted"}) return _Response( 200, { "result": { "attempted_count": 1, "sent_count": 1, "failed_count": 0, "outcome_unknown_count": 0, "skipped_count": 0, "delivery_mode": "synchronous", "results": [{"job_id": "not-retained", "status": "smtp_accepted"}], } }, ) if path.endswith("/append-sent"): return _Response( 200, { "result": { "pending_count": 1, "processed_count": 1, "appended_count": 1, "failed_count": 0, "skipped_count": 0, "results": [{"job_id": "not-retained", "status": "appended"}], } }, ) raise AssertionError(f"unexpected POST {path}") def get(self, path: str, **kwargs: Any) -> _Response: if path.endswith("/report"): return _Response( 200, { "cards": { "jobs_total": 1, "sent": 1, "smtp_accepted": 1, "failed": 0, "outcome_unknown": 0, "retryable": 0, "needs_attention": 0, "imap_appended": 1, "imap_failed": 0, }, "status_counts": { "send": {"smtp_accepted": 1}, "imap": {"appended": 1}, }, }, ) raise AssertionError(f"unexpected GET {path}") def _settings(): return runner.TestbedSettings( smtp_host="127.0.0.1", smtp_port=3025, imap_host="127.0.0.1", imap_port=3143, username="campaign-test@govoplan.test", password="local-test-password", sender="campaign-test@govoplan.test", recipient="campaign-test@govoplan.test", sent_folder="Sent", provider_timeout_seconds=5, ) @pytest.mark.parametrize("host", ["localhost", "mail.test", "192.168.1.20", "8.8.8.8"]) def test_testbed_rejects_hostnames_and_non_loopback_addresses(host: str) -> None: settings = _settings() rejected = runner.TestbedSettings( smtp_host=host, smtp_port=settings.smtp_port, imap_host=settings.imap_host, imap_port=settings.imap_port, username=settings.username, password=settings.password, sender=settings.sender, recipient=settings.recipient, sent_folder=settings.sent_folder, provider_timeout_seconds=settings.provider_timeout_seconds, ) with pytest.raises(runner.AcceptanceError, match="literal loopback|restricted to the loopback"): rejected.assert_local_testbed() @pytest.mark.parametrize("host", ["127.0.0.1", "127.8.9.10", "::1"]) def test_testbed_accepts_literal_loopback_and_preserves_it_in_profile(host: str) -> None: settings = _settings() accepted = runner.TestbedSettings( smtp_host=host, smtp_port=settings.smtp_port, imap_host=host, imap_port=settings.imap_port, username=settings.username, password=settings.password, sender=settings.sender, recipient=settings.recipient, sent_folder=settings.sent_folder, provider_timeout_seconds=settings.provider_timeout_seconds, ) accepted.assert_local_testbed() profile = runner._profile_payload(accepted, name="Literal loopback") assert profile["smtp"]["host"] == host assert profile["imap"]["host"] == host def test_campaign_acceptance_orchestration_retains_only_profile_reference_and_safe_evidence() -> None: client = _AcceptanceClient() def snapshot_probe(_version_id: str): assert client.campaign_json is not None return client.campaign_json, { "mail_profile_id": "profile-1", "smtp_transport_revision": "opaque-smtp-revision", "imap_transport_revision": "opaque-imap-revision", "delivery": {"imap_append_sent": {"enabled": True, "folder": "Sent"}}, } audit = { "campaign.created": 1, "campaign.validated": 1, "campaign.messages_built": 1, "campaign.sent_now": 1, "campaign.send_now_rejected": 1, "campaign.append_sent_enqueued": 1, } evidence, subject = runner.execute_campaign_scenario( client, {"Authorization": "not-retained"}, fixture_path=FIXTURE_PATH, profile_id="profile-1", settings=_settings(), scenario="success", snapshot_probe=snapshot_probe, audit_probe=lambda _campaign_id, _version_id: audit, append_sent=True, repeat_send=True, ) evidence["provider_verification"] = { "inbox_increment": 1, "sent_increment": 1, "unique_subject_matches_in_inbox": 1, "unique_subject_matches_in_sent": 1, } runner._assert_success_evidence(evidence) runner._assert_evidence_safe( { "schema_version": runner.EVIDENCE_SCHEMA, "coverage": { "smtp_acceptance": True, "partial_envelope_refusal": False, "post_data_connection_loss_outcome_unknown": False, "source_artifact_provenance": False, "worker_restart_interruption": False, }, "success": evidence, }, settings=_settings(), ) assert subject.startswith("[GovOPlaN acceptance ") assert client.send_calls == 2 assert client.campaign_json is not None assert client.campaign_json["server"] == {"mail_profile_id": "profile-1"} assert "credentials" not in json.dumps(client.campaign_json).casefold() serialized = json.dumps(evidence, sort_keys=True) assert "not-retained" not in serialized assert "local-test-password" not in serialized def test_campaign_boundary_rejects_resolved_transport_material() -> None: with pytest.raises(runner.AcceptanceError, match="forbidden transport material"): runner.assert_campaign_boundary( {"server": {"mail_profile_id": "profile-1"}}, { "mail_profile_id": "profile-1", "smtp_transport_revision": "smtp-revision", "imap_transport_revision": "imap-revision", "smtp": {"host": "should-not-be-here"}, }, profile_id="profile-1", ) def test_success_projection_fails_closed_on_inconsistent_campaign_report() -> None: evidence = { "send": { "attempted_count": 1, "sent_count": 1, "failed_count": 0, "outcome_unknown_count": 0, "skipped_count": 0, "delivery_mode": "synchronous", "statuses": {"smtp_accepted": 1}, }, "append_sent": { "pending_count": 1, "processed_count": 1, "appended_count": 1, "failed_count": 0, "skipped_count": 0, "statuses": {"appended": 1}, }, "report": { "cards": { "jobs_total": 1, "sent": 0, "smtp_accepted": 0, "failed": 0, "outcome_unknown": 0, "needs_attention": 0, "imap_appended": 0, "imap_failed": 0, }, "send_status_counts": {}, "imap_status_counts": {}, }, "provider_verification": { "inbox_increment": 1, "sent_increment": 1, "unique_subject_matches_in_inbox": 1, "unique_subject_matches_in_sent": 1, }, "campaign_mail_boundary": { "profile_reference_only": True, "smtp_revision_frozen": True, "imap_revision_frozen": True, "resolved_transport_material_present": False, }, } with pytest.raises(runner.AcceptanceError, match="report does not agree"): runner._assert_success_evidence(evidence) def test_evidence_projection_rejects_unknown_status_keys() -> None: with pytest.raises(runner.AcceptanceError, match="unsupported status"): runner._send_evidence( { "delivery_mode": "synchronous", "results": [{"status": "provider diagnostic: recipient@example.test"}], } ) with pytest.raises(runner.AcceptanceError, match="durable attempt status"): runner._durable_state_evidence( { "job_count": 1, "send_status_counts": {"sending": 1}, "attempt_status_counts": {"provider-secret": 1}, "unfinished_attempt_count": 1, } ) with pytest.raises(runner.AcceptanceError, match="synchronous delivery mode"): runner._send_evidence( { "delivery_mode": "provider diagnostic: recipient@example.test", "results": [], } ) with pytest.raises(runner.AcceptanceError, match="unsupported"): runner._report_evidence( { "cards": {}, "status_counts": { "send": {"smtp_accepted": 1, "provider-secret": 1}, "imap": {}, }, } ) def _open_fault_smtp(endpoint): client = smtplib.SMTP(endpoint.host, endpoint.port, timeout=5) client.ehlo() client.login("acceptance-user", "acceptance-password") return client def test_explicit_temporary_smtp_response_occurs_after_complete_data() -> None: with runner.smtp_fault_endpoint("temporary_data_response") as endpoint: client = _open_fault_smtp(endpoint) try: with pytest.raises(smtplib.SMTPDataError) as captured: client.sendmail( "sender@example.test", ["recipient@example.test"], b"Subject: temporary\r\n\r\nmessage", ) finally: client.close() assert captured.value.smtp_code == 451 assert endpoint.evidence() == { "connection_count": 1, "accepted_rcpt_commands": 1, "refused_rcpt_commands": 0, "data_transactions": 1, } def test_partial_recipient_refusal_retains_one_accepted_envelope() -> None: with runner.smtp_fault_endpoint("partial_recipient_refusal") as endpoint: client = _open_fault_smtp(endpoint) try: refused = client.sendmail( "sender@example.test", ["accepted@example.test", "refused@example.test"], b"Subject: partial\r\n\r\nmessage", ) finally: client.quit() assert set(refused) == {"refused@example.test"} assert endpoint.evidence() == { "connection_count": 1, "accepted_rcpt_commands": 1, "refused_rcpt_commands": 1, "data_transactions": 1, } def test_post_data_disconnect_is_a_real_ambiguous_protocol_boundary() -> None: with runner.smtp_fault_endpoint("post_data_disconnect") as endpoint: client = _open_fault_smtp(endpoint) try: with pytest.raises(smtplib.SMTPServerDisconnected): client.sendmail( "sender@example.test", ["recipient@example.test"], b"Subject: ambiguous\r\n\r\nmessage", ) finally: client.close() assert endpoint.wait_for_data(1) assert endpoint.evidence()["data_transactions"] == 1 def test_partial_refusal_fixture_adds_a_second_distinct_recipient() -> None: raw, _subject = runner.materialize_campaign_fixture( FIXTURE_PATH, profile_id="profile-1", settings=_settings(), scenario="partial_envelope_refusal", run_token="0123456789ab", additional_envelope_recipient=True, ) recipients = raw["entries"]["inline"][0]["to"] assert len(recipients) == 2 assert recipients[0]["email"] != recipients[1]["email"] assert recipients[1]["email"].endswith("@govoplan.test") def test_fixture_contains_no_transport_credentials() -> None: raw = json.loads(FIXTURE_PATH.read_text(encoding="utf-8")) runner._assert_no_forbidden_campaign_keys(raw) assert raw["server"] == {"mail_profile_id": "00000000-0000-4000-8000-000000000001"} config = load_campaign_config(FIXTURE_PATH) assert config.server.mail_profile_id == "00000000-0000-4000-8000-000000000001" def test_fixture_composition_versions_are_complete_and_exact() -> None: versions = { "core": "0.1.13", "access": "0.1.11", "audit": "0.1.8", "campaigns": "0.1.11", "mail": "0.1.10", "files": "0.1.9", } assert runner.required_composition_versions(FIXTURE_PATH, versions) == { "core": "0.1.13", "access": "0.1.11", "audit": "0.1.8", "campaigns": "0.1.11", "mail": "0.1.10", } def test_fixture_composition_fails_closed_when_a_required_version_is_missing() -> None: with pytest.raises(runner.AcceptanceError, match="versions are unavailable"): runner.required_composition_versions( FIXTURE_PATH, { "core": "0.1.13", "access": "0.1.11", "campaigns": "0.1.11", "mail": "0.1.10", }, ) def test_testbed_documentation_distinguishes_proven_and_open_failure_drills() -> None: testbed = (REPOSITORY_ROOT / "dev" / "mail-testbed" / "README.md").read_text(encoding="utf-8") runbook = (REPOSITORY_ROOT / "docs" / "CAMPAIGN_DELIVERY_RUNBOOK.md").read_text(encoding="utf-8") assert "second ordinary send must be rejected before another provider effect" in testbed assert "connection loss after complete DATA is frozen" in testbed assert "dedicated OS process" in testbed assert "Redis/Celery delivery" in runbook assert "broker redelivery" in runbook assert "celery_broker_redelivery" in testbed assert "raw provider diagnostics" in runbook