govoplan-mail/tests/test_jmap.py

from __future__ import annotations

import base64
import json
import unittest
import urllib.error
from types import SimpleNamespace
from unittest.mock import patch

from govoplan_mail.backend.config import JmapConfig, JmapServerConfig
from govoplan_mail.backend.sending.jmap import (
    JMAP_CORE_CAPABILITY,
    JMAP_MAIL_CAPABILITY,
    JmapAuthenticationError,
    JmapCapabilityError,
    JmapConfigurationError,
    discover_jmap,
    get_jmap_email_changes,
    get_jmap_message,
    list_jmap_folders,
    list_jmap_messages,
    test_jmap_connection,
)


def _response(payload: dict) -> SimpleNamespace:
    return SimpleNamespace(
        status=200,
        headers={"Content-Type": "application/json"},
        body=json.dumps(payload).encode("utf-8"),
    )


def _session(*, api_url: str = "https://jmap.example.test/api") -> dict:
    return {
        "capabilities": {
            JMAP_CORE_CAPABILITY: {"maxCallsInRequest": 32},
            JMAP_MAIL_CAPABILITY: {},
        },
        "accounts": {
            "account-1": {
                "name": "Example",
                "isPersonal": True,
                "isReadOnly": False,
                "accountCapabilities": {JMAP_MAIL_CAPABILITY: {}},
            }
        },
        "primaryAccounts": {JMAP_MAIL_CAPABILITY: "account-1"},
        "username": "reader@example.test",
        "apiUrl": api_url,
        "downloadUrl": "https://jmap.example.test/download/{accountId}/{blobId}/{name}",
        "uploadUrl": "https://jmap.example.test/upload/{accountId}",
        "eventSourceUrl": "https://jmap.example.test/events",
        "state": "session-state-1",
    }


def _mailboxes() -> list[dict]:
    return [
        {
            "id": "mb-inbox",
            "name": "Inbox",
            "parentId": None,
            "role": "inbox",
            "sortOrder": 10,
            "isSubscribed": True,
            "totalEmails": 2,
            "unreadEmails": 1,
        },
        {
            "id": "mb-projects",
            "name": "Projects",
            "parentId": None,
            "role": None,
            "sortOrder": 20,
            "isSubscribed": True,
            "totalEmails": 1,
            "unreadEmails": 0,
        },
        {
            "id": "mb-project-2026",
            "name": "2026",
            "parentId": "mb-projects",
            "role": None,
            "sortOrder": 1,
            "isSubscribed": True,
            "totalEmails": 1,
            "unreadEmails": 0,
        },
        {
            "id": "mb-sent",
            "name": "Sent",
            "parentId": None,
            "role": "sent",
            "sortOrder": 30,
            "isSubscribed": True,
            "totalEmails": 4,
            "unreadEmails": 0,
        },
    ]


def _email(email_id: str = "email-1", *, detail: bool = False) -> dict:
    value = {
        "id": email_id,
        "threadId": "thread-1",
        "mailboxIds": {"mb-inbox": True},
        "keywords": {"$seen": False, "$flagged": True},
        "size": 1234,
        "receivedAt": "2026-08-22T10:30:00Z",
        "sentAt": "2026-08-22T10:29:00Z",
        "messageId": ["message-1@example.test"],
        "from": [{"name": "Sender", "email": "sender@example.test"}],
        "to": [{"name": "Reader", "email": "reader@example.test"}],
        "cc": [],
        "subject": "A governed message",
        "hasAttachment": True,
        "preview": "A bounded preview",
    }
    if detail:
        value.update(
            {
                "replyTo": [{"email": "reply@example.test"}],
                "bcc": [],
                "textBody": [{"partId": "text", "type": "text/plain"}],
                "htmlBody": [{"partId": "html", "type": "text/html"}],
                "bodyValues": {
                    "text": {"value": "Plain body", "isTruncated": False},
                    "html": {"value": "<p>HTML body</p>", "isTruncated": False},
                },
                "attachments": [
                    {
                        "partId": "attachment",
                        "blobId": "blob-1",
                        "name": "evidence.pdf",
                        "type": "application/pdf",
                        "size": 44,
                    }
                ],
            }
        )
    return value


