"""Tests for gmail-fetch-attachments.py. Covers: - Pure helpers: safe_filename, collect_attachments, load_client_creds - File I/O: load_mcp_env, load_refresh_token (tmp_path + monkeypatch on module-level constants CLAUDE_CONFIG and TOKEN_DIR) - HTTP wrappers: refresh_access_token, gmail_get (monkeypatch on urllib.request.urlopen) - Argparse: --help / missing-args via subprocess Strategy mirrors test_cmail_action.py: import the script via importlib (filename has hyphens), mock at external boundaries, no integration test for main() — the components are tested individually. """ from __future__ import annotations import importlib.util import json import subprocess import sys from pathlib import Path from unittest.mock import MagicMock import pytest SCRIPT_PATH = Path(__file__).resolve().parent.parent / "gmail-fetch-attachments.py" def _load_module(): spec = importlib.util.spec_from_file_location( "gmail_fetch_attachments", str(SCRIPT_PATH) ) mod = importlib.util.module_from_spec(spec) spec.loader.exec_module(mod) return mod @pytest.fixture(scope="module") def gfa(): return _load_module() def _mock_urlopen_response(payload): """Build a MagicMock mimicking urllib.request.urlopen()'s context-manager response.""" mock_resp = MagicMock() mock_resp.read.return_value = json.dumps(payload).encode() mock_resp.__enter__ = MagicMock(return_value=mock_resp) mock_resp.__exit__ = MagicMock(return_value=False) return mock_resp # --------------------------------------------------------------------------- # safe_filename — pure # --------------------------------------------------------------------------- class TestSafeFilename: def test_normal_clean_filename(self, gfa): assert gfa.safe_filename("report.pdf") == "report.pdf" def test_boundary_forward_slash_replaced_with_underscore(self, gfa): assert gfa.safe_filename("foo/bar.txt") == "foo_bar.txt" def test_boundary_backslash_replaced_with_underscore(self, gfa): assert gfa.safe_filename("foo\\bar.txt") == "foo_bar.txt" def test_boundary_path_traversal_stripped(self, gfa): # "../etc/passwd" -> after slash replace: ".._etc_passwd" # While loop strips leading "..": "_etc_passwd" assert gfa.safe_filename("../etc/passwd") == "_etc_passwd" def test_boundary_dotfile_preserved(self, gfa): # The fix Craig requested: single-dot prefixes survive so dotfiles # like .gitignore aren't silently renamed. assert gfa.safe_filename(".gitignore") == ".gitignore" assert gfa.safe_filename(".env.local") == ".env.local" def test_boundary_empty_string(self, gfa): assert gfa.safe_filename("") == "" @pytest.mark.parametrize("input_name,expected", [ ("..", ""), # single ".." stripped, leaves empty ("...", "."), # one strip leaves a single dot ("....", ""), # two strips leave empty (".....", "."), # two strips leave one dot ]) def test_boundary_only_dots(self, gfa, input_name, expected): assert gfa.safe_filename(input_name) == expected def test_boundary_double_dot_followed_by_name_stripped(self, gfa): assert gfa.safe_filename("..foo") == "foo" def test_boundary_middle_dotdot_preserved(self, gfa): # Only LEADING ".." gets stripped. Mid-string ".." stays. # "foo..bar" has no leading dots, so it's preserved as-is. assert gfa.safe_filename("foo..bar") == "foo..bar" # --------------------------------------------------------------------------- # collect_attachments — pure # --------------------------------------------------------------------------- class TestCollectAttachments: def test_normal_single_attachment(self, gfa): payload = { "parts": [ {"mimeType": "text/plain", "body": {"size": 100}}, {"filename": "doc.pdf", "mimeType": "application/pdf", "body": {"attachmentId": "abc123", "size": 5000}}, ] } result = gfa.collect_attachments(payload) assert result == [{ "filename": "doc.pdf", "attachmentId": "abc123", "size": 5000, "mimeType": "application/pdf", }] def test_boundary_nested_multipart_recursion(self, gfa): payload = { "parts": [ {"mimeType": "multipart/mixed", "parts": [ {"mimeType": "multipart/alternative", "parts": [ {"filename": "deep.pdf", "mimeType": "application/pdf", "body": {"attachmentId": "deep1", "size": 100}}, ]}, ]}, ] } result = gfa.collect_attachments(payload) assert len(result) == 1 assert result[0]["filename"] == "deep.pdf" assert result[0]["attachmentId"] == "deep1" def test_boundary_no_attachments_returns_empty(self, gfa): payload = { "parts": [ {"mimeType": "text/plain", "body": {"size": 100}}, {"mimeType": "text/html", "body": {"size": 200}}, ] } assert gfa.collect_attachments(payload) == [] def test_boundary_inline_image_no_filename_skipped(self, gfa): # Inline images embedded via cid: typically have an attachmentId # but no filename. The "user-visible attachments" heuristic skips # them so they don't litter the output dir as image001.png. payload = { "parts": [ {"mimeType": "image/png", "body": {"attachmentId": "inline1", "size": 500}}, ] } assert gfa.collect_attachments(payload) == [] def test_boundary_empty_filename_skipped(self, gfa): # Empty-string filename also skipped (truthy check). payload = { "parts": [ {"filename": "", "mimeType": "image/png", "body": {"attachmentId": "empty1", "size": 500}}, ] } assert gfa.collect_attachments(payload) == [] def test_boundary_filename_without_attachment_id_skipped(self, gfa): # A part with a filename but no attachmentId isn't a separately # downloadable attachment — it's inline content with a name. payload = { "parts": [ {"filename": "fake.txt", "mimeType": "text/plain", "body": {"size": 100}}, ] } assert gfa.collect_attachments(payload) == [] def test_boundary_multiple_attachments_at_different_depths(self, gfa): payload = { "parts": [ {"filename": "top.pdf", "mimeType": "application/pdf", "body": {"attachmentId": "top1", "size": 100}}, {"mimeType": "multipart/mixed", "parts": [ {"filename": "nested.txt", "mimeType": "text/plain", "body": {"attachmentId": "nested1", "size": 50}}, ]}, ] } result = gfa.collect_attachments(payload) names = sorted(r["filename"] for r in result) assert names == ["nested.txt", "top.pdf"] def test_boundary_default_mimetype_when_missing(self, gfa): payload = { "parts": [ {"filename": "x.bin", "body": {"attachmentId": "x1", "size": 10}}, ] } result = gfa.collect_attachments(payload) assert result[0]["mimeType"] == "application/octet-stream" def test_error_empty_payload(self, gfa): assert gfa.collect_attachments({}) == [] def test_error_payload_with_null_parts(self, gfa): # Defensive: parts = None falls through to empty list via `or []`. payload = {"parts": None} assert gfa.collect_attachments(payload) == [] # --------------------------------------------------------------------------- # load_client_creds — pure # --------------------------------------------------------------------------- class TestLoadClientCreds: def test_normal_both_credentials_present(self, gfa): env = {"GOOGLE_CLIENT_ID": "cid123", "GOOGLE_CLIENT_SECRET": "secret456"} assert gfa.load_client_creds(env) == ("cid123", "secret456") def test_error_missing_client_id(self, gfa): env = {"GOOGLE_CLIENT_SECRET": "secret456"} with pytest.raises(SystemExit): gfa.load_client_creds(env) def test_error_missing_client_secret(self, gfa): env = {"GOOGLE_CLIENT_ID": "cid123"} with pytest.raises(SystemExit): gfa.load_client_creds(env) def test_error_empty_client_id(self, gfa): env = {"GOOGLE_CLIENT_ID": "", "GOOGLE_CLIENT_SECRET": "secret456"} with pytest.raises(SystemExit): gfa.load_client_creds(env) def test_error_empty_client_secret(self, gfa): env = {"GOOGLE_CLIENT_ID": "cid123", "GOOGLE_CLIENT_SECRET": ""} with pytest.raises(SystemExit): gfa.load_client_creds(env) # --------------------------------------------------------------------------- # load_mcp_env — file I/O via tmp_path + monkeypatch CLAUDE_CONFIG # --------------------------------------------------------------------------- class TestLoadMcpEnv: @staticmethod def _write_config(tmp_path, monkeypatch, gfa, content): config_path = tmp_path / ".claude.json" config_path.write_text(json.dumps(content)) monkeypatch.setattr(gfa, "CLAUDE_CONFIG", config_path) return config_path def test_normal_personal_profile_with_env(self, monkeypatch, gfa, tmp_path): self._write_config(tmp_path, monkeypatch, gfa, { "mcpServers": { "google-docs-personal": { "env": {"GOOGLE_CLIENT_ID": "cid", "GOOGLE_CLIENT_SECRET": "sec"} } } }) env = gfa.load_mcp_env("personal") assert env == {"GOOGLE_CLIENT_ID": "cid", "GOOGLE_CLIENT_SECRET": "sec"} def test_boundary_server_present_no_env_key(self, monkeypatch, gfa, tmp_path): self._write_config(tmp_path, monkeypatch, gfa, { "mcpServers": {"google-docs-work": {}} }) assert gfa.load_mcp_env("work") == {} def test_boundary_env_explicitly_null(self, monkeypatch, gfa, tmp_path): # The `or {}` defends against null env. Returns empty dict, not None. self._write_config(tmp_path, monkeypatch, gfa, { "mcpServers": {"google-docs-personal": {"env": None}} }) assert gfa.load_mcp_env("personal") == {} def test_error_config_file_missing(self, monkeypatch, gfa, tmp_path): monkeypatch.setattr(gfa, "CLAUDE_CONFIG", tmp_path / "nope.json") with pytest.raises(SystemExit): gfa.load_mcp_env("personal") def test_error_server_not_in_config(self, monkeypatch, gfa, tmp_path): self._write_config(tmp_path, monkeypatch, gfa, { "mcpServers": {"google-docs-personal": {"env": {}}} }) with pytest.raises(SystemExit): gfa.load_mcp_env("work") # --------------------------------------------------------------------------- # load_refresh_token — file I/O via tmp_path + monkeypatch TOKEN_DIR # --------------------------------------------------------------------------- class TestLoadRefreshToken: @staticmethod def _setup_token(tmp_path, monkeypatch, gfa, profile=None, content=None): token_dir = tmp_path / "google-docs-mcp" token_dir.mkdir() if profile: (token_dir / profile).mkdir() token_path = token_dir / profile / "token.json" else: token_path = token_dir / "token.json" if content is not None: token_path.write_text(json.dumps(content)) monkeypatch.setattr(gfa, "TOKEN_DIR", token_dir) return token_path def test_normal_no_profile_token_at_root(self, monkeypatch, gfa, tmp_path): self._setup_token(tmp_path, monkeypatch, gfa, content={"refresh_token": "rt-root"}) assert gfa.load_refresh_token({}) == "rt-root" def test_boundary_with_profile_subdir(self, monkeypatch, gfa, tmp_path): self._setup_token(tmp_path, monkeypatch, gfa, profile="personal", content={"refresh_token": "rt-personal"}) assert gfa.load_refresh_token( {"GOOGLE_MCP_PROFILE": "personal"} ) == "rt-personal" def test_boundary_explicit_empty_profile_falls_back_to_root( self, monkeypatch, gfa, tmp_path): # GOOGLE_MCP_PROFILE="" is treated the same as the key being missing — # both fall back to TOKEN_DIR/token.json. Pinning both shapes so a # future refactor that drops `or ""` doesn't silently break this. self._setup_token(tmp_path, monkeypatch, gfa, content={"refresh_token": "rt-root"}) assert gfa.load_refresh_token({"GOOGLE_MCP_PROFILE": ""}) == "rt-root" def test_error_token_file_missing(self, monkeypatch, gfa, tmp_path): token_dir = tmp_path / "google-docs-mcp" token_dir.mkdir() monkeypatch.setattr(gfa, "TOKEN_DIR", token_dir) with pytest.raises(SystemExit): gfa.load_refresh_token({}) def test_error_no_refresh_token_field_in_file(self, monkeypatch, gfa, tmp_path): self._setup_token(tmp_path, monkeypatch, gfa, content={"access_token": "at-only"}) with pytest.raises(SystemExit): gfa.load_refresh_token({}) # --------------------------------------------------------------------------- # refresh_access_token — mocked urllib # --------------------------------------------------------------------------- class TestRefreshAccessToken: def test_normal_returns_access_token(self, monkeypatch, gfa): mock_urlopen = MagicMock( return_value=_mock_urlopen_response({"access_token": "at-new"}) ) monkeypatch.setattr(gfa.urllib.request, "urlopen", mock_urlopen) result = gfa.refresh_access_token("rt-val", "cid-val", "sec-val") assert result == "at-new" # Verify the request shape: URL, body grant_type and refresh_token. req = mock_urlopen.call_args[0][0] assert req.full_url == gfa.OAUTH_TOKEN_URL body = req.data.decode() assert "grant_type=refresh_token" in body assert "refresh_token=rt-val" in body assert "client_id=cid-val" in body def test_error_response_missing_access_token(self, monkeypatch, gfa): mock_urlopen = MagicMock( return_value=_mock_urlopen_response({"error": "invalid_grant"}) ) monkeypatch.setattr(gfa.urllib.request, "urlopen", mock_urlopen) with pytest.raises(SystemExit): gfa.refresh_access_token("rt", "cid", "sec") # --------------------------------------------------------------------------- # gmail_get — mocked urllib # --------------------------------------------------------------------------- class TestGmailGet: def test_normal_returns_parsed_json_with_bearer_header(self, monkeypatch, gfa): mock_urlopen = MagicMock( return_value=_mock_urlopen_response({"id": "msg123", "snippet": "hi"}) ) monkeypatch.setattr(gfa.urllib.request, "urlopen", mock_urlopen) result = gfa.gmail_get("/messages/msg123", "at-token") assert result == {"id": "msg123", "snippet": "hi"} req = mock_urlopen.call_args[0][0] assert req.full_url == f"{gfa.GMAIL_API}/messages/msg123" # urllib.request.Request lowercases header names except the first # char via .capitalize() → "Authorization" stays as "Authorization". assert req.headers["Authorization"] == "Bearer at-token" # --------------------------------------------------------------------------- # Argparse — black-box subprocess sanity check # --------------------------------------------------------------------------- class TestArgparseShape: def test_normal_help_lists_all_required_args(self): result = subprocess.run( [sys.executable, str(SCRIPT_PATH), "--help"], capture_output=True, text=True, ) assert result.returncode == 0 for flag in ("--profile", "--message-id", "--output-dir"): assert flag in result.stdout def test_error_no_args_exits_nonzero(self): result = subprocess.run( [sys.executable, str(SCRIPT_PATH)], capture_output=True, text=True, ) assert result.returncode != 0