diff --git a/tests/_openai_record_replay_proxy.py b/tests/_openai_record_replay_proxy.py index 60832188a75..1413b1ab746 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 5e34621d019..f2624c79575 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)