class _JmapProvider:
    def __init__(self, *, query_states: list[str] | None = None) -> None:
        self.requests: list[tuple[str, str, dict[str, str], dict | None]] = []
        self.query_states = list(query_states or ["query-state-1"])

    def __call__(self, url: str, **kwargs):
        body = json.loads(kwargs["body"].decode("utf-8")) if kwargs.get("body") else None
        self.requests.append((url, kwargs.get("method", "GET"), kwargs.get("headers", {}), body))
        if kwargs.get("method") == "GET":
            return _response(_session())
        method, arguments, call_id = body["methodCalls"][0]
        if method == "Mailbox/get":
            payload = {"accountId": "account-1", "state": "mailbox-state-1", "list": _mailboxes(), "notFound": []}
        elif method == "Email/query":
            state = self.query_states.pop(0) if self.query_states else "query-state-1"
            payload = {
                "accountId": "account-1",
                "queryState": state,
                "canCalculateChanges": True,
                "position": arguments["position"],
                "ids": ["email-1", "email-2"][arguments["position"] : arguments["position"] + arguments["limit"]],
                "total": 2,
                "limit": arguments["limit"],
            }
        elif method == "Email/get":
            detail = bool(arguments.get("fetchTextBodyValues"))
            payload = {
                "accountId": "account-1",
                "state": "email-state-1",
                "list": [_email(email_id, detail=detail) for email_id in arguments["ids"]],
                "notFound": [],
            }
        elif method == "Email/changes":
            payload = {
                "accountId": "account-1",
                "oldState": arguments["sinceState"],
                "newState": "email-state-2",
                "hasMoreChanges": False,
                "created": ["email-new"],
                "updated": ["email-1"],
                "destroyed": ["email-old"],
            }
        else:  # pragma: no cover - fixture guard
            raise AssertionError(method)
        return _response({"methodResponses": [[method, payload, call_id]], "sessionState": "session-state-1"})


