hermes-proton/tests/test_mail.py

"""Tests for the proton-mail Hermes skill — IMAP/SMTP via Proton Mail Bridge.

All tests mock imaplib and smtplib. Real Bridge integration is behind a
pytest.mark.skipif guard for systems with Bridge running.
"""

import json
import pytest
from unittest.mock import patch, MagicMock, call
from skills.proton_mail import tools as mail_tools


# ── Fixtures ──────────────────────────────────────────────────────────────

@pytest.fixture
def mock_imap():
    """Mock IMAP4_SSL connection returning a logged-in session."""
    with patch("skills.proton_mail.tools.imaplib.IMAP4_SSL") as mock:
        inst = mock.return_value
        inst.login.return_value = ("OK", [b"Logged in"])
        inst.select.return_value = ("OK", [b"42"])
        inst.search.return_value = ("OK", [b"1 2 3"])
        # Default fetch: one entry (UID 1, basic headers)
        inst.fetch.return_value = (
            "OK",
            [_make_fetch_response(1, "Default Subject", "default@test.com",
                                  "Mon, 1 Jan 2024 10:00:00 +0000")],
        )
        # Default uid fetch
        inst.uid.return_value = (
            "OK",
            [_make_fetch_response(1, "Default Subject", "default@test.com",
                                  "Mon, 1 Jan 2024 10:00:00 +0000")],
        )
        inst.logout.return_value = ("OK", [b"Bye"])
        yield inst


@pytest.fixture
def mock_smtp():
    """Mock SMTP connection returning a logged-in session."""
    with patch("skills.proton_mail.tools.smtplib.SMTP") as mock:
        inst = mock.return_value
        inst.starttls.return_value = None
        inst.login.return_value = None
        inst.send_message.return_value = ({}, "test-message-id@bridge")
        yield inst


@pytest.fixture
def mock_env_vars():
    """Set Proton Bridge env vars for tests."""
    with patch.dict("os.environ", {
        "PROTONMAIL_ACCOUNT": "test@proton.me",
        "PROTONMAIL_BRIDGE_PASSWORD": "bridge-password-123",
    }):
        yield


# ── Helpers ───────────────────────────────────────────────────────────────

def _make_fetch_response(uid: int, subject: str, sender: str, date: str,
                          body: str = "Hello world") -> tuple:
    """Build a realistic IMAP FETCH response tuple.

    Returns (uid_data, rfc822_data) as a single-element list per IMAP
    protocol — each list item is (header_bytes, message_bytes).
    """
    headers = (
        f"Subject: {subject}\r\n"
        f"From: {sender}\r\n"
        f"To: recipient@proton.me\r\n"
        f"Date: {date}\r\n"
        f"Message-ID: <msg{uid}@proton.me>\r\n"
        f"MIME-Version: 1.0\r\n"
        f"Content-Type: text/plain; charset=utf-8\r\n"
        f"Content-Transfer-Encoding: 7bit\r\n"
        f"\r\n"
    ).encode("utf-8")
    msg_bytes = headers + body.encode("utf-8")
    uid_wrapper = f"1 (UID {uid} FLAGS (\\Seen))\r\n".encode("utf-8")
    return (uid_wrapper, msg_bytes)


# ── Bridge Status ─────────────────────────────────────────────────────────

class TestBridgeStatus:
    def test_bridge_running(self, mock_env_vars):
        """Bridge status returns 'running' when IMAP port is reachable."""
        with patch("skills.proton_mail.tools._check_port_open",
                   return_value=True):
            result = json.loads(mail_tools.proton_mail_bridge_status({}))
            assert result["status"] == "running"

    def test_bridge_not_running(self, mock_env_vars):
        """Bridge status returns 'stopped' when IMAP port is unreachable."""
        with patch("skills.proton_mail.tools._check_port_open",
                   return_value=False):
            result = json.loads(mail_tools.proton_mail_bridge_status({}))
            assert result["status"] == "stopped"

    def test_missing_env_vars(self):
        """Bridge status returns error when credentials are missing."""
        with patch.dict("os.environ", {}, clear=True):
            result = json.loads(mail_tools.proton_mail_bridge_status({}))
            assert "error" in result


# ── Mail List ─────────────────────────────────────────────────────────────

