#!/usr/bin/env python3 """Re-encode a recorded cassette so its response body is gzip-compressed and the response advertises ``Content-Encoding: gzip``. Why this exists --------------- ``decode_compressed_response=True`` in :mod:`tests.vcr_config` strips ``Content-Encoding`` from the headers on record (after VCR transparently gunzips the body), so 89/89 cassettes in ``tests/cassettes/`` carry NO encoding header and a plaintext body. That hid issue #769 — every RPC failed in production with ``Error -3 while decompressing data: incorrect header check`` because :func:`notebooklm._streaming_post.stream_post_with_size_cap` rebuilt the response with the upstream ``Content-Encoding: gzip`` header attached to *already-decoded* body bytes, triggering a second gzip-decode inside :class:`httpx.Response.__init__`. Cassettes that don't carry the header on replay simply skip that branch and pass. This script ports a cassette to the production wire shape: gzip the response body, restore the ``Content-Encoding: gzip`` header, and strip the now-meaningless ``Transfer-Encoding`` and ``Content-Length`` headers (the originals described the chunked / decoded wire form, not the gzipped buffer we hand to vcrpy on replay). Run from the repo root:: uv run python tests/scripts/inject_gzip_into_cassette.py \\ tests/cassettes/artifacts_revise_slide.yaml \\ tests/cassettes/gzip_coverage/artifacts_revise_slide_gzipped.yaml Idempotent: re-running on a cassette that already advertises ``Content-Encoding: gzip`` rewrites it from scratch using the gzipped body that vcrpy has already deserialized to ``bytes``, so a second pass produces a byte-identical file. Library use: Other tooling can ``from tests.scripts.inject_gzip_into_cassette import inject_gzip_into_cassette`` and pass a cassette dict in memory. """ from __future__ import annotations import argparse import gzip import sys from pathlib import Path from typing import Any import yaml # Use the SAFE loader: cassettes are arbitrary YAML on disk and a malicious # ``!!python/object/apply`` tag would execute code under the full Loader. # ``!!binary`` (the encoding we rely on for gzipped bodies) is part of the # core YAML schema so it works under ``CSafeLoader`` / ``SafeLoader``. try: from yaml import CSafeDumper as Dumper from yaml import CSafeLoader as Loader except ImportError: # pragma: no cover — libyaml is bundled with PyYAML wheels from yaml import SafeDumper as Dumper # type: ignore[assignment] from yaml import SafeLoader as Loader # Headers stripped from the response when we rewrite the body. The recorded # Transfer-Encoding / Content-Length values described the chunked / decoded # wire form; once we re-gzip the body, neither matches reality and leaving # them in confuses httpx on replay. Content-Encoding is included so the # rewrite always re-adds the canonical-cased header regardless of how the # cassette was originally recorded. _STRIP_RESPONSE_HEADERS = frozenset( { "transfer-encoding", "content-length", "content-encoding", } ) _CONTENT_ENCODING_HEADER = "Content-Encoding" def _has_content_encoding(headers: dict[str, list[str]]) -> bool: """Case-insensitive check for ``Content-Encoding`` in a cassette header dict.""" return any(k.lower() == "content-encoding" for k in headers) def _coerce_body_to_bytes(body_string: Any) -> bytes: """Normalize the recorded ``response.body.string`` to ``bytes``. PyYAML deserializes ``!!binary`` blobs as :class:`bytes` and plain block scalars as :class:`str`. The cassette format permits either, so accept both and return raw bytes. """ if isinstance(body_string, bytes): return body_string if isinstance(body_string, str): return body_string.encode("utf-8") raise TypeError( f"Unexpected response.body.string type {type(body_string).__name__}; expected bytes or str." ) def _inject_into_response(response: dict[str, Any]) -> bool: """Mutate ``response`` in place so the body is gzip-compressed and ``Content-Encoding: gzip`` is set. Returns ``True`` when the body was (re)written; ``False`` only when the response has no ``body.string`` to process (missing key or ``None``). A response that already advertises ``Content-Encoding`` is gunzipped first and then re-gzipped from the decoded form so the rewrite is idempotent — still a rewrite, still ``True``. """ body = response.get("body") or {} if "string" not in body: return False raw = body["string"] if raw is None: return False body_bytes = _coerce_body_to_bytes(raw) headers = response.setdefault("headers", {}) already_encoded = _has_content_encoding(headers) if already_encoded: # The body on disk is already gzipped — gunzip first so the # rewrite produces a byte-identical file on a second pass. try: body_bytes = gzip.decompress(body_bytes) except OSError as exc: # pragma: no cover — defensive raise ValueError( "Response already advertises Content-Encoding but body is not valid gzip." ) from exc # ``gzip.compress`` writes one byte that varies across Python releases # (the OS field at offset 9): 3.10 picks an OS-dependent constant # while 3.11+ pins it to ``0xff`` (unknown). Pin ``mtime=0`` and # overwrite the OS field so the same source cassette produces a # byte-identical artifact under every Python in the CI matrix # (``3.10`` – ``3.14``). ``0xff`` is the documented "unknown" value # from RFC 1952 §2.3.1 and round-trips through every gzip decoder. gzipped = bytearray(gzip.compress(body_bytes, mtime=0)) gzipped[9] = 0xFF body["string"] = bytes(gzipped) # Drop any case-variant of the headers in _STRIP_RESPONSE_HEADERS # (including Content-Encoding) so the rewrite re-adds the canonical # header name below. for key in list(headers): if key.lower() in _STRIP_RESPONSE_HEADERS: del headers[key] headers[_CONTENT_ENCODING_HEADER] = ["gzip"] return True def inject_gzip_into_cassette(cassette: dict[str, Any]) -> int: """Rewrite every batchexecute-style response in ``cassette`` to be gzip-encoded. Only responses whose request is a POST to a path containing ``batchexecute`` are touched — those are the ones that flow through :func:`notebooklm._streaming_post.stream_post_with_size_cap` and exercise the rebuild step that bit #769. SPA-shell GET responses are left alone so the cassette diff stays focused. Returns the number of responses that were rewritten. """ rewritten = 0 for interaction in cassette.get("interactions", []) or []: request = interaction.get("request") or {} if request.get("method", "").upper() != "POST": continue uri = request.get("uri", "") or "" if "batchexecute" not in uri: continue response = interaction.get("response") or {} if _inject_into_response(response): rewritten += 1 return rewritten def _load_cassette(path: Path) -> dict[str, Any]: with path.open("r", encoding="utf-8") as fh: return yaml.load(fh, Loader=Loader) def _dump_cassette(cassette: dict[str, Any], path: Path) -> None: path.parent.mkdir(parents=True, exist_ok=True) with path.open("w", encoding="utf-8") as fh: yaml.dump(cassette, fh, Dumper=Dumper) def _main(argv: list[str]) -> int: parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("source", type=Path, help="cassette to read") parser.add_argument( "destination", type=Path, help="destination path (may equal source for in-place rewrite)", ) args = parser.parse_args(argv) cassette = _load_cassette(args.source) rewritten = inject_gzip_into_cassette(cassette) if rewritten == 0: print( f"warning: no batchexecute POST interactions found in {args.source}", file=sys.stderr, ) return 1 _dump_cassette(cassette, args.destination) print(f"rewrote {rewritten} response(s) in {args.destination}") return 0 if __name__ == "__main__": # pragma: no cover raise SystemExit(_main(sys.argv[1:]))