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

Latest change 7a284f40b2637792d0599c38c4cac0342da676d7 - Merge live EC2 ops copies into scripts/ and tools/ by Ólafur Búi Ólafsson

#!/usr/bin/env python3
"""Unit tests for scripts/akurai-mail-server merged logic: structured anti-spam
policy lists and the delivery hook (Multica issue creation + pi-bun push)."""
import http.server, importlib.machinery, importlib.util, json, os, shutil, tempfile, threading, unittest
from pathlib import Path

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

HOOK_ENV = ("AKURAI_MAIL_PIBUN_URL", "AKURAI_MAIL_PIBUN_TOKEN", "AKURAI_MAIL_MULTICA_URL", "AKURAI_MAIL_MULTICA_TOKEN",
            "AKURAI_MAIL_KLARA_ROUTES", "AKURAI_MAIL_MULTICA_WORKSPACE_ID", "AKURAI_MAIL_KLARA_PROJECT_ID", "AKURAI_MAIL_HOOK_SEEN")


class Recorder(http.server.BaseHTTPRequestHandler):
    requests = []

    def do_POST(self):
        body = self.rfile.read(int(self.headers.get("content-length", 0)))
        Recorder.requests.append((self.path, self.headers.get("authorization"), json.loads(body)))
        payload = json.dumps({"identifier": "OLI-1"}).encode()
        self.send_response(200)
        self.send_header("content-length", str(len(payload)))
        self.end_headers()
        self.wfile.write(payload)

    def log_message(self, *args):
        pass


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 HookTests(unittest.TestCase):
    def setUp(self):
        self.saved = {k: os.environ.get(k) for k in HOOK_ENV}
        for k in HOOK_ENV:
            os.environ.pop(k, None)
        base = HERE / ".test-tmp"
        base.mkdir(exist_ok=True)
        self.tmp = Path(tempfile.mkdtemp(dir=base))
        os.environ["AKURAI_MAIL_HOOK_SEEN"] = str(self.tmp / "seen.json")
        Recorder.requests = []
        self.httpd = http.server.HTTPServer(("127.0.0.1", 0), Recorder)
        threading.Thread(target=self.httpd.serve_forever, daemon=True).start()
        self.base_url = "http://127.0.0.1:%d" % self.httpd.server_port

    def tearDown(self):
        self.httpd.shutdown()
        self.httpd.server_close()
        shutil.rmtree(HERE / ".test-tmp", ignore_errors=True)
        for k, v in self.saved.items():
            if v is None:
                os.environ.pop(k, None)
            else:
                os.environ[k] = v

    def write_message(self, sender):
        path = self.tmp / "msg"
        path.write_text("From: %s\nTo: oli@olibuijr.com\nSubject: Hello\nMessage-ID: <m1@example.com>\n\nPlease do the thing.\n" % sender)
        return path

    def test_pibun_skipped_without_token(self):
        os.environ["AKURAI_MAIL_PIBUN_URL"] = self.base_url + "/api/ingest-mail"
        self.assertTrue(srv.post_pibun("oli@olibuijr.com"))
        self.assertEqual(Recorder.requests, [])

    def test_pibun_posts_account_with_token(self):
        os.environ["AKURAI_MAIL_PIBUN_URL"] = self.base_url + "/api/ingest-mail"
        os.environ["AKURAI_MAIL_PIBUN_TOKEN"] = "t0k"
        self.assertTrue(srv.post_pibun("oli@olibuijr.com"))
        self.assertEqual(Recorder.requests, [("/api/ingest-mail", "Bearer t0k", {"account": "oli@olibuijr.com"})])

    def configure_multica(self):
        os.environ.update({
            "AKURAI_MAIL_MULTICA_URL": self.base_url,
            "AKURAI_MAIL_MULTICA_TOKEN": "mt",
            "AKURAI_MAIL_MULTICA_WORKSPACE_ID": "ws",
            "AKURAI_MAIL_KLARA_PROJECT_ID": "proj",
            "AKURAI_MAIL_KLARA_ROUTES": json.dumps({"oli@olibuijr.com": "agent-1"}),
        })

    def test_multica_issue_from_klara_once(self):
        self.configure_multica()
        path = self.write_message("Klara <klara.rakel@icloud.com>")
        self.assertTrue(srv.post_multica("oli@olibuijr.com", path))
        self.assertTrue(srv.post_multica("oli@olibuijr.com", path))
        self.assertEqual(len(Recorder.requests), 1)
        url, auth, body = Recorder.requests[0]
        self.assertEqual(url, "/api/issues?workspace_id=ws")
        self.assertEqual(auth, "Bearer mt")
        self.assertEqual(body["assignee_id"], "agent-1")
        self.assertEqual(body["title"], "Email from Klara: Hello")

    def test_multica_ignores_other_senders(self):
        self.configure_multica()
        path = self.write_message("someone@example.com")
        self.assertTrue(srv.post_multica("oli@olibuijr.com", path))
        self.assertEqual(Recorder.requests, [])


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