From e2d2c656b46c7c885c9acf7cd24a48f12e048500 Mon Sep 17 00:00:00 2001 From: Thor Whalen <1906276+thorwhalen@users.noreply.github.com> Date: Wed, 19 Aug 2026 12:48:13 +0100 Subject: [PATCH 1/2] feat: authenticated file access (drive=), silent-HTML guard, GDFiles Three related gaps that together made private-Drive ingest impossible. Authenticated single-file fetch (#5) get_bytes gains a keyword-only drive=. With it, the download goes through the authenticated API (_download_via_api -> GetContentIOBuffer, streamed and joined; byte-exact, no temp file). Without it, behaviour is unchanged. get_bytes also now accepts a bare file id, not just a URL. Silent-HTML failure (#6) Drive serves its sign-in interstitial with HTTP 200, so the public path was returning the login page as if it were the file -- plausible-looking bytes that only fail later, in whatever parses them. _looks_like_html now sniffs the content type and body prefix, and NotPubliclyShared is raised with an actionable message. allow_html=True opts out for genuine HTML downloads. GDFiles (#7) A file-level Mapping keyed by file id or Drive URL (normalised through _extract_file_id), values bytes over the authenticated API. Iteration requires a folder_url scope -- listing a whole Drive is unbounded and paginated, so __iter__/__len__ raise NotImplementedError naming the fix rather than silently paginating. Lookup works either way. __contains__ answers from a metadata probe, never a download; a genuine 404 becomes KeyError while 5xx/auth failures propagate as themselves. Metadata without downloading get_metadata(url, drive=...) plus GDFiles.metadata: title/fileSize/mimeType/ modifiedDate, so a caller can decide whether an 18MB fetch is worth it. fileSize is coerced to int; absent for Google-native files. Also - GDReader.__getitem__ went through GetContentString(...).encode('latin-1'), which decodes as utf-8 first and so corrupted (or raised on) every real binary. It now shares _download_via_api. - Folder traversal extracted to _iter_folder_files, shared by GDFiles and GDReader instead of duplicated. - .gitignore covers service-account keys, client_secrets.json, tokens. - README: private-file quickstart, GDFiles, and the service-account setup steps (including sharing the folder with the account's client_email). 48 new tests, all offline: a fake GoogleDrive and a fake requests.Session cover the HTML guard, the authenticated paths, metadata, and GDFiles. Closes #5 Closes #6 Closes #7 Claude-Session: https://claude.ai/code/session_01Vb8moaBd5Yg8b4Abo77nHf --- .gitignore | 15 + README.md | 116 ++++- pydrivedol/__init__.py | 24 +- pydrivedol/base.py | 471 +++++++++++++++-- pydrivedol/tests/test_authenticated_access.py | 482 ++++++++++++++++++ 5 files changed, 1055 insertions(+), 53 deletions(-) create mode 100644 pydrivedol/tests/test_authenticated_access.py diff --git a/.gitignore b/.gitignore index b7ce764..0a1bc34 100644 --- a/.gitignore +++ b/.gitignore @@ -113,3 +113,18 @@ venv.bak/ # PyCharm .idea + +# Google credentials — NEVER commit these +client_secrets.json +client_secret*.json +credentials.json +settings.yaml +token.json +token.pickle +mycreds.txt +*-key.json +*_key.json +*service*account*.json +*serviceaccount*.json +sa-key*.json +.credentials/ diff --git a/README.md b/README.md index f56c402..5daa406 100644 --- a/README.md +++ b/README.md @@ -10,6 +10,8 @@ ## Features - 🚀 **Simple public downloads** - No API setup required for public files +- 🔒 **Private files too** - Pass an authenticated `drive=` to reach files shared only with you +- 🗂️ **Files as dict** - `GDFiles` keyed by file id *or* Drive URL - 📂 **Folder as dict** - Browse folders with `dict`-like interface - 💾 **Read/write operations** - Full CRUD support through mapping protocol - 🔄 **Recursive traversal** - Control depth with `max_levels` @@ -55,6 +57,62 @@ get_bytes(url, local_path="/path/to/save.pdf") get_bytes(url, use_cache=True) # Uses ~/.cache/pydrivedol/cached/ ``` +If the file is **not** publicly shared, Google answers with its HTML sign-in page — under +HTTP 200, so it looks like a successful download. pydrivedol refuses to hand that back as +file content and raises `NotPubliclyShared` instead, telling you to authenticate: + +```python +from pydrivedol import get_bytes, NotPubliclyShared + +try: + content = get_bytes(url) +except NotPubliclyShared: + content = get_bytes(url, drive=drive) # see below +``` + +(If you really are downloading an HTML file, pass `allow_html=True`.) + +### Private Files (Requires API Setup) + +Authenticate once, then pass the `drive` wherever bytes are needed: + +```python +from pydrivedol import drive_from_service_account, get_bytes, get_metadata, GDFiles + +drive = drive_from_service_account("service-account-key.json") + +# Is it worth downloading? Metadata is cheap — it fetches no content. +info = get_metadata(url, drive=drive) +info["title"], info["fileSize"], info["mimeType"], info["modifiedDate"] +# ('client_export.xlsx', 18512345, 'application/vnd...sheet', '2026-08-01T12:00:00.000Z') + +# Same get_bytes, now authenticated — local_path= and use_cache= work as before +content = get_bytes(url, drive=drive) +``` + +`fileSize` comes back as an `int` (Drive sends it as a string), and is **absent** for +Google-native files — Sheets/Docs/Slides have no stored byte size. + +### Files as a Mapping: `GDFiles` + +When files arrive as *links* rather than as a folder listing, key by the link: + +```python +files = GDFiles(drive) + +content = files[url] # a Drive file URL... +content = files[file_id] # ...or the bare id: same entry +url in files # metadata probe, never a download +files.metadata(url) # name / size / mimeType / modifiedDate + +# Iteration needs a scope — an unscoped GDFiles addresses the whole Drive +scoped = GDFiles(drive, folder_url=folder_url) +list(scoped) # file ids +``` + +Unscoped, `iter()` and `len()` raise `NotImplementedError` (with a message naming +`folder_url=`) rather than silently paginating your entire Drive. Lookup works either way. + ### Working with Folders (Requires API Setup) ```python @@ -151,6 +209,40 @@ reader = GDReader(folder_url) That's it! You only need to authenticate once. +### Headless / server: use a service account instead + +The browser flow above is unusable from a script, a server, or an agent. For those, create a +**service account** — a robot identity with its own key file and no interactive login: + +1. [Google Cloud Console](https://console.cloud.google.com/) → your project → + "APIs & Services" → "Credentials" +2. "Create Credentials" → **Service account**. Any name; no roles needed. +3. Open the new service account → "Keys" → "Add key" → "Create new key" → **JSON**. The key + file downloads once and cannot be re-downloaded. +4. Copy the service account's **`client_email`** (it looks like + `something@your-project.iam.gserviceaccount.com`). +5. In Google Drive, **share the file or folder with that `client_email`** — Viewer to read, + Editor to write. This is the step people forget: the service account is a separate + identity, and a file shared with *you* is not shared with *it*. +6. Point pydrivedol at the key file: + +```python +from pydrivedol import drive_from_service_account + +drive = drive_from_service_account("service-account-key.json") +``` + +Keep the key file out of version control (this repo's `.gitignore` covers the usual names) and +out of the repo entirely if you can — read its path from an environment variable. + +Read-only by token, if you want the extra guarantee: + +```python +drive = drive_from_service_account( + key_file, scopes=("https://www.googleapis.com/auth/drive.readonly",) +) +``` + ## Advanced Usage ### Control Recursion Depth @@ -281,11 +373,14 @@ pydrivedol follows the `dol` package patterns: ``` Helper Functions - └─ get_bytes() # Simple public downloads (no API) - -API-Based Classes - └─ GDReader (Mapping) # Read-only folder access - └─ GDStore (MutableMapping) # Read-write folder access + └─ get_bytes(url) # Simple public downloads (no API) + └─ get_bytes(url, drive=...) # Authenticated: reaches private files + └─ get_metadata(url, drive=...) # Name/size/mimeType/date, no download + +API-Based Classes + └─ GDFiles (Mapping) # Files, keyed by file id or file URL + └─ GDReader (Mapping) # Read-only folder access, keyed by path + └─ GDStore (MutableMapping) # Read-write folder access ``` **Design Principles:** @@ -410,6 +505,17 @@ pip install pydrive2 2. Verify folder sharing settings 3. Re-authenticate: delete saved credentials and run again +### `NotPubliclyShared` from `get_bytes` + +The file is not shared "anyone with the link", so the unauthenticated endpoint cannot reach +it. Either make it public, or authenticate and pass `drive=` (see *Private Files* above). For +a service account, remember the file/folder must be shared with the account's `client_email`. + +### Downloaded file won't open / `BadZipFile` on an xlsx + +You are almost certainly holding Google's sign-in page rather than the file. Recent versions +raise `NotPubliclyShared` for exactly this; if you are pinned to an older one, upgrade. + ### Tests are skipped Check environment variables: diff --git a/pydrivedol/__init__.py b/pydrivedol/__init__.py index e745cfc..46b58fd 100644 --- a/pydrivedol/__init__.py +++ b/pydrivedol/__init__.py @@ -5,6 +5,9 @@ Key Features: - get_bytes(url): Download bytes from public Google Drive URLs (no API setup needed) +- get_bytes(url, drive=...): Same, but authenticated -- reaches *private* files +- get_metadata(url, drive=...): Name/size/mimeType/modifiedDate without downloading +- GDFiles: Read-only Mapping of files, keyed by file id or file URL - GDReader: Read-only Mapping interface to a Google Drive folder - GDStore: Read-write MutableMapping interface to a Google Drive folder @@ -17,11 +20,24 @@ Usage: - Simple download (no API needed) + Simple download (no API needed) -- public files only. If the file is not public, + Google answers with its HTML sign-in page (under HTTP 200); pydrivedol raises + NotPubliclyShared rather than handing that page back as the file's content. >>> content = get_bytes(url) # doctest: +SKIP >>> path = get_bytes(url, local_path=True) # doctest: +SKIP + Private files: authenticate once, then pass the drive around + + >>> drive = drive_from_service_account('service-account-key.json') # doctest: +SKIP + >>> get_metadata(url, drive=drive)['fileSize'] # decide before downloading # doctest: +SKIP + >>> content = get_bytes(url, drive=drive) # doctest: +SKIP + + Files as a Mapping, keyed by file id or URL + + >>> files = GDFiles(drive) # doctest: +SKIP + >>> content = files[url] # doctest: +SKIP + Read folder >>> reader = GDReader(folder_url) # doctest: +SKIP @@ -43,8 +59,12 @@ """ from pydrivedol.base import ( - get_bytes, # simple download function + get_bytes, # simple download function (authenticated when given drive=) + NotPubliclyShared, # raised when a public download returns Google's sign-in page # Needing API setup: + get_metadata, # a file's name/size/mimeType/modifiedDate, without downloading it + DEFAULT_METADATA_FIELDS, # the metadata fields get_metadata asks for by default + GDFiles, # read-only Mapping of files, keyed by file id or file URL GDReader, # read-only Mapping interface to a Google Drive folder GDStore, # read-write MutableMapping interface to a Google Drive folder drive_from_service_account, # headless auth: build a GoogleDrive from an SA key diff --git a/pydrivedol/base.py b/pydrivedol/base.py index 0448079..97d060b 100644 --- a/pydrivedol/base.py +++ b/pydrivedol/base.py @@ -1,5 +1,15 @@ """ -Base objects for pydrivedol: simple download function, GDReader, GDStore. +Base objects for pydrivedol: download functions, GDFiles, GDReader, GDStore. + +Two access levels, both returning ``bytes``: + +- **Public, no setup**: :func:`get_bytes` against Google's unauthenticated download endpoint. + It raises :class:`NotPubliclyShared` rather than handing back the HTML sign-in page Drive + serves (with HTTP 200) when the file is not shared publicly. +- **Authenticated**: pass a PyDrive2 ``GoogleDrive`` -- from :func:`drive_from_service_account` + for headless use -- as ``drive=`` to :func:`get_bytes` / :func:`get_metadata`, or build a + :class:`GDFiles` (file-level Mapping, keyed by file id or URL), :class:`GDReader` / + :class:`GDStore` (folder-level, keyed by relative path). """ @@ -23,6 +33,21 @@ _PYDRIVE2_AVAILABLE = False +# ============================================================================= +# Errors +# ============================================================================= + + +class NotPubliclyShared(RuntimeError): + """Raised when an unauthenticated download returns a sign-in page instead of file content. + + Google serves its HTML sign-in / permission interstitial with HTTP **200**, so without an + explicit check that page body would be returned as if it were the file: plausible-looking + bytes that only blow up much later, in whatever tries to parse them. Catch this to fall + back to an authenticated fetch (``get_bytes(url, drive=...)``). + """ + + # ============================================================================= # Helper Functions # ============================================================================= @@ -59,6 +84,48 @@ def _extract_folder_id(url: str) -> Optional[str]: return match.group(1) if match else None +#: A bare Drive file id: URL-safe base64-ish, and long enough not to be a typo'd URL. +_FILE_ID_PATTERN = re.compile(r"^[A-Za-z0-9_-]{10,}$") + + +def _resolve_file_id(url_or_id: str) -> str: + """Normalise a file key -- a Drive file URL *or* a bare file id -- to a file id. + + URLs are parsed with :func:`_extract_file_id`; anything else is accepted only if it looks + like a Drive id, so that a malformed URL still fails loudly instead of being sent to the + API as a nonsense id. + + >>> _resolve_file_id('https://drive.google.com/file/d/1AbCdEfGhIjKlMnOp/view') + '1AbCdEfGhIjKlMnOp' + >>> _resolve_file_id('1AbCdEfGhIjKlMnOp') + '1AbCdEfGhIjKlMnOp' + """ + file_id = _extract_file_id(url_or_id) + if file_id: + return file_id + if _FILE_ID_PATTERN.match(url_or_id): + return url_or_id + raise ValueError( + f"Not a Google Drive file URL or file id: {url_or_id!r}. Expected something like " + "'https://drive.google.com/file/d//view' or the bare ''." + ) + + +def _is_not_found(error: BaseException) -> bool: + """Whether a Drive API error means "no such file" rather than a transient failure. + + Used to keep the Mapping contract honest: a genuinely missing file becomes ``KeyError``, + while a 5xx / auth / network failure is left to propagate as itself. + + >>> _is_not_found(Exception('')) + True + >>> _is_not_found(Exception('')) + False + """ + text = str(error).lower() + return "404" in text or "not found" in text or "notfound" in text + + def _resolve_cache_dir(use_cache: Union[bool, str]) -> Optional[str]: """Get cache directory path, creating if needed.""" if use_cache is False: @@ -76,6 +143,55 @@ def _get_cached_path(file_id: str, cache_dir: str) -> str: return os.path.join(cache_dir, file_id) +# ============================================================================= +# HTML-interstitial detection (the "silent success" guard) +# ============================================================================= + +#: Content type Drive declares for its sign-in / confirmation pages. +_HTML_CONTENT_TYPE = "text/html" +#: Document prefixes that mark a payload as an HTML page rather than file bytes. +_HTML_BODY_PREFIXES = (b" bool: + """Whether a downloaded payload is an HTML page rather than raw file bytes. + + True when the declared content type is HTML, or when the body itself opens with an HTML + document prefix. Leading whitespace and a UTF-8 BOM are tolerated; matching is + case-insensitive. + + >>> _looks_like_html(b'id,name', 'text/csv') + False + >>> _looks_like_html(b' ...') + True + >>> _looks_like_html(b'anything at all', 'text/html; charset=utf-8') + True + """ + if _HTML_CONTENT_TYPE in content_type.lower(): + return True + head = content[:_HTML_SNIFF_NBYTES].lstrip(b"\xef\xbb\xbf").lstrip().lower() + return head.startswith(_HTML_BODY_PREFIXES) + + +def _not_publicly_shared_message(file_id: str) -> str: + """Compose the actionable error text for an HTML payload from the public endpoint.""" + return ( + f"Google Drive returned an HTML page, not file content, for file id {file_id!r}. " + "That page is Google's sign-in / permission interstitial, served with HTTP 200, so the " + "download 'succeeded' with the wrong bytes. It means the file is not shared publicly " + "and the unauthenticated endpoint cannot reach it (it can also be Drive's large-file " + "confirmation page).\n" + "Fix -- authenticate and pass a drive= client:\n" + " from pydrivedol import drive_from_service_account, get_bytes\n" + " drive = drive_from_service_account('service-account-key.json')\n" + " content = get_bytes(url, drive=drive)\n" + "...having shared the file (or its folder) with the service account's client_email.\n" + "If you really are downloading an HTML file, pass allow_html=True." + ) + + # ============================================================================= # Simple Download (No API Required) # ============================================================================= @@ -86,48 +202,71 @@ def get_bytes( *, local_path: Union[bool, str] = False, use_cache: Union[bool, str] = False, + drive=None, + allow_html: bool = False, ) -> Union[bytes, str]: """ - Download bytes from public Google Drive URL. + Download bytes from a Google Drive URL. - Works with publicly shared files, no API setup required. + Without ``drive`` this uses Google's public download endpoint -- no API setup, but it only + reaches files shared "anyone with the link", and it raises :class:`NotPubliclyShared` if + Drive answers with its sign-in page instead of the file. Pass an authenticated ``drive`` + (see :func:`drive_from_service_account`) to reach **private** files shared with that + identity. Args: - url: Google Drive shared link + url: Google Drive file link (a bare file id is accepted too). local_path: False (return bytes), True (save to temp), or str (save to path) use_cache: False (no cache), True (use ~/.cache/pydrivedol/cached/), or str (use dir) + drive: an authenticated PyDrive2 ``GoogleDrive``. When given, the download goes through + the API; ``None`` (default) keeps the public, unauthenticated behaviour exactly. + allow_html: by default an HTML payload from the *public* endpoint raises, because it is + Google's login page masquerading as file content. Set True only when the file you + are downloading genuinely is HTML. Ignored on the authenticated path. Returns: bytes if local_path is False or str, filepath str if local_path is True + Raises: + NotPubliclyShared: the public endpoint returned a sign-in page; pass ``drive=``. + >>> content = get_bytes(url) # doctest: +SKIP >>> path = get_bytes(url, local_path=True) # doctest: +SKIP >>> content = get_bytes(url, local_path='/tmp/file.txt') # doctest: +SKIP + + A private file, shared with a service account: + + >>> drive = drive_from_service_account('service-account-key.json') # doctest: +SKIP + >>> content = get_bytes(private_url, drive=drive) # doctest: +SKIP """ - file_id = _extract_file_id(url) - if not file_id: - raise ValueError(f"Could not extract file ID from URL: {url}") + file_id = _resolve_file_id(url) # Check cache cache_dir = _resolve_cache_dir(use_cache) - if cache_dir: - cached_file = _get_cached_path(file_id, cache_dir) - if os.path.exists(cached_file): - content = Path(cached_file).read_bytes() - return _handle_local_path_output(content, local_path, cached_file) + cached_file = _get_cached_path(file_id, cache_dir) if cache_dir else None + if cached_file and os.path.exists(cached_file): + content = Path(cached_file).read_bytes() + return _handle_local_path_output(content, local_path, cached_file) # Download - content = _download_from_drive(file_id) + if drive is not None: + content = _download_via_api(drive, file_id) + else: + content = _download_from_drive(file_id, allow_html=allow_html) # Cache if requested - if cache_dir: + if cached_file: Path(cached_file).write_bytes(content) return _handle_local_path_output(content, local_path, None) -def _download_from_drive(file_id: str) -> bytes: - """Download file content from Google Drive.""" +def _download_from_drive(file_id: str, *, allow_html: bool = False) -> bytes: + """Download file content from Google Drive's public (unauthenticated) endpoint. + + Guards against the silent-success failure: Drive answers an unauthorised request with its + HTML sign-in page under HTTP 200, which would otherwise be returned as the file's content. + """ download_url = f"https://drive.google.com/uc?export=download&id={file_id}" session = requests.Session() @@ -144,7 +283,13 @@ def _download_from_drive(file_id: str) -> bytes: if response.status_code != 200: raise RuntimeError(f"Download failed. Status: {response.status_code}") - return response.content + content = response.content + if not allow_html and _looks_like_html( + content, response.headers.get("Content-Type", "") + ): + raise NotPubliclyShared(_not_publicly_shared_message(file_id)) + + return content def _handle_local_path_output( @@ -233,6 +378,125 @@ def drive_from_service_account( return GoogleDrive(gauth) +# ============================================================================= +# Authenticated access (needs a ``drive``) +# ============================================================================= + +#: Drive's mimeType for a folder -- the marker that a listing entry should be recursed into. +_FOLDER_MIMETYPE = "application/vnd.google-apps.folder" + +#: Metadata fields fetched by default: enough to decide whether a download is worth making. +DEFAULT_METADATA_FIELDS = ( + "id", + "title", + "mimeType", + "fileSize", + "modifiedDate", + "alternateLink", +) + +#: Metadata the Drive API returns as decimal strings but that are natural numbers. +_INTEGER_METADATA_FIELDS = ("fileSize", "quotaBytesUsed", "version") + + +def _download_via_api(drive, file_id: str) -> bytes: + """Download a file's content through the authenticated Drive API. + + Streams the content chunk by chunk into memory via PyDrive2's ``GetContentIOBuffer`` -- + byte-exact (no text decode/encode round trip, which corrupts binaries) and with no + temporary file on disk. + + ``drive`` is injected (a PyDrive2 ``GoogleDrive``) so this stays pure and unit-testable. + """ + gfile = drive.CreateFile({"id": file_id}) + return b"".join(chunk for chunk in gfile.GetContentIOBuffer() if chunk) + + +def _normalized_metadata(gfile, fields) -> dict: + """Pick ``fields`` out of a fetched PyDrive2 file, coercing numeric strings to ``int``.""" + metadata = {field: gfile[field] for field in fields if field in gfile} + for field in _INTEGER_METADATA_FIELDS: + value = metadata.get(field) + if isinstance(value, str) and value.isdigit(): + metadata[field] = int(value) + return metadata + + +def get_metadata(url_or_id: str, *, drive, fields=DEFAULT_METADATA_FIELDS) -> dict: + """Fetch a Drive file's metadata **without downloading its content**. + + The cheap half of :func:`get_bytes`: use it to decide *whether* to download -- an 18MB + spreadsheet is not something you fetch just to learn its name. + + Args: + url_or_id: a Drive file URL or a bare file id. + drive: an authenticated PyDrive2 ``GoogleDrive`` (see + :func:`drive_from_service_account`). + fields: which metadata fields to request; ``None`` fetches everything Drive offers. + The default, :data:`DEFAULT_METADATA_FIELDS`, covers name (``title``), ``fileSize``, + ``mimeType`` and ``modifiedDate``. + + Returns: + A plain ``dict`` of the requested fields that the file actually has. ``fileSize`` is + returned as an ``int`` (Drive sends it as a string), and is **absent** for + Google-native files -- Sheets/Docs/Slides have no stored byte size. + + >>> drive = drive_from_service_account('service-account-key.json') # doctest: +SKIP + >>> info = get_metadata(url, drive=drive) # doctest: +SKIP + >>> info['title'], info['fileSize'] # doctest: +SKIP + ('client_export.xlsx', 18512345) + """ + file_id = _resolve_file_id(url_or_id) + gfile = drive.CreateFile({"id": file_id}) + if fields: + gfile.FetchMetadata(fields=",".join(fields)) + wanted = fields + else: + gfile.FetchMetadata(fetch_all=True) + wanted = tuple(gfile.keys()) + return _normalized_metadata(gfile, wanted) + + +def _iter_folder_files( + drive, + folder_id: str, + *, + max_levels: Optional[int] = None, + include_hidden: bool = False, + prefix: str = "", + level: int = 0, +): + """Yield ``(relative_path, file_id)`` for every file under ``folder_id``. + + Recurses into subfolders up to ``max_levels`` (``None`` = unlimited, ``0`` = this folder + only). ``drive`` is injected so the traversal is shared by :class:`GDFiles` and + :class:`GDReader` rather than duplicated in each. + """ + if max_levels is not None and level > max_levels: + return + + query = f"'{folder_id}' in parents and trashed=false" + for item in drive.ListFile({"q": query}).GetList(): + name = item["title"] + if not include_hidden and name.startswith("."): + continue + + item_path = os.path.join(prefix, name) if prefix else name + + if item["mimeType"] == _FOLDER_MIMETYPE: + if max_levels is None or level < max_levels: + yield from _iter_folder_files( + drive, + item["id"], + max_levels=max_levels, + include_hidden=include_hidden, + prefix=item_path, + level=level + 1, + ) + else: + yield (item_path, item["id"]) + + # Office → Google-native editor MIME types (the target when ``convert=True``). GOOGLE_MIME = { ".xlsx": "application/vnd.google-apps.spreadsheet", @@ -299,6 +563,134 @@ def _upload_converting( return gfile +_UNSCOPED_ITERATION_MESSAGE = ( + "GDFiles has no folder scope, so it cannot be listed: an unscoped instance addresses the " + "whole Drive, which is unbounded and paginated. Lookup (files[key], key in files, " + "files.get(key)) works without a scope; to iterate, construct with " + "GDFiles(drive, folder_url='https://drive.google.com/drive/folders/')." +) + + +class GDFiles(Mapping): + """Read-only Mapping of Google Drive **files**, keyed by file id or file URL. + + The file-level sibling of :class:`GDReader`. Where ``GDReader`` is scoped to a folder and + keyed by relative path, ``GDFiles`` is keyed by whatever identifies a single file: a bare + file id, or any Drive file URL -- both normalise to the same key through + :func:`_extract_file_id`, so ``files[url]`` and ``files[file_id]`` are one entry. Values + are ``bytes``, fetched over the authenticated API, so **private** files work as long as + they are shared with the authenticated identity. + + That is the shape a caller has when files arrive as *links* -- the usual Drive sharing + idiom -- rather than as a folder listing. + + ``drive`` is required and injected: a file-level view is pointless without auth, since its + whole reason to exist is reaching files the public endpoint cannot. + + **Iteration requires a scope.** Unscoped, this mapping covers the entire Drive, which is + unbounded and paginated; enumerating it is never what a caller wants, so offering it would + be a trap. ``__iter__``/``__len__`` therefore raise ``NotImplementedError`` unless + ``folder_url`` is given, in which case they yield that folder's file **ids** (honouring + ``max_levels`` and ``include_hidden``, the same traversal :class:`GDReader` uses). Lookup + always works, scoped or not -- it is the primary use case and needs no listing. + + >>> drive = drive_from_service_account('service-account-key.json') # doctest: +SKIP + >>> files = GDFiles(drive) # doctest: +SKIP + >>> files.metadata(url)['fileSize'] # cheap: no download # doctest: +SKIP + >>> content = files[url] # or files[file_id] # doctest: +SKIP + >>> url in files # metadata probe, not a download # doctest: +SKIP + True + + Scoped, so it can be listed: + + >>> scoped = GDFiles(drive, folder_url=folder_url) # doctest: +SKIP + >>> list(scoped) # file ids # doctest: +SKIP + """ + + def __init__( + self, + drive, + *, + folder_url: Optional[str] = None, + max_levels: Optional[int] = None, + include_hidden: bool = False, + ): + """ + Initialize a file-level mapping. + + Args: + drive: an authenticated PyDrive2 ``GoogleDrive`` (see + :func:`drive_from_service_account`). Required. + folder_url: optional Drive folder URL scoping ``__iter__``/``__len__``. Without it + those raise; lookup is unaffected. + max_levels: recursion depth for the scope (None=infinite, 0=files only) + include_hidden: include files starting with '.' + """ + self._drive = drive + self.folder_url = folder_url + self.folder_id = _extract_folder_id(folder_url) if folder_url else None + if folder_url and not self.folder_id: + raise ValueError(f"Invalid folder URL: {folder_url}") + self.max_levels = max_levels + self.include_hidden = include_hidden + self._file_cache = None + + @property + def _file_ids(self): + """Cached file ids of the scoping folder; raises if there is no scope.""" + if self.folder_id is None: + raise NotImplementedError(_UNSCOPED_ITERATION_MESSAGE) + if self._file_cache is None: + listing = _iter_folder_files( + self._drive, + self.folder_id, + max_levels=self.max_levels, + include_hidden=self.include_hidden, + ) + # dict.fromkeys: dedupe (Drive files can have several parents) keeping order + self._file_cache = list(dict.fromkeys(file_id for _, file_id in listing)) + return self._file_cache + + def _refresh_cache(self): + """Forget the cached folder listing; the next iteration re-lists.""" + self._file_cache = None + + def __iter__(self): + return iter(self._file_ids) + + def __len__(self): + return len(self._file_ids) + + def __contains__(self, key: str) -> bool: + """Whether the file exists and is reachable -- a metadata probe, never a download.""" + try: + self.metadata(key, fields=("id",)) + except ValueError: + return False # not a usable file key at all + except Exception as error: + if _is_not_found(error): + return False + raise # auth / network / server failures are not "absent" + return True + + def __getitem__(self, key: str) -> bytes: + """The file's content as bytes, downloaded over the authenticated API.""" + file_id = _resolve_file_id(key) + try: + return _download_via_api(self._drive, file_id) + except Exception as error: + if _is_not_found(error): + raise KeyError(f"File not found or not accessible: {key}") from error + raise + + def metadata(self, key: str, *, fields=DEFAULT_METADATA_FIELDS) -> dict: + """Metadata (name, size, mimeType, modifiedDate) for ``key``, without downloading it. + + Thin method form of :func:`get_metadata` -- see it for the field semantics. + """ + return get_metadata(key, drive=self._drive, fields=fields) + + class GDReader(Mapping): """ Read-only Mapping to Google Drive folder. @@ -356,25 +748,15 @@ def __init__( self._file_cache = None def _list_files(self, folder_id: str, prefix: str = "", level: int = 0): - """Recursively list files in folder.""" - if self.max_levels is not None and level > self.max_levels: - return - - query = f"'{folder_id}' in parents and trashed=false" - file_list = self._drive.ListFile({"q": query}).GetList() - - for item in file_list: - name = item["title"] - if not self.include_hidden and name.startswith("."): - continue - - item_path = os.path.join(prefix, name) if prefix else name - - if item["mimeType"] == "application/vnd.google-apps.folder": - if self.max_levels is None or level < self.max_levels: - yield from self._list_files(item["id"], item_path, level + 1) - else: - yield (item_path, item["id"]) + """Recursively list ``(relative_path, file_id)`` pairs under ``folder_id``.""" + yield from _iter_folder_files( + self._drive, + folder_id, + max_levels=self.max_levels, + include_hidden=self.include_hidden, + prefix=prefix, + level=level, + ) @property def _files(self): @@ -397,19 +779,16 @@ def __contains__(self, key): return key in self._files def __getitem__(self, key: str) -> bytes: - """Get file content as bytes.""" + """Get file content as bytes. + + Goes through :func:`_download_via_api`, which is byte-exact: the previous + ``GetContentString(...).encode('latin-1')`` route decoded as utf-8 first, so it + corrupted (or raised on) every real binary -- xlsx, pdf, images. + """ if key not in self._files: raise KeyError(f"File not found: {key}") - file_id = self._files[key] - gfile = self._drive.CreateFile({"id": file_id}) - - # Download as bytes - content = gfile.GetContentString(mimetype="application/octet-stream") - if isinstance(content, str): - content = content.encode("latin-1") - - return content + return _download_via_api(self._drive, self._files[key]) def get_url( self, diff --git a/pydrivedol/tests/test_authenticated_access.py b/pydrivedol/tests/test_authenticated_access.py new file mode 100644 index 0000000..b987643 --- /dev/null +++ b/pydrivedol/tests/test_authenticated_access.py @@ -0,0 +1,482 @@ +"""Unit tests for authenticated file access -- no real Drive, no network, no credentials. + +Everything here runs off two fakes: a ``FakeDrive`` implementing the slice of the PyDrive2 +``GoogleDrive`` API that ``pydrivedol`` actually calls, and a ``FakeSession`` standing in for +``requests.Session``. That covers all three concerns end to end: + +- the silent-HTML guard on the public download path (``NotPubliclyShared``), +- the authenticated download / metadata path (``drive=``), +- the file-level ``GDFiles`` mapping. +""" + +import re + +import pytest + +import pydrivedol.base as base +from pydrivedol import ( + GDFiles, + GDReader, + NotPubliclyShared, + get_bytes, + get_metadata, +) + +FILE_ID = '1AbCdEfGhIjKlMnOpQrStUvWxYz012345' +FILE_URL = f'https://drive.google.com/file/d/{FILE_ID}/view?usp=sharing' +FOLDER_ID = '1FoLdErIdAbCdEfGhIjKlMnOpQrStUvW' +FOLDER_URL = f'https://drive.google.com/drive/folders/{FOLDER_ID}' + +# A real xlsx starts with the zip magic; nothing about it looks like HTML. +XLSX_BYTES = b'PK\x03\x04\x14\x00\x08\x08\x08\x00' + b'\xe2\x80\x93' * 100 + +# Abridged shape of what Drive actually serves (with HTTP 200) for a non-public file. +GOOGLE_SIGNIN_HTML = ( + b'' + b'Sign in - Google Accounts' + b'You need access. Ask for access, or switch to an account with access.' +) + + +# ============================================================================= +# Fakes +# ============================================================================= + + +class _FakeList: + """What PyDrive2's ``ListFile`` returns: something with ``GetList()``.""" + + def __init__(self, items): + self._items = items + + def GetList(self): + return list(self._items) + + +class FakeGDriveFile(dict): + """The slice of PyDrive2's ``GoogleDriveFile`` that pydrivedol calls.""" + + def __init__(self, meta, drive): + super().__init__(meta or {}) + self._drive = drive + + def GetContentIOBuffer(self, **kwargs): + self._drive.calls.append(('GetContentIOBuffer', self['id'])) + content = self._drive.content_of(self['id']) + # PyDrive2 yields chunks; two of them here so the join is actually exercised. + midpoint = len(content) // 2 + return iter([content[:midpoint], content[midpoint:]]) + + def FetchMetadata(self, fields=None, fetch_all=False): + self._drive.calls.append(('FetchMetadata', self['id'], fields, fetch_all)) + self.update(self._drive.metadata_of(self['id'])) + + +class FakeDrive: + """A stand-in ``GoogleDrive``: ``files`` map ids to metadata dicts (with ``_content``).""" + + def __init__(self, files=None, listings=None, errors=None): + self.files = dict(files or {}) + self.listings = dict(listings or {}) # folder_id -> list of listing entries + self.errors = dict(errors or {}) # file_id -> exception to raise + self.calls = [] + + def _check(self, file_id): + if file_id in self.errors: + raise self.errors[file_id] + if file_id not in self.files: + raise Exception(f'') + + def content_of(self, file_id): + self._check(file_id) + return self.files[file_id].get('_content', b'') + + def metadata_of(self, file_id): + self._check(file_id) + return {k: v for k, v in self.files[file_id].items() if not k.startswith('_')} + + def CreateFile(self, meta=None): + self.calls.append(('CreateFile', dict(meta or {}))) + return FakeGDriveFile(meta, self) + + def ListFile(self, param): + folder_id = re.match(r"'([^']+)' in parents", param['q']).group(1) + self.calls.append(('ListFile', folder_id)) + return _FakeList(self.listings.get(folder_id, [])) + + +class FakeResponse: + """A ``requests`` response carrying whatever bytes/content-type the test wants.""" + + def __init__(self, content, *, content_type='application/octet-stream', status_code=200): + self.content = content + self.status_code = status_code + self.headers = {'Content-Type': content_type} + self.cookies = {} + + @property + def text(self): + return self.content.decode('utf-8', 'replace') + + +class FakeSession: + def __init__(self, response): + self.response = response + self.gets = [] + + def get(self, url, **kwargs): + self.gets.append((url, kwargs)) + return self.response + + +@pytest.fixture +def fake_drive(): + """A drive holding one binary file, one Google-native Sheet, and a two-level folder tree.""" + return FakeDrive( + files={ + FILE_ID: { + 'id': FILE_ID, + 'title': 'client_export.xlsx', + 'mimeType': ( + 'application/vnd.openxmlformats-officedocument.spreadsheetml.sheet' + ), + 'fileSize': '18512345', # Drive sends sizes as strings + 'modifiedDate': '2026-08-01T12:00:00.000Z', + 'alternateLink': FILE_URL, + '_content': XLSX_BYTES, + }, + 'nativeSheetId0123456789': { + 'id': 'nativeSheetId0123456789', + 'title': 'A Google Sheet', + 'mimeType': 'application/vnd.google-apps.spreadsheet', + # no fileSize: Google-native files have no stored byte size + 'modifiedDate': '2026-08-02T12:00:00.000Z', + '_content': b'', + }, + 'nestedFileId0123456789': { + 'id': 'nestedFileId0123456789', + 'title': 'nested.csv', + '_content': b'a,b', + }, + }, + listings={ + FOLDER_ID: [ + {'id': FILE_ID, 'title': 'client_export.xlsx', 'mimeType': 'application/x'}, + {'id': 'hiddenFileId012345678', 'title': '.hidden', 'mimeType': 'application/x'}, + { + 'id': 'subFolderId0123456789', + 'title': 'sub', + 'mimeType': 'application/vnd.google-apps.folder', + }, + ], + 'subFolderId0123456789': [ + {'id': 'nestedFileId0123456789', 'title': 'nested.csv', 'mimeType': 'text/csv'}, + ], + }, + ) + + +def _use_session(monkeypatch, response): + """Point ``base.requests.Session`` at a fake returning ``response``; return the fake.""" + session = FakeSession(response) + monkeypatch.setattr(base.requests, 'Session', lambda: session) + return session + + +def _forbid_network(monkeypatch): + """Make any use of ``requests.Session`` an outright test failure.""" + + def _boom(): + raise AssertionError('the authenticated path must not touch the public endpoint') + + monkeypatch.setattr(base.requests, 'Session', _boom) + + +# ============================================================================= +# Key resolution +# ============================================================================= + + +def test_resolve_file_id_accepts_urls_and_bare_ids(): + assert base._resolve_file_id(FILE_URL) == FILE_ID + assert base._resolve_file_id(FILE_ID) == FILE_ID + assert base._resolve_file_id(f'https://drive.google.com/open?id={FILE_ID}') == FILE_ID + + +@pytest.mark.parametrize('bad', ['', 'nope', 'not a url', FOLDER_URL, 'https://example.com/x']) +def test_resolve_file_id_rejects_non_file_keys(bad): + with pytest.raises(ValueError): + base._resolve_file_id(bad) + + +# ============================================================================= +# The silent-HTML guard +# ============================================================================= + + +@pytest.mark.parametrize( + 'content, content_type, expected', + [ + (XLSX_BYTES, 'application/octet-stream', False), + (b'id,name', 'text/csv', False), + (b'', '', False), + (GOOGLE_SIGNIN_HTML, 'text/html; charset=utf-8', True), + (GOOGLE_SIGNIN_HTML, '', True), # body alone is enough + (b'anything at all', 'text/html; charset=utf-8', True), # content type alone is enough + (b' \n\t', '', True), # leading whitespace tolerated + (b'\xef\xbb\xbf', '', True), # BOM tolerated + (b'', '', True), # lowercase + ], +) +def test_looks_like_html(content, content_type, expected): + assert base._looks_like_html(content, content_type) is expected + + +def test_get_bytes_raises_instead_of_returning_the_sign_in_page(monkeypatch): + """The bug this guards: HTTP 200 + HTML body was returned as if it were the file.""" + _use_session(monkeypatch, FakeResponse(GOOGLE_SIGNIN_HTML, content_type='text/html')) + + with pytest.raises(NotPubliclyShared) as excinfo: + get_bytes(FILE_URL) + + message = str(excinfo.value) + assert FILE_ID in message + assert 'drive=' in message # says how to fix it + assert 'allow_html=True' in message # ...and how to opt out + + +def test_get_bytes_detects_html_even_when_content_type_lies(monkeypatch): + _use_session( + monkeypatch, FakeResponse(GOOGLE_SIGNIN_HTML, content_type='application/octet-stream') + ) + with pytest.raises(NotPubliclyShared): + get_bytes(FILE_URL) + + +def test_get_bytes_allow_html_opts_out(monkeypatch): + _use_session(monkeypatch, FakeResponse(GOOGLE_SIGNIN_HTML, content_type='text/html')) + assert get_bytes(FILE_URL, allow_html=True) == GOOGLE_SIGNIN_HTML + + +def test_get_bytes_public_path_still_returns_real_content(monkeypatch): + """No false positives: ordinary binary payloads pass through untouched.""" + session = _use_session(monkeypatch, FakeResponse(XLSX_BYTES)) + assert get_bytes(FILE_URL) == XLSX_BYTES + assert FILE_ID in session.gets[0][0] + + +def test_get_bytes_public_path_still_raises_on_bad_status(monkeypatch): + _use_session(monkeypatch, FakeResponse(b'nope', status_code=403)) + with pytest.raises(RuntimeError): + get_bytes(FILE_URL) + + +# ============================================================================= +# Authenticated get_bytes +# ============================================================================= + + +def test_get_bytes_with_drive_downloads_via_api(fake_drive, monkeypatch): + _forbid_network(monkeypatch) + assert get_bytes(FILE_URL, drive=fake_drive) == XLSX_BYTES + assert ('CreateFile', {'id': FILE_ID}) in fake_drive.calls + assert ('GetContentIOBuffer', FILE_ID) in fake_drive.calls + + +def test_get_bytes_with_drive_accepts_a_bare_file_id(fake_drive, monkeypatch): + _forbid_network(monkeypatch) + assert get_bytes(FILE_ID, drive=fake_drive) == XLSX_BYTES + + +def test_get_bytes_with_drive_honours_local_path(fake_drive, monkeypatch, tmp_path): + _forbid_network(monkeypatch) + target = tmp_path / 'sub' / 'export.xlsx' + content = get_bytes(FILE_URL, drive=fake_drive, local_path=str(target)) + assert content == XLSX_BYTES + assert target.read_bytes() == XLSX_BYTES + + +def test_get_bytes_with_drive_honours_use_cache(fake_drive, monkeypatch, tmp_path): + _forbid_network(monkeypatch) + cache_dir = str(tmp_path / 'cache') + + assert get_bytes(FILE_URL, drive=fake_drive, use_cache=cache_dir) == XLSX_BYTES + downloads = [c for c in fake_drive.calls if c[0] == 'GetContentIOBuffer'] + assert len(downloads) == 1 + + # Second call is served from cache -- the drive is not touched again. + assert get_bytes(FILE_URL, drive=fake_drive, use_cache=cache_dir) == XLSX_BYTES + downloads = [c for c in fake_drive.calls if c[0] == 'GetContentIOBuffer'] + assert len(downloads) == 1 + + +def test_get_bytes_without_drive_never_calls_the_api(monkeypatch, fake_drive): + """drive=None must keep the public behaviour exactly.""" + _use_session(monkeypatch, FakeResponse(XLSX_BYTES)) + get_bytes(FILE_URL) + assert fake_drive.calls == [] + + +# ============================================================================= +# Metadata without downloading +# ============================================================================= + + +def test_get_metadata_returns_the_decision_making_fields(fake_drive): + info = get_metadata(FILE_URL, drive=fake_drive) + assert info['title'] == 'client_export.xlsx' + assert info['modifiedDate'] == '2026-08-01T12:00:00.000Z' + assert 'spreadsheetml' in info['mimeType'] + + +def test_get_metadata_coerces_file_size_to_int(fake_drive): + """Drive sends fileSize as a string; callers compare it against a threshold.""" + assert get_metadata(FILE_URL, drive=fake_drive)['fileSize'] == 18512345 + + +def test_get_metadata_does_not_download(fake_drive): + get_metadata(FILE_URL, drive=fake_drive) + assert not any(c[0] == 'GetContentIOBuffer' for c in fake_drive.calls) + + +def test_get_metadata_requests_only_the_named_fields(fake_drive): + get_metadata(FILE_URL, drive=fake_drive, fields=('id', 'fileSize')) + (fetch,) = [c for c in fake_drive.calls if c[0] == 'FetchMetadata'] + assert fetch[2] == 'id,fileSize' + assert fetch[3] is False + + +def test_get_metadata_fields_none_fetches_everything(fake_drive): + info = get_metadata(FILE_URL, drive=fake_drive, fields=None) + (fetch,) = [c for c in fake_drive.calls if c[0] == 'FetchMetadata'] + assert fetch[3] is True + assert info['title'] == 'client_export.xlsx' + + +def test_get_metadata_omits_file_size_for_google_native_files(fake_drive): + info = get_metadata('nativeSheetId0123456789', drive=fake_drive) + assert info['mimeType'] == 'application/vnd.google-apps.spreadsheet' + assert 'fileSize' not in info + + +# ============================================================================= +# GDFiles +# ============================================================================= + + +def test_gdfiles_keyed_by_url_or_id(fake_drive): + files = GDFiles(fake_drive) + assert files[FILE_URL] == XLSX_BYTES + assert files[FILE_ID] == XLSX_BYTES + + +def test_gdfiles_unscoped_iteration_raises_with_an_actionable_message(fake_drive): + files = GDFiles(fake_drive) + for call in (lambda: list(files), lambda: len(files)): + with pytest.raises(NotImplementedError) as excinfo: + call() + assert 'folder_url' in str(excinfo.value) + + +def test_gdfiles_scoped_iteration_yields_file_ids(fake_drive): + files = GDFiles(fake_drive, folder_url=FOLDER_URL) + assert list(files) == [FILE_ID, 'nestedFileId0123456789'] + assert len(files) == 2 + + +def test_gdfiles_scope_respects_max_levels(fake_drive): + files = GDFiles(fake_drive, folder_url=FOLDER_URL, max_levels=0) + assert list(files) == [FILE_ID] + + +def test_gdfiles_scope_respects_include_hidden(fake_drive): + files = GDFiles(fake_drive, folder_url=FOLDER_URL, max_levels=0, include_hidden=True) + assert list(files) == [FILE_ID, 'hiddenFileId012345678'] + + +def test_gdfiles_scoped_values_are_the_file_contents(fake_drive): + files = GDFiles(fake_drive, folder_url=FOLDER_URL) + assert dict(files) == {FILE_ID: XLSX_BYTES, 'nestedFileId0123456789': b'a,b'} + + +def test_gdfiles_rejects_a_bad_folder_url(fake_drive): + with pytest.raises(ValueError): + GDFiles(fake_drive, folder_url='https://example.com/not-a-folder') + + +def test_gdfiles_contains_probes_metadata_rather_than_downloading(fake_drive): + files = GDFiles(fake_drive) + assert FILE_URL in files + assert not any(c[0] == 'GetContentIOBuffer' for c in fake_drive.calls) + + +def test_gdfiles_contains_is_false_for_missing_or_unusable_keys(fake_drive): + files = GDFiles(fake_drive) + assert 'noSuchFileId0123456789' not in files + assert 'not a drive key' not in files + + +def test_gdfiles_contains_propagates_non_404_failures(fake_drive): + fake_drive.errors[FILE_ID] = Exception('') + files = GDFiles(fake_drive) + with pytest.raises(Exception, match='500'): + FILE_URL in files # noqa: B015 -- the point is that it raises + + +def test_gdfiles_getitem_raises_keyerror_for_a_missing_file(fake_drive): + files = GDFiles(fake_drive) + with pytest.raises(KeyError): + files['noSuchFileId0123456789'] + + +def test_gdfiles_getitem_propagates_non_404_failures(fake_drive): + """A 5xx must not be disguised as 'absent' -- that would hide real outages.""" + fake_drive.errors[FILE_ID] = Exception('') + files = GDFiles(fake_drive) + with pytest.raises(Exception, match='500'): + files[FILE_URL] + + +def test_gdfiles_getitem_rejects_an_unusable_key(fake_drive): + with pytest.raises(ValueError): + GDFiles(fake_drive)['not a drive key'] + + +def test_gdfiles_metadata_method_delegates(fake_drive): + files = GDFiles(fake_drive) + assert files.metadata(FILE_URL)['fileSize'] == 18512345 + assert files.metadata(FILE_URL, fields=('id',)) == {'id': FILE_ID} + + +def test_gdfiles_needs_no_pydrive2_import(fake_drive, monkeypatch): + """The drive is injected, so GDFiles must not require PyDrive2 to be importable.""" + monkeypatch.setattr(base, '_PYDRIVE2_AVAILABLE', False) + assert GDFiles(fake_drive)[FILE_URL] == XLSX_BYTES + + +# ============================================================================= +# GDReader: binary reads must be byte-exact +# ============================================================================= + + +def test_gdreader_getitem_returns_exact_bytes(fake_drive, monkeypatch): + """Regression: GetContentString(...).encode('latin-1') mangled every real binary.""" + monkeypatch.setattr(base, '_init_google_drive', _no_oauth) + reader = GDReader(FOLDER_URL, drive=fake_drive) + assert reader['client_export.xlsx'] == XLSX_BYTES + assert reader['sub/nested.csv'] == b'a,b' + + +def test_gdreader_listing_is_unchanged_by_the_shared_traversal(fake_drive, monkeypatch): + monkeypatch.setattr(base, '_init_google_drive', _no_oauth) + assert list(GDReader(FOLDER_URL, drive=fake_drive)) == [ + 'client_export.xlsx', + 'sub/nested.csv', + ] + assert list(GDReader(FOLDER_URL, drive=fake_drive, max_levels=0)) == [ + 'client_export.xlsx' + ] + + +def _no_oauth(*args, **kwargs): + raise AssertionError('must not run the OAuth flow when drive= is provided') From b5ef2c12548c28d078dbd82565df84d1647668d0 Mon Sep 17 00:00:00 2001 From: Thor Whalen <1906276+thorwhalen@users.noreply.github.com> Date: Wed, 19 Aug 2026 12:49:24 +0100 Subject: [PATCH 2/2] test: make GDReader key assertions platform-independent GDReader builds keys with os.path.join, so the nested-file key uses a backslash on Windows. The Windows CI job is continue-on-error, but a red check is still noise. Claude-Session: https://claude.ai/code/session_01Vb8moaBd5Yg8b4Abo77nHf --- pydrivedol/tests/test_authenticated_access.py | 9 +++++++-- 1 file changed, 7 insertions(+), 2 deletions(-) diff --git a/pydrivedol/tests/test_authenticated_access.py b/pydrivedol/tests/test_authenticated_access.py index b987643..c7f6966 100644 --- a/pydrivedol/tests/test_authenticated_access.py +++ b/pydrivedol/tests/test_authenticated_access.py @@ -9,6 +9,7 @@ - the file-level ``GDFiles`` mapping. """ +import os import re import pytest @@ -459,19 +460,23 @@ def test_gdfiles_needs_no_pydrive2_import(fake_drive, monkeypatch): # ============================================================================= +# GDReader keys are built with os.path.join, so the separator is the platform's. +NESTED_KEY = os.path.join('sub', 'nested.csv') + + def test_gdreader_getitem_returns_exact_bytes(fake_drive, monkeypatch): """Regression: GetContentString(...).encode('latin-1') mangled every real binary.""" monkeypatch.setattr(base, '_init_google_drive', _no_oauth) reader = GDReader(FOLDER_URL, drive=fake_drive) assert reader['client_export.xlsx'] == XLSX_BYTES - assert reader['sub/nested.csv'] == b'a,b' + assert reader[NESTED_KEY] == b'a,b' def test_gdreader_listing_is_unchanged_by_the_shared_traversal(fake_drive, monkeypatch): monkeypatch.setattr(base, '_init_google_drive', _no_oauth) assert list(GDReader(FOLDER_URL, drive=fake_drive)) == [ 'client_export.xlsx', - 'sub/nested.csv', + NESTED_KEY, ] assert list(GDReader(FOLDER_URL, drive=fake_drive, max_levels=0)) == [ 'client_export.xlsx'