from __future__ import annotations import importlib.util import json 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="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 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.10", "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.10", "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.10", "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 "post-DATA connection loss" in testbed assert "real worker restart" in testbed assert "does not simulate a connection loss after SMTP DATA" in runbook assert "raw provider diagnostics" in runbook