govoplan-campaign/tests/test_example_campaigns.py

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()