Blame for scripts/verify_gdrive.py

"""Verification harness for the Google Drive delivery path.

Every check runs against an in-memory stand-in for Drive, so the whole upload,
sharing, token-refresh and caching path is verified with no credentials, no
network, and nothing uploaded to a real account:

    venv/Scripts/python.exe -m scripts.verify_gdrive

The invariants here are the ones that would otherwise fail quietly in
production -- a second copy of every report accumulating in Drive, a report
uploaded but never shared, a link that stops resolving after a rebuild, or a
revoked grant reported as a transient error and retried forever.
"""

import json
import logging
import os
from pathlib import Path
import sys
import tempfile
import time

sys.path.insert(0, str(Path(__file__).resolve().parent.parent))

from scripts import cli  # noqa: E402
from scripts.checks import Report, banner  # noqa: E402
import core.config as config  # noqa: E402
import storage.gdrive as gdrive  # noqa: E402
from storage.gdrive import (  # noqa: E402
    API_BASE,
    TOKEN_URL,
    UPLOAD_BASE,
    DriveAuthError,
    DriveError,
    DriveStore,
    ensure_uploaded,
)

PDF_BYTES = b"%PDF-1.7\n" + b"x" * 512 + b"\n%%EOF\n"

class FakeResponse:
    """Minimal stand-in for a ``requests`` response."""

    def __init__(self, status_code=200, payload=None, text=""):
        self.status_code = status_code
        self._payload = payload
        self.text = text or (json.dumps(payload) if payload is not None else "")

    def json(self):
        if self._payload is None:
            raise ValueError("no JSON body")
        return self._payload


class FakeDrive:
    """An in-memory Drive: files with names, bytes, parents and permissions."""

    def __init__(self, token_responses=None):
        self.files = {}
        self.permissions = {}
        self.calls = []
        self.next_id = 1
        self.token_responses = list(token_responses or [])
        self.token_calls = 0

    # -- transport ---------------------------------------------------------

    def post(self, url, data=None, json=None, timeout=None, headers=None):
        """Handles the token endpoint; everything else goes through request()."""
        if url == TOKEN_URL:
            self.token_calls += 1
            if self.token_responses:
                return self.token_responses.pop(0)
            return FakeResponse(200, {"access_token": "at-%d" % self.token_calls, "expires_in": 3600})
        return self.request("POST", url, data=data, json=json, headers=headers)

    def request(self, method, url, headers=None, params=None, data=None, json=None, timeout=None):
        params = params or {}
        self.calls.append((method, url, params, headers or {}))

        if method == "GET" and url == "%s/files" % API_BASE:
            return self._list(params.get("q", ""))
        if method == "POST" and url == "%s/files" % UPLOAD_BASE:
            return self._create(data)
        if method == "PATCH" and url.startswith("%s/files/" % UPLOAD_BASE):
            return self._update(url.rsplit("/", 1)[-1], data)
        if url.endswith("/permissions"):
            file_id = url.rsplit("/", 2)[-2]
            if method == "GET":
                return FakeResponse(200, {"permissions": self.permissions.get(file_id, [])})
            if method == "POST":
                grant = dict(json or {})
                grant["id"] = "perm-%d" % (len(self.permissions.get(file_id, [])) + 1)
                self.permissions.setdefault(file_id, []).append(grant)
                return FakeResponse(200, {"id": grant["id"]})
        return FakeResponse(404, {"error": {"message": "unrouted %s %s" % (method, url)}})

    # -- behaviour ---------------------------------------------------------

    def _list(self, query):
        name = None
        if "name = '" in query:
            name = query.split("name = '", 1)[1].split("'", 1)[0]
        parent = None
        if " in parents" in query:
            parent = query.split("'", 1)[1].split("'", 1)[0] if query.startswith("'") else None
            for part in query.split(" and "):
                if part.endswith(" in parents"):
                    parent = part.split("'")[1]
        matches = [
            {"id": fid, "name": meta["name"]}
            for fid, meta in self.files.items()
            if meta["name"] == name and (parent is None or parent in meta.get("parents", []))
        ]
        return FakeResponse(200, {"files": matches})

    def _create(self, body):
        metadata = {}
        if body:
            text = body.decode("utf-8", "replace")
            start = text.find("{")
            end = text.find("}", start)
            if start >= 0 and end > start:
                metadata = json.loads(text[start:end + 1])
        file_id = "file-%d" % self.next_id
        self.next_id += 1
        self.files[file_id] = {
            "name": metadata.get("name", "unnamed"),
            "parents": metadata.get("parents", []),
            "bytes": body or b"",
        }
        return FakeResponse(200, {
            "id": file_id,
            "name": self.files[file_id]["name"],
            "webViewLink": "https://drive.google.com/file/d/%s/view" % file_id,
        })

    def _update(self, file_id, body):
        if file_id not in self.files:
            return FakeResponse(404, {"error": {"message": "no such file"}})
        self.files[file_id]["bytes"] = body or b""
        self.files[file_id]["updates"] = self.files[file_id].get("updates", 0) + 1
        return FakeResponse(200, {
            "id": file_id,
            "name": self.files[file_id]["name"],
            "webViewLink": "https://drive.google.com/file/d/%s/view" % file_id,
        })


