"""E2: REAL Gmail send integration test. This module performs an ACTUAL authenticated send through the Gmail API (and, if configured, real XOAUTH2 SMTP) using a real Google account's OAuth2 refresh token. It is the only place in the suite that talks to Google's live servers. SAFETY / GATING: - Every test is skipped unless ALL of GMAIL_TEST_ACCOUNT_EMAIL, GMAIL_TEST_ACCOUNT_REFRESH_TOKEN, GMAIL_TEST_ACCOUNT_CLIENT_ID and GMAIL_TEST_ACCOUNT_CLIENT_SECRET are set (see app/core/config.py). CI runs without these vars will report every test in this file as SKIPPED, never as a failure or an error. - No fake/mocked transport is used here — this exercises the real google-api-python-client / smtplib code paths end-to-end. - The test account emails ITSELF (To == the same GMAIL_TEST_ACCOUNT_EMAIL), so no third party ever receives test traffic. - Subjects/bodies are tagged with a unique run id and an "[automated integration test]" marker so they are easy to find/delete in the mailbox afterward, and never resemble real bulk-sending content. HOW TO RUN LOCALLY: 1. Create/rotate a Google OAuth2 client (Desktop or Web) with the Gmail API enabled and the scopes this app requests (see gmail_service.py GMAIL_SCOPES) — https://console.cloud.google.com/apis/credentials 2. Obtain a refresh token for a real Gmail test mailbox (run the app's own /gmail/oauth/google/start flow once, or use Google's OAuth Playground). 3. Export the four env vars and run: export GMAIL_TEST_ACCOUNT_EMAIL="youraccount@gmail.com" export GMAIL_TEST_ACCOUNT_REFRESH_TOKEN="1//..." export GMAIL_TEST_ACCOUNT_CLIENT_ID="....apps.googleusercontent.com" export GMAIL_TEST_ACCOUNT_CLIENT_SECRET="...." pytest tests/test_integration_real_gmail.py -v -s """ import secrets import time import uuid from datetime import datetime, timezone import pytest from app.core.config import settings def _gmail_creds_configured() -> bool: return bool(settings.GMAIL_TEST_ACCOUNT_EMAIL and settings.GMAIL_TEST_ACCOUNT_REFRESH_TOKEN and settings.GMAIL_TEST_ACCOUNT_CLIENT_ID and settings.GMAIL_TEST_ACCOUNT_CLIENT_SECRET) requires_real_gmail = pytest.mark.skipif( not _gmail_creds_configured(), reason="GMAIL_TEST_ACCOUNT_* env vars not configured — set them to run the " "real Gmail send integration test (see module docstring).", ) pytestmark = requires_real_gmail @pytest.fixture(scope="module") def gmail_client(): """A real, authenticated GmailClient (API mode) for the configured test mailbox. Built directly from env-provided credentials — no DB row needed.""" from app.services.gmail_service import GmailClient client_config = { "client_id": settings.GMAIL_TEST_ACCOUNT_CLIENT_ID, "client_secret": settings.GMAIL_TEST_ACCOUNT_CLIENT_SECRET, } return GmailClient( account_email=settings.GMAIL_TEST_ACCOUNT_EMAIL, access_token="", # force a real refresh on first use refresh_token=settings.GMAIL_TEST_ACCOUNT_REFRESH_TOKEN, client_config=client_config, sending_method="api", ) @pytest.fixture(scope="module") def gmail_client_smtp(): from app.services.gmail_service import GmailClient client_config = { "client_id": settings.GMAIL_TEST_ACCOUNT_CLIENT_ID, "client_secret": settings.GMAIL_TEST_ACCOUNT_CLIENT_SECRET, } return GmailClient( account_email=settings.GMAIL_TEST_ACCOUNT_EMAIL, access_token="", refresh_token=settings.GMAIL_TEST_ACCOUNT_REFRESH_TOKEN, client_config=client_config, sending_method="smtp", ) def _run_tag() -> str: return f"{uuid.uuid4().hex[:8]}-{int(time.time())}" # --------------------------------------------------------------------------- # Real OAuth2 token refresh # --------------------------------------------------------------------------- class TestRealTokenRefresh: def test_refresh_access_token_against_google(self): """Exchanges the real refresh token for a fresh access token directly against Google's token endpoint — proves the credentials are valid BEFORE any send is attempted.""" from app.services.gmail_service import refresh_access_token client = {"client_id": settings.GMAIL_TEST_ACCOUNT_CLIENT_ID, "client_secret": settings.GMAIL_TEST_ACCOUNT_CLIENT_SECRET} result = refresh_access_token(client, settings.GMAIL_TEST_ACCOUNT_REFRESH_TOKEN) assert result.get("access_token"), ( "Google did not return an access_token — refresh token may be " "expired/revoked or the client id/secret do not match the token's " "original client.") assert result.get("expires_in", 0) > 0 # --------------------------------------------------------------------------- # Real connection / identity verification # --------------------------------------------------------------------------- class TestRealConnectionVerification: def test_verify_connection_reports_real_identity(self, gmail_client): out = gmail_client.verify_connection() assert out["ok"] is True, out assert out["email_address"].lower() == settings.GMAIL_TEST_ACCOUNT_EMAIL.lower() def test_get_profile_returns_real_gmail_metadata(self, gmail_client): prof = gmail_client.get_profile() assert prof["email_address"].lower() == settings.GMAIL_TEST_ACCOUNT_EMAIL.lower() assert isinstance(prof["messages_total"], int) assert prof["messages_total"] >= 0 # --------------------------------------------------------------------------- # Real send via Gmail API (users.messages.send) # --------------------------------------------------------------------------- class TestRealSendViaGmailApi: def test_send_plain_text_message_via_api(self, gmail_client): from app.services import message_builder tag = _run_tag() mime = message_builder.build_mime_message( from_email=settings.GMAIL_TEST_ACCOUNT_EMAIL, to_email=settings.GMAIL_TEST_ACCOUNT_EMAIL, # self-send only subject=f"[automated integration test] plain {tag}", text_body=f"Real send integration test (plain). Run tag: {tag}", ) out = gmail_client.send_mime_api(mime) assert out["ok"] is True, out assert out["message_id"], "Gmail did not return a real message id" assert out["thread_id"] assert "SENT" in out.get("label_ids", []) or out.get("raw_response") def test_send_html_message_with_variables_substituted_via_api(self, gmail_client): """Full real pipeline slice: VariableEngine substitution -> MIME -> real Gmail API send -> real message id, proving the substitution output Google actually accepts is not just internally self-consistent but valid RFC 5322/2045 MIME that a live mail server will take.""" from app.services import message_builder from app.services.variable_engine import GeneratedValues, VariableEngine tag = _run_tag() engine = VariableEngine({}) g = GeneratedValues( invoice_number=engine.generate_invoice_number(), amount=engine.generate_amount(), shipping=engine.generate_shipping(), invoice_date=datetime.now(timezone.utc).date().isoformat(), ) recipient = {"Email": settings.GMAIL_TEST_ACCOUNT_EMAIL, "FirstName": "Integration", "LastName": "Test", "CompanyName": "QA Corp"} raw_html = (f"

Hi %FirstName%, this is a real send integration test " f"(run {tag}). Invoice %RandomInvoiceNumber%, total %RandomTotal%.

") result = engine.substitute(raw_html, recipient, g, custom_fields={}) assert "%FirstName%" not in result.text assert "Integration" in result.text assert g.invoice_number in result.text mime = message_builder.build_mime_message( from_email=settings.GMAIL_TEST_ACCOUNT_EMAIL, to_email=settings.GMAIL_TEST_ACCOUNT_EMAIL, subject=f"[automated integration test] html {tag}", html_body=result.text, text_body=f"Real send integration test (html fallback). Run tag: {tag}", unsubscribe_url=f"{settings.BASE_URL}/unsubscribe?token=integrationtest", unsubscribe_mailto=f"{settings.GMAIL_TEST_ACCOUNT_EMAIL}?subject=unsubscribe", ) out = gmail_client.send_mime_api(mime) assert out["ok"] is True, out assert out["message_id"] def test_send_with_real_pdf_attachment_via_api(self, gmail_client): """Real WeasyPrint PDF generation -> real MIME attachment -> real Gmail API accept. Confirms the fixed multipart/mixed structure (bug #14) is actually deliverable, not just locally well-formed.""" from app.services import message_builder tag = _run_tag() pdf_bytes = message_builder.html_to_pdf( f"

Integration Test Invoice

Run tag: {tag}

") assert pdf_bytes[:5] == b"%PDF-" mime = message_builder.build_mime_message( from_email=settings.GMAIL_TEST_ACCOUNT_EMAIL, to_email=settings.GMAIL_TEST_ACCOUNT_EMAIL, subject=f"[automated integration test] pdf attachment {tag}", html_body=f"

See attached invoice. Run tag: {tag}

", text_body=f"See attached invoice. Run tag: {tag}", attachments=[{"filename": f"IntegrationTest_{tag}.pdf", "content": pdf_bytes, "mime_type": "application/pdf"}], ) out = gmail_client.send_mime_api(mime) assert out["ok"] is True, out assert out["message_id"] def test_message_appears_in_sent_folder(self, gmail_client): """After a real send, Gmail's own SENT label must be present on the message the API returned — the strongest possible proof this is a genuine delivery, not a simulated one.""" from app.services import message_builder tag = _run_tag() mime = message_builder.build_mime_message( from_email=settings.GMAIL_TEST_ACCOUNT_EMAIL, to_email=settings.GMAIL_TEST_ACCOUNT_EMAIL, subject=f"[automated integration test] sent-folder-check {tag}", text_body=f"Verifying SENT label. Run tag: {tag}", ) out = gmail_client.send_mime_api(mime) assert out["ok"] is True, out msg_id = out["message_id"] service = gmail_client._api() # brief wait for Gmail's index to settle fetched = None for _ in range(5): fetched = service.users().messages().get( userId="me", id=msg_id, format="minimal").execute() if "SENT" in (fetched.get("labelIds") or []): break time.sleep(1) assert fetched is not None assert "SENT" in (fetched.get("labelIds") or []), fetched # --------------------------------------------------------------------------- # Real send via authenticated XOAUTH2 SMTP (no password, real smtp.gmail.com) # --------------------------------------------------------------------------- class TestRealSendViaSmtp: def test_send_plain_text_message_via_xoauth2_smtp(self, gmail_client_smtp): from app.services import message_builder tag = _run_tag() mime = message_builder.build_mime_message( from_email=settings.GMAIL_TEST_ACCOUNT_EMAIL, to_email=settings.GMAIL_TEST_ACCOUNT_EMAIL, subject=f"[automated integration test] smtp {tag}", text_body=f"Real XOAUTH2 SMTP integration test. Run tag: {tag}", ) out = gmail_client_smtp.send_mime_smtp(mime) assert out["ok"] is True, out assert "250" in out.get("raw_response", "") # real SMTP accept code # --------------------------------------------------------------------------- # Real send through the full task pipeline (send_one) against the real DB + # a REAL Gmail account row (real refresh token stored, encrypted at rest) # --------------------------------------------------------------------------- @pytest.mark.usefixtures("test_app") class TestRealSendThroughFullPipeline: """Exercises send_one() end-to-end exactly as a worker would: real DB rows, real encrypted OAuthToken, real Gmail API transport — the ONLY thing not "real" is that this is a self-send test mailbox rather than production recipients. This is the strongest possible proof the whole system works, not a mockup.""" def test_send_one_task_performs_a_real_accepted_send(self, db_engine): import importlib import secrets as _secrets from datetime import timedelta from sqlalchemy.orm import Session import app.tasks.send_tasks as send_tasks_mod importlib.reload(send_tasks_mod) # rebind SyncSession to the test DB from app.models import (Campaign, CampaignRecipient, EmailTemplate, GmailAccount, OAuthToken, Recipient, RecipientList, SendJob, SubjectLine, User) tag = _run_tag() with Session(db_engine, expire_on_commit=False) as db: u = User(email=f"realgmail{_secrets.token_hex(4)}@x.com", password_hash="x" * 60, full_name="Real Gmail IT") db.add(u) db.commit() rl = RecipientList(name=f"real-{tag}", user_id=u.id) db.add(rl) db.commit() # self-send only: recipient IS the test mailbox rec = Recipient(list_id=rl.id, email=settings.GMAIL_TEST_ACCOUNT_EMAIL, first_name="Integration", last_name="Test", unsub_token=_secrets.token_urlsafe(24)) db.add(rec) db.commit() subj = SubjectLine(user_id=u.id, subject=f"[automated integration test] pipeline {tag}") db.add(subj) tpl = EmailTemplate(user_id=u.id, name=f"real-tpl-{tag}", subject_hint="hint", body_html=f"

Hi %FirstName%, real pipeline send. " f"Invoice %RandomInvoiceNumber%. Run {tag}.

") db.add(tpl) db.commit() # Real Gmail account row with the real (encrypted-at-rest) refresh token acc = GmailAccount(user_id=u.id, email=settings.GMAIL_TEST_ACCOUNT_EMAIL, status="authorized", sending_method="api", is_active=True) db.add(acc) db.commit() db.add(OAuthToken(account_id=acc.id, access_token="", refresh_token=settings.GMAIL_TEST_ACCOUNT_REFRESH_TOKEN, expires_at=datetime.now(timezone.utc) - timedelta(seconds=1))) db.commit() camp = Campaign(user_id=u.id, name=f"real-camp-{tag}", status="running", recipient_list_id=rl.id, subject_set_ids=[subj.id], template_ids=[tpl.id], account_ids=[acc.id], send_interval_seconds=0, unsubscribe_enabled=True) db.add(camp) db.commit() cr = CampaignRecipient(campaign_id=camp.id, recipient_id=rec.id, email=rec.email, status="pending", seq=1, generated_subject_id=subj.id, generated_template_id=tpl.id, generated_invoice_number="INV-REAL-0001", assigned_account_id=acc.id) db.add(cr) db.commit() job = SendJob(campaign_id=camp.id, campaign_recipient_id=cr.id, status="queued", max_attempts=4) db.add(job) db.commit() cid, crid = camp.id, cr.id # We must supply the real client_config since this account has no # oauth_config_id row — patch resolve_client_config_for_account for # this one call only, exactly the way the app resolves a server-level # default config in production. real_cfg = {"client_id": settings.GMAIL_TEST_ACCOUNT_CLIENT_ID, "client_secret": settings.GMAIL_TEST_ACCOUNT_CLIENT_SECRET} orig_resolve = send_tasks_mod.build_gmail_client_from_account def _patched_build(account, oauth_token, client_config=None): return orig_resolve(account, oauth_token, client_config=real_cfg) send_tasks_mod.build_gmail_client_from_account = _patched_build try: out = send_tasks_mod.send_one(cid, crid) finally: send_tasks_mod.build_gmail_client_from_account = orig_resolve assert out["ok"] is True, out assert out["status"] == "sent" assert out["message_id"], "no real Gmail message id returned" with Session(db_engine, expire_on_commit=False) as db: cr = db.get(CampaignRecipient, crid) assert cr.status == "sent" assert cr.msg_id == out["message_id"] assert cr.sent_at is not None