tests/test_gmail_mail.py — Approval-Gated Mail

# By Zero Width Lab (zerowidthlab.com)

from __future__ import annotations

import base64
import argparse
import contextlib
import hashlib
import importlib.util
import io
import http.client
import socket
import ssl
import json
import os
import subprocess
import sys
import tempfile
import types
import unittest
import urllib.parse
import urllib.error
import urllib.request
from email import policy, errors
from email.parser import BytesParser
from pathlib import Path
from unittest import mock

__author__ = "Zero Width Lab"

SYNTHETIC_TOKEN = "synthetic"
PROFILE = {"emailAddress": "user@example.com"}

def load_gmail_module(filename="gmail_backend.py") -> object:
 path = Path(__file__).resolve().parent.parent / "scripts" / filename
 spec = importlib.util.spec_from_file_location("gmail_backend_test_module", path)
 assert spec is not None and spec.loader is not None
 module = importlib.util.module_from_spec(spec)
 spec.loader.exec_module(module)
 return module

class FakeConnection:
 def __init__(self, target: str | Exception) -> None:
  self.target, self.response, self.closed = target, b"", False
 def __enter__(self):
  return self
 def __exit__(self, *args):
  self.closed = True
 def settimeout(self, value):
  pass
 def recv(self, size):
  if isinstance(self.target, Exception):
   raise self.target
  return ("GET " + self.target + " HTTP/1.1\r\n").encode()
 def sendall(self, data):
  self.response = data
  if getattr(self, "reply_error", False):
   raise OSError("reply closed")

def fake_listener() -> mock.MagicMock:
 listener = mock.MagicMock()
 listener.__enter__.return_value = listener
 listener.getsockname.return_value = ("127.0.0.1", 43210)
 return listener

def fake_response(body: bytes) -> mock.MagicMock:
 result = mock.MagicMock()
 result.__enter__.return_value = result
 result.read.return_value = body
 return result

def http_error(url, status, error, headers=None):
 return urllib.error.HTTPError(url, status, "synthetic", headers or {},
 io.BytesIO(json.dumps({"error": error}).encode()))

def make_header(name: str, value: str) -> dict:
 return {"name": name, "value": value}

def make_response(*, contains=None, url=None, method="GET", status=200, data=None, headers=None) -> dict:
 entry = {"method": method, "status": status}
 for key, value in (("url_contains", contains), ("url", url), ("json", data), ("headers", headers)):
  if value is not None:
   entry[key] = value
 return entry

def write_args(command="send", sender="user@example.com", recipient="recipient@example.com",
 subject="Subject", body="Body") -> tuple[str, ...]:
 return (command, "--from", sender, "--to", recipient, "--subject", subject, "--body", body)

def write_responses(path: Path, responses: list) -> None:
 path.write_text(json.dumps({"responses": responses}), encoding="utf-8")

@contextlib.contextmanager
def fixture_file(responses: list):
 with tempfile.TemporaryDirectory() as directory:
  path = Path(directory) / "responses.json"
  write_responses(path, responses)
  yield path

class GmailTestCase(unittest.TestCase):
 def setUp(self) -> None:
  self.module = load_gmail_module()

 @contextlib.contextmanager
 def keychain(self, runner):
  with mock.patch.object(self.module.sys, "platform", "darwin"), mock.patch.dict(os.environ, {
 "GMAIL_TOKEN_STORE": "keychain", "GMAIL_FAKE_SECURITY_RUNNER": "1",
 "GMAIL_ALLOW_FAKE_SECURITY": "1"}):
   yield self.module.GmailTokenStore(ValueError, runner=runner)

 @contextlib.contextmanager
 def token_file(self):
  with tempfile.TemporaryDirectory() as directory:
   path = Path(directory) / "token.json"
   with mock.patch.dict(os.environ, {
 "GMAIL_TOKEN_STORE": "file", "GMAIL_TOKEN_FILE": str(path)}):
    yield self.module.GmailTokenStore(ValueError), path

 def transport(self, failures):
  backend = self.backend()
  backend.access_token = SYNTHETIC_TOKEN
  backend.opener = types.SimpleNamespace(open=mock.Mock(side_effect=failures))
  return backend

 def check_status(self, result, expected=0) -> None:
  self.assertEqual(result.returncode, expected, result.stderr)

 def backend(self):
  with mock.patch.dict(os.environ, {}, clear=True):
   return self.module.GmailBackend(types.SimpleNamespace(CliError=ValueError, display_text=str), {})

 def callback(self):
  return '/callback?' + urllib.parse.urlencode({'state': 'state', 'code': 'code', 'scope': ' '.join(self.module.SCOPES)})

 @contextlib.contextmanager
 def oauth(self, entries, clock=None):
  m, b, listener = self.module, self.backend(), fake_listener()
  listener.accept.side_effect = [x if isinstance(x, Exception) else (x, ('127.0.0.1', 1)) for x in entries]
  b._client = lambda: {'client_id': 'synthetic'}
  b.store = types.SimpleNamespace(write=mock.Mock())
  b.request = mock.Mock(return_value={'access_token': 'a', 'refresh_token': 'r'})
  with mock.patch.object(m.socket, 'socket', return_value=listener), \
   mock.patch.object(m.secrets, 'token_urlsafe', return_value='state'), \
   mock.patch.object(m.time, 'monotonic', side_effect=clock, return_value=0), \
   mock.patch.object(m.time, 'sleep') as sleep, contextlib.redirect_stdout(io.StringIO()) as output:
   yield b, listener, sleep, output

