tests/test_desktop_mail_assistant.py — Desktop Mail Assistant

#!/usr/bin/env python3
"""Synthetic tests for Desktop Mail Assistant.

By Zero Width Lab (zerowidthlab.com)
"""

from __future__ import annotations

import importlib.util
import json
import os
import subprocess
import sys
import tempfile
import unittest
from pathlib import Path
from unittest import mock

__author__ = "Zero Width Lab"

ROOT = Path(__file__).resolve().parents[1]
SCRIPT = ROOT / "scripts" / "desktop_mail_assistant.py"
FIXTURE = ROOT / "tests" / "fixtures" / "mail_responses.json"
SPEC = importlib.util.spec_from_file_location("desktop_mail_assistant", SCRIPT)
assert SPEC and SPEC.loader
tool = importlib.util.module_from_spec(SPEC)
SPEC.loader.exec_module(tool)


def cli(*args: str, fixture: bool = True) -> subprocess.CompletedProcess[str]:
    env = os.environ.copy()
    env.pop(tool.FAKE_RESPONSES, None)
    env.pop(tool.FAKE_ACK, None)
    if fixture:
        env[tool.FAKE_RESPONSES] = str(FIXTURE)
        env[tool.FAKE_ACK] = "1"
    return subprocess.run([sys.executable, str(SCRIPT), *args], env=env, capture_output=True, text=True, timeout=10)


