From 7a0c1855829bc544f2dd49b70b0bfe9b72346c48 Mon Sep 17 00:00:00 2001 From: mateo-berri <277851410+mateo-berri@users.noreply.github.com> Date: Fri, 5 Jun 2026 19:08:41 +0000 Subject: [PATCH] test(ci): make the image-gen record/replay proxy report cache mode and per-request HIT/MISS The recorder could come up pointed at a missing or unreachable cassette redis and silently forward every request live; the health check still passed and the process logged nothing, so a CI run looked identical whether it replayed from the cassette or paid OpenAI for a fresh call every commit. There was no way to tell from the logs whether the 24h caching was actually happening. It now announces its mode at startup (REPLAY when the cassette redis is reachable, PASSTHROUGH when CASSETTE_REDIS_URL is unset, DEGRADED when it is set but the redis is unreachable) and logs a HIT/MISS line per request. _cache_set returns whether the write landed so a mid-run redis failure surfaces as a warning instead of masquerading as a successful record. Adds unit tests covering the three startup modes and the HIT/MISS/not-recorded request paths; both new behaviors were mutation-checked. --- tests/_openai_record_replay_proxy.py | 53 +++++++++++++++++-- .../test_openai_record_replay_proxy.py | 51 ++++++++++++++++++ 2 files changed, 99 insertions(+), 5 deletions(-) diff --git a/tests/_openai_record_replay_proxy.py b/tests/_openai_record_replay_proxy.py index 60832188a75d..1413b1ab7462 100644 --- a/tests/_openai_record_replay_proxy.py +++ b/tests/_openai_record_replay_proxy.py @@ -19,6 +19,11 @@ capture and the next run past that point re-records live and catches provider contract drift, exactly matching the lapse-after-write contract in ``tests/_vcr_redis_persister.py``. + +The process logs its mode at startup (REPLAY when the cassette redis is +reachable, PASSTHROUGH or DEGRADED otherwise) and a HIT/MISS line per request, +so a CI run shows whether it served from the cassette or went live instead of +silently degrading. """ from __future__ import annotations @@ -26,9 +31,13 @@ import base64 import hashlib import json +import logging import os from typing import Awaitable, Callable, List, Optional, Tuple +_LOGGER = logging.getLogger("openai_record_replay") +_LOGGER.setLevel(logging.INFO) + CASSETTE_TTL_SECONDS = 24 * 60 * 60 RECORD_KEY_PREFIX = "litellm:openai:record:" RECORDER_REDIS_URL_ENV = "CASSETTE_REDIS_URL" @@ -112,12 +121,21 @@ async def handle(self, method: str, path: str, body: bytes, fetch_upstream: Fetc key = self.record_key(method, path, body) cached = self._cache_get(key) if cached is not None: + _LOGGER.info("HIT replayed from cassette: %s %s", method, path) return cached status, headers, resp_body = await fetch_upstream() sanitized = _sanitize_headers(headers) - if 200 <= status < 300: - self._cache_set(key, status, sanitized, resp_body) + if not (200 <= status < 300): + _LOGGER.info("MISS forwarded live, not cached (status=%s): %s %s", status, method, path) + elif self._cache_set(key, status, sanitized, resp_body): + _LOGGER.info("MISS forwarded live and recorded: %s %s", method, path) + else: + _LOGGER.warning( + "MISS forwarded live but NOT recorded (redis unset or unreachable): %s %s", + method, + path, + ) return status, sanitized, resp_body def _cache_get(self, key: str) -> Optional[UpstreamResult]: @@ -138,9 +156,9 @@ def _cache_get(self, key: str) -> Optional[UpstreamResult]: return None return status, headers, resp_body - def _cache_set(self, key: str, status: int, headers: Headers, body: bytes) -> None: + def _cache_set(self, key: str, status: int, headers: Headers, body: bytes) -> bool: if self._redis is None: - return + return False payload = json.dumps( { "status": status, @@ -150,8 +168,31 @@ def _cache_set(self, key: str, status: int, headers: Headers, body: bytes) -> No ) try: self._redis.set(key, payload, ex=self._ttl_seconds) + return True except Exception: - pass + return False + + def log_startup_mode(self) -> None: + if self._redis is None: + _LOGGER.warning( + "PASSTHROUGH: %s unset, every request goes live to %s and nothing is cached", + RECORDER_REDIS_URL_ENV, + self.upstream_base_url, + ) + return + try: + self._redis.ping() + except Exception as exc: + _LOGGER.warning( + "DEGRADED to live: %s set but cassette redis unreachable (%s); nothing is cached", + RECORDER_REDIS_URL_ENV, + type(exc).__name__, + ) + return + _LOGGER.info( + "REPLAY mode: cassette redis reachable, recordings expire %ss after write (no refresh on read)", + self._ttl_seconds, + ) def _build_default_redis_client(): @@ -186,6 +227,7 @@ def create_app(recorder: Optional[OpenAIRecordReplay] = None, http_client=None): @contextlib.asynccontextmanager async def lifespan(_app): + recorder.log_startup_mode() try: yield finally: @@ -231,6 +273,7 @@ async def fetch_upstream() -> UpstreamResult: import uvicorn + logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(name)s %(message)s") parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--host", default="0.0.0.0") parser.add_argument("--port", type=int, default=8090) diff --git a/tests/llm_translation/test_openai_record_replay_proxy.py b/tests/llm_translation/test_openai_record_replay_proxy.py index 5e34621d0197..f2624c795752 100644 --- a/tests/llm_translation/test_openai_record_replay_proxy.py +++ b/tests/llm_translation/test_openai_record_replay_proxy.py @@ -1,6 +1,7 @@ from __future__ import annotations import asyncio +import logging import os import sys @@ -214,3 +215,53 @@ def test_app_lifespan_leaves_injected_http_client_open(): pass assert client.closed is False + + +def test_handle_logs_miss_then_hit(caplog): + """Each request self-reports so a CI run shows cassette vs live.""" + recorder = _recorder() + upstream = _Upstream() + body_in = b'{"model":"gpt-image-1"}' + + with caplog.at_level(logging.INFO, logger="openai_record_replay"): + _run(recorder.handle("POST", "/v1/images/generations", body_in, upstream)) + _run(recorder.handle("POST", "/v1/images/generations", body_in, upstream)) + + messages = [r.getMessage() for r in caplog.records] + assert any("MISS forwarded live and recorded" in m for m in messages) + assert any("HIT replayed from cassette" in m for m in messages) + + +def test_handle_warns_when_recording_not_persisted(caplog): + """A redis failure must surface loudly, not look like a successful record.""" + recorder = _recorder(_BoomRedis()) + upstream = _Upstream() + + with caplog.at_level(logging.WARNING, logger="openai_record_replay"): + _run(recorder.handle("POST", "/v1/images/generations", b'{"model":"gpt-image-1"}', upstream)) + + assert any(r.levelno == logging.WARNING and "NOT recorded" in r.getMessage() for r in caplog.records) + + +def test_log_startup_mode_distinguishes_replay_from_passthrough(caplog): + """Startup must announce whether the recorder will actually cache.""" + with caplog.at_level(logging.INFO, logger="openai_record_replay"): + OpenAIRecordReplay(None).log_startup_mode() + _recorder().log_startup_mode() + + emitted = [(r.levelno, r.getMessage()) for r in caplog.records] + assert any(lvl == logging.WARNING and "PASSTHROUGH" in m for lvl, m in emitted) + assert any(lvl == logging.INFO and "REPLAY mode" in m for lvl, m in emitted) + + +class _UnreachableRedis: + def ping(self): + raise ConnectionError("redis offline") + + +def test_log_startup_mode_warns_when_redis_configured_but_unreachable(caplog): + """A configured-but-dead redis must warn, not look like it will cache.""" + with caplog.at_level(logging.WARNING, logger="openai_record_replay"): + _recorder(_UnreachableRedis()).log_startup_mode() + + assert any(r.levelno == logging.WARNING and "DEGRADED" in r.getMessage() for r in caplog.records)