tests/test_m365_mail.py — Approval-Gated Mail

# By Zero Width Lab (zerowidthlab.com)

from __future__ import annotations

import argparse
import ast
import contextlib
import hashlib
import importlib.util
import io
import json
import os
import re
import subprocess
import sys
import tempfile
import unittest
import urllib.request
import urllib.parse
from pathlib import Path
from unittest import mock

__author__ = "Zero Width Lab"

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

USER = 'user@example.com'
GRAPH_USER = 'https://graph.microsoft.com/v1.0/users/user%40example.com'

class M365MailCliTests(unittest.TestCase):
 def setUp(self) -> None:
  self.package_root = Path(__file__).resolve().parent.parent
  self.script_path = self.package_root / "scripts" / "m365_mail.py"
  fixtures = self.package_root / "tests" / "fixtures"
  self.fixture_path = fixtures / "graph-responses.json"
  self.empty_fixture_path = fixtures / "empty-responses.json"

 def load_module(self):
  spec = importlib.util.spec_from_file_location("m365_test_module", self.script_path)
  self.assertIsNotNone(spec)
  self.assertIsNotNone(spec.loader)
  module = importlib.util.module_from_spec(spec)
  spec.loader.exec_module(module)
  return module

 def send_args(self, subject="Subject", body="Body", to="to@example.com",
 sender=USER, **fields):
  values = {"from": sender, "to": to, "subject": subject, "body": body, **fields}
  return tuple(item for key, value in values.items()
 for item in ("--" + key.replace("_", "-"), value) if value is not None)

 def message_body(self, subject="Subject", body="Body", to=("to@example.com",),
 cc=(), bcc=(), save=True):
  message = {"subject": subject, "body": {"contentType": "Text", "content": body}}
  for key, addresses in (("to", to), ("cc", cc), ("bcc", bcc)):
   message[key + "Recipients"] = [
 {"emailAddress": {"address": address}} for address in addresses
 ]
  return {"message": message, **({"saveToSentItems": True} if save else {})}

 def successful_json(self, result):
  self.assertEqual(result.returncode, 0, result.stderr)
  return json.loads(result.stdout)

 @contextlib.contextmanager
 def temporary_path(self, name):
  with tempfile.TemporaryDirectory() as directory:
   yield Path(directory) / name

 def assert_fields(self, record, fields):
  for path, expected in fields.items():
   if path == "length":
    actual = len(record)
   elif path == "ids":
    actual = [item["id"] for item in record]
   elif path == "all":
    actual = record
   else:
    actual = record
    for key in path.split("."):
     actual = actual[int(key) if key.isdecimal() else key]
   self.assertEqual(actual, expected, path)

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

 def assert_refused(self, result, reason=None, no_transport=True, clean=True):
  self.assertEqual(result.returncode, 1)
  lowered = result.stderr.lower()
  if reason is not None:
   self.assertIn(reason, lowered)
  if no_transport:
   self.assertNotIn("fake transport has no matching", lowered)
  if clean:
   self.assertNotIn("traceback", lowered)

 @contextlib.contextmanager
 def fake_saved_message(self, module, output):
  args = argparse.Namespace(output_path=str(output), message_id="message-save",
 max_chars=100, full=False, json=True,
 user=USER)
  with mock.patch.object(module, "get_message", return_value={"id": "message-save"}), \
   mock.patch.object(module, "truncate_body"), \
   mock.patch.object(module, "markdown_message", return_value="content"):
   yield args

 def cli_env(self, fixture=None, allow_fake=True):
  env = os.environ.copy()
  for name in ("M365_TENANT_ID", "M365_CLIENT_ID", "M365_CLIENT_SECRET",
 "M365_DEFAULT_USER", "M365_MAIL_ALLOW_FAKE", "MAIL_PROVIDER"):
   env.pop(name, None)
  env["M365_MAIL_FAKE_TRANSPORT"] = str(fixture or self.fixture_path)
  if allow_fake:
   env["M365_MAIL_ALLOW_FAKE"] = "1"
  return env

 def invoke(self, command, env):
  return subprocess.run(command, cwd=self.package_root, capture_output=True,
 text=True, check=False, env=env)

 def run_cli(self, *arguments, include_user=True, fixture=None, json_output=True,
 allow_fake=True):
  command = [sys.executable, str(self.script_path)]
  if json_output:
   command.append("--json")
  if include_user:
   command.extend(["--user", USER])
  return self.invoke([*command, *arguments], self.cli_env(fixture, allow_fake))

 def dry_run(self, *arguments, **options):
  return self.run_cli('send', *arguments, '--dry-run', fixture=self.empty_fixture_path, **options)

 def approval_for(self, *arguments: str) -> str:
  preview = self.dry_run(*arguments)
  result = self.successful_json(preview)
  digest = result["approvalDigest"]
  self.assertRegex(digest, r"^[0-9a-f]{64}$")
  payload = {key: result[key] for key in ("method", "url", "jsonBody")}
  encoded = json.dumps(payload, sort_keys=True, ensure_ascii=False,
 separators=(",", ":")).encode("utf-8")
  expected = hashlib.sha256(encoded).hexdigest()
  self.assertEqual(digest, expected)
  return digest

 def approved_send(self, *arguments: str) -> subprocess.CompletedProcess[str]:
  return self.run_cli("send", *arguments, "--approve", self.approval_for(*arguments))

 def test_metadata_queries(self) -> None:
  cases = (
 (("recent", "--count", "2"), {
 "length": 2, "0.id": "message-002", "0.subject": "Second fixture message",
 "0.conversationId": "conversation-001", "0.isRead": False,
 "0.hasAttachments": True,
 }),
 (("recent", "--count", "1"), {"ids": ["message-bounded"]}),
 (("search", "needle phrase", "--count", "3"), {
 "ids": ["message-search"], "0.subject": "Needle phrase result",
 }),
 (("thread", "conversation-thread", "--full"), {
 "ids": ["thread-older", "thread-newer"],
 "0.body": "Older thread message.", "1.body": "Newer thread message.",
 }),
 (("attachments", "message-attachments"), {"all": [{
 "id": "attachment-001", "name": "example.txt", "size": 42,
 "contentType": "text/plain",
 }]}),
 (("verify-sent", "--subject", "Standalone verified subject", "--to",
 "verified@example.com", "--count", "11"), {
 "id": "sent-standalone-message", "sentDateTime": "2026-04-03T10:11:12Z",
 "to.0.address": "verified@example.com",
 }),
 (("verify-sent", "--subject", "Subject only verification", "--count", "12"), {
 "id": "sent-subject-only-message", "cc.0.address": "copy@example.com",
 }),
 )
  for arguments, fields in cases:
   with self.subTest(arguments=arguments):
    record = self.successful_json(self.run_cli(*arguments))
    self.assert_fields(record, fields)

 def test_redirect_handler_rejects_unapproved_https_host_offline(self) -> None:
  module = self.load_module()
  handler = module.ApprovedRedirectHandler()
  request = urllib.request.Request("https://graph.microsoft.com/v1.0/users")

  with self.assertRaises(module.CliError):
   handler.redirect_request(request, None, 302, 'Found', {}, 'https://example.invalid/redirect')

 def test_default_opener_keeps_user_selected_https_proxy_offline(self) -> None:
  env = os.environ.copy()
  for name in ("https_proxy", "NO_PROXY", "no_proxy", "M365_MAIL_FAKE_TRANSPORT"):
   env.pop(name, None)
  env["HTTPS_PROXY"] = "http://127.0.0.1:9"
  with mock.patch.dict(os.environ, env, clear=True):
   module = self.load_module()
   handlers = module.GraphClient({}).opener.handlers
  proxies = [handler for handler in handlers
 if isinstance(handler, urllib.request.ProxyHandler)]
  self.assertEqual(len(proxies), 1, handlers)
  self.assertEqual(proxies[0].proxies.get("https"), "http://127.0.0.1:9")
  for kind in (urllib.request.HTTPSHandler, module.ApprovedRedirectHandler):
   with self.subTest(handler=kind):
    self.assertTrue(any(isinstance(handler, kind) for handler in handlers))

 def test_read_converts_html_and_honors_truncation_and_full(self) -> None:
  for flags, truncated, present in (
 (("--max-chars", "12"), True, ("Hello & welc",)),
 (("--full",), False, ("Hello & welcome.", "Second line has more text.")),
 ):
   with self.subTest(flags=flags):
    record = self.successful_json(self.run_cli("read", "message-html", *flags))
    self.assertEqual(record["bodyTruncated"], truncated)
    self.assertNotIn("Ignore this text", record["body"])
    if truncated:
     self.assertEqual(record["body"], present[0])
    else:
     for text in present:
      self.assertIn(text, record["body"])

 def test_save_writes_markdown_only_to_explicit_path(self) -> None:
  with self.temporary_path("explicit-message.md") as output_path:
   temp_path = output_path.parent

   result = self.run_cli("save", "message-save", str(output_path))

   self.assertEqual(result.returncode, 0, result.stderr)
   self.assertEqual(sorted((path.name for path in temp_path.iterdir())), ['explicit-message.md'])
   markdown = output_path.read_text(encoding="utf-8")
   self.contains(markdown, "# Saved fixture", "Saved body text.")
   self.assertEqual(json.loads(result.stdout)["saved"], str(output_path))
   if os.name == "posix":
    self.assertEqual(output_path.stat().st_mode & 0o777, 0o600)

 def test_save_refuses_existing_file_symlink_and_missing_parent_before_transport(self) -> None:
  for kind in ("file", "symlink", "parent"):
   with self.subTest(kind=kind), tempfile.TemporaryDirectory() as temp_dir:
    target = Path(temp_dir) / "target.md"
    output = Path(temp_dir) / "output.md"
    if kind != "parent":
     target.write_text("original", encoding="utf-8")
     if kind == "symlink":
      output.symlink_to(target)
     else:
      output = target
    else:
     output = Path(temp_dir) / "missing" / "message.md"
    result = self.run_cli('save', 'message-save', str(output), fixture=self.empty_fixture_path)
    reason = "parent" if kind == "parent" else "already exists"
    self.assert_refused(result, reason, clean=False)
    if kind != "parent":
     self.assertEqual(target.read_text(encoding="utf-8"), "original")
    if kind == "symlink":
     self.assertTrue(output.is_symlink())

 def test_save_unlinks_partial_file_after_write_error(self) -> None:
  module = self.load_module()

  original_fdopen = os.fdopen

  class FailedWrite:
   def __init__(self, descriptor: int) -> None:
    self.stream = original_fdopen(descriptor, "w", encoding="utf-8")

   def __enter__(self) -> "FailedWrite":
    return self

   def __exit__(self, *arguments: object) -> None:
    self.stream.close()

   def write(self, content: str) -> None:
    self.stream.write(content[:1])
    raise OSError("synthetic write failure")

  with self.temporary_path("partial.md") as output_path:
   with self.fake_saved_message(module, output_path) as args, \
    mock.patch.object(module.os, "fdopen", side_effect=lambda fd, *a, **k: FailedWrite(fd)), \
    self.assertRaisesRegex(module.CliError, "synthetic write failure"):
    module.command_save(args, object(), {})
   self.assertFalse(os.path.lexists(output_path))

 @unittest.skipUnless(os.name == "posix", "requires POSIX O_NOFOLLOW")
 def test_save_open_flags_refuse_existing_file_and_dangling_symlink(self) -> None:
  module = self.load_module()
  with tempfile.TemporaryDirectory() as temp_dir:
   existing = Path(temp_dir) / "existing.md"
   existing.write_text("original", encoding="utf-8")
   dangling = Path(temp_dir) / "dangling.md"
   dangling.symlink_to(Path(temp_dir) / "absent.md")
   for output_path in (existing, dangling):
    with self.subTest(output_path=output_path):
     with mock.patch.object(module.os.path, "lexists", return_value=False), \
      self.fake_saved_message(module, output_path) as args, \
      mock.patch.object(module.os, "open", wraps=os.open) as guarded_open, \
      self.assertRaises(module.CliError):
      module.command_save(args, object(), {})
     flags = guarded_open.call_args.args[1]
     self.assertTrue(flags & os.O_CREAT)
     self.assertTrue(flags & os.O_EXCL)
     self.assertTrue(flags & os.O_NOFOLLOW)
   self.assertEqual(existing.read_text(encoding="utf-8"), "original")
   self.assertTrue(dangling.is_symlink())
   self.assertFalse(dangling.exists())

 def test_save_body_limits_and_truncation_notice(self) -> None:
  for flags in (("--max-chars", "12"), ("--full",), ()):
   with self.subTest(flags=flags), self.temporary_path("message.md") as output_path:
    result = self.run_cli("save", "message-html", str(output_path), *flags)
    self.assertEqual(result.returncode, 0, result.stderr)
    markdown = output_path.read_text(encoding="utf-8")
    if flags == ("--max-chars", "12"):
     self.assertIn("Hello & welc", markdown)
     self.assertNotIn("Hello & welcome.", markdown)
     self.assertIn(
 "Body truncated to 12 characters; rerun with --full for the complete text.",
 markdown,
 )
    else:
     self.contains(markdown, "Hello & welcome.", "Second line has more text.")
     self.assertNotIn("Body truncated to", markdown)

 def test_quoted_search_terms_escape_quotes_and_backslashes_in_exact_url(self) -> None:
  summary_fields = (
 "id,subject,from,receivedDateTime,sentDateTime,isRead,hasAttachments,"
 "conversationId,toRecipients,ccRecipients,bccRecipients"
 )
  for command, term, search_value in (
 ("search", 'a"b\\c', 'a\\"b\\\\c'),
 ("from", 'a"b\\c', 'from:a\\"b\\\\c'),
 ):
   with self.subTest(command=command):
    query = urllib.parse.urlencode({
 "$select": summary_fields,
 "$top": 1,
 "$search": f'"{search_value}"',
 })
    expected_url = GRAPH_USER + f"/messages?{query}"
    with tempfile.TemporaryDirectory() as temp_dir:
     fixture = Path(temp_dir) / "exact-url.json"
     fixture.write_text(json.dumps({"responses": [{
 "method": "GET", "url": expected_url,
 "status": 200, "json": {"value": []},
 }]}), encoding="utf-8")
     result = self.run_cli(command, term, '--count', '1', fixture=fixture)
    self.assertEqual(self.successful_json(result), [])

 def test_approved_send_and_reply_records(self) -> None:
  for reply, subject, body, recipients, day, identifier in (
 (False, "Exact approved subject", "Exact approved body.",
 "first@example.com, second@example.com", "01", "sent-new-message"),
 (True, "Exact reply subject", "Exact approved reply body.",
 "reply@example.com", "02", "sent-reply-message"),
 ):
   with self.subTest(reply=reply):
    fields = {"reply_to_message_id": "reply-source"} if reply else {
 "cc": "copy@example.com", "bcc": "hidden@example.com",
 }
    sent = self.successful_json(self.approved_send(
 *self.send_args(subject, body, recipients, **fields),
 ))
    self.assertTrue(sent["verifiedInSentItems"])
    self.assert_fields(sent, {
 "message.id": identifier, "message.sentDateTime": f"2026-04-{day}T10:11:12Z",
 })
    message = sent["message"]
    if reply:
     self.assertEqual(message["conversationId"], "conversation-reply")
    else:
     self.assertTrue(sent["sent"])
     for field, addresses in (
 ("to", ["first@example.com", "second@example.com"]),
 ("cc", ["copy@example.com"]), ("bcc", ["hidden@example.com"]),
 ):
      with self.subTest(field=field):
       self.assertEqual([item["address"] for item in message[field]], addresses)

 def test_exact_send_and_reply_previews(self) -> None:
  cases = (
 ("Dry run subject", "Exact dry run body.",
 ("first@example.com", "second@example.com"), ("copy@example.com",),
 ("hidden@example.com",), None, "/sendMail"),
 ("Dry run reply subject", "Exact dry run reply body.",
 ("reply@example.com",), (), (), "reply-source", "/messages/reply-source/reply"),
 )
  for subject, body, to, cc, bcc, reply, endpoint in cases:
   with self.subTest(reply=reply):
    fields = {"reply_to_message_id": reply} if reply else {}
    if cc:
     fields["cc"] = cc[0]
    if bcc:
     fields["bcc"] = bcc[0]
    preview = self.successful_json(self.dry_run(
 *self.send_args(subject, body, ", ".join(to), **fields),
 include_user=False,
 ))
    self.assertEqual(preview, {
 "dryRun": True, "method": "POST",
 "url": GRAPH_USER + endpoint,
 "approvalDigest": preview["approvalDigest"],
 "jsonBody": self.message_body(subject, body, to, cc, bcc, save=not reply),
 })
    if reply:
     self.assertNotIn("saveToSentItems", preview["jsonBody"])

 def test_send_dry_run_resolves_mailbox_from_user(self) -> None:
  result = self.dry_run(*self.send_args(
 "User mailbox dry run subject", "Exact user mailbox dry run body.",
 "recipient@example.com", sender=None,
 ))

  preview = self.successful_json(result)
  self.assertEqual(preview['url'], GRAPH_USER + '/sendMail')

 def test_text_dry_run_prints_approval_digest(self) -> None:
  result = self.run_cli("send", *self.send_args(), "--dry-run",
 fixture=self.empty_fixture_path, json_output=False)
  self.assertEqual(result.returncode, 0, result.stderr)
  self.contains(result.stdout, "Method: POST", GRAPH_USER + '/sendMail',
 '"content": "Body"')
  self.assertRegex(result.stdout, r"approval digest: [0-9a-f]{64}")

 def test_json_dry_run_escapes_zero_width_subject(self) -> None:
  subject = "Visible" + chr(0x200B) + "subject"
  arguments = self.send_args(subject=subject)
  result = self.dry_run(*arguments)
  preview = self.successful_json(result)
  self.assertIn("Visible\\u200bsubject", result.stdout)
  self.assertNotIn(chr(0x200B), result.stdout)
  self.assertEqual(preview["jsonBody"]["message"]["subject"], subject)
  self.assertEqual(preview["approvalDigest"], self.approval_for(*arguments))

 def test_unverified_send_outcomes_warn_against_retry(self) -> None:
  cases = (
 ("accepted@example.com", "Accepted without verification",
 "Exact accepted body.", "submission was accepted"),
 ("uncertain@example.com", "Unconfirmed transport result",
 "Exact approved body with an unconfirmed result.",
 "delivery state may be uncertain"),
 )
  for recipient, subject, body, reason in cases:
   with self.subTest(recipient=recipient):
    result = self.approved_send(*self.send_args(subject, body, recipient))
    self.assert_refused(result, reason, no_transport=False)
    self.assertIn("do not resend automatically", result.stderr.lower())

 def test_missing_mailbox_is_rejected(self) -> None:
  result = self.run_cli("recent", include_user=False)

  self.assert_refused(result, "user", no_transport=False)

 def test_invalid_arguments_are_rejected(self) -> None:
  cases = [
 ("recent", "--count", "0"),
 ("read", "message-001", "--max-chars", "0"),
 ("send", "--from", "sender@example.com"),
 ("send", *self.send_args("Conflicting body inputs", "Inline body",
 "recipient@example.com", "sender@example.com"),
 "--body-file", str(self.fixture_path), "--approve", "0" * 64),
 ]
  for arguments in cases:
   with self.subTest(arguments=arguments):
    result = self.run_cli(*arguments)
    self.assert_refused(result, no_transport=False)
    self.assertTrue(result.stderr.strip())

 def test_send_approval_and_conversation_refusals(self) -> None:
  conversation = (*self.send_args("Reply subject", "Reply body", "reply@example.com"),
 "--in-reply-to", "conversation-preview")
  cases = (
 (self.send_args("Approved text required", "This body has not been approved.",
 "recipient@example.com", "sender@example.com"), "approv", True),
 ((*self.send_args(), "--approve", "é"), "mismatched approval digest", True),
 (("--to", "recipient@example.com", "--subject", "Subject", "--body",
 "Body", "--approved"), None, False),
 ((*conversation, "--dry-run"), "--in-reply-to is unsupported", True),
 ((*conversation, "--approve", "0" * 64), "--in-reply-to is unsupported", True),
 )
  for arguments, reason, clean in cases:
   with self.subTest(arguments=arguments):
    result = self.run_cli("send", *arguments, fixture=self.empty_fixture_path)
    self.assert_refused(result, reason, clean=clean)
    if reason is None:
     self.assertIn("--approve <digest>", result.stderr)
    elif reason == "--in-reply-to is unsupported":
     self.contains(result.stderr, "run thread CONVERSATION_ID",
 "--reply-to-message-id ID")

 def test_send_digest_mismatch_refuses_each_bound_field_before_transport(self) -> None:
  baseline = dict(to="first@example.com", cc="copy@example.com", bcc="hidden@example.com")
  digest = self.approval_for(*self.send_args(**baseline))
  reply_digest = self.approval_for(*self.send_args(**baseline),
 "--reply-to-message-id", "reply-source")
  changes = (
 ({"sender": "other@example.com"}, digest),
 *(({field: "different@example.com"}, digest) for field in ("to", "cc", "bcc")),
 ({"subject": "Different subject"}, digest),
 ({"body": "Different body"}, digest),
 ({"reply_to_message_id": "reply-source"}, digest),
 ({"reply_to_message_id": "other-message"}, reply_digest),
 )
  for fields, approval in changes:
   with self.subTest(fields=fields):
    result = self.run_cli("send", *self.send_args(**(baseline | fields)),
 "--approve", approval, include_user=False,
 fixture=self.empty_fixture_path)
    self.assert_refused(result, "digest", clean=False)

 def test_body_file_digest_binds_exact_content(self) -> None:
  with self.temporary_path("body.txt") as body_path:
   body_path.write_text("Original body", encoding="utf-8")
   arguments = (
 "--from", USER, "--to", "first@example.com",
 "--subject", "File subject", "--body-file", str(body_path),
 )
   digest = self.approval_for(*arguments)
   body_path.write_text("Changed body", encoding="utf-8")
   result = self.run_cli('send', *arguments, '--approve', digest, fixture=self.empty_fixture_path)
   self.assert_refused(result, 'digest', clean=False)

 def test_send_dry_run_reads_body_file_once_and_hashes_sent_bytes(self) -> None:
  module = self.load_module()
  with self.temporary_path("body.txt") as body_path:
   body_path.write_text("First read body", encoding="utf-8")
   original_read = Path.read_text
   reads = 0

   def counted_read(path: Path, *args: object, **kwargs: object) -> str:
    nonlocal reads
    if path == body_path:
     reads += 1
     if reads > 1:
      raise AssertionError("body file read more than once")
    return original_read(path, *args, **kwargs)

   arguments = argparse.Namespace(
 approved=False, approve=None, dry_run=True, from_user=USER,
 user=None, to=["to@example.com"], cc=None, bcc=None,
 subject="File subject", body=None, body_file=str(body_path),
 reply_to_message_id=None, in_reply_to=None, json=True,
 )
   output = io.StringIO()
   with mock.patch.object(Path, "read_text", autospec=True, side_effect=counted_read), \
    contextlib.redirect_stdout(output):
    module.command_send(arguments, object(), {})
   preview = json.loads(output.getvalue())
   self.assertEqual(reads, 1)
   self.assertEqual(preview["jsonBody"]["message"]["body"]["content"], "First read body")
   self.assertEqual(preview['approvalDigest'], module.approval_digest(preview))

 def test_digest_binds_canonical_request_fields(self) -> None:
  module = self.load_module()
  request = {
 "method": "POST",
 "url": GRAPH_USER + '/sendMail',
 "jsonBody": self.message_body(),
 }
  original = module.approval_digest(request)
  changes = (
 (("method",), "PATCH"), (("url",), request["url"] + "x"),
 (("url",), request["url"].replace("user%40example.com", "other%40example.com")),
 (("jsonBody", "message", "body", "contentType"), "HTML"),
 (("jsonBody", "message", "body", "content"), "Other body"),
 (("jsonBody", "saveToSentItems"), False),
 )
  for path, value in changes:
   with self.subTest(path=path):
    changed = json.loads(json.dumps(request))
    target = changed
    for key in path[:-1]:
     target = target[key]
    target[path[-1]] = value
    self.assertNotEqual(module.approval_digest(changed), original)

 def test_http_errors_are_clear_and_throttling_surfaces_retry_after(self) -> None:
  for count in ("7", "13"):
   with self.subTest(count=count):
    result = self.run_cli("recent", "--count", count)
    self.assert_refused(result, no_transport=False)
    if count == "7":
     self.assertTrue(result.stderr.strip())
     self.assertTrue(
 "503" in result.stderr or "fixture failure" in result.stderr.lower(),
 result.stderr,
 )
    else:
     for value in ("429", "Retry-After", "30"):
      self.assertIn(value, result.stderr)

 def test_fake_transport_requires_explicit_allow_flag(self) -> None:
  result = self.run_cli("recent", "--count", "2", allow_fake=False)
  self.assert_refused(result, no_transport=False)
  self.assertIn("M365_MAIL_ALLOW_FAKE", result.stderr)

 def test_sources_contain_no_private_identifiers_or_unapproved_guids(self) -> None:
  listing_path = self.package_root / "packaging" / "skills-outpost-listing.md"
  if (self.package_root / "packaging").is_dir():
   self.assertTrue(listing_path.is_file())
  required = (
 "SKILL.md", "README.md", "SETUP.md", "LICENSE.txt", "source-package.json",
 "references/graph-api.md", "agents/m365-mail-agent.md",
 "scripts/m365_mail.py", "tests/test_m365_mail.py",
 "tests/fixtures/graph-responses.json",
 )
  expected_paths = {self.package_root / name for name in required}
  source_paths = [self.package_root / name for name in required[:5]]
  if listing_path.is_file():
   source_paths.append(listing_path)
  for directory, pattern in (
 ("references", "*.md"), ("agents", "*.md"), ("scripts", "*.py"),
 ("tests", "test_*.py"), ("tests/fixtures", "*.json"), ("packaging", "*.json"),
 ):
   source_paths.extend(sorted((self.package_root / directory).glob(pattern)))
  self.assertTrue(
 expected_paths.issubset(set(source_paths)),
 "One or more required source files are missing",
 )

  # Public canary uses placeholder patterns, never private identifiers or obfuscated terms.
  guid_lengths = (8, 4, 4, 4, 12)
  guid_pattern = "-".join(f"[0-9a-fA-F]{{{size}}}" for size in guid_lengths)
  placeholder = "-".join("0" * size for size in guid_lengths)
  email_pattern = r"[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Za-z]{2,}"
  allowed_email_suffix = "@example.com"
  disallowed_names = (
 "Open" + "AI", "Anth" + "ropic", "Chat" + "GP" + "T",
 "Gem" + "ini", "Clau" + "de", "Cod" + "ex",
 "Cop" + "ilot", "Son" + "net", "Op" + "us",
 )
  token_pattern = r"\b" + "GP" + "T" + r"\b"

  for path in source_paths:
   self.assertTrue(path.resolve().is_relative_to(self.package_root.resolve()))
   text = path.read_text(encoding="utf-8")

   patterns = [(name, re.escape(name), re.I) for name in disallowed_names]
   for name, pattern, flags in [*patterns, ("model token", token_pattern, 0)]:
    with self.subTest(path=path, name=name):
     self.assertIsNone(re.search(pattern, text, flags))

   for match in re.findall(email_pattern, text):
    with self.subTest(path=path, email=match):
     self.assertTrue(
 match.lower().endswith(allowed_email_suffix),
 f"non-{allowed_email_suffix} email address in {path}",
 )

   if path.parent != self.package_root / "tests" and not path.is_relative_to(
 self.package_root / "tests"
 ):
    for match in re.findall(guid_pattern, text):
     with self.subTest(path=path, guid=match):
      self.assertEqual(match, placeholder)

 def test_hygiene_has_no_environment_selected_input(self) -> None:
  for path in (self.package_root / "tests").glob("test_*.py"):
   self.assertTrue(path.resolve().is_relative_to(self.package_root.resolve()))
   for node in ast.walk(ast.parse(path.read_text(encoding="utf-8"))):
    if isinstance(node, ast.Call):
     called = ast.unparse(node.func)
     self.assertNotIn(called, ("os.environ.get", "os.getenv"))
    if isinstance(node, ast.Subscript):
     self.assertNotEqual(ast.unparse(node.value), "os.environ")

 def test_sent_correlation_exact_sets_and_duplicate_subjects(self) -> None:
  module = self.load_module()
  record = {"id": "first", "subject": "Subject", "body": "unrelated",
 "sentDateTime": "old", "to": [{"address": " A@example.com "},
 {"address": "b@example.com"}, {"address": "A@example.com"}],
 "cc": [{"address": "cc@example.com"}], "bcc": [{"address": "bcc@example.com"}]}
  duplicate = {**record, "id": "second"}
  for subject, to, cc, bcc, success in (
 ("Subject", ["B@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)):
   with self.subTest(subject=subject, to=to, cc=cc, bcc=bcc):
    result = module.match_sent_record([record, duplicate], subject, to, cc, bcc, None)
    self.assertEqual(result, record if success else None)
  self.assertEqual(module.match_sent_record([record, duplicate], "Subject", None, None, None,
 {"first"}), duplicate)
  client = mock.Mock(fake_path=True)
  with mock.patch.object(module, "sent_items", return_value=[]) as sent:
   with self.assertRaisesRegex(module.CliError, "not found"):
    module.verify_sent_record(client, USER, "Subject", None, None, None, 1, retries=1)
   sent.assert_called_once_with(client, USER, 1)

def main() -> None:
 parser = argparse.ArgumentParser(
 description="Run the Approval-Gated Mail standard library test suite.",
 epilog=SIGNATURE,
 )
 _, remaining = parser.parse_known_args()
 unittest.main(argv=[sys.argv[0], *remaining])

if __name__ == "__main__":
 main()

License: MIT (read it).

Back to the listing