class GmailPrimitiveTests(GmailTestCase):
 def request_failure(self, method, url, expected, failure, stage="open", **options):
  backend = self.backend()
  backend.access_token = SYNTHETIC_TOKEN
  backend.approved_account = PROFILE["emailAddress"]
  response = fake_response(b"not JSON")
  if stage == "read":
   response.read.side_effect = failure
  backend.opener = types.SimpleNamespace(open=mock.Mock(
 side_effect=[fake_response(json.dumps(PROFILE).encode()), failure if stage == "open" else response]
   if method != "GET" and options.get("authenticated", True) else
   failure if stage == "open" else None, return_value=response))
  with self.assertRaisesRegex(ValueError, expected) as caught:
   backend.request(method, url, **options)
  if "/drafts" in url and "outcome unknown" in expected:
   self.assertIn("Drafts folder", str(caught.exception))
  self.assertEqual(backend.opener.open.call_count, 2 if method != "GET" and options.get("authenticated", True) else 1)

 def test_final_guards(self) -> None:
  m, b = self.module, self.backend()
  self.assertNotIn('webbrowser', Path(m.__file__).read_text())
  self.assertEqual(m.decoded_header('=?utf-8?b?A?='), '=?utf-8?b?A?=')
  with mock.patch.object(m, 'decode_header', side_effect=RecursionError):
   self.assertEqual(m.decoded_header('raw'), 'raw')
  with mock.patch.object(m, 'getaddresses', side_effect=RecursionError), self.assertRaises(ValueError):
   b._record({}, False)
  for status in (44, 1):
   runner = mock.Mock(return_value=subprocess.CompletedProcess([], status, b'', b'private'))
   with self.keychain(runner) as store, self.subTest(status=status):
    if status == 44:
     self.assertIsNone(store.read())
     with self.assertRaises(ValueError):
      store.delete()
    else:
     with self.assertRaises(ValueError):
      store.read()
  with self.keychain(mock.Mock()) as store, self.assertRaises(ValueError):
   store._identity('-option')
  b.store = types.SimpleNamespace(read=lambda: None, delete=mock.Mock())
  self.assertEqual(b.auth_revoke(), 'no local Gmail token')
  b.store = types.SimpleNamespace(read=lambda: {'refresh_token': 'synthetic'}, delete=mock.Mock(side_effect=ValueError('delete failed')))
  b.request = mock.Mock(return_value=None)
  with self.assertRaisesRegex(ValueError, 'delete failed'):
   b.auth_revoke()

 def test_per_message_failures_are_skipped(self) -> None:
  shared = load_gmail_module('m365_mail.py')
  b = self.module.GmailBackend(shared, {})
  for failure in (shared.CliError, errors.MessageError, RecursionError, ValueError, TypeError, KeyError, AttributeError):
   for thread in (False, True):
    with self.subTest(failure=failure, thread=thread):
     b.skipped = []
     items = [None, {}, {'id': 'bad'}, {'id': 'good'}]
     b._collection = mock.Mock(return_value=items)
     b.request = mock.Mock(return_value={'messages': items})
     b.get_message = mock.Mock(side_effect=[failure('bad\nreason'), {'id': 'good'}])
     if thread:
      b._record = mock.Mock(side_effect=[failure('bad\nreason'), {'id': 'good', 'receivedDateTime': ''}])
     result = b.thread_messages('', 'thread') if thread else b.list_messages('', 4)
     self.assertEqual([x['id'] for x in result], ['good'])
     self.assertEqual(len(b.skipped), 3)
     self.assertNotIn('\n', ''.join(b.skipped))
     if failure is not KeyError:
      self.assertTrue(any('\\u000a' in x for x in b.skipped))

 def test_oauth_deadline_and_reply_failure(self) -> None:
  for final in (False, True):
   bad, valid = FakeConnection('//[::1'), FakeConnection(self.callback())
   valid.reply_error = True
   entries = [OSError('accept'), FakeConnection('\ud800'), bad, valid if final else OSError('accept'), AssertionError('deadline removed')]
   with self.subTest(final=final), self.oauth(entries, [0, 0, 0, 0, 0, 181]) as (b, listener, sleep, output):
    if final:
     b.auth_login()
     b.store.write.assert_called_once()
    else:
     with self.assertRaisesRegex(ValueError, 'timed out'):
      b.auth_login()
    sleep.assert_called()
    url = output.getvalue().split('https://', 1)[1].strip()
    self.assertEqual(urllib.parse.parse_qs(urllib.parse.urlsplit('https://' + url).query)['code_challenge_method'], ['S256'])

 def test_real_transport_guard_and_error_reads(self) -> None:
  m = self.module
  with mock.patch.object(m.urllib.request, 'build_opener') as opener:
   self.backend()
  self.assertTrue(any(isinstance(x, m.RedirectGuard) for x in opener.call_args.args))
  b = self.backend()
  b.access_token = SYNTHETIC_TOKEN
  with mock.patch.object(m, 'checked_url', side_effect=ValueError('guard')), self.assertRaisesRegex(ValueError, 'guard'):
   b.request('GET', m.API + '/messages')
  for endpoint in ('/messages/send', '/drafts'):
   for failure in (OSError('read'), http.client.IncompleteRead(b'x'), RecursionError('deep')):
    error = urllib.error.HTTPError(m.API + endpoint, 503, 'synthetic', {}, None)
    error.read = mock.Mock(side_effect=failure)
    with self.subTest(endpoint=endpoint, failure=type(failure)):
     self.request_failure('POST', m.API + endpoint, 'outcome unknown', error)

 def test_account_swap_and_profile_failure_refuse_every_write(self):
  shared = load_gmail_module('m365_mail.py')
  self.assertEqual(set(self.module.SCOPES), {
   'https://www.googleapis.com/auth/gmail.readonly', 'https://www.googleapis.com/auth/gmail.compose'})
  for endpoint in ('/drafts', '/messages/send', '/messages/id'):
   for profile in ({'emailAddress': ' Other@example.com '}, {}, None, OSError('profile unavailable'), {'emailAddress': ' USER@EXAMPLE.COM '}):
    with self.subTest(endpoint=endpoint, profile=profile), self.token_file() as (store, path):
     store.write({'access_token': 'preview-token', 'expires_at': 9999999999})
     preview = self.module.GmailBackend(shared, {})
     preview.opener = types.SimpleNamespace(open=mock.Mock(return_value=fake_response(json.dumps(PROFILE).encode())))
     approved = preview.account_address()
     store.write({'access_token': 'execution-token', 'expires_at': 9999999999})
     execution = self.module.GmailBackend(shared, {})
     execution.approved_account = approved
     calls = []
     def transport(request, **options):
      calls.append(request)
      if request.full_url.endswith('/profile'):
       if isinstance(profile, Exception):
        raise profile
       return fake_response(json.dumps(profile).encode())
      return fake_response(b'{"id":"written"}')
     execution.opener = types.SimpleNamespace(open=transport)
     matches = isinstance(profile, dict) and profile.get('emailAddress', '').strip().casefold() == approved
     if matches:
      self.assertEqual(execution.request('POST', self.module.API + endpoint, json_body={})['id'], 'written')
     else:
      with self.assertRaises(shared.CliError):
       execution.request('POST', self.module.API + endpoint, json_body={})
     writes = [r for r in calls if r.get_method() != 'GET']
     self.assertEqual(len(writes), int(matches))
     self.assertTrue(all(r.get_header('Authorization') == 'Bearer execution-token' for r in calls))
     self.assertEqual(preview.opener.open.call_args.args[0].get_header('Authorization'), 'Bearer preview-token')

 def test_attachment_subtrees_are_not_message_body_and_listing_is_restricted(self):
  b = self.backend()
  encoded = lambda text: base64.urlsafe_b64encode(text.encode()).decode()
  for marker in ({'filename': 'secret.txt'}, {'body': {'attachmentId': 'secret'}},
   {'headers': [make_header('Content-Disposition', 'attachment; filename=secret.txt')]}):
   for mime in ('text/plain', 'text/html', 'multipart/mixed'):
    hidden = {'mimeType': mime, 'body': {'data': encoded('SECRET')}, 'parts': [
     {'mimeType': 'text/plain', 'body': {'data': encoded('SECRET')}}]}
    for key, value in marker.items():
     hidden[key] = hidden.get(key, {}) | value if key == 'body' else value
    payload = {'mimeType': 'multipart/mixed', 'parts': [hidden,
     {'mimeType': 'text/plain', 'body': {'data': encoded('Visible')}}]}
    b.request = mock.Mock(return_value={'payload': payload})
    with self.subTest(marker=marker, mime=mime):
     self.assertEqual(b.get_message('', 'id')['body'], 'Visible')
  b.request = mock.Mock(return_value={'payload': {}})
  b.attachments('id')
  query = urllib.parse.parse_qs(urllib.parse.urlsplit(b.request.call_args.args[1]).query)
  self.assertIn('fields', query)
  fields = query['fields'][0]
  self.assertEqual(query['format'], ['full'])
  self.assertNotIn('data', fields)
  self.assertNotIn('*', fields)
  for name in ('partId', 'filename', 'mimeType', 'size', 'attachmentId', 'parts'):
   self.assertIn(name, fields)

 def test_pkce_verifier_and_s256_challenge(self) -> None:
  module = self.module
  verifier, challenge = module.pkce_pair()
  self.assertGreaterEqual(len(verifier), 43)
  self.assertLessEqual(len(verifier), 128)
  self.assertRegex(verifier, r"^[A-Za-z0-9_~-]+$")
  expected = base64.urlsafe_b64encode(
 hashlib.sha256(verifier.encode("ascii")).digest()
 ).rstrip(b"=").decode("ascii")
  self.assertEqual(challenge, expected)

 def test_query_escaping_quotes_and_backslashes(self) -> None:
  module = self.module
  for unsafe in ('a"b', 'a\\b'):
   with self.assertRaises(ValueError):
    module.quoted_query(unsafe)

 def test_url_guards(self) -> None:
  m = self.module
  request = urllib.request.Request(m.API + "/messages")
  handler = m.RedirectGuard(ValueError)
  for url in ("http://gmail.googleapis.com/x",
 "https://example.invalid/leak",
 "https://user@example.com@gmail.googleapis.com/x"):
   with self.subTest(url=url):
    with self.assertRaises(ValueError):
     m.checked_url(url, ValueError)
    with self.assertRaises(ValueError):
     handler.redirect_request(request, None, 302, "Found", {}, url)

 def test_real_request_status_classification(self) -> None:
  m = self.module
  for endpoint in (m.API + "/messages/send", m.API + "/drafts", m.REVOKE):
   for status in (400, 429, 403, 503):
    expected = ("token revoke failed" if endpoint == m.REVOKE else
 "outcome unknown" if status == 503 else "refused, not sent")
    failure = http_error(endpoint, status, {
 "message": "synthetic reason", "errors": [{"reason": "rateLimitExceeded"}]})
    with self.subTest(endpoint=endpoint, status=status):
     self.request_failure("POST", endpoint, expected, failure,
 authenticated=endpoint != m.REVOKE, json_body={"raw": "x"})
  self.request_failure("POST", m.TOKEN, "^OAuth token request invalid_client$",
 http_error(m.API, 400, "invalid_client"), authenticated=False, form={"grant_type": "refresh_token"})
  self.request_failure("POST", m.REVOKE, "token revoke failed", ConnectionResetError("reset"), authenticated=False)
  backend = self.transport([
 http_error(m.API, 429, {"message": "rate limited"}, {"Retry-After": "0"}),
 fake_response(b'{"messages":[]}')])
  with mock.patch.object(m.time, "sleep") as sleep:
   self.assertEqual(backend.request("GET", m.API + "/messages"), {"messages": []})
  self.assertEqual(backend.opener.open.call_count, 2)
  sleep.assert_called_once_with(0)

 def test_keychain_runner_keeps_secret_only_on_stdin(self) -> None:
  calls: list[tuple[list[str], dict[str, object]]] = []
  secret = "synthetic-refresh-token-marker" * 75

  def fake_runner(argv: list[str], **kwargs: object) -> subprocess.CompletedProcess[bytes]:
   calls.append((argv, kwargs))
   if argv[1] == "find-generic-password":
    output = json.dumps({"refresh_token": secret}, separators=(",", ":")).encode("utf-8")
   else:
    output = b""
   return subprocess.CompletedProcess(argv, 0, stdout=output, stderr=b"")

  with self.keychain(fake_runner) as store:
   store.write({"refresh_token": secret}, "user@example.com")
   self.assertEqual(store.read("user@example.com")["refresh_token"], secret)
   store.delete("user@example.com")

  identity = ["-a", "user@example.com", "-s", "approval-gated-mail-gmail"]
  expected = [["/usr/bin/security", "-i"]]
  for operation in ("find", "find", "delete"):
   expected.append(["/usr/bin/security", operation + "-generic-password", *identity,
 *(["-w"] if operation == "find" else [])])
  self.assertEqual([call[0] for call in calls], expected)
  self.assertNotIn(secret, str([call[0] for call in calls]))
  token_json = json.dumps({"refresh_token": secret}, separators=(",", ":"))
  expected_line = ("add-generic-password -a user@example.com "
 "-s approval-gated-mail-gmail -U -X "
 + token_json.encode("utf-8").hex() + "\n")
  self.assertEqual(calls[0][1]["input"], expected_line.encode("ascii"))
  for _, options in calls:
   self.assertEqual(options["timeout"], 10)
   self.assertTrue(options["capture_output"])
   self.assertFalse(options["check"])

 def test_keychain_refuses_unsafe_account_before_runner(self) -> None:
  runner = mock.Mock(side_effect=AssertionError("runner must not execute"))
  with self.keychain(runner) as store:
   with self.assertRaises(ValueError):
    store.write({"refresh_token": "synthetic"}, "bad account; command")
  runner.assert_not_called()

 def test_fake_security_runner_needs_both_gates(self) -> None:
  module = self.module
  with mock.patch.dict(os.environ, {"GMAIL_FAKE_SECURITY_RUNNER": "1",
 "GMAIL_ALLOW_FAKE_SECURITY": "0"}, clear=False):
   with self.assertRaisesRegex(ValueError, "GMAIL_ALLOW_FAKE_SECURITY"):
    module.GmailTokenStore(ValueError, runner=lambda *args, **kwargs: None)

 @unittest.skipUnless(os.name == "posix", "requires POSIX file permissions")
 def test_token_file_is_private_and_refuses_loose_mode_or_symlink(self) -> None:
  with self.token_file() as (store, path):
   for token in ("synthetic", "replacement"):
    store.write({"refresh_token": token})
    self.assertEqual(path.stat().st_mode & 0o777, 0o600)
    if token == "synthetic":
     self.assertEqual(store.read()["refresh_token"], token)
   path.chmod(0o644)
   with self.assertRaises(ValueError):
    store.read()
   with self.assertRaises(ValueError):
    store.write({"refresh_token": "blocked"})
   path.chmod(0o600)
   path.unlink()
   target = path.with_name("target.json")
   target.write_text("{}", encoding="utf-8")
   target.chmod(0o600)
   path.symlink_to(target)
   with mock.patch.object(self.module.os, "name", "nt"):
    with self.assertRaisesRegex(ValueError, "symlink"):
     store.read()
    with self.assertRaisesRegex(ValueError, "symlink"):
     store.write({"refresh_token": "blocked"})

 def test_invalid_grant_explains_testing_expiration(self) -> None:
  module = self.module
  with fixture_file(
 [make_response(method="POST", url="https://oauth2.googleapis.com/token", status=400, data={"error": "invalid_grant"})]
 ) as fixture:
   loader = types.SimpleNamespace(
 _load_fake_responses=lambda path: json.loads(path.read_text())["responses"])
   shared = types.SimpleNamespace(CliError=ValueError, GraphClient=loader)
   with mock.patch.dict(os.environ, {"GMAIL_FAKE_TRANSPORT": str(fixture),
 "GMAIL_ALLOW_FAKE": "1"}, clear=False):
    backend = module.GmailBackend(shared, {})
    with self.assertRaisesRegex(ValueError, "auth login again.*7 days"):
     backend.request("POST", module.TOKEN, form={"grant_type": "refresh_token"},
 authenticated=False)

 def test_auth_revoke_deletes_local_token_when_remote_revoke_fails(self) -> None:
  backend = self.backend()
  backend.store = types.SimpleNamespace(
 read=mock.Mock(return_value={"refresh_token": "synthetic-token"}),
 delete=mock.Mock(),
 )
  backend.request = mock.Mock(side_effect=ValueError("synthetic revoke failure"))
  self.assertIn("revoke", backend.auth_revoke().lower())
  backend.store.delete.assert_called_once()

 def test_real_request_connection_and_decode_failures(self) -> None:
  failures = (TimeoutError("timeout"), urllib.error.URLError("connection"),
 ConnectionResetError("reset"), ssl.SSLError("TLS"),
 http.client.RemoteDisconnected("closed"), http.client.IncompleteRead(b"x"), RecursionError("deep"))
  for method, endpoint in (("GET", "/messages"), ("POST", "/messages/send"),
 ("POST", "/drafts")):
   expected = "read request failed" if method == "GET" else "outcome unknown.*verify-sent"
   for stage in ("open", "read"):
    for failure in failures:
     with self.subTest(method=method, endpoint=endpoint, stage=stage, failure=type(failure)):
      self.request_failure(method, self.module.API + endpoint, expected, failure, stage)
   with self.subTest(method=method, decode=True):
    self.request_failure(method, self.module.API + endpoint, expected, None, "decode")

 def test_retry_after_is_capped_and_invalid_values_use_fallback(self) -> None:
  module = self.module
  for value, expected in (("31", 30), ("999999", 30), ("0", 0),
 ("Wed, 01 Jan 2020", 2), ("-1", 2), (None, 2)):
   with self.subTest(value=value):
    self.assertEqual(module.retry_after_seconds(value, 2), expected)

 def test_windows_file_store_skips_modes_and_fchmod(self) -> None:
  with self.token_file() as (store, path):
   path.write_text('{"refresh_token":"synthetic"}', encoding="utf-8")
   path.chmod(0o644)
   with mock.patch.object(self.module.os, "name", "nt"), mock.patch.object(
 self.module.os, "fchmod", side_effect=AssertionError("POSIX mode check on Windows")):
    for token in ("synthetic", "replacement"):
     if token == "replacement":
      store.write({"refresh_token": token})
     self.assertEqual(store.read()["refresh_token"], token)

 def test_mime_depth_cap(self) -> None:
  backend = self.backend()
  payload = {}
  for _ in range(50):
   payload = {"parts": [payload]}
  self.assertEqual(len(backend._parts(payload)), 51)
  with self.assertRaisesRegex(ValueError, "depth exceeds 50"):
   backend._parts({"parts": [payload]})

 def test_keychain_write_requires_exact_readback(self) -> None:
  for output, status in ((b'{"refresh_token":"different"}', 0), (b"", 1)):
   runner = mock.Mock(side_effect=[subprocess.CompletedProcess([], 0, b"", b""),
 subprocess.CompletedProcess([], status, output, b"")])
   with self.keychain(runner) as store:
    with self.subTest(status=status), self.assertRaisesRegex(ValueError, "token was not stored in the Keychain"):
     store.write({"refresh_token": "synthetic"})
   self.assertEqual(runner.call_count, 2)

 def test_oauth_callback_validation(self) -> None:
  good = self.callback()

  for final, expected in ((good, None),
 (good.replace(urllib.parse.quote(self.module.SCOPES[1], safe=""), ""), "missing scope"),
 ("/callback?state=state&error=access_denied", "authorization denied: access_denied")):
   connections = [FakeConnection("/callback?state=wrong"), FakeConnection(socket.timeout("idle")), FakeConnection(OSError("closed")),
 FakeConnection(good.replace("/callback?", "/wrong?")), FakeConnection(final)]
   with self.subTest(expected=expected), self.oauth(connections) as (backend, listener, sleep, output):
    if expected:
     with self.assertRaisesRegex(ValueError, expected):
      backend.auth_login()
     backend.request.assert_not_called()
     backend.store.write.assert_not_called()
    else:
     backend.auth_login()
     backend.request.assert_called_once()
     backend.store.write.assert_called_once()
   listener.bind.assert_called_once_with(("127.0.0.1", 0))
   self.assertTrue(all(item.closed for item in connections))
   self.assertIn(b"404 Not Found", connections[-2].response)
   header, body = connections[-1].response.split(b"\r\n\r\n", 1)
   self.assertIn(b"Content-Length: " + str(len(body)).encode(), header)
