test(campaign): add simple announcement acceptance fixture

tests/test_example_campaigns.py

@@ -0,0 +1,207 @@
+from __future__ import annotations
+
+import json
+import shutil
+import subprocess
+import sys
+import tempfile
+import unittest
+from pathlib import Path
+
+
+REPOSITORY_ROOT = Path(__file__).resolve().parents[1]
+FIXTURE_ROOT = REPOSITORY_ROOT / "examples" / "simple-announcement"
+
+
+_ISOLATED_ACCEPTANCE_PROGRAM = r"""
+from __future__ import annotations
+
+import hashlib
+import importlib.abc
+import json
+import socket
+import sys
+from email import policy
+from email.parser import BytesParser
+from pathlib import Path
+
+
+source_root = Path(sys.argv[1]).resolve()
+fixture_root = Path(sys.argv[2]).resolve()
+sys.path.insert(0, str(source_root))
+
+
+class AbsentOptionalModuleFinder(importlib.abc.MetaPathFinder):
+    absent_roots = {"govoplan_files", "govoplan_mail"}
+
+    def find_spec(self, fullname, path=None, target=None):
+        if fullname.partition(".")[0] in self.absent_roots:
+            raise ModuleNotFoundError(
+                f"{fullname} is intentionally absent in the Campaign fixture check",
+                name=fullname,
+            )
+        return None
+
+
+sys.meta_path.insert(0, AbsentOptionalModuleFinder())
+
+
+def deny_network(*args, **kwargs):
+    raise AssertionError("the Campaign validate/build fixture must not open a network connection")
+
+
+class NoNetworkSocket(socket.socket):
+    def connect(self, address):
+        deny_network(address)
+
+    def connect_ex(self, address):
+        deny_network(address)
+
+
+socket.create_connection = deny_network
+socket.socket = NoNetworkSocket
+
+from govoplan_campaign.backend.campaign import (  # noqa: E402
+    load_campaign_config,
+    load_campaign_json,
+    validate_campaign_config,
+)
+from govoplan_campaign.backend.messages import build_campaign_messages  # noqa: E402
+
+
+metadata = json.loads((fixture_root / "fixture.json").read_text(encoding="utf-8"))
+assert metadata["required_modules"] == ["core", "access", "campaigns"]
+assert metadata["absent_optional_modules"] == ["files", "mail"]
+assert metadata["external_effects"] == "forbidden"
+
+campaign_file = fixture_root / metadata["campaign_file"]
+raw_campaign = load_campaign_json(campaign_file)
+
+
+def iter_keys(value):
+    if isinstance(value, dict):
+        for key, nested in value.items():
+            yield key.casefold()
+            yield from iter_keys(nested)
+    elif isinstance(value, list):
+        for nested in value:
+            yield from iter_keys(nested)
+
+
+forbidden_key_fragments = ("credential", "imap", "mail_profile", "password", "secret", "smtp")
+assert not [
+    key
+    for key in iter_keys(raw_campaign)
+    if any(fragment in key for fragment in forbidden_key_fragments)
+]
+
+config = load_campaign_config(campaign_file)
+assert config.server.mail_profile_id is None
+assert not config.attachments.global_
+
+validation = validate_campaign_config(config, campaign_file=campaign_file, check_files=True)
+assert validation.ok
+assert validation.error_count == 0
+assert validation.warning_count == 0
+assert validation.entries_count == metadata["expected"]["entries_count"]
+
+
+def build(output_name):
+    return build_campaign_messages(
+        config,
+        campaign_file=campaign_file,
+        output_dir=fixture_root.parent / output_name,
+        write_eml=True,
+    )
+
+
+first = build("build-first")
+second = build("build-second")
+expected = metadata["expected"]
+for result in (first, second):
+    assert result.report.campaign_id == expected["campaign_id"]
+    assert result.report.entries_count == expected["entries_count"]
+    assert result.report.built_count == expected["built_count"]
+    assert result.report.build_failed_count == 0
+    assert result.report.queueable_count == expected["queueable_count"]
+    assert len(result.built_messages) == 1
+
+    built = result.built_messages[0]
+    assert built.mime is not None
+    assert built.draft.subject == expected["subject"]
+    assert built.draft.validation_status.value == "ready"
+    assert built.draft.send_status.value == "draft"
+    assert built.draft.imap_status.value == "not_requested"
+    assert built.draft.attachment_count == expected["attachment_count"]
+    assert not built.draft.attachments
+    assert not built.draft.issues
+    assert built.draft.from_ is not None
+    assert built.draft.from_.email == "announcements@example.test"
+    assert [address.email for address in built.draft.to] == ["recipient@example.test"]
+    assert built.mime["Subject"] == expected["subject"]
+    assert "Hello Example Recipient" in built.mime.get_content()
+    assert list(built.mime.iter_attachments()) == []
+
+
+def normalized_eml(path_value):
+    message = BytesParser(policy=policy.default).parsebytes(Path(path_value).read_bytes())
+    del message["Date"]
+    del message["Message-ID"]
+    return message.as_bytes(policy=policy.default)
+
+
+first_eml = normalized_eml(first.report.messages[0].eml_path)
+second_eml = normalized_eml(second.report.messages[0].eml_path)
+assert first_eml == second_eml
+assert not any(
+    name == root or name.startswith(root + ".")
+    for name in sys.modules
+    for root in ("govoplan_files", "govoplan_mail")
+)
+
+print(json.dumps({
+    "campaign_id": first.report.campaign_id,
+    "built_count": first.report.built_count,
+    "queueable_count": first.report.queueable_count,
+    "normalized_eml_sha256": hashlib.sha256(first_eml).hexdigest(),
+}, sort_keys=True))
+"""
+
+
+class CampaignExampleAcceptanceTests(unittest.TestCase):
+    def test_simple_announcement_validates_and_builds_without_mail_or_files(self) -> None:
+        self.assertTrue((FIXTURE_ROOT / "campaign.json").is_file())
+        self.assertTrue((FIXTURE_ROOT / "fixture.json").is_file())
+
+        with tempfile.TemporaryDirectory(prefix="govoplan-campaign-acceptance-", dir="/tmp") as temp_dir:
+            workspace = Path(temp_dir).resolve()
+            self.assertNotIn(REPOSITORY_ROOT, workspace.parents)
+            isolated_fixture = workspace / "simple-announcement"
+            shutil.copytree(FIXTURE_ROOT, isolated_fixture)
+
+            completed = subprocess.run(
+                [
+                    sys.executable,
+                    "-I",
+                    "-c",
+                    _ISOLATED_ACCEPTANCE_PROGRAM,
+                    str(REPOSITORY_ROOT / "src"),
+                    str(isolated_fixture),
+                ],
+                cwd=workspace,
+                check=False,
+                capture_output=True,
+                text=True,
+                timeout=30,
+            )
+
+        self.assertEqual(completed.returncode, 0, completed.stderr or completed.stdout)
+        evidence = json.loads(completed.stdout)
+        self.assertEqual(evidence["campaign_id"], "simple-announcement")
+        self.assertEqual(evidence["built_count"], 1)
+        self.assertEqual(evidence["queueable_count"], 1)
+        self.assertRegex(evidence["normalized_eml_sha256"], r"^[0-9a-f]{64}$")
+
+
+if __name__ == "__main__":
+    unittest.main()