AkurAI-Mail / scripts/test-akurai-mail-server.py

Latest change 275d4fa11a39ecc30e2caf604d5a0d499910e8db - Remove retired Multica/pi-bun delivery hook by Ólafur Búi Ólafsson

#!/usr/bin/env python3
"""Unit tests for scripts/akurai-mail-server: structured anti-spam policy lists
and the service/command surface (the retired delivery hook stays retired)."""
import importlib.machinery, importlib.util, subprocess, sys, unittest
from pathlib import Path

HERE = Path(__file__).resolve().parent
SERVER = HERE / "akurai-mail-server"
loader = importlib.machinery.SourceFileLoader("akurai_mail_server", str(SERVER))
spec = importlib.util.spec_from_loader(loader.name, loader)
srv = importlib.util.module_from_spec(spec)
loader.exec_module(srv)


class PolicyListTests(unittest.TestCase):
    def test_add_creates_structured_entries(self):
        policy = {"blocklist": ["old@example.com"]}
        srv.update_policy_list(policy, "blocklist", "add", ["New@Example.com"], {"reason": "spam"})
        self.assertEqual([e["value"] for e in policy["blocklist"]], ["new@example.com", "old@example.com"])
        self.assertTrue(all(e["disposition"] == "block" for e in policy["blocklist"]))
        self.assertEqual(policy["blocklist"][0]["reason"], "spam")

    def test_remove_and_set(self):
        policy = {"allowlist": []}
        srv.update_policy_list(policy, "allowlist", "add", ["a@example.com", "b@example.com"])
        srv.update_policy_list(policy, "allowlist", "remove", ["a@example.com"])
        self.assertEqual([e["value"] for e in policy["allowlist"]], ["b@example.com"])
        self.assertEqual(policy["allowlist"][0]["disposition"], "allow")
        srv.update_policy_list(policy, "allowlist", "set", ["c@example.com"])
        self.assertEqual([e["value"] for e in policy["allowlist"]], ["c@example.com"])

    def test_domain_list_and_plain_keyword_list(self):
        policy = {"domainBlocklist": [], "subjectKeywords": ["viagra"]}
        srv.update_policy_list(policy, "domainBlocklist", "add", ["@Spam.Example"])
        self.assertEqual(policy["domainBlocklist"][0]["kind"], "domain")
        self.assertEqual(policy["domainBlocklist"][0]["value"], "spam.example")
        srv.update_policy_list(policy, "subjectKeywords", "add", ["Lottery"])
        self.assertEqual(policy["subjectKeywords"], ["lottery", "viagra"])

    def test_result_survives_normalisation(self):
        data = {"spamPolicy": {"blocklist": []}}
        srv.update_policy_list(data["spamPolicy"], "blocklist", "add", ["x@example.com"])
        policy = srv.spam_policy(data)
        self.assertEqual(policy["blocklist"][0]["value"], "x@example.com")

    def test_invalid_sender_rejected(self):
        with self.assertRaises(ValueError):
            srv.update_policy_list({"blocklist": []}, "blocklist", "add", ["not-an-address"])


class RemovedHookTests(unittest.TestCase):
    """The delivery hook and its watcher unit were retired; nothing of them may remain."""

    def test_hook_commands_are_rejected(self):
        for command in ("hook-watch", "hook-notify"):
            result = subprocess.run([sys.executable, str(SERVER), command], capture_output=True, text=True, timeout=10)
            self.assertEqual(result.returncode, 2, command)
            self.assertIn("invalid choice", result.stderr)

    def test_services_do_not_report_the_hook_unit(self):
        queried = []

        def fake_run(args):
            queried.append(args[-1])
            return subprocess.CompletedProcess(args, 0, stdout="active\n")

        original, srv.run = srv.run, fake_run
        try:
            status = srv.services()
        finally:
            srv.run = original
        self.assertNotIn("akurai-mail-hook", queried)
        self.assertEqual(set(status), {"postfix", "dovecot", "opendkim", "nginx", "akurai-mail-ui", "fail2ban"})
        self.assertTrue(all(v == "active" for v in status.values()))


if __name__ == "__main__":
    unittest.main()