class TestMailList:
    def test_list_returns_messages(self, mock_imap, mock_env_vars):
        """List returns parsed messages with subject, from, date, uid."""
        mock_imap.search.return_value = ("OK", [b"1 2 3"])
        mock_imap.fetch.side_effect = [
            ("OK", [_make_fetch_response(1, "Subject A", "a@test.com", "Mon, 1 Jan 2024 10:00:00 +0000")]),
            ("OK", [_make_fetch_response(2, "Subject B", "b@test.com", "Tue, 2 Jan 2024 11:00:00 +0000")]),
            ("OK", [_make_fetch_response(3, "Subject C", "c@test.com", "Wed, 3 Jan 2024 12:00:00 +0000")]),
        ]

        result = json.loads(mail_tools.proton_mail_list({"folder": "INBOX", "limit": 3}))

        assert result["success"] is True
        assert len(result["messages"]) == 3
        assert result["messages"][0]["subject"] == "Subject A"
        assert result["messages"][0]["from"] == "a@test.com"
        assert result["messages"][2]["uid"] == 3
        assert result["folder"] == "INBOX"

    def test_list_empty_mailbox(self, mock_imap, mock_env_vars):
        """List returns empty array when no messages match."""
        mock_imap.search.return_value = ("OK", [b""])

        result = json.loads(mail_tools.proton_mail_list({"folder": "INBOX", "limit": 10}))

        assert result["success"] is True
        assert result["messages"] == []

    def test_list_custom_folder(self, mock_imap, mock_env_vars):
        """List selects a non-INBOX folder."""
        mock_imap.search.return_value = ("OK", [b"1 2"])

        result = json.loads(mail_tools.proton_mail_list({"folder": "Sent", "limit": 5}))

        mock_imap.select.assert_called_with('"Sent"', readonly=True)
        assert result["folder"] == "Sent"

    def test_list_limits_results(self, mock_imap, mock_env_vars):
        """List respects the limit parameter."""
        mock_imap.search.return_value = ("OK", [b"1 2 3 4 5 6 7 8 9 10"])

        result = json.loads(mail_tools.proton_mail_list({"folder": "INBOX", "limit": 3}))

        assert len(result["messages"]) == 3

    def test_list_imap_error(self, mock_imap, mock_env_vars):
        """List returns error on IMAP failure."""
        mock_imap.search.return_value = ("NO", [b"Search failed"])

        result = json.loads(mail_tools.proton_mail_list({"folder": "INBOX", "limit": 10}))

        assert "error" in result
        assert result["success"] is False


# ── Mail Read ─────────────────────────────────────────────────────────────

class TestMailRead:
    def test_read_returns_full_message(self, mock_imap, mock_env_vars):
        """Read returns subject, from, to, date, body, and uid."""
        mock_imap.uid.return_value = (
            "OK",
            [_make_fetch_response(42, "Fancy Subject", "alice@test.com",
                                  "Thu, 4 Jan 2024 14:00:00 +0000",
                                  "This is the message body.\nWith two lines.")]
        )

        result = json.loads(mail_tools.proton_mail_read({"uid": 42, "folder": "INBOX"}))

        assert result["success"] is True
        assert result["uid"] == 42
        assert result["subject"] == "Fancy Subject"
        assert result["from"] == "alice@test.com"
        assert result["body"] == "This is the message body.\nWith two lines."

    def test_read_missing_uid(self, mock_env_vars):
        """Read returns error when uid parameter is missing."""
        result = json.loads(mail_tools.proton_mail_read({"folder": "INBOX"}))

        assert "error" in result

    def test_read_uid_not_found(self, mock_imap, mock_env_vars):
        """Read returns error when UID fetch returns empty."""
        mock_imap.uid.return_value = ("OK", [None])

        result = json.loads(mail_tools.proton_mail_read({"uid": 999, "folder": "INBOX"}))

        assert "error" in result


# ── Mail Search ───────────────────────────────────────────────────────────

