scripts/verify_gcal_e2e.py

"""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())