class DesktopMailAssistantTests(unittest.TestCase):
    def test_read_commands_use_fixture(self):
        for args in (("accounts",), ("mailboxes", "--account", "Example"),
                     ("recent", "--account", "Example", "--mailbox", "Inbox"),
                     ("search", "--account", "Example", "--mailbox", "Inbox", "--subject", "Hello")):
            with self.subTest(args=args):
                result = cli(*args)
                self.assertEqual(result.returncode, 0, result.stderr)
                self.assertIn("synthetic osascript fixture", result.stderr)

    def test_read_body_limit_is_partial(self):
        args = ("read", "--account", "Example", "--mailbox", "Inbox", "--message-id", "42", "--max-body-chars", "4")
        result = cli(*args)
        self.assertEqual(result.returncode, 2)
        self.assertIn("abcd", result.stdout)
        self.assertNotIn("efghij", result.stdout)
        self.assertIn("skipped", result.stdout)
        self.assertEqual(cli(*args, "--allow-partial").returncode, 0)

    def test_user_text_only_in_argv(self):
        marker = 'subject "quoted" ' + chr(10) + ' end tell'
        completed = subprocess.CompletedProcess([], 0, json.dumps({"messages": [], "incomplete": False, "skipped": []}), "")
        with mock.patch.object(tool.subprocess, "run", return_value=completed) as runner:
            tool.mail_call("search", "Account", "Inbox", "2", "subject", marker)
        call = runner.call_args.args[0]
        self.assertEqual(call[0], "/usr/bin/osascript")
        self.assertEqual(call[-1], marker)
        self.assertNotIn(marker, call[4])
        self.assertEqual(runner.call_args.kwargs["timeout"], tool.TIMEOUT)

    def test_send_digest_and_verification(self):
        with tempfile.TemporaryDirectory() as directory:
            body = Path(directory) / "body.txt"
            body.write_text("Body\nsecond line", encoding="utf-8")
            args = ("send", "--from", "from.invalid", "--to", "to.invalid", "--subject", "Hello", "--body-file", str(body))
            preview = cli(*args, "--dry-run")
            self.assertEqual(preview.returncode, 0, preview.stderr)
            self.assertIn("body: Body\\u000asecond line", preview.stdout)
            approval = preview.stdout.split("approval_digest: ")[1].strip()
            self.assertEqual(len(approval), 64)
            self.assertEqual(cli(*args).returncode, 1)
            self.assertEqual(cli(*args, "--approve", "0" * 64).returncode, 1)
            sent = cli(*args, "--approve", approval)
            self.assertEqual(sent.returncode, 0, sent.stderr)
            self.assertIn("sent_verified: true", sent.stdout)
            body.write_text("Changed", encoding="utf-8")
            self.assertEqual(cli(*args, "--approve", approval).returncode, 1)

    def test_draft_gate(self):
        with tempfile.TemporaryDirectory() as directory:
            body = Path(directory) / "body.txt"
            body.write_text("Draft", encoding="utf-8")
            args = ("draft", "--from", "from.invalid", "--to", "to.invalid", "--subject", "Hello", "--body-file", str(body))
            preview = cli(*args, "--dry-run")
            approval = preview.stdout.split("approval_digest: ")[1].strip()
            self.assertEqual(cli(*args, "--approve", approval).returncode, 0)

    def test_fake_fixture_required_before_transport(self):
        with mock.patch.object(tool.subprocess, "run") as runner, mock.patch.dict(os.environ, {tool.FAKE_RESPONSES: "/missing/fixture", tool.FAKE_ACK: "1"}):
            with self.assertRaises(tool.CliError):
                tool.mail_call("accounts")
            runner.assert_not_called()

    def test_unsupported_platform_before_io(self):
        with mock.patch.object(tool.sys, "platform", "win32"), mock.patch.object(tool.Path, "open") as opened, mock.patch.object(tool.subprocess, "run") as runner:
            for command in (("accounts",), ("send", "--from", "from.invalid", "--to", "to.invalid", "--subject", "Hi", "--body-file", "missing", "--dry-run")):
                self.assertEqual(tool.main(list(command)), 1)
            with self.assertRaisesRegex(tool.CliError, "macOS only"):
                tool.body_from_file("missing")
            with self.assertRaises(tool.CliError):
                tool.mail_call("accounts")
            opened.assert_not_called()
            runner.assert_not_called()

    def test_bounded_body_and_fixture_inputs(self):
        with tempfile.TemporaryDirectory() as directory:
            path = Path(directory) / "large.txt"
            for content, error in ((b"x" * (tool.BODY_BYTES + 1), "bytes"), (b"x" * 100_001, "characters"), (b"\xff", "cannot read")):
                path.write_bytes(content)
                with self.assertRaisesRegex(tool.CliError, error):
                    tool.body_from_file(str(path))
            path.write_bytes(b"x" * (tool.FIXTURE_BYTES + 1))
            with mock.patch.dict(os.environ, {tool.FAKE_RESPONSES: str(path), tool.FAKE_ACK: "1"}), mock.patch.object(tool.subprocess, "run") as runner:
                with self.assertRaisesRegex(tool.CliError, "bytes"):
                    tool.mail_call("accounts")
                runner.assert_not_called()

    def test_malformed_response_shapes(self):
        invalid = {"accounts": [{}, {"accounts": [{}]}, {"accounts": [{"name": "A", "addresses": [1]}]}],
                   "mailboxes": [{}, {"mailboxes": [1]}], "recent": [{}, {"messages": [{}], "incomplete": False, "skipped": []}],
                   "search": [{"messages": [], "incomplete": "false", "skipped": []}], "read": [{"message": {"id": 42, "body": "x"}}],
                   "draft": [{"draft_saved": 1}], "send": [{"send_accepted": True, "started_at": "yesterday"}],
                   "verify": [{"verified": True, "matching_sent_messages": 0}, {"verified": True, "matching_sent_messages": True}]}
        for operation, responses in invalid.items():
            for response in responses:
                with self.subTest(operation=operation, response=response), self.assertRaisesRegex(tool.CliError, "malformed"):
                    tool.validate_response(operation, response)
        with mock.patch.object(tool.subprocess, "run", return_value=subprocess.CompletedProcess([], 0, '{"accounts":[{}]}', "")):
            with self.assertRaisesRegex(tool.CliError, "malformed"):
                tool.mail_call("accounts")
        with tempfile.TemporaryDirectory() as directory:
            path = Path(directory) / "fixture.json"
            path.write_text('{"accounts":{"accounts":[{}]}}', encoding="utf-8")
            with mock.patch.dict(os.environ, {tool.FAKE_RESPONSES: str(path), tool.FAKE_ACK: "1"}):
                with self.assertRaisesRegex(tool.CliError, "malformed"):
                    tool.mail_call("accounts")

    def synthetic_jxa(self, operation, failure="optional"):
        # The fixed JXA runs in Node with a synthetic Application, never Mail.
        harness = r"""
function broken() { throw new Error('synthetic property failure'); }
const failure = process.argv[1];
const operation = process.argv[2];
const payload = {from: 'sender.invalid', to: [], cc: [], bcc: [], subject: '', body: 'Body', started_at: new Date().toISOString()};
const message = {id: failure === 'id' ? broken : () => '42', subject: operation === 'verify' ? () => '' : broken, sender: operation === 'verify' ? () => 'sender.invalid' : broken, dateReceived: broken,
  content: () => 'Body', toRecipients: () => [], ccRecipients: () => [], bccRecipients: () => [], dateSent: broken};
const inbox = {name: failure === 'mailbox' ? broken : () => 'Inbox', messages: () => [message, {...message, id: () => '43'}]};
const good = {name: failure === 'account' ? broken : () => 'Good', emailAddresses: () => ['sender.invalid'], mailboxes: () => [inbox], sentMailbox: () => inbox};
const bad = {name: () => 'BrokenOptional', emailAddresses: broken};
function Application() { return {accounts: () => [bad, good],
  OutgoingMessage: () => ({toRecipients: [], ccRecipients: [], bccRecipients: [], save: () => true, send: () => true}),
  outgoingMessages: []}; }
const args = ['send', 'draft', 'verify'].includes(operation) ? [operation, JSON.stringify(payload)] : [operation, 'Good', 'Inbox', operation === 'read' ? '42' : '5', 'subject', ''];
process.stdout.write(run(args));
"""
        return subprocess.run(["node", "-e", tool.JXA + harness, failure, operation],
                              capture_output=True, text=True, timeout=10)

    def test_jxa_optional_properties_tolerate_errors(self):
        for operation in ("accounts", "recent", "search", "send", "draft", "verify", "read"):
            with self.subTest(operation=operation):
                result = self.synthetic_jxa(operation)
                self.assertEqual(result.returncode, 0, result.stderr)
                response = tool.validate_response(operation, json.loads(result.stdout))
                if operation == "accounts":
                    self.assertEqual(response["accounts"][0]["addresses"], [])
                    self.assertEqual(response["accounts"][1]["addresses"], ["sender.invalid"])
                elif operation in ("recent", "search"):
                    self.assertEqual(response["messages"][0], {"id": "42", "subject": "", "sender": "", "date": ""})

    def test_jxa_required_properties_remain_strict(self):
        for failure in ("account", "mailbox", "id"):
            with self.subTest(failure=failure):
                result = self.synthetic_jxa("recent", failure)
                self.assertNotEqual(result.returncode, 0)
                self.assertIn("synthetic property failure", result.stderr)

    def test_setup_documents_argv_size_limit(self):
        setup = (ROOT / "SETUP.md").read_text(encoding="utf-8")
        self.assertIn("single argument", setup)
        self.assertIn("OS argument-size limit", setup)
        self.assertIn("could not launch or decode osascript response", setup)

    def test_deep_json_and_invalid_transport_encoding(self):
        nested = "[" * 2000 + "0" + "]" * 2000
        with mock.patch.object(tool.subprocess, "run", return_value=subprocess.CompletedProcess([], 0, nested, "")):
            with self.assertRaisesRegex(tool.CliError, "invalid JSON|malformed"):
                tool.mail_call("accounts")
        with mock.patch.object(tool.subprocess, "run", side_effect=UnicodeDecodeError("utf-8", b"\xff", 0, 1, "invalid")):
            with self.assertRaisesRegex(tool.CliError, "decode"):
                tool.mail_call("accounts")
        with tempfile.TemporaryDirectory() as directory:
            path = Path(directory) / "fixture.json"
            path.write_text(nested, encoding="utf-8")
            with mock.patch.dict(os.environ, {tool.FAKE_RESPONSES: str(path), tool.FAKE_ACK: "1"}):
                with self.assertRaisesRegex(tool.CliError, "invalid fake response"):
                    tool.mail_call("accounts")

    def test_non_finite_json_constants_rejected(self):
        for constant in ("NaN", "Infinity", "-Infinity"):
            text = '{"accounts": [], "extra": ' + constant + '}'
            with mock.patch.object(tool.subprocess, "run", return_value=subprocess.CompletedProcess([], 0, text, "")):
                with self.assertRaisesRegex(tool.CliError, "invalid JSON"):
                    tool.mail_call("accounts")
            with tempfile.TemporaryDirectory() as directory:
                path = Path(directory) / "fixture.json"
                path.write_text('{"accounts":' + text + '}', encoding="utf-8")
                with mock.patch.dict(os.environ, {tool.FAKE_RESPONSES: str(path), tool.FAKE_ACK: "1"}):
                    with self.assertRaisesRegex(tool.CliError, "non-finite"):
                        tool.mail_call("accounts")

    def test_output_escaping(self):
        self.assertEqual(tool.display_text("x" + chr(27) + chr(10) + "y"), "x\\u001b\\u000ay")

    def test_timeout_message(self):
        with mock.patch.object(tool.subprocess, "run", side_effect=subprocess.TimeoutExpired([], 1)):
            with self.assertRaisesRegex(tool.CliError, "permission prompt"):
                tool.mail_call("accounts")


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

License: MIT (read it).

Back to the listing