def _store(fake, **kwargs):
    """Builds a store wired to a fake Drive."""
    options = {
        "client_id": "cid",
        "client_secret": "secret",
        "refresh_token": "rt-original",
        "folder_id": "",
        "session": fake,
    }
    options.update(kwargs)
    return DriveStore(**options)


def _write_pdf(directory: Path, body: bytes = PDF_BYTES) -> Path:
    """Writes a PDF into a directory and returns its path."""
    directory.mkdir(parents=True, exist_ok=True)
    path = directory / "WIPRO_report.pdf"
    path.write_bytes(body)
    return path


def check_configuration(r: Report) -> None:
    """Configuration and credential handling."""
    saved = {k: os.environ.pop(k, None) for k in
             ("GDRIVE_CLIENT_ID", "GDRIVE_CLIENT_SECRET", "GDRIVE_REFRESH_TOKEN", "GDRIVE_TOKEN_FILE")}
    try:
        with tempfile.TemporaryDirectory() as tmp:
            os.environ["GDRIVE_TOKEN_FILE"] = str(Path(tmp) / "token.json")
            r.check("config.not_configured_reports_false", not gdrive.credentials_present(),
                   "no credentials present")
            r.raises("config.missing_credentials_raise_auth_error", DriveAuthError,
                    lambda: DriveStore(client_id="", client_secret="", refresh_token=""))

            os.environ["GDRIVE_CLIENT_ID"] = "cid"
            os.environ["GDRIVE_CLIENT_SECRET"] = "secret"
            os.environ["GDRIVE_REFRESH_TOKEN"] = "rt"
            r.check("config.env_credentials_detected", gdrive.credentials_present(),
                   "client id, secret and refresh token seen")

            os.environ.pop("GDRIVE_REFRESH_TOKEN")
            path = gdrive.save_refresh_token("rt-from-file")
            r.check("config.token_file_is_read_back",
                   gdrive.credentials_present() and gdrive._stored_token() == "rt-from-file",
                   "token file %s supplies the grant" % Path(path).name)

            store = _store(FakeDrive(), refresh_token=None)
            r.check("config.token_file_beats_env", store.refresh_token == "rt-from-file",
                   "the persisted token is preferred over the environment")
    finally:
        for key, value in saved.items():
            if value is None:
                os.environ.pop(key, None)
            else:
                os.environ[key] = value