class TestMailSearch:
    def test_search_by_subject(self, mock_imap, mock_env_vars):
        """Search finds messages matching a subject query."""
        mock_imap.search.return_value = ("OK", [b"2 3"])
        mock_imap.fetch.side_effect = [
            ("OK", [_make_fetch_response(2, "Meeting at 3pm", "b@test.com", "Tue, 2 Jan 2024 11:00:00 +0000")]),
            ("OK", [_make_fetch_response(3, "Meeting notes", "c@test.com", "Wed, 3 Jan 2024 12:00:00 +0000")]),
        ]

        result = json.loads(mail_tools.proton_mail_search({
            "query": "Meeting", "folder": "INBOX", "limit": 10
        }))

        assert result["success"] is True
        assert len(result["messages"]) == 2
        assert all("Meeting" in m["subject"] for m in result["messages"])

    def test_search_from_sender(self, mock_imap, mock_env_vars):
        """Search filters by sender."""
        mock_imap.search.return_value = ("OK", [b"1"])
        mock_imap.fetch.return_value = (
            "OK",
            [_make_fetch_response(1, "Hello", "specific@test.com", "Mon, 1 Jan 2024 10:00:00 +0000")]
        )

        result = json.loads(mail_tools.proton_mail_search({
            "query": "specific@test.com", "folder": "INBOX",
            "limit": 10, "field": "from"
        }))

        assert result["success"] is True
        assert result["messages"][0]["from"] == "specific@test.com"

    def test_search_no_results(self, mock_imap, mock_env_vars):
        """Search returns empty when nothing matches."""
        mock_imap.search.return_value = ("OK", [b""])

        result = json.loads(mail_tools.proton_mail_search({
            "query": "zzzzzxxxxx", "folder": "INBOX", "limit": 10
        }))

        assert result["success"] is True
        assert result["messages"] == []

    def test_search_requires_query(self, mock_env_vars):
        """Search returns error when query is missing or too short."""
        result = json.loads(mail_tools.proton_mail_search({
            "folder": "INBOX", "limit": 10
        }))
        assert "error" in result


# ── Mail Send ─────────────────────────────────────────────────────────────

class TestMailSend:
    def test_send_plain_text(self, mock_smtp, mock_env_vars):
        """Send delivers a plain text email."""
        result = json.loads(mail_tools.proton_mail_send({
            "to": "recipient@example.com",
            "subject": "Test Subject",
            "body": "Hello from Proton skill!",
        }))

        assert result["success"] is True
        assert "message_id" in result

    def test_send_with_cc(self, mock_smtp, mock_env_vars):
        """Send includes CC recipients."""
        result = json.loads(mail_tools.proton_mail_send({
            "to": "primary@example.com",
            "cc": "cc@example.com",
            "subject": "Cc Test",
            "body": "CC included.",
        }))

        assert result["success"] is True
        assert mock_smtp.send_message.called

    def test_send_requires_recipients(self, mock_env_vars):
        """Send returns error when 'to' is missing."""
        result = json.loads(mail_tools.proton_mail_send({
            "subject": "No Recipient",
            "body": "Where does this go?",
        }))
        assert "error" in result

    def test_send_requires_subject(self, mock_env_vars):
        """Send returns error with a helpful message when subject is missing."""
        result = json.loads(mail_tools.proton_mail_send({
            "to": "recipient@example.com",
            "body": "No subject",
        }))
        assert "error" in result


# ── Mail Reply ────────────────────────────────────────────────────────────

class TestMailReply:
    def test_reply_sets_thread_headers(self, mock_imap, mock_smtp, mock_env_vars):
        """Reply reads original, sets In-Reply-To and References headers."""
        # Original message fetch
        mock_imap.fetch.return_value = (
            "OK",
            [_make_fetch_response(10, "Original Thread", "original@test.com",
                                  "Fri, 5 Jan 2024 09:00:00 +0000",
                                  "This is the original email.")]
        )

        result = json.loads(mail_tools.proton_mail_reply({
            "uid": 10,
            "body": "Thanks for your email!",
        }))

        assert result["success"] is True

    def test_reply_requires_uid(self, mock_env_vars):
        """Reply returns error when uid is missing."""
        result = json.loads(mail_tools.proton_mail_reply({
            "body": "Missing uid",
        }))
        assert "error" in result

    def test_reply_requires_body(self, mock_env_vars):
        """Reply returns error when body is missing."""
        result = json.loads(mail_tools.proton_mail_reply({
            "uid": 10,
        }))
        assert "error" in result