tests/test_mail_and_files_connector.py — Mail and Files Connector

#!/usr/bin/env python3
# By Zero Width Lab (zerowidthlab.com)
"""Synthetic subprocess tests for the read-only stdio server."""
from __future__ import annotations

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

__author__ = "Zero Width Lab"
ROOT = Path(__file__).resolve().parents[1]
SCRIPT = ROOT / "scripts/mail_and_files_connector.py"
FIXTURE = ROOT / "tests/fixtures/graph.json"
META = {"io.modelcontextprotocol/protocolVersion": "2026-07-28",
        "io.modelcontextprotocol/clientCapabilities": {}}


def request(number: int, method: str, params: dict | None = None) -> dict:
    return {"jsonrpc": "2.0", "id": number, "method": method,
            "params": {"_meta": dict(META), **(params or {})}}


class ConnectorTests(unittest.TestCase):
    def run_server(self, messages: list[dict], fixture: str | None = None,
                   allow_partial: bool = False) -> tuple[list[dict], str, int]:
        env = os.environ.copy()
        env.update({"GRAPH_TENANT_ID": "tenant-test", "GRAPH_CLIENT_ID": "secret-client",
                    "GRAPH_CLIENT_SECRET": "secret-password", "MAIL_DEFAULT_USER": "user@example.com",
                    "SHAREPOINT_SITE": "tenant.sharepoint.com:/sites/test",
                    "GRAPH_FAKE_RESPONSES": fixture or str(FIXTURE),
                    "GRAPH_FAKE_TRANSPORT_TEST_ACK": "zero-width-lab-test-suite-only"})
        command = [sys.executable, str(SCRIPT)] + (["--allow-partial"] if allow_partial else [])
        result = subprocess.run(command, input="".join(json.dumps(x) + "\n" for x in messages),
                                text=True, capture_output=True, env=env, cwd=ROOT, timeout=10, check=False)
        self.assertNotIn("secret-password", result.stdout + result.stderr)
        self.assertNotIn("secret-client", result.stdout + result.stderr)
        self.assertNotIn("secret-token", result.stdout + result.stderr)
        return [json.loads(line) for line in result.stdout.splitlines()], result.stderr, result.returncode

    def test_discover_and_deterministic_listing(self) -> None:
        output, warning, status = self.run_server([request(1, "server/discover"), request(2, "tools/list"),
                                                   request(3, "tools/list")])
        self.assertEqual(status, 0)
        self.assertIn("fake transport", warning)
        discover = output[0]["result"]
        self.assertEqual(discover["supportedVersions"], ["2026-07-28", "2025-11-25", "2025-06-18"])
        self.assertEqual(discover["_meta"]["io.modelcontextprotocol/serverInfo"],
                         {"name": "mail-and-files-connector", "version": "0.1.0"})
        self.assertEqual(discover["resultType"], "complete")
        self.assertEqual(discover["ttlMs"], 0)
        self.assertEqual(discover["cacheScope"], "private")
        self.assertEqual(output[1]["result"], output[2]["result"])
        self.assertEqual([x["name"] for x in output[1]["result"]["tools"]],
                         ["mail_recent", "mail_search", "mail_read", "mail_thread",
                          "files_search", "files_list", "file_versions"])
        self.assertEqual(output[1]["result"]["cacheScope"], "private")
        self.assertEqual(output[1]["result"]["resultType"], "complete")
        self.assertTrue(all("allow_partial" not in tool["inputSchema"]["properties"]
                            for tool in output[1]["result"]["tools"]))

    def test_mail_success_partial_and_error(self) -> None:
        calls = [request(1, "tools/call", {"name": "mail_recent", "arguments": {"limit": 1}}),
                 request(2, "tools/call", {"name": "mail_recent", "arguments": {}}),
                 request(3, "tools/call", {"name": "mail_read", "arguments": {"id": "m1", "body_limit": 4}}),
                 request(4, "tools/call", {"name": "mail_search", "arguments": {}})]
        output, _, status = self.run_server(calls)
        self.assertEqual(status, 0)
        self.assertTrue(output[0]["result"]["isError"])
        self.assertEqual(output[0]["result"]["structuredContent"]["skipped"], ["additional pages"])
        self.assertTrue(output[1]["result"]["isError"])
        self.assertTrue(output[1]["result"]["structuredContent"]["incomplete"])
        self.assertEqual(output[1]["result"]["resultType"], "complete")
        for response in output:
            self.assertNotIn("ttlMs", response["result"])
            self.assertNotIn("cacheScope", response["result"])
        self.assertEqual(output[2]["result"]["structuredContent"]["item"]["body"]["content"], "synt")
        self.assertTrue(output[3]["result"]["isError"])

    def test_files_and_versions(self) -> None:
        output, _, status = self.run_server([
            request(1, "tools/call", {"name": "files_list", "arguments": {}}),
            request(2, "tools/call", {"name": "file_versions", "arguments": {"id": "f1"}})])
        self.assertEqual(status, 0)
        self.assertEqual(output[0]["result"]["structuredContent"]["items"][0]["name"], "report.txt")
        self.assertNotIn("secret-link", json.dumps(output))
        self.assertTrue(output[0]["result"]["structuredContent"]["incomplete"])
        self.assertEqual(output[1]["result"]["structuredContent"]["items"][0]["id"], "v1")

    def test_unsupported_version(self) -> None:
        message = request(1, "tools/list")
        message["params"]["_meta"]["io.modelcontextprotocol/protocolVersion"] = "1900-01-01"
        output, _, _ = self.run_server([message])
        self.assertEqual(output[0]["error"]["code"], -32022)
        self.assertEqual(output[0]["error"]["data"],
                         {"supported": ["2026-07-28", "2025-11-25", "2025-06-18"],
                          "requested": "1900-01-01"})
        legacy, _, _ = self.run_server([{"jsonrpc": "2.0", "id": 4, "method": "initialize",
                                         "params": {"protocolVersion": "1900-01-01"}}])
        self.assertEqual(legacy[0]["error"]["data"], output[0]["error"]["data"])

    def test_allow_partial_startup_flag(self) -> None:
        output, _, status = self.run_server(
            [request(1, "tools/call", {"name": "mail_recent", "arguments": {}})], allow_partial=True)
        self.assertEqual(status, 0)
        self.assertFalse(output[0]["result"]["isError"])
        self.assertTrue(output[0]["result"]["structuredContent"]["incomplete"])
        self.assertEqual(output[0]["result"]["structuredContent"]["skipped"], ["additional pages"])

    def test_unknown_arguments_rejected_without_request(self) -> None:
        output, _, status = self.run_server([
            request(1, "tools/call", {"name": "mail_read", "arguments": {"id": "m1", "limit": 1}}),
            request(2, "tools/call", {"name": "mail_recent", "arguments": {"unexpected": 1}}),
            request(3, "tools/call", {"name": "mail_recent", "arguments": {}})])
        self.assertEqual(status, 0)
        self.assertIn("unknown tool argument", output[0]["result"]["content"][0]["text"])
        self.assertTrue(output[0]["result"]["isError"])
        self.assertTrue(output[1]["result"]["isError"])
        self.assertIn("structuredContent", output[2]["result"])

    def test_malformed_json_line_and_internal_error_keep_server_alive(self) -> None:
        env = os.environ.copy()
        env.update({"GRAPH_FAKE_RESPONSES": str(FIXTURE),
                    "GRAPH_FAKE_TRANSPORT_TEST_ACK": "zero-width-lab-test-suite-only"})
        result = subprocess.run([sys.executable, str(SCRIPT)],
                                input='{bad json\n' + json.dumps(request(2, "tools/list")) + '\n',
                                text=True, capture_output=True, env=env, cwd=ROOT, timeout=10, check=False)
        responses = [json.loads(line) for line in result.stdout.splitlines()]
        self.assertEqual(result.returncode, 0)
        self.assertEqual(responses[0]["error"]["code"], -32700)
        self.assertIsNone(responses[0]["id"])
        self.assertEqual(responses[1]["id"], 2)

        spec = importlib.util.spec_from_file_location("connector_internal_test", SCRIPT)
        self.assertIsNotNone(spec)
        self.assertIsNotNone(spec.loader)
        module = importlib.util.module_from_spec(spec)
        spec.loader.exec_module(module)
        stdout = io.StringIO()
        with mock.patch.object(module, "GraphClient", return_value=object()), \
             mock.patch.object(module, "dispatch", side_effect=RuntimeError("synthetic")), \
             mock.patch.object(sys, "stdin", io.TextIOWrapper(io.BytesIO((json.dumps(request(17, "tools/list")) + "\n").encode()), encoding="utf-8")), \
             mock.patch.object(sys, "stdout", stdout), mock.patch.object(sys, "stderr", io.StringIO()), \
             mock.patch.object(sys, "argv", [str(SCRIPT)]):
            self.assertEqual(module.main(), 0)
        internal = json.loads(stdout.getvalue())
        self.assertEqual(internal["id"], 17)
        self.assertEqual(internal["error"]["code"], -32603)

    def test_legacy_initialize_and_call(self) -> None:
        output, _, _ = self.run_server([
            {"jsonrpc": "2.0", "id": 1, "method": "initialize", "params": {"protocolVersion": "2025-11-25"}},
            {"jsonrpc": "2.0", "method": "notifications/initialized"},
            {"jsonrpc": "2.0", "id": 2, "method": "tools/list", "params": {}}])
        self.assertEqual(output[0]["result"]["protocolVersion"], "2025-11-25")
        self.assertEqual(len(output[1]["result"]["tools"]), 7)

    def test_second_legacy_version_and_redirect_guards(self) -> None:
        output, _, _ = self.run_server([
            {"jsonrpc": "2.0", "id": 1, "method": "initialize", "params": {"protocolVersion": "2025-06-18"}}])
        self.assertEqual(output[0]["result"]["protocolVersion"], "2025-06-18")
        spec = importlib.util.spec_from_file_location("connector", SCRIPT)
        self.assertIsNotNone(spec)
        self.assertIsNotNone(spec.loader)
        module = importlib.util.module_from_spec(spec)
        spec.loader.exec_module(module)
        module.guard_url("https://tenant.sharepoint.com/content", True)
        with self.assertRaises(module.ConnectorError):
            module.guard_url("https://tenant.sharepoint.com/content")
        with self.assertRaises(module.ConnectorError):
            module.guard_url("http://graph.microsoft.com/resource")
        with self.assertRaises(module.ConnectorError):
            module.guard_url("https://nested.tenant.sharepoint.com/content", True)

    def test_missing_fixture_fails_before_request(self) -> None:
        output, warning, status = self.run_server([request(1, "tools/call", {"name": "mail_recent"})],
                                                   str(ROOT / "tests/fixtures/nonexistent.json"))
        self.assertEqual(output, [])
        self.assertEqual(status, 1)
        self.assertIn("fixture missing", warning)

    def load_module(self):
        spec = importlib.util.spec_from_file_location("connector_audit", SCRIPT)
        module = importlib.util.module_from_spec(spec)
        spec.loader.exec_module(module)
        return module

    def test_remote_paths_rejected_before_any_request(self):
        module = self.load_module()
        client = mock.Mock()
        invalid = ("../escape", "/absolute", "a/../b", "a/./b", "a\\b", "%2e%2e/escape",
                   "%252e%252e/escape", "C:/escape", "a//b")
        with mock.patch.dict(os.environ, {"SHAREPOINT_SITE": "tenant.sharepoint.com:/sites/test", "SHAREPOINT_DRIVE": ""}):
            for folder in invalid:
                with self.subTest(folder=folder), self.assertRaises(module.ConnectorError):
                    module.tool_call(client, "files_list", {"folder": folder})
            client.get.assert_not_called()
            for key in ("SHAREPOINT_SITE", "SHAREPOINT_DRIVE"):
                for value in ("..", "a/b", "a\\b", "%2e%2e", "/absolute"):
                    with mock.patch.dict(os.environ, {key: value}), self.assertRaises(module.ConnectorError):
                        module.drive_base(client)
            client.get.assert_not_called()
        for tool, arguments in (("file_versions", {"id": "../escape"}), ("files_search", {"query": ""}),
                                ("mail_read", {"id": "%2fescape"})):
            with self.assertRaises(module.ConnectorError):
                module.tool_call(client, tool, arguments)
        client.get.assert_not_called()
        with mock.patch.dict(os.environ, {"SHAREPOINT_DRIVE": "drive-test"}):
            client.get.return_value = {"value": []}
            module.tool_call(client, "files_list", {"folder": "reports/year"})
            self.assertEqual(client.get.call_args.args[0], "/drives/drive-test/root:/reports/year:/children")

    def test_malformed_provider_data_rejected(self):
        module = self.load_module()
        for data in ({"value": [None]}, {"value": [{}], "@odata.nextLink": 2}, {"value": "wrong"}):
            with self.assertRaises(module.ConnectorError):
                module.items(data, 1)
        client = mock.Mock()
        with mock.patch.dict(os.environ, {"MAIL_DEFAULT_USER": "user@example.com"}):
            for body in (None, "bad", {"content": 7}, {"content": "x", "contentType": "unknown"}):
                client.get.return_value = {"body": body}
                with self.assertRaisesRegex(module.ConnectorError, "body is invalid"):
                    module.tool_call(client, "mail_read", {"id": "m1"})
        with self.assertRaises(ValueError):
            module.parse_json('{"value": NaN}')
        with mock.patch.object(module.json, "loads", side_effect=RecursionError), self.assertRaisesRegex(ValueError, "nesting"):
            module.parse_json("[]")
        with self.assertRaises(module.ConnectorError):
            module.guard_url("https://[invalid/metadata")

    def test_fixture_read_is_bounded_and_schema_checked(self):
        module = self.load_module()
        env = {"GRAPH_FAKE_RESPONSES": "synthetic.json", "GRAPH_FAKE_TRANSPORT_TEST_ACK": module.ACK}
        for raw in (b"x" * (module.MAX_RESPONSE_BYTES + 1), b'{"responses":[{}]}', b'{"responses":null}'):
            handle = mock.mock_open(read_data=raw)
            with mock.patch.dict(os.environ, env), mock.patch("builtins.open", handle), self.assertRaises(module.ConnectorError):
                module.GraphClient()
            handle().read.assert_called_once_with(module.MAX_RESPONSE_BYTES + 1)

    def test_multibyte_request_limit(self):
        module = self.load_module()
        stdout = io.StringIO()
        with mock.patch.object(module, "GraphClient", return_value=object()), \
             mock.patch.object(sys, "stdin", io.TextIOWrapper(io.BytesIO(("é" * 500001 + "\n").encode()), encoding="utf-8")), \
             mock.patch.object(sys, "stdout", stdout), mock.patch.object(sys, "argv", [str(SCRIPT)]):
            self.assertEqual(module.main(), 0)
        self.assertIn("size limit", stdout.getvalue())

    def test_invalid_utf8_stdin_continues(self):
        env = os.environ.copy()
        env.pop("PYTHONIOENCODING", None)
        env.update({"GRAPH_FAKE_RESPONSES": str(FIXTURE),
                    "GRAPH_FAKE_TRANSPORT_TEST_ACK": "zero-width-lab-test-suite-only",
                    "PYTHONUTF8": "0", "PYTHONCOERCECLOCALE": "0"})
        good = (json.dumps(request("é", "tools/list"), ensure_ascii=False) + "\n").encode("utf-8")
        for locale in ("en_US.UTF-8", "C"):
            env.update({"LC_ALL": locale, "LC_CTYPE": locale})
            for bad in (b"\xff\xfe\n", b'{"id":"\xff"}\n'):
                with self.subTest(locale=locale, bad=bad):
                    result = subprocess.run([sys.executable, str(SCRIPT)], input=bad + good,
                                            capture_output=True, env=env, cwd=ROOT, timeout=10, check=False)
                    self.assertEqual(result.returncode, 0, result.stderr)
                    result.stdout.decode("ascii")
                    self.assertNotIn(b"Traceback", result.stderr)
                    responses = [json.loads(line) for line in result.stdout.splitlines()]
                    self.assertEqual(responses[0]["error"]["code"], -32700)
                    self.assertEqual(responses[1]["id"], "é")

    def test_percent_search_url_and_path_guards(self):
        module = self.load_module()
        with mock.patch.dict(os.environ, {"SHAREPOINT_DRIVE": "drive-test",
                             "GRAPH_FAKE_RESPONSES": "", "GRAPH_FAKE_TRANSPORT_TEST_ACK": ""}):
            client = module.GraphClient()
            client.token = "synthetic-token"
            response = mock.MagicMock()
            response.status = 200
            response.read.return_value = b'{"value":[]}'
            response.__enter__.return_value = response
            with mock.patch.object(module.OPENER, "open", return_value=response) as opened:
                module.tool_call(client, "files_search", {"query": "100% plan"})
                self.assertEqual(opened.call_args.args[0].full_url,
                                 "https://graph.microsoft.com/v1.0/drives/drive-test/root/search(q='100%25%20plan')?%24top=10")
        for path in ("/a/../x", "/a/./x", "/a\\x"):
            with self.subTest(path=path), self.assertRaises(module.ConnectorError):
                module.guard_url(module.ROOT + path)
        for identifier in ("../x", "%2e%2e", "a\\x"):
            with self.assertRaises(module.ConnectorError):
                module.segment(identifier)
        client = mock.Mock()
        client.get.return_value = {"id": "../x"}
        with mock.patch.dict(os.environ, {"SHAREPOINT_DRIVE": "", "SHAREPOINT_SITE": "site-test"}), \
             self.assertRaises(module.ConnectorError):
            module.drive_base(client)

    def test_invalid_success_json_has_precise_error(self):
        module = self.load_module()
        with mock.patch.dict(os.environ, {"GRAPH_FAKE_RESPONSES": "", "GRAPH_FAKE_TRANSPORT_TEST_ACK": ""}):
            client = module.GraphClient()
        response = mock.MagicMock()
        response.status = 200
        response.read.return_value = b"<html>invalid</html>"
        response.__enter__.return_value = response
        with mock.patch.object(module.OPENER, "open", return_value=response), \
             self.assertRaisesRegex(module.ConnectorError, "Graph returned invalid JSON"):
            client.request("GET", module.ROOT + "/me")

    def test_help_signature(self) -> None:
        result = subprocess.run([sys.executable, str(SCRIPT), "--help"], text=True,
                                capture_output=True, timeout=10, check=False)
        self.assertEqual(result.returncode, 0)
        self.assertTrue(result.stdout.strip().endswith("By Zero Width Lab (zerowidthlab.com)"))

    def test_oversized_input_is_rejected_and_server_continues(self) -> None:
        env = os.environ.copy()
        env["GRAPH_FAKE_RESPONSES"] = str(FIXTURE)
        env["GRAPH_FAKE_TRANSPORT_TEST_ACK"] = "zero-width-lab-test-suite-only"
        oversized = "x" * 1_000_001 + "\n"
        result = subprocess.run([sys.executable, str(SCRIPT)], input=oversized +
                                json.dumps(request(2, "tools/list")) + "\n", text=True,
                                capture_output=True, env=env, cwd=ROOT, timeout=10, check=False)
        responses = [json.loads(line) for line in result.stdout.splitlines()]
        self.assertEqual(result.returncode, 0)
        self.assertEqual(responses[0]["error"]["code"], -32600)
        self.assertEqual(len(responses[1]["result"]["tools"]), 7)


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

License: MIT (read it).

Back to the listing