test(pineapple): add live Google Drive accessibility tests; refactor resolve

Sweep all 30 catalog resources (segmentation/benchmark/weights) against
Google Drive to guarantee the curated data stays downloadable. The new
TestPineappleLiveAccess checks headers only (no body download): each file
ID must still resolve to a binary attachment whose filename matches the
catalog and is not a tiny placeholder; transient Google Drive throttling
(HTML quota page) is treated as unavailable, not a failure.

Extract _resolve_gdrive_response from _download_from_gdrive so the live
tests exercise the same production resolution path, including the
large-file virus-scan-warning confirmation.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

Author: Elarwei001

gget/gget_pineapple.py

@@ -269,11 +269,15 @@ def _parse_gdrive_form(html_text: str) -> tuple[str | None, dict[str, str]]:
     return action, params
 
 
-def _download_from_gdrive(file_id: str, dest_path: str, verbose: bool = True) -> None:
-    """Download a (potentially large) file from Google Drive by file ID."""
-    session = requests.Session()
-    session.headers.update({"User-Agent": "Mozilla/5.0 (compatible; gget)"})
-
+def _resolve_gdrive_response(session: requests.Session, file_id: str) -> requests.Response:
+    """Resolve a Google Drive file ID to a streaming response for the actual file.
+
+    Small files download directly; large files first return an HTML
+    "can't scan for viruses" warning page whose form must be submitted to
+    obtain the real file. The returned response is opened with ``stream=True``,
+    so the body is not fetched until the caller iterates it. Callers that only
+    need the headers (e.g. accessibility checks) must ``close()`` the response.
+    """
     response = session.get(
         PINEAPPLE_GDRIVE_URL,
         params={"id": file_id},
@@ -282,8 +286,6 @@ def _download_from_gdrive(file_id: str, dest_path: str, verbose: bool = True) ->
     )
     response.raise_for_status()
 
-    # Small files download directly; large files first return an HTML
-    # "can't scan for viruses" warning page that must be confirmed.
     content_type = response.headers.get("Content-Type", "")
     if "text/html" in content_type:
         action, params = _parse_gdrive_form(response.text)
@@ -302,6 +304,16 @@ def _download_from_gdrive(file_id: str, dest_path: str, verbose: bool = True) ->
                 )
                 response.raise_for_status()
 
+    return response
+
+
+def _download_from_gdrive(file_id: str, dest_path: str, verbose: bool = True) -> None:
+    """Download a (potentially large) file from Google Drive by file ID."""
+    session = requests.Session()
+    session.headers.update({"User-Agent": "Mozilla/5.0 (compatible; gget)"})
+
+    response = _resolve_gdrive_response(session, file_id)
+
     with open(dest_path, "wb") as fh:
         for chunk in response.iter_content(chunk_size=32768):
             if chunk:

tests/test_pineapple.py

@@ -3,6 +3,7 @@
 from unittest.mock import patch
 
 import gget.gget_pineapple as gget_pineapple
+import requests
 from gget.gget_pineapple import (
     _catalog_row,
     _parse_gdrive_form,
@@ -87,5 +88,104 @@ def test_download_invokes_gdrive(self, mock_dl):
         self.assertEqual(df.iloc[0]["name"], "vicar_2021")
 
 
+class TestPineappleLiveAccess(unittest.TestCase):
+    """Live data test: verify EVERY curated Pineapple resource is still
+    downloadable from Google Drive (issue #161).
+
+    This hits Google Drive over the network but never downloads the (multi-GB)
+    bodies. Each file ID is resolved through the *same* production code path as
+    a real download (``_resolve_gdrive_response``, including the large-file
+    virus-scan-warning confirmation), and only the response headers are checked,
+    so the whole catalog can be swept cheaply.
+
+    Purpose: if someone edits the catalog, or upstream repoints/removes a file,
+    this fails loudly -- the Google-Drive-reported filename must still match the
+    catalog and the ID must still resolve to a binary download rather than an
+    error/quota HTML page.
+    """
+
+    _CATALOGS = {
+        "segmentation": gget_pineapple._SEGMENTATION,
+        "benchmark": gget_pineapple._BENCHMARK,
+        "weights": gget_pineapple._WEIGHTS,
+    }
+
+    # 1 MB floor: catches an ID repointed to a tiny placeholder/error file.
+    # NOT tied to the catalog's size_gb, which is only approximate (e.g.
+    # livecell_2021 lists 3.26 GB but the real file is ~1.81 GB).
+    _MIN_BYTES = 1_000_000
+
+    def _assert_resource_headers(self, response, expected, name):
+        """Header-only assertions for a resolved resource (no body download)."""
+        self.assertEqual(response.status_code, 200, f"{name}: unexpected status code")
+        self.assertIn(
+            "octet-stream",
+            response.headers.get("Content-Type", ""),
+            f"{name}: expected a binary download, got Content-Type "
+            f"{response.headers.get('Content-Type', '')!r}",
+        )
+        disposition = response.headers.get("Content-Disposition", "")
+        self.assertIn(
+            f'filename="{expected["filename"]}"',
+            disposition,
+            f"{name}: Google Drive filename does not match the catalog "
+            f"(expected {expected['filename']!r}, Content-Disposition={disposition!r}). "
+            f"The file ID may have been repointed upstream.",
+        )
+        length = response.headers.get("Content-Length")
+        self.assertIsNotNone(length, f"{name}: response is missing Content-Length")
+        self.assertGreater(
+            int(length),
+            self._MIN_BYTES,
+            f"{name}: file is implausibly small ({length} bytes) -- possible placeholder",
+        )
+
+    def test_live_all_resources_accessible(self):
+        session = requests.Session()
+        session.headers.update({"User-Agent": "Mozilla/5.0 (compatible; gget)"})
+
+        total = sum(len(cat) for cat in self._CATALOGS.values())
+        verified = 0
+        unavailable = []  # (name, reason) -- transient: throttling / network, not a failure
+
+        for category, catalog in self._CATALOGS.items():
+            for name, info in catalog.items():
+                expected = _catalog_row(category, name, info)
+                with self.subTest(category=category, name=name):
+                    try:
+                        response = gget_pineapple._resolve_gdrive_response(
+                            session, expected["google_drive_id"]
+                        )
+                    except requests.RequestException as exc:
+                        unavailable.append((name, f"network error: {exc}"))
+                        continue
+                    try:
+                        # Google Drive serves a transient HTML "download quota
+                        # exceeded" page under load. Treat as unavailable (not a
+                        # failure) and move on; a genuinely missing file would have
+                        # raised a 404/410 in _resolve_gdrive_response above.
+                        if "text/html" in response.headers.get("Content-Type", ""):
+                            unavailable.append((name, "Google Drive HTML page (quota throttling?)"))
+                            continue
+                        self._assert_resource_headers(response, expected, name)
+                        verified += 1
+                    finally:
+                        response.close()
+
+        # Only skip when the ENTIRE sweep was transiently unavailable (throttling /
+        # network) -- never when a real assertion failed (those are already recorded
+        # as subTest failures and must keep the build red).
+        if verified == 0 and len(unavailable) == total:
+            self.skipTest(
+                f"Could not verify any of the {total} resources (all transiently "
+                f"unavailable): {unavailable}"
+            )
+        if unavailable:
+            print(
+                f"\n[pineapple live] verified {verified}/{total}; "
+                f"{len(unavailable)} transiently unavailable: {unavailable}"
+            )
+
+
 if __name__ == "__main__":
     unittest.main()