def check_tokens(r: Report) -> None:
    """Access-token lifecycle: the part that keeps a long-running server alive."""

    fake = FakeDrive()
    store = _store(fake)
    first = store.access_token()
    second = store.access_token()
    r.check("token.access_token_is_cached", first == second and fake.token_calls == 1,
           "two calls, one refresh (%s)" % first)

    store._expires_at = time.time() - 1
    third = store.access_token()
    r.check("token.expiry_triggers_refresh", third != first and fake.token_calls == 2,
           "expired token replaced without intervention")

    fake = FakeDrive(token_responses=[
        FakeResponse(400, {"error": "invalid_grant",
                           "error_description": "Token has been expired or revoked."}),
    ])
    store = _store(fake)
    r.raises("token.invalid_grant_is_terminal", DriveAuthError, store.access_token,
            "revoked grant surfaces as DriveAuthError")

    # A fresh store, because the rejection above consumed the queued response.
    store = _store(FakeDrive(token_responses=[
        FakeResponse(400, {"error": "invalid_grant",
                           "error_description": "Token has been expired or revoked."}),
    ]))
    message = ""
    try:
        store.access_token()
    except DriveAuthError as exc:
        message = str(exc)
    r.check("token.invalid_grant_says_how_to_fix",
           "scripts.gdrive_auth" in message and "Testing status" in message,
           "names the re-auth command and the 7-day cause")

    fake = FakeDrive(token_responses=[FakeResponse(503, {}, text="backend error")])
    store = _store(fake)
    r.raises("token.transient_failure_is_not_auth_error", DriveError, store.access_token,
            "a 503 is retryable, not a lost grant")
    fake = FakeDrive(token_responses=[FakeResponse(503, {}, text="backend error")])
    store = _store(fake)
    try:
        store.access_token()
    except DriveAuthError:
        r.check("token.transient_failure_not_misreported", False, "503 raised DriveAuthError")
    except DriveError:
        r.check("token.transient_failure_not_misreported", True,
               "DriveError, so the caller may retry")

    saved = os.environ.get("GDRIVE_TOKEN_FILE")
    with tempfile.TemporaryDirectory() as tmp:
        token_file = Path(tmp) / "token.json"
        os.environ["GDRIVE_TOKEN_FILE"] = str(token_file)
        try:
            fake = FakeDrive(token_responses=[FakeResponse(200, {
                "access_token": "at-rotated", "expires_in": 3600, "refresh_token": "rt-rotated"})])
            store = _store(fake)
            store.access_token()
            persisted = json.loads(token_file.read_text(encoding="utf-8")).get("refresh_token")
            r.check("token.rotated_refresh_token_persisted",
                   store.refresh_token == "rt-rotated" and persisted == "rt-rotated",
                   "a rotated grant survives a restart")
        finally:
            if saved is None:
                os.environ.pop("GDRIVE_TOKEN_FILE", None)
            else:
                os.environ["GDRIVE_TOKEN_FILE"] = saved

    # A 401 mid-session: one silent retry, no error surfaced to the caller.
    fake = FakeDrive()
    calls = {"n": 0}
    original = fake.request

    def flaky(method, url, **kwargs):
        calls["n"] += 1
        if calls["n"] == 1:
            return FakeResponse(401, {"error": {"message": "Invalid Credentials"}})
        return original(method, url, **kwargs)

    fake.request = flaky
    store = _store(fake)
    store.find_by_name("WIPRO_report.pdf")
    r.check("token.401_retried_once_with_a_fresh_token",
           calls["n"] == 2 and fake.token_calls == 2,
           "one retry after re-minting the access token")


def check_upload(r: Report) -> None:
    """Upload, replacement and sharing."""

    with tempfile.TemporaryDirectory() as tmp:
        pdf = _write_pdf(Path(tmp))

        fake = FakeDrive()
        store = _store(fake)
        first = store.upload(pdf)
        r.check("upload.creates_when_absent", len(fake.files) == 1 and first.file_id in fake.files,
               "one file created (%s)" % first.file_id)
        body = fake.files[first.file_id]["bytes"]
        r.check("upload.multipart_carries_metadata_and_pdf",
               b"WIPRO_report.pdf" in body and b"%PDF-1.7" in body,
               "%d bytes, metadata part and PDF part present" % len(body))

        second = store.upload(pdf)
        r.check("upload.replaces_rather_than_duplicating",
               len(fake.files) == 1 and second.file_id == first.file_id,
               "second upload reused %s" % second.file_id)
        r.check("upload.replacement_is_a_media_patch",
               fake.files[first.file_id].get("updates") == 1,
               "existing file updated in place, so the link still resolves")

        r.check("share.link_sharing_is_the_default",
               any(p.get("type") == "anyone" and p.get("role") == "reader"
                   for p in fake.permissions.get(first.file_id, [])),
               "anyone-with-link reader granted on upload")
        r.check("share.existing_grant_is_not_recreated",
               len(fake.permissions.get(first.file_id, [])) == 1,
               "one permission after two uploads")

        unshared = _store(FakeDrive())
        result = unshared.upload(pdf, share=False)
        r.check("share.can_be_declined_explicitly", result.shared is False,
               "share=False leaves the file private")

        r.check("links.preview_link_is_embeddable",
               first.preview_link == "https://drive.google.com/file/d/%s/preview" % first.file_id,
               first.preview_link)
        r.check("links.direct_link_points_at_the_file",
               first.file_id in first.direct_link and "export=download" in first.direct_link,
               first.direct_link)

        fake = FakeDrive()
        store = _store(fake, folder_id="folder-123")
        placed = store.upload(pdf)
        r.check("upload.folder_id_is_applied",
               fake.files[placed.file_id]["parents"] == ["folder-123"],
               "file parented to the configured folder")

        fake = FakeDrive()
        store = _store(fake)
        store.find_by_name("O'Brien & Sons_report.pdf")
        query = [c for c in fake.calls if c[0] == "GET"][0][2].get("q", "")
        r.check("upload.name_query_escapes_quotes", "O\\'Brien" in query,
               "an apostrophe cannot break the query")

        missing = _store(FakeDrive())
        r.raises("upload.missing_file_is_refused", DriveError,
                lambda: missing.upload(Path(tmp) / "nope.pdf"))