class JmapTransportTests(unittest.TestCase):
    def config(self, **overrides) -> JmapConfig:
        return JmapConfig(
            session_url="https://jmap.example.test/.well-known/jmap",
            password="access-token",
            **overrides,
        )

    def test_server_configuration_rejects_embedded_credentials_and_normalizes_origins(self) -> None:
        with self.assertRaisesRegex(ValueError, "embedded credentials"):
            JmapServerConfig(session_url="https://user:secret@example.test/jmap")
        config = JmapServerConfig(
            session_url="https://jmap.example.test/.well-known/jmap",
            allowed_api_origins=["https://api.example.test/path", "https://api.example.test"],
        )
        self.assertEqual(config.allowed_api_origins, ["https://api.example.test"])

    def test_discovery_selects_primary_mail_account_and_bearer_auth(self) -> None:
        provider = _JmapProvider()
        with patch("govoplan_mail.backend.sending.jmap.fetch_http", side_effect=provider):
            result = discover_jmap(self.config())
        self.assertEqual(result.account_id, "account-1")
        self.assertIn(JMAP_MAIL_CAPABILITY, result.account_capabilities)
        self.assertEqual(provider.requests[0][2]["Authorization"], "Bearer access-token")
        self.assertNotIn("access-token", repr(result))

    def test_basic_authentication_is_supported_without_exposing_credentials(self) -> None:
        provider = _JmapProvider()
        config = self.config(auth_scheme="basic", username="reader")
        with patch("govoplan_mail.backend.sending.jmap.fetch_http", side_effect=provider):
            test_jmap_connection(jmap_config=config)
        expected = base64.b64encode(b"reader:access-token").decode("ascii")
        self.assertEqual(provider.requests[0][2]["Authorization"], f"Basic {expected}")

    def test_cross_origin_api_url_is_fail_closed_unless_allowlisted(self) -> None:
        provider = _JmapProvider()
        provider.requests = []
        with patch(
            "govoplan_mail.backend.sending.jmap.fetch_http",
            return_value=_response(_session(api_url="https://api.example.test/jmap")),
        ):
            with self.assertRaisesRegex(JmapConfigurationError, "unapproved origin"):
                discover_jmap(self.config())

    def test_lists_hierarchical_mailboxes_with_roles_and_counts(self) -> None:
        provider = _JmapProvider()
        with patch("govoplan_mail.backend.sending.jmap.fetch_http", side_effect=provider):
            result = list_jmap_folders(jmap_config=self.config())
        self.assertEqual(result.protocol, "jmap")
        self.assertEqual(result.detected_folder_mappings["inbox"], "Inbox")
        self.assertEqual(result.detected_sent_folder, "Sent")
        self.assertIn("Projects/2026", [item.name for item in result.folders])
        inbox = next(item for item in result.folders if item.name == "Inbox")
        self.assertEqual((inbox.message_count, inbox.unseen_count), (2, 1))

    def test_query_search_and_get_share_protocol_neutral_message_shape(self) -> None:
        provider = _JmapProvider()
        with patch("govoplan_mail.backend.sending.jmap.fetch_http", side_effect=provider):
            result = list_jmap_messages(
                jmap_config=self.config(),
                folder="INBOX",
                limit=1,
                query="governed",
            )
        self.assertEqual(result.folder, "Inbox")
        self.assertEqual(result.total_count, 2)
        self.assertEqual(result.uidvalidity, "query-state-1")
        self.assertEqual(result.messages[0].uid, "email-1")
        self.assertEqual(result.messages[0].from_header, "Sender <sender@example.test>")
        query_call = next(request[3]["methodCalls"][0] for request in provider.requests if request[3] and request[3]["methodCalls"][0][0] == "Email/query")
        self.assertEqual(query_call[1]["filter"]["conditions"][1], {"text": "governed"})

    def test_changed_query_state_restarts_the_page_without_skipping(self) -> None:
        provider = _JmapProvider(query_states=["new-state", "newer-state"])
        with patch("govoplan_mail.backend.sending.jmap.fetch_http", side_effect=provider):
            result = list_jmap_messages(
                jmap_config=self.config(),
                folder="Inbox",
                limit=1,
                offset=1,
                expected_query_state="old-state",
            )
        self.assertTrue(result.cursor_reset)
        self.assertEqual(result.offset, 0)
        query_positions = [
            request[3]["methodCalls"][0][1]["position"]
            for request in provider.requests
            if request[3] and request[3]["methodCalls"][0][0] == "Email/query"
        ]
        self.assertEqual(query_positions, [1, 0])

    def test_detail_returns_bounded_body_values_and_attachment_metadata(self) -> None:
        provider = _JmapProvider()
        with patch("govoplan_mail.backend.sending.jmap.fetch_http", side_effect=provider):
            result = get_jmap_message(
                jmap_config=self.config(),
                folder="Inbox",
                email_id="email-1",
            )
        self.assertEqual(result.message.body_text, "Plain body")
        self.assertEqual(result.message.body_html, "<p>HTML body</p>")
        self.assertEqual(result.message.attachments[0].filename, "evidence.pdf")
        detail_call = provider.requests[-1][3]["methodCalls"][0][1]
        self.assertEqual(detail_call["maxBodyValueBytes"], 1024 * 1024)

    def test_incremental_changes_are_bounded_and_preserve_server_state(self) -> None:
        provider = _JmapProvider()
        with patch("govoplan_mail.backend.sending.jmap.fetch_http", side_effect=provider):
            result = get_jmap_email_changes(
                jmap_config=self.config(),
                since_state="email-state-1",
                max_changes=25,
            )
        self.assertEqual(result.new_state, "email-state-2")
        self.assertEqual(result.created, ("email-new",))
        change_call = provider.requests[-1][3]["methodCalls"][0][1]
        self.assertEqual(change_call["maxChanges"], 25)

    def test_authentication_and_expired_change_state_have_distinct_diagnostics(self) -> None:
        http_error = urllib.error.HTTPError(
            "https://jmap.example.test/.well-known/jmap",
            401,
            "Unauthorized",
            {},
            None,
        )
        with patch("govoplan_mail.backend.sending.jmap.fetch_http", side_effect=http_error):
            with self.assertRaisesRegex(JmapAuthenticationError, "authentication failed"):
                discover_jmap(self.config())

        def expired(url: str, **kwargs):
            if kwargs.get("method") == "GET":
                return _response(_session())
            body = json.loads(kwargs["body"])
            call_id = body["methodCalls"][0][2]
            return _response(
                {
                    "methodResponses": [
                        ["error", {"type": "cannotCalculateChanges"}, call_id]
                    ]
                }
            )

        with patch("govoplan_mail.backend.sending.jmap.fetch_http", side_effect=expired):
            with self.assertRaisesRegex(JmapCapabilityError, "full refresh"):
                get_jmap_email_changes(
                    jmap_config=self.config(),
                    since_state="expired-state",
                )


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