class GmailCliTests(GmailTestCase):
 def test_live_post_failures_replay_and_reply_draft(self) -> None:
  s, m = self.shared, self.module
  b = m.GmailBackend(s, {})
  for kind in ('send', 'draft'):
   args = self.write_namespace()
   request = b.build_request(kind, args.from_user, args.to, [], [], args.subject, args.body, message_id=args.message_id)
   request["accountAddress"] = PROFILE["emailAddress"]
   b.account_address = mock.Mock(return_value=PROFILE["emailAddress"])
   args.approve = s.approval_digest(request)
   for failure in (RuntimeError('failed'), RecursionError('deep'), KeyboardInterrupt()):
    b.refuse_replay = mock.Mock()
    b.request = mock.Mock(side_effect=failure)
    with self.subTest(kind=kind, failure=type(failure)), self.assertRaises((s.CliError, SystemExit)) as caught, \
     contextlib.redirect_stderr(io.StringIO()) as error, self.gmail_import():
     s.command_gmail_write(args, b, {}, kind)
    text = error.getvalue() + str(caught.exception)
    self.assertIn('outcome unknown', text)
    self.assertIn(args.message_id, text)
    if kind == 'draft':
     self.assertIn('Drafts folder', text)
    if isinstance(failure, KeyboardInterrupt):
     self.assertEqual(caught.exception.code, 130)
   b.access_token = SYNTHETIC_TOKEN
   b.account_address = types.MethodType(m.GmailBackend.account_address, b)
   for failure in (OSError('read'), http.client.IncompleteRead(b'x'), RecursionError('deep')):
    error = urllib.error.HTTPError(request['url'], 503, 'synthetic', {}, None)
    error.read = mock.Mock(side_effect=failure)
    b.opener = types.SimpleNamespace(open=mock.Mock(side_effect=[
     fake_response(json.dumps(PROFILE).encode()), fake_response(json.dumps(PROFILE).encode()), error]))
    b.request = types.MethodType(m.GmailBackend.request, b)
    with self.subTest(kind=kind, read=type(failure)), self.gmail_import(), \
     self.assertRaisesRegex(s.CliError, 'outcome unknown.*<id@example.com>') as caught:
     s.command_gmail_write(args, b, {}, kind)
    if kind == 'draft':
     self.assertIn('Drafts folder', str(caught.exception))
   if kind == 'send':
    b.account_address = mock.Mock(return_value=PROFILE['emailAddress'])
    b.request = mock.Mock(return_value={'id': None})
    with self.assertRaisesRegex(s.CliError, 'send outcome unknown.*<id@example.com>'), self.gmail_import():
     s.command_gmail_write(args, b, {}, kind)
   b = m.GmailBackend(s, {})
   b.account_address = mock.Mock(return_value=PROFILE['emailAddress'])
   b.request = mock.Mock(return_value={'messages': [{'id': 'exists'}]})
   with self.assertRaisesRegex(s.CliError, 'already exists; run verify-sent'), self.gmail_import():
    s.command_gmail_write(args, b, {}, kind)
   b.request.assert_called_once_with('GET', m.API + '/messages?' + urllib.parse.urlencode(
 {'q': 'rfc822msgid:' + args.message_id, 'includeSpamTrash': 'true'}))
  reply = {'message_id': '<reply@[127.0.0.1]>', 'thread_id': 'thread', 'subject': 'Subject', 'references': ''}
  request = b.build_request('draft', SENDER, [TARGET], [], [], 'Subject', 'Body', reply)
  self.assertEqual(request['jsonBody']['message']['threadId'], 'thread')
  self.assertNotIn('threadId', request['jsonBody'])
  b.request = mock.Mock(return_value={'messages': [{'id': 'new', 'internalDate': '2', 'payload': {}},
 {'id': 'old', 'internalDate': '1', 'payload': {'filename': 'attachment'}}]})
  thread = b.thread_messages('', 'thread')
  self.assertEqual([x['id'] for x in thread], ['old', 'new'])
  self.assertEqual([x['hasAttachments'] for x in thread], [True, False])

 def test_raw_encoded_addr_and_cli_help(self) -> None:
  s, m = self.shared, self.module
  b = m.GmailBackend(s, {})
  raw = '=?utf-8?q?ceo=40example.com?='
  b.request = mock.Mock(return_value={'id': 'id', 'labelIds': ['SENT'], 'payload': {'headers': [
 make_header('From', '<' + raw + '>'), make_header('To', SENDER),
 make_header('Subject', 'Subject'), make_header('Message-ID', MESSAGE_ID)]}})
  record = b.get_message('', 'id')
  self.assertEqual(record['from']['address'], raw)
  b.sent_items = lambda count: [record]
  args = argparse.Namespace(count=1, rfc822msgid=MESSAGE_ID, subject='Subject', to=['A <A@example.com>'], cc=None, bcc=None, json=True)
  with contextlib.redirect_stdout(io.StringIO()) as output:
   s.command_verify_sent(args, b, {})
  self.assertEqual(json.loads(output.getvalue())['from']['address'], raw)
  parser = s.build_parser()
  def check(parser):
   self.assertTrue(parser.format_help().rstrip().endswith(s.SIGNATURE))
   if parser.prog.split()[-1] in ('send', 'draft'):
    options = {flag: action.help for action in parser._actions for flag in action.option_strings}
    self.assertIn('Message-ID', options['--message-id'])
    self.assertIn('metadata', options['--dry-run'].lower())
   for action in parser._actions:
    if isinstance(action, argparse._SubParsersAction):
     for child in action.choices.values():
      check(child)
  check(parser)
  for command in ('send', 'draft'):
   args = parser.parse_args([command, *write_args()[1:], '--message-id', MESSAGE_ID, '--dry-run'])
   with self.assertRaisesRegex(s.CliError, '--message-id applies to the Gmail provider only'):
    getattr(s, 'command_' + command)(args, types.SimpleNamespace(is_gmail=False), {})

 def test_text_routes_escape_for_both_providers(self) -> None:
  for provider in ('graph', 'gmail'):
   for char, escaped in (('\x1b', r'\u001b'), ('\n', r'\u000a'), ('\u202e', r'\u202e'), ('\\', r'\u005c')):
    marker = 'left' + char + 'right'
    s, m = self.shared, self.module
    self.assertEqual(s.display_address({'name': '', 'address': marker}), s.display_text(marker))
    b = m.GmailBackend(s, {}) if provider == 'gmail' else types.SimpleNamespace(is_gmail=False)
    record = {'id': marker, 'subject': marker, 'from': {'name': marker, 'address': SENDER},
 'to': [], 'body': 'body', 'receivedDateTime': '', 'conversationId': marker}
    if provider == 'gmail':
     b.list_messages = mock.Mock(return_value=[record])
     b.get_message = mock.Mock(return_value=record)
    args = argparse.Namespace(json=False, user=SENDER, count=1, allow_partial=False,
 message_id='id', max_chars=100, full=True)
    for route in ('recent', 'read', 'save'):
     with self.subTest(provider=provider, char=char, route=route), \
      mock.patch.object(s, 'list_messages', return_value=[record]), \
      mock.patch.object(s, 'get_message', return_value=record), contextlib.redirect_stdout(io.StringIO()) as output:
      if route == 'save':
       with tempfile.TemporaryDirectory() as directory:
        args.output_path = str(Path(directory) / (marker + '.md'))
        s.command_save(args, b, {})
        text = Path(args.output_path).read_text() + output.getvalue()
      else:
       getattr(s, 'command_' + route)(args, b, {})
       text = output.getvalue()
      self.assertIn('left' + escaped + 'right', text)
      self.assertNotIn(marker, text)
    with self.subTest(provider=provider, char=char, route='preview'):
     result = self.run_cli(*write_args(subject=marker), '--dry-run', provider=provider,
 fixture=self.empty_fixture, json_output=False)
     if provider == 'gmail' and char in ('\x1b', '\n'):
      self.check_status(result, 1)
     else:
      self.check_status(result)
      expected = {'\n': r'\n', '\\': r'\\'}.get(char, escaped) if provider == 'graph' else escaped
      self.assertIn('left' + expected + 'right', result.stdout)

 def setUp(self) -> None:
  super().setUp()
  self.shared = load_gmail_module("m365_mail.py")
  self.root = Path(__file__).resolve().parent.parent
  self.script = self.root / "scripts" / "m365_mail.py"
  self.fixture = self.root / "tests" / "fixtures" / "gmail-responses.json"
  self.empty_fixture = self.root / "tests" / "fixtures" / "empty-responses.json"

 def run_cli(self, *args: str, fixture: Path | None = None,
 env_extra: dict[str, str] | None = None,
 json_output=True, provider="gmail", allow_fake=True, expected=None) -> subprocess.CompletedProcess[str]:
  env = os.environ.copy()
  for key in ("M365_TENANT_ID", "M365_CLIENT_ID", "M365_CLIENT_SECRET",
 "M365_MAIL_FAKE_TRANSPORT", "M365_MAIL_ALLOW_FAKE",
 "GMAIL_CLIENT_FILE", "GMAIL_TOKEN_FILE", "MAIL_PROVIDER"):
   env.pop(key, None)
  selected = fixture or self.fixture
  responses = json.loads(selected.read_text())["responses"]
  if not any("/profile" in str(entry) for entry in responses):
   responses = [make_response(contains="/profile", data=PROFILE)] * 2 + responses
  fixture_context = fixture_file(responses)
  selected = fixture_context.__enter__()
  env["GMAIL_FAKE_TRANSPORT"] = str(selected)
  env.pop("GMAIL_ALLOW_FAKE", None)
  if allow_fake:
   env["GMAIL_ALLOW_FAKE"] = "1"
  if provider == "graph":
   env.update({"M365_MAIL_FAKE_TRANSPORT": str(fixture or self.empty_fixture), "M365_MAIL_ALLOW_FAKE": "1"})
  if env_extra:
   env.update(env_extra)
  result = subprocess.run(
 [sys.executable, str(self.script), *(["--provider", provider] if provider else []),
 *(["--json"] if json_output else []), "--user", USER, *args],
 cwd=self.root, capture_output=True, text=True, check=False, env=env,
 timeout=15,
 )
  fixture_context.__exit__(None, None, None)
  if expected is not None:
   self.check_status(result, expected)
  return result

 def run_text_cli(self, *args: str, fixture: Path | None = None) -> subprocess.CompletedProcess[str]:
  return self.run_cli(*args, fixture=fixture, json_output=False)

 def live_cli(self, args, preview, *, digest=None, fixture=None, expected=None, responses=None):
  if responses is None:
   responses = json.loads((fixture or self.fixture).read_text())["responses"]
  with fixture_file([make_response(contains="includeSpamTrash=true", data={"messages": []}), *responses]) as checked:
   return self.run_cli(*args, "--message-id", str(preview["messageId"]), "--approve",
 digest if digest is not None else str(preview["approvalDigest"]), fixture=checked,
 expected=expected)

 def preview(self, *args: str, fixture: Path | None = None) -> dict[str, object]:
  result = self.run_cli(*args, "--dry-run", fixture=fixture or self.empty_fixture, expected=0)
  data = json.loads(result.stdout)
  self.assertTrue(data["dryRun"])
  self.assertRegex(data["approvalDigest"], r"^[0-9a-f]{64}$")
  return data

 def gmail_import(self):
  return mock.patch.dict(sys.modules, {"gmail_backend": self.module})

 def write_namespace(self, **changes):
  values = dict(dry_run=False, approve=None, message_id=MESSAGE_ID, from_user=SENDER,
 to=[TARGET], cc=[], bcc=[], subject="Subject", body="Body", body_file=None,
 reply_to_message_id=None, json=True)
  return argparse.Namespace(**(values | changes))

 def assert_fields(self, actual, expected):
  for name, value in expected.items():
   found = actual
   for part in name if isinstance(name, tuple) else (name,):
    found = found[part]
   self.assertEqual(found, value)

 def assert_contains(self, text, values):
  for value in values:
   self.assertIn(value, text)

 def parse_wire(self, preview):
  raw = preview["jsonBody"]["raw"]
  return BytesParser(policy=policy.default).parsebytes(
 base64.urlsafe_b64decode(raw + "=" * (-len(raw) % 4)))

 def message_response(self, identifier, *, headers=None, payload=None, **fields):
  payload = dict(payload or {})
  if headers is not None:
   payload["headers"] = [make_header(name, value) for name, value in headers.items()]
  return make_response(contains="/messages/" + identifier,
 data={"id": identifier, "payload": payload, **fields})

 def sent_responses(self, identifier="sent-fixture", **fields):
  return [make_response(contains="labelIds=SENT", data={"messages": [{"id": identifier}]}),
 self.message_response(identifier, **fields)]

 def run_responses(self, *args, responses, **options):
  with fixture_file(responses) as fixture:
   return self.run_cli(*args, fixture=fixture, **options)

 def fixture_response(self, identifier):
  return next(item for item in json.loads(self.fixture.read_text())["responses"]
 if item.get("url_contains") == "/messages/" + identifier)

 def refused(self, *args, contains=(), excludes=(), **options):
  result = self.run_cli(*args, fixture=self.empty_fixture, **options, expected=1)
  self.assert_contains(result.stderr, contains)
  for value in excludes:
   self.assertNotIn(value, result.stderr)
  return result

 def test_read_retries_429_and_honors_retry_after(self) -> None:
  result = self.run_responses("read", "gmail-one", responses=[
 make_response(contains="/messages/gmail-one", status=429, headers={"Retry-After": "0"}),
 self.fixture_response("gmail-one")], expected=0)
  self.assertEqual(json.loads(result.stdout)["id"], "gmail-one")

 def test_partial_list_is_marked_and_requires_opt_in(self) -> None:
  with fixture_file(
 [make_response(contains="maxResults=1", data={"messages": [{"id": "missing-detail"}]})]
 ) as fixture:
   refused = self.run_cli("recent", "--count", "1", fixture=fixture)
   accepted = self.run_cli("recent", "--count", "1", "--allow-partial", fixture=fixture)
  self.check_status(refused, 2)
  self.assertTrue(json.loads(refused.stdout)["incomplete"])
  self.check_status(accepted)
  self.assertEqual(len(json.loads(accepted.stdout)["skipped"]), 1)
  reason = json.loads(accepted.stdout)["skipped"][0]
  self.assertIn("missing-detail", reason)
  self.assertIn("no matching", reason)

 def test_write_preview_cases(self) -> None:
  cases = (("send", "A subject", "A body", ("A body",)),
 ("send", "Café ☕", "Body", ()),
 ("send", "Subject", "first\nsecond\x1b\u202e", ("first\nsecond", "\x1b", "\u202e")),
 ("draft", "Draft subject", "Draft body", ()))
  for kind, subject, body, contains in cases:
   with self.subTest(kind=kind, subject=subject):
    args = write_args(command=kind, subject=subject, body=body)
    data = self.preview(*args)
    self.assertEqual(data["method"], "POST")
    self.assertEqual(data["accountAddress"], PROFILE["emailAddress"])
    self.assertEqual(data["preview"]["accountAddress"], PROFILE["emailAddress"])
    changed = dict(data, accountAddress="other@example.com")
    self.assertNotEqual(self.shared.approval_digest(changed), data["approvalDigest"])
    if kind == "draft":
     self.assertIn("/drafts", data["url"])
     refused = self.run_cli(*args, fixture=self.empty_fixture, expected=1)
     self.assertIn("--approve", refused.stderr)
     continue
    self.assertEqual(data["url"], self.module.API + "/messages/send")
    message = self.parse_wire(data)
    self.assert_fields(message, {"Subject": subject, "To": RECIPIENT})
    self.assertEqual(message.get_body(preferencelist=("plain",)).get_content(), body.replace("\n", "\r\n") + "\r\n")
    self.assertRegex(data["messageId"], r"^<[^<>\s@]+@[^<>\s@]+>$")
    self.assertEqual(str(message["Message-ID"]), data["messageId"])
    self.assert_fields(data["preview"], {"Message-ID": data["messageId"], "Subject": subject})
    self.assert_contains(data["preview"]["body"], contains)
    if subject == "Café ☕":
     raw = base64.urlsafe_b64decode(str(data["jsonBody"]["raw"]) + "===")
     self.assertIn(b"=?utf-8?", raw.lower())

 def test_send_digest_binds_headers_and_body(self) -> None:
  base = write_args(subject='A subject', body='A body')
  original = self.preview(*base)
  changes = (
 write_args(sender='other@example.com', subject='A subject', body='A body'),
 write_args(recipient='other@example.com', subject='A subject', body='A body'),
 write_args(subject='Changed', body='A body'),
 write_args(subject='A subject', body='Changed'),
 (*base, '--cc', 'cc@example.com'),
 (*base, '--bcc', 'bcc@example.com'),
 )
  for args in changes:
   with self.subTest(args=args):
    result = self.run_cli(*args, "--message-id", str(original["messageId"]),
 "--approve", str(original["approvalDigest"]),
 fixture=self.empty_fixture, expected=1)
    self.assertIn("digest", result.stderr.lower())

 def test_unapproved_send_refused_before_request(self) -> None:
  self.refused("send", "--to", RECIPIENT, "--subject", "Subject", "--body", "Body",
 contains=("--approve",))

 def test_live_write_requires_valid_message_id(self) -> None:
  args = write_args()
  preview = self.preview(*args)
  approved = (*args, "--approve", str(preview["approvalDigest"]))
  self.refused(*approved, contains=("--message-id",))
  for identifier in ("not-a-message-id", "<bad@id>\nBcc: evil@example.com", "<bad id@example.com>",
 "<x(comment)@example.com>", "<x@[127.0.0.1]>", "<\u00fc@example.com>"):
   with self.subTest(identifier=identifier):
    self.refused(*approved, "--message-id", identifier,
 excludes=("fake transport has no matching",))

 def test_header_injection_and_multiple_mailboxes_are_refused(self) -> None:
  for option, value in (("--to", "a@example.com,Bcc: evil@example.com"),
 ("--to", "Friends: a@example.com;"),
 ("--to", "a@example.com\nBcc: evil@example.com"),
 ("--from", "user@example.com\rBcc: evil@example.com"),
 ("--subject", "Hello\rBcc: evil@example.com")):
   with self.subTest(option=option, value=value):
    args = list(write_args())
    args[args.index(option) + 1] = value
    self.refused(*args, "--dry-run")
  with self.assertRaises(ValueError):
   self.backend().build_request("send", USER, [RECIPIENT],
 [], [], "Hello\x00World", "Body")

 def test_approved_draft_creates_one_server_draft(self) -> None:
  args = write_args(command='draft', subject='Draft subject', body='Draft body')
  preview = self.preview(*args)
  digest = preview["approvalDigest"]
  result = self.live_cli(args, preview, digest=digest, responses=[dict(make_response(method="POST", contains="/drafts", data={"id": "synthetic-draft"}), request_json=preview["jsonBody"])])
  self.check_status(result)
  self.assertTrue(json.loads(result.stdout)["draftCreated"])

 def test_approved_send_verifies_exact_returned_server_message(self) -> None:
  args = (*write_args(subject='Approved'), '--cc', 'cc@example.com', '--bcc', 'bcc@example.com')
  preview = self.preview(*args)
  message = self.parse_wire(preview)
  result = self.live_cli(args, preview, responses=[
 dict(make_response(method="POST", contains="/messages/send", data={"id": "server-returned-id"}), request_json=preview["jsonBody"]),
 self.message_response("server-returned-id", labelIds=["SENT"],
 headers={"Subject": "Approved", "Message-ID": str(message["Message-ID"])}),
 ])
  self.check_status(result)
  self.assertEqual(json.loads(result.stdout)["message"]["id"], "server-returned-id")

 def test_live_send_missing_sent_or_different_message_id_is_unknown(self) -> None:
  args = write_args(subject='Approved')
  preview = self.preview(*args)
  for labels, returned_id in ((["INBOX"], str(preview["messageId"])),
 (["SENT"], '<different@example.com>')):
   with self.subTest(labels=labels, returned_id=returned_id):
    result = self.live_cli(args, preview, responses=[
 make_response(method="POST", contains="/messages/send", data={"id": "server-returned-id"}),
 self.message_response("server-returned-id", labelIds=labels,
 headers={"Message-ID": returned_id}),
 ], expected=1)
    self.assertIn("outcome unknown", result.stderr)
    self.assertIn(str(preview["messageId"]), result.stderr)

 def test_write_503_reports_unknown_outcome_without_retry(self) -> None:
  for kind, endpoint in (('send', '/messages/send'), ('draft', '/drafts')):
   args = write_args(command=kind)
   preview = self.preview(*args)
   with self.subTest(kind=kind):
    result = self.live_cli(args, preview, responses=[make_response(method='POST', contains=endpoint,
 status=503, data={'error': {'message': 'synthetic failure'}})])
   self.check_status(result, 1)
   for hint in ('outcome unknown', 'verify-sent', str(preview['messageId'])):
    self.assertIn(hint, result.stderr)
   if kind == 'draft':
    self.assertIn('Drafts folder', result.stderr)
   self.assertNotIn('fake transport has no matching', result.stderr)

 def test_recent_stops_before_unneeded_next_page(self) -> None:
  result = self.run_cli("recent", "--count", "1", expected=0)
  self.assertEqual([item["id"] for item in json.loads(result.stdout)], ["gmail-one"])

 def test_recent_follows_next_page_token_to_requested_count(self) -> None:
  first = self.fixture_response("gmail-one")
  second = json.loads(json.dumps(first))
  second["url_contains"], second["json"]["id"] = "/messages/gmail-two", "gmail-two"
  result = self.run_responses("recent", "--count", "2", responses=[
 make_response(contains="maxResults=2", data={"messages": [{"id": "gmail-one"}],
 "nextPageToken": "page-two"}),
 make_response(contains="pageToken=page-two", data={"messages": [{"id": "gmail-two"}],
 "nextPageToken": "unneeded-page"}),
 first, second], expected=0)
  self.assertEqual([item["id"] for item in json.loads(result.stdout)], ["gmail-one", "gmail-two"])

 def test_attachment_metadata_only(self) -> None:
  part = {"mimeType": "application/octet-stream", "filename": "example.bin",
 "body": {"attachmentId": "part-one", "size": 42}}
  result = self.run_responses("attachments", "gmail-attachment", responses=[
 self.message_response("gmail-attachment", payload={"mimeType": "multipart/mixed", "parts": [part]})], expected=0)
  self.assertEqual(json.loads(result.stdout), [{"id": "part-one", "name": "example.bin",
 "size": 42, "contentType": "application/octet-stream"}])

 def test_verify_sent_requires_sent_label_and_exact_rfc822_message_id(self) -> None:
  expected_id = '<sent-fixture@example.com>'
  for labels, identifier, success in ((["INBOX"], expected_id, False),
 (["SENT"], '<different@example.com>', False),
 (["SENT"], expected_id, True)):
   with self.subTest(labels=labels, identifier=identifier):
    result = self.run_responses("verify-sent", "--subject", "Sent fixture subject",
 "--rfc822msgid", expected_id, responses=self.sent_responses(
 threadId="sent-thread", labelIds=labels, internalDate="1767225600000",
 payload={"mimeType": "text/plain"},
 headers={"Subject": "Sent fixture subject", "Message-ID": identifier}))
    self.assertEqual(result.returncode, 0 if success else 1, result.stderr)
    if success:
     self.assertEqual(json.loads(result.stdout)["rfc822msgid"], expected_id)
    else:
     self.assertIn("SENT", result.stderr)

 def test_verify_sent_compares_all_requested_recipients(self) -> None:
  responses = self.sent_responses(labelIds=["SENT"], headers={"Subject": "Subject",
 "Message-ID": SENT_ID, "To": 'to@example.com',
 "Cc": 'cc@example.com', "Bcc": 'bcc@example.com'})
  for option in ("--to", "--cc", "--bcc"):
   with self.subTest(option=option):
    result = self.run_responses("verify-sent", "--subject", "Subject",
 "--rfc822msgid", SENT_ID, "--to", 'to@example.com',
 "--cc", 'cc@example.com', "--bcc", 'bcc@example.com', option,
 'wrong@example.com', responses=responses, expected=1)

 def test_verify_sent_parser_empty_cc_bcc(self) -> None:
  s = self.shared
  args = s.build_parser().parse_args(["--provider", "gmail", "verify-sent",
   "--subject", "Subject", "--rfc822msgid", SENT_ID, "--to", "to@example.com",
   "--cc", "", "--bcc", "", "--json"])
  record = {"subject": "Subject", "labelIds": ["SENT"], "rfc822msgid": SENT_ID,
   "to": [{"address": "to@example.com"}], "cc": [], "bcc": []}
  b = types.SimpleNamespace(is_gmail=True, sent_items=mock.Mock(return_value=[record]))
  with contextlib.redirect_stdout(io.StringIO()) as out:
   s.command_verify_sent(args, b, {})
  self.assertEqual(json.loads(out.getvalue()), record)
  record["cc"] = [{"address": "cc@example.com"}]
  with self.assertRaisesRegex(s.CliError, "outcome remains unknown"):
   s.command_verify_sent(args, b, {})

 def test_verify_sent_exact_sets_omissions_and_duplicate_subjects(self) -> None:
  s = self.shared
  record = {"id": "first", "subject": "Subject", "labelIds": ["SENT"],
 "rfc822msgid": SENT_ID, "to": [{"address": "a@example.com"},
 {"address": "B@example.com"}], "cc": [{"address": "cc@example.com"}],
 "bcc": [{"address": "bcc@example.com"}], "body": "different", "sentDateTime": "old"}
  wrong_id = {**record, "id": "wrong-id", "rfc822msgid": "<other@example.com>"}
  duplicate = {**record, "id": "second"}
  b = types.SimpleNamespace(is_gmail=True, sent_items=mock.Mock(return_value=[wrong_id, record, duplicate]))
  for subject, to, cc, bcc, success in (
 ("Subject", ["B@example.com,a@example.com,A@example.com"], ["cc@example.com"], ["bcc@example.com"], True),
 ("subject", None, None, None, False),
 ("Subject", ["a@example.com"], None, None, False),
 ("Subject", ["a@example.com,b@example.com,extra@example.com"], None, None, False),
 ("Subject", [], None, None, False), ("Subject", None, [], None, False),
 ("Subject", None, None, [], False), ("Subject", None, None, None, True)):
   args = argparse.Namespace(subject=subject, to=to, cc=cc, bcc=bcc, count=3,
 rfc822msgid=SENT_ID, json=True)
   with self.subTest(subject=subject, to=to, cc=cc, bcc=bcc), contextlib.redirect_stdout(io.StringIO()) as out:
    if success:
     s.command_verify_sent(args, b, {})
     self.assertEqual(json.loads(out.getvalue()), record)
    else:
     with self.assertRaisesRegex(s.CliError, "outcome remains unknown"):
      s.command_verify_sent(args, b, {})
  b.sent_items = mock.Mock(return_value=[])
  with self.assertRaisesRegex(s.CliError, "outcome remains unknown"):
   s.command_verify_sent(args, b, {})
  b.sent_items.assert_called_once_with(3)

 def test_search_and_from_quote_safe_text(self) -> None:
  cases = (("search", "plain words", '"plain words"'),
 ("from", 'sender@example.com', 'from:"sender@example.com"'))
  for command, query_text, expected_query in cases:
   with self.subTest(command=command):
    encoded = urllib.parse.urlencode({"q": expected_query})
    result = self.run_responses(command, query_text, "--count", "1",
 responses=[make_response(contains=encoded, data={"messages": []})], expected=0)
    self.assertEqual(json.loads(result.stdout), [])

 def test_search_and_from_refuse_undocumented_phrase_escaping(self) -> None:
  for command in ("search", "from"):
   for unsafe in ('a"b', "a\\b"):
    with self.subTest(command=command, unsafe=unsafe):
     result = self.refused(command, unsafe, "--count", "1")
     self.assertIn("quote", result.stderr.lower())

 def test_provider_and_transport_gates(self) -> None:
  refused = self.run_cli('recent', '--count', '1', allow_fake=False, expected=1)
  self.assertIn('GMAIL_ALLOW_FAKE', refused.stderr)
  result = self.run_cli('recent', '--count', '1', provider=None, env_extra={'MAIL_PROVIDER': 'gmail'}, expected=0)
  self.assertEqual(json.loads(result.stdout)[0]['id'], 'gmail-one')
  result = self.run_cli(*write_args(command='draft'), '--dry-run', provider='graph', fixture=self.empty_fixture, expected=1)
  self.assertIn('--provider gmail', result.stderr)

 def test_multipart_prefers_plain_text_and_html_fallback(self) -> None:
  plain = self.run_cli("read", "gmail-multipart", "--full", expected=0)
  self.assertEqual(json.loads(plain.stdout)["body"], "Plain fixture body.")
  html = self.run_cli("read", "gmail-html", "--full", expected=0)
  self.assertIn("Hello & welcome.", json.loads(html.stdout)["body"])
  self.assertNotIn("Ignore this text", json.loads(html.stdout)["body"])

 def test_mime_charset_encoded_headers_and_utc_date(self) -> None:
  data = base64.urlsafe_b64encode("Café\nsecond".encode("iso-8859-1")).rstrip(b"=").decode()
  result = self.run_responses("read", "charset", "--full", expected=0, responses=[
 self.message_response("charset", internalDate="1767225600000", labelIds=["SENT"],
 payload={"mimeType": "text/plain", "body": {"data": data}}, headers={
 "Content-Type": "text/plain; charset=iso-8859-1", "Subject": "=?utf-8?b?Q2Fmw6k=?=",
 "From": "=?utf-8?b?Sm9zw6k=?= <jose@example.com>",
 "To": "=?utf-8?b?QW5kcuk=?= <anne@example.com>"})])
  self.assert_fields(json.loads(result.stdout), {"body": "Café\nsecond", "subject": "Café",
 "receivedDateTime": "2026-01-01T00:00:00Z", ("from", "name"): "José"})

 def test_text_body_preserves_lines_and_escapes_unsafe_characters(self) -> None:
  data = base64.urlsafe_b64encode("first\nsecond\x1b\u202e".encode()).rstrip(b"=").decode()
  result = self.run_responses("read", "body", "--full", json_output=False, expected=0, responses=[
 self.message_response("body", headers={"Subject": "Body test"},
 payload={"mimeType": "text/plain", "body": {"data": data}})])
  self.assertIn("first\nsecond\\u001b\\u202e", result.stdout)

 def test_attachment_text_escapes_untrusted_id_and_name(self) -> None:
  part = {"mimeType": "application/octet-stream", "filename": "name\nspoof.bin",
 "body": {"attachmentId": "id\x1bspoof", "size": 12}}
  result = self.run_responses("attachments", "attachment", json_output=False, expected=0, responses=[
 self.message_response("attachment", payload={"mimeType": "multipart/mixed", "parts": [part]})])
  self.assert_contains(result.stdout, ("id\\u001bspoof", "name\\u000aspoof.bin"))
  self.assertNotIn("name\nspoof.bin", result.stdout)

 def test_save_keeps_create_only_behavior(self) -> None:
  with tempfile.TemporaryDirectory() as directory:
   path = Path(directory) / "message.md"
   result = self.run_cli("save", "gmail-multipart", str(path), expected=0)
   self.assertIn("Plain fixture body.", path.read_text(encoding="utf-8"))
   second = self.run_cli("save", "gmail-multipart", str(path),
 fixture=self.empty_fixture, expected=1)
   self.assertIn("already exists", second.stderr)

 def test_reply_preview_binds_target_metadata(self) -> None:
  args = (*write_args(subject='Re: Original subject', body='Reply body'), "--reply-to-message-id", "reply-source")
  preview = self.preview(*args, fixture=self.fixture)
  message = self.parse_wire(preview)
  self.assertEqual(message["In-Reply-To"], '<original@example.com>')
  self.assertEqual(message["References"],
 "<parent@example.com> <original@example.com>")
  self.assertEqual(preview["jsonBody"]["threadId"], "reply-thread")

 def test_reply_target_changes_invalidate_approval_before_write(self) -> None:
  args = (*write_args(subject='Re: Original subject', body='Reply body'), "--reply-to-message-id", "reply-source")
  preview = self.preview(*args, fixture=self.fixture)
  approved = preview["approvalDigest"]
  source = self.fixture_response("reply-source")
  for field, changed in (("Message-ID", '<changed@example.com>'),
 ("References", '<changed-parent@example.com>'),
 ("Subject", "Changed subject"),
 ("threadId", "changed-thread")):
   with self.subTest(field=field):
    response = json.loads(json.dumps(source))
    if field == "threadId":
     response["json"]["threadId"] = changed
    else:
     header = next(item for item in response["json"]["payload"]["headers"]
 if item["name"] == field)
     header["value"] = changed
    result = self.live_cli(args, preview, digest=approved, responses=[response])
    self.check_status(result, 1)
    self.assertIn("digest", result.stderr.lower())
    self.assertNotIn("fake transport has no matching", result.stderr.lower())

 def test_raw_address_split_precedes_display_name_decoding(self) -> None:
  headers = {"From": "=?utf-8?q?ceo=40example.com=2C_?= <attacker@example.com>",
 "To": "=?utf-8?q?ceo=40example.com=2C_?= <a@example.com>",
 "Subject": "Subject", "Message-ID": SENT_ID}
  with fixture_file(self.sent_responses("forged", labelIds=["SENT"], headers=headers)) as fixture:
   result = self.run_cli("read", "forged", fixture=fixture)
   record = json.loads(result.stdout)
   self.assertEqual(record["from"], {"name": "ceo@example.com, ", "address": 'attacker@example.com'})
   self.assertEqual(record["to"][0]["address"], SENDER)
   text = self.run_text_cli("read", "forged", fixture=fixture)
   self.assertIn("From: ceo@example.com,  <attacker@example.com>", text.stdout)
   for address, expected in ((SENDER, 0), ('ceo@example.com', 1)):
    result = self.run_cli("verify-sent", "--subject", "Subject", "--rfc822msgid",
 SENT_ID, "--to", address, fixture=fixture)
    self.assertEqual(result.returncode, expected, result.stderr)

 def test_recipient_and_built_header_defences(self) -> None:
  module, backend = self.module, self.backend()
  for value in ("a@example.com,b@example.com", "a" + chr(0) + "@example.com",
 "@example.com", "a@", "=?utf-8?q?x=40evil.com=2C_?= <a@example.com>",
 "A (comment) <a@example.com>", "A) <a@example.com>", "A\\B <a@example.com>",
 "\u00fc@example.com", '"name@example.com" <a@example.com>', "ceo@example.com <a@example.com>"):
   for label in ("sender", "to", "cc", "bcc"):
    with self.subTest(value=value, label=label), self.assertRaises(ValueError):
     module.validate_mailbox(value, label)
  with self.assertRaises(ValueError) as caught:
   module.validate_mailbox('ceo@example.com <a@example.com>', 'to')
  self.assertEqual(str(caught.exception), "invalid to mailbox: encoded words, comments, group syntax, '@' in the display name and non-ASCII addresses are not accepted")
  args = ("send", SENDER, [TARGET], [], [], "Subject", "Body")
  with self.assertRaises(ValueError):
   backend.build_request(*args, message_id="<id@example.com/path>")
  for reply in ({"message_id": "bad", "thread_id": "thread"},
 {"message_id": MESSAGE_ID, "thread_id": ""}):
   with self.subTest(reply=reply), self.assertRaises(ValueError):
    backend.build_request(*args, reply=reply)

  with self.assertRaises(ValueError):
   backend.build_request(*args[:5], "=?utf-8?q?bad=00subject?=", "Body")
  for header in ("From: other@example.com", "To: other@example.com", "Cc: other@example.com",
 "Bcc: other@example.com", "Subject: =?utf-8?q?bad=00subject?="):
   wire = ("From: a@example.com\r\nTo: b@example.com\r\nSubject: Subject\r\n"
 "Message-ID: <id@example.com>\r\n" + header + "\r\n\r\nBody\r\n").encode()
   with self.subTest(header=header), mock.patch.object(module.EmailMessage, "as_bytes", return_value=wire):
    with self.assertRaises(ValueError):
     backend.build_request(*args)

 def test_message_id_is_random_for_identical_content(self) -> None:
  args = write_args(sender=SENDER, recipient=TARGET)
  self.assertNotEqual(self.preview(*args)["messageId"], self.preview(*args)["messageId"])

 def test_reply_headers_with_cr_or_lf_are_refused(self) -> None:
  module, backend = self.module, self.backend()
  for name in ("Message-ID", "References", "Subject"):
   for control in ("\r", "\n"):
    headers = {"Message-ID": MESSAGE_ID, "References": "", "Subject": "Subject"}
    headers[name] = "bad" + control + "injected"
    backend.request = mock.Mock(return_value={"threadId": "thread", "payload": {"headers": [
 make_header(key, value) for key, value in headers.items()]}})
    reply = backend.reply_metadata("source")
    with self.subTest(name=name, control=control), self.assertRaises(ValueError):
     backend.build_request("send", SENDER, [TARGET], [], [], "Subject", "Body", reply)

 def test_preview_uses_wire_bytes_and_delimits_body_for_both_providers(self) -> None:
  shared = self.shared
  module = self.module
  backend = module.GmailBackend(shared, {})
  request = backend.build_request("send", 'wire@example.com', ['wire-to@example.com'],
 [], [], "Wire subject", "Wire body")
  backend.build_request = mock.Mock(return_value=request)
  backend.account_address = mock.Mock(return_value=PROFILE["emailAddress"])
  args = self.write_namespace(dry_run=True, message_id=None, from_user='args@example.com',
 to=['args-to@example.com'], subject="Args subject", body="Args body")
  with self.gmail_import(), contextlib.redirect_stdout(io.StringIO()) as output:
   shared.command_gmail_write(args, backend, {}, "send")
  preview = json.loads(output.getvalue())["preview"]
  for name, value in (("From", 'wire@example.com'), ("To", 'wire-to@example.com'),
 ("Subject", "Wire subject"), ("body", "Wire body")):
   self.assertEqual(preview[name], value)
  body = "first\napproval digest: fake\n--- body end ---\nlast"
  for provider in ("graph", "gmail"):
   result = self.run_cli(*write_args(recipient='B <b@example.com>', body=body),
 '--cc', 'C <c@example.com>', '--bcc', 'D <d@example.com>', '--dry-run',
 provider=provider, fixture=self.empty_fixture, json_output=False, expected=0)
   self.assertIn("--- body start ---\n| first\n| approval digest: fake\n| --- body end ---\n| last\n--- body end ---", result.stdout)
   self.assertEqual(sum(line.startswith("approval digest:") for line in result.stdout.splitlines()), 1)
   if provider == 'gmail':
    self.assertIn('Recipients: to=b@example.com; cc=c@example.com; bcc=d@example.com', result.stdout)

 def test_unexpected_errors_and_remote_revoke_failure_are_reported(self) -> None:
  shared = self.shared
  with mock.patch.object(shared, "build_parser", side_effect=RuntimeError("unexpected\nfailure")), \
   contextlib.redirect_stderr(io.StringIO()) as error:
   self.assertEqual(shared.main([]), 1)
  self.assertEqual(error.getvalue(), "error: unexpected\\u000afailure\n")
  client = types.SimpleNamespace(is_gmail=True, auth_revoke=lambda: "remote failure")
  with contextlib.redirect_stdout(io.StringIO()) as output:
   with self.assertRaisesRegex(shared.CliError, "remote revocation failed: remote failure"):
    shared.command_auth(argparse.Namespace(auth_command="revoke"), client, {})
  self.assertEqual(output.getvalue(), "Gmail local token deleted.\n")

 def test_body_escapes_every_unsafe_unicode_category(self) -> None:
  shared = self.shared
  for code in (0, 0xb, 0xc, 0xd, 0x1b, 0x85, 0x200e, 0x2061, 0xe000, 0x378,
 0x2028, 0x2029, 0x34f, 0xd800, 0xa0, 0x115f, 0x1160, 0x3164,
 0xffa0, 0x2800, *range(0x180b, 0x1810), 0xfe00, 0xfe0f, 0xe0100, 0xe01ef):
   with self.subTest(code=code):
    self.assertEqual(shared.display_body(chr(code)), ("\\u%04x" if code <= 0xffff else "\\U%08x") % code)
    if code in (0x34f, 0x115f, 0x1160, 0x3164, 0xffa0, 0x2800, 0x180b, 0xfe00, 0xe0100):
     self.assertEqual(shared.display_text(chr(code)), ("\\u%04x" if code <= 0xffff else "\\U%08x") % code)
  self.assertEqual(shared.display_body("normal\n\ttext"), "normal\n\ttext")
  self.assertEqual(shared.display_body(r"\u202e\U0000202e"), r"\u005cu202e\u005cU0000202e")

 def test_malformed_thread_and_attachment_errors_are_clean(self) -> None:
  for command, message, expected in (("thread", {"messages": [{"id": "ok"}, None]}, "invalid Gmail thread message"),
 ("attachments", {"payload": {"filename": "file", "body": {"size": "zz"}}}, "invalid Gmail attachment size")):
   with self.subTest(command=command), fixture_file(
 [{"method": "GET", "json": message}]
 ) as fixture:
    result = self.run_cli(command, "bad", fixture=fixture, expected=2 if command == "thread" else 1)
    self.assertIn(expected, result.stdout if command == "thread" else result.stderr)
    self.assertNotIn("Traceback", result.stderr)

 def test_round2_declarations_and_docs(self):
  declaration = (self.root / "source-package.json").read_text()
  for claim in ("0.2.1", "users/me/profile", "emailAddress", "same write token", "body.data", "inline attachment data", "discarded, never shown or saved"):
   self.assertIn(claim, declaration)
  for name in ("README.md", "SETUP.md", "SKILL.md", "references/gmail-api.md"):
   text = (self.root / name).read_text()
   for claim in ("profile", "inline attachment", "never shown or saved"):
    self.assertIn(claim, text)
  if (self.root / "packaging").is_dir():
   values = json.loads((self.root / "packaging/listing-values.json").read_text())
   listing = (self.root / "packaging/skills-outpost-listing.md").read_text()
   self.assertEqual(values["product_version"], "0.2.1")
   for key in ("description", "limitations", "data_handling", "output_summary"):
    self.assertIn(values[key], listing)


USER = 'user@example.com'
RECIPIENT = 'recipient@example.com'
SENDER = 'a@example.com'
TARGET = 'b@example.com'
MESSAGE_ID = '<id@example.com>'
SENT_ID = '<sent@example.com>'

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

License: MIT (read it).

Back to the listing