"""End-to-end Google Calendar verification: agent-extracted event -> real push -> API readback -> cleanup. Manual-run only (needs a real per-user token and the google libs); never imported by CI. One-time token bootstrap: connect in the app (Step 2a), then DevTools -> Application -> Local Storage -> copy the `gcal_token` value into a file. python scripts/verify_gcal_e2e.py --token-file tok.json [--check-only] [--keep] """ from __future__ import annotations import argparse import json import os import sys import uuid from datetime import datetime from pathlib import Path os.environ.setdefault("USE_STUB_EXTRACTOR", "1") sys.path.insert(0, str(Path(__file__).resolve().parents[1])) from dateutil import parser as dtparser # noqa: E402 from calendar_out import gcal # noqa: E402 # The canonical appointment-confirmation sample (kept in sync with # tests/test_agent.py::CANON — copied because CI test modules aren't a package). CANON = ( "Thank you for scheduling your appointment with Primary Care of Manhattan. " "We look forward to seeing you!\n" "\n" "Date: Monday, June 22, 2026\n" "Time: 10:30 AM\n" "Duration: Approx. 30–45 min\n" "(Please arrive 15 minutes early to complete intake forms)\n" "\n" "\U0001f4cd 112A West 72nd Street\n" "New York, NY 10023\n" "(Upper West Side — 72nd & Columbus)\n" ) EXPECT_START = "2026-06-22T10:15:00" EXPECT_LOCATION = "112A West 72nd Street, New York, NY 10023" EXPECT_REMINDER = 60 _results: list[tuple[bool, str]] = [] def _check(ok: bool, label: str) -> bool: _results.append((ok, label)) print(("PASS " if ok else "FAIL ") + label) return ok def _bare_creds(token_json: str): """Access-token-only Credentials: works for API calls while the token is fresh, but cannot refresh.""" from google.oauth2.credentials import Credentials return Credentials(token=json.loads(token_json).get("token")) def _enable_bare_token_fallback(token: str) -> None: """Space-minted tokens may lack client_secret (it exists only in the Space's env, and GOOGLE_OAUTH_CLIENT_SECRET isn't set locally). google-auth then refuses to build refresh-capable creds — fall back to using the bare access token directly (valid ~1h after minting).""" try: gcal._creds_from_token_json(token) except ValueError as e: if "client_secret" not in str(e): raise print("note: token has no client_secret and none in env -> using the " "access token directly (no refresh; re-mint if it expires)") gcal._creds_from_token_json = _bare_creds def main() -> int: ap = argparse.ArgumentParser(description=__doc__) ap.add_argument("--token-file", default=os.environ.get("GCAL_TOKEN_FILE", "")) ap.add_argument("--check-only", action="store_true", help="only liveness-check the token; no event is created") ap.add_argument("--keep", action="store_true", help="leave the test event in the calendar (default: delete)") ap.add_argument("--calendar-id", default="primary") args = ap.parse_args() if not args.token_file or not Path(args.token_file).exists(): print("ERROR: pass --token-file (or set GCAL_TOKEN_FILE) pointing at the " "localStorage gcal_token JSON") return 1 token = Path(args.token_file).read_text(encoding="utf-8").strip() _enable_bare_token_fallback(token) # 1. liveness check (same call the /oauth2/check endpoint makes) res = gcal.check_token(token) if not _check(res["ok"], f"check_token: {res if not res['ok'] else 'token is live'}"): return 1 if res.get("refreshed_token"): token = res["refreshed_token"] Path(args.token_file).write_text(token, encoding="utf-8") print(" (token was refreshed; token file updated)") if args.check_only: return 0 # 2. agent extraction (stub mode = deterministic) + invariants from server.agent import run_agent plan = run_agent(CANON, now=datetime(2026, 6, 12, 9, 0)) if not _check(len(plan.events) == 1, f"agent extracted 1 event (got {len(plan.events)})"): return 1 ev = plan.events[0] _check(ev.start == EXPECT_START, f"start == {EXPECT_START} (got {ev.start})") _check(ev.location == EXPECT_LOCATION, f"location == {EXPECT_LOCATION!r} (got {ev.location!r})") _check(ev.reminder_minutes == EXPECT_REMINDER, f"reminder == {EXPECT_REMINDER} (got {ev.reminder_minutes})") # 3. push with a nonce title so readback/cleanup can never touch a real event nonce = f"e2e-{uuid.uuid4().hex[:6]}" ev.title = f"{ev.title} [{nonce}]" links = gcal.push_events_with_token(token, [ev], calendar_id=args.calendar_id) _check(bool(links and links[0]), f"push returned an event link: {links[0] if links else '-'}") # 4. read it back through the API and compare what actually landed from googleapiclient.discovery import build creds = gcal._creds_from_token_json(token) svc = build("calendar", "v3", credentials=creds) found = svc.events().list( calendarId=args.calendar_id, q=nonce, singleEvents=True, timeMin="2026-06-21T00:00:00Z", timeMax="2026-06-23T00:00:00Z", ).execute().get("items", []) if _check(len(found) == 1, f"readback found exactly 1 event (got {len(found)})"): got = found[0] _check(nonce in got.get("summary", ""), f"summary carries nonce (got {got.get('summary')!r})") _check(got.get("location") == EXPECT_LOCATION, f"location landed (got {got.get('location')!r})") want = dtparser.isoparse(gcal._dt_field(ev.start)["dateTime"]) have = dtparser.isoparse(got["start"]["dateTime"]) # compare instants — the API echoes in the calendar's zone _check(want == have or want.replace(tzinfo=None) == have.replace(tzinfo=None), f"start instant matches (sent {want.isoformat()}, got {have.isoformat()})") overrides = (got.get("reminders") or {}).get("overrides") or [] _check(any(o.get("minutes") == EXPECT_REMINDER for o in overrides), f"reminder override {EXPECT_REMINDER} min landed (got {overrides})") if not args.keep: svc.events().delete(calendarId=args.calendar_id, eventId=got["id"]).execute() print(f" (test event {got['id']} deleted)") else: print(f" (kept: {got.get('htmlLink')})") failures = [label for ok, label in _results if not ok] print(f"\n{'PASS' if not failures else 'FAIL'}: " f"{len(_results) - len(failures)}/{len(_results)} checks passed") return 0 if not failures else 1 if __name__ == "__main__": raise SystemExit(main())