def check_caching(r: Report) -> None:
    """The sidecar: what stops every request re-uploading the same report."""

    original_output = config.OUTPUT_DIR
    with tempfile.TemporaryDirectory() as tmp:
        config.OUTPUT_DIR = Path(tmp)
        try:
            ticker = "WIPRO"
            pdf = _write_pdf(config.stock_dir(ticker))

            fake = FakeDrive()
            store = _store(fake)
            first = ensure_uploaded(pdf, ticker, store=store)
            sidecar = gdrive.sidecar_path(ticker)
            record = json.loads(sidecar.read_text(encoding="utf-8"))
            r.check("cache.sidecar_written", sidecar.exists() and record.get("file_id") == first.file_id,
                   "%s records the file id and links" % sidecar.name)
            r.check("cache.sidecar_records_content_hash", len(record.get("pdf_sha256", "")) == 64,
                   "identity is the PDF's own hash")

            uploads_before = len(fake.calls)
            again = ensure_uploaded(pdf, ticker, store=store)
            r.check("cache.unchanged_pdf_is_not_reuploaded",
                   len(fake.calls) == uploads_before and again.file_id == first.file_id,
                   "no Drive calls for an unchanged report")

            pdf.write_bytes(PDF_BYTES + b"revised\n")
            changed = ensure_uploaded(pdf, ticker, store=store)
            r.check("cache.changed_pdf_is_reuploaded",
                   changed.file_id == first.file_id and fake.files[first.file_id].get("updates") == 1,
                   "same file id, new bytes -- the shared link is stable")
            new_record = json.loads(sidecar.read_text(encoding="utf-8"))
            r.check("cache.hash_advances_with_the_rebuild",
                   new_record["pdf_sha256"] != record["pdf_sha256"],
                   "sidecar tracks the new content")

            forced_before = fake.files[first.file_id].get("updates")
            ensure_uploaded(pdf, ticker, store=store, force=True)
            r.check("cache.force_reuploads_identical_bytes",
                   fake.files[first.file_id].get("updates") == forced_before + 1,
                   "force=True bypasses the hash check")

            gdrive.sidecar_path(ticker).write_text("{ not json", encoding="utf-8")
            recovered = ensure_uploaded(pdf, ticker, store=store)
            r.check("cache.corrupt_sidecar_is_survivable", bool(recovered.file_id),
                   "an unreadable sidecar re-uploads instead of crashing")
        finally:
            config.OUTPUT_DIR = original_output


def main() -> int:
    """Runs every check and returns a process exit code."""
    cli.setup(logging.CRITICAL)
    banner("Google Drive delivery verification (offline)")

    report = Report()
    for title, group in (("Configuration", check_configuration),
                         ("Token handling", check_tokens),
                         ("Upload and sharing", check_upload),
                         ("Upload caching", check_caching)):
        report.section(title)
        group(report)

    print(report.render())
    return report.finish()


if __name__ == "__main__":
    sys.exit(main())