class TestGmailClientSendApi: def _client(self): cfg = {"client_id": "c", "client_secret": "s"} with patch.object(GmailClient, "_ensure_valid"): c = GmailClient("acct@gmail.com", "AT", "RT", cfg, sending_method="api") return c def test_send_mime_api_returns_provider_response(self): c = self._client() from email.mime.text import MIMEText msg = MIMEText("body", "html") msg["From"] = "acct@gmail.com" msg["To"] = "to@x.com" sent = {"id": "MSGID123", "threadId": "TH123", "labelIds": ["SENT"]} fake_service = MagicMock() fake_service.users().messages().send.return_value.execute.return_value = sent c._service = fake_service out = c.send_mime_api(msg) assert out["ok"] is True assert out["message_id"] == "MSGID123" assert out["thread_id"] == "TH123" # the raw MIME was base64url-encoded in the API body body = fake_service.users().messages().send.call_args[1]["body"] assert set(body) == {"raw"} def test_send_mime_smtp_xoauth2(self): c = self._client() c._creds = MagicMock() c._creds.token = "ACCESSTOKEN" c._creds.valid = True from email.mime.text import MIMEText msg = MIMEText("x") msg["To"] = "to@x.com" recorded = {} class FakeSMTP: def __init__(self, host, port, timeout=30): recorded["host"], recorded["port"] = host, port def __enter__(self): return self def __exit__(self, *a): return False def ehlo(self, *a): return (250, b"ok") def starttls(self, context=None): recorded["starttls"] = True def docmd(self, cmd, arg=""): recorded["auth"] = (cmd, arg) return (235, b"accepted") def sendmail(self, frm, to, data): recorded["sent"] = (frm, to, data) return {} # none refused import smtplib with patch.object(gs, "smtplib", types.SimpleNamespace(SMTP=FakeSMTP)) if False else patch("smtplib.SMTP", FakeSMTP): out = c.send_mime_smtp(msg) assert out["ok"] is True assert recorded["host"] == "smtp.gmail.com" and recorded["port"] == 587 assert recorded["starttls"] is True cmd, arg = recorded["auth"] assert cmd == "AUTH" and arg.startswith("XOAUTH2 ") import base64 decoded = base64.b64decode(arg.split(" ", 1)[1]).decode() assert decoded == "user=acct@gmail.com\x01auth=Bearer ACCESSTOKEN\x01\x01" frm, to, data = recorded["sent"] assert frm == "acct@gmail.com" and to == ["to@x.com"] def test_send_mime_smtp_refused_recipient(self): c = self._client() c._creds = MagicMock() c._creds.token = "T" c._creds.valid = True from email.mime.text import MIMEText msg = MIMEText("x") msg["To"] = "refused@x.com" class FakeSMTPRefused: def __init__(self, *a, **k): pass def __enter__(self): return self def __exit__(self, *a): return False def ehlo(self, *a): return (250, b"ok") def starttls(self, context=None): pass def docmd(self, cmd, arg=""): return (235, b"ok") def sendmail(self, frm, to, data): return {"refused@x.com": (550, b"no such user")} # refused! with patch("smtplib.SMTP", FakeSMTPRefused): out = c.send_mime_smtp(msg) # refused recipients = permanent failure, NO fake success assert out["ok"] is False assert out["error_class"] == "permanent" def test_verify_connection_ok(self): c = self._client() fake_service = MagicMock() fake_service.users().getProfile.return_value.execute.return_value = { "emailAddress": "acct@gmail.com", "historyId": "1", "messagesTotal": 10, "threadsTotal": 2} c._service = fake_service v = c.verify_connection() assert v["ok"] is True and v["email_address"] == "acct@gmail.com" def test_verify_connection_failure_reports_honestly(self): c = self._client() fake_service = MagicMock() fake_service.users().getProfile.return_value.execute.side_effect = RuntimeError("401 invalid grant") c._service = fake_service v = c.verify_connection() assert v["ok"] is False assert "401" in v["detail"] def test_invalid_method_falls_back_to_api(self): with patch.object(GmailClient, "_ensure_valid"): c = GmailClient("a@x.com", "T", "R", {"client_id": "c", "client_secret": "s"}, sending_method="carrier-pigeon") assert c.method == "api"