tests(vcr): only persist cassette on test pass to avoid poisoning cache

A test that fails (incl. all the failing retries before a passing one)
can otherwise overwrite a known-good cassette with a 'bad luck'
recording. Tests like test_prompt_caching, which assert on provider
state across two calls, can produce a 200 response that semantically
fails the assertion — the 2xx filter doesn't catch this because the
HTTP layer is fine.

- pytest_runtest_makereport hook attaches each phase report to the
  pytest item.
- _vcr_outcome_gate fixture (combining the verbose-mode reporter)
  reads the call-phase outcome at teardown and informs the persister
  via mark_test_outcome_for_cassette before vcrpy's Cassette.__exit__
  triggers save_cassette.
- save_cassette consults the per-key 'did the test pass?' flag and
  short-circuits when False, leaving any prior good recording intact.
- Defaults to passed=True when no marker is present so non-test
  usage of the persister still works.

tests/_vcr_redis_persister.py

@@ -13,6 +13,25 @@ REDIS_KEY_PREFIX = "litellm:vcr:cassette:"
 _log = logging.getLogger(__name__)
 
 
+# Per-process map: cassette key -> "did the test that produced this cassette
+# pass?". The conftest's pytest_runtest_makereport hook sets True when the test
+# body succeeds; save_cassette consults it to avoid persisting recordings from
+# failed runs. We key by the redis cache key so retries (which may produce a
+# fresh cassette object each time but write to the same key) interleave
+# correctly.
+_passed_by_cassette_key: dict[str, bool] = {}
+
+
+def mark_test_outcome_for_cassette(cassette_path: str, passed: bool) -> None:
+    """Record whether the test that owns ``cassette_path`` passed.
+
+    Called from a pytest hook in conftest. The recorded value is consulted by
+    ``save_cassette`` so failed-attempt recordings (e.g. a flaky test that
+    asserts on provider state) don't poison the cache for future runs.
+    """
+    _passed_by_cassette_key[redis_key_for(cassette_path)] = passed
+
+
 def redis_key_for(cassette_path: str) -> str:
     rel = os.path.relpath(str(cassette_path))
     if rel.endswith(".yaml"):
@@ -96,10 +115,26 @@ def make_redis_persister(
 
         @staticmethod
         def save_cassette(cassette_path, cassette_dict, serializer):
+            key = redis_key_for(cassette_path)
+            # Only persist successful runs. A failed test (incl. all the failed
+            # retries before a passing one) would otherwise poison the cache —
+            # e.g. a flaky test that observes provider state across two calls
+            # could capture a "bad luck" response that deterministically fails
+            # every future replay. We default to True if the hook didn't run
+            # (e.g. cassette saved outside a test context) so non-test usage
+            # still works.
+            passed = _passed_by_cassette_key.pop(key, True)
+            if not passed:
+                _log.info(
+                    "VCR redis save skipped for %s; test did not pass — "
+                    "leaving any prior cassette intact",
+                    cassette_path,
+                )
+                return
             data = serialize(cassette_dict, serializer)
             payload = data.encode("utf-8") if isinstance(data, str) else data
             try:
-                redis_client.set(redis_key_for(cassette_path), payload, ex=ttl_seconds)
+                redis_client.set(key, payload, ex=ttl_seconds)
             except _transient_errors as exc:
                 # Cassette persistence is a cache, not test correctness. A Redis
                 # outage on save should not fail an otherwise-passing test —

tests/llm_responses_api_testing/conftest.py

@@ -17,6 +17,7 @@ from tests._vcr_redis_persister import (  # noqa: E402
     filter_non_2xx_response,
     format_vcr_verdict,
     make_redis_persister,
+    mark_test_outcome_for_cassette,
     patch_vcrpy_aiohttp_record_path,
     vcr_verbose_enabled,
 )
@@ -104,16 +105,41 @@ def pytest_recording_configure(config, vcr):
     patch_vcrpy_aiohttp_record_path()
 
 
-@pytest.fixture(autouse=True)
-def _vcr_hit_miss_report(request, vcr):
-    """When LITELLM_VCR_VERBOSE=1, print a one-line cassette verdict per test.
+@pytest.hookimpl(hookwrapper=True)
+def pytest_runtest_makereport(item, call):
+    """Attach each phase's report to the item so fixture teardown can read it.
 
-    Runs after the `vcr` fixture (which yields the active Cassette), so we can
-    inspect play_count / dirty / len in teardown."""
+    Used by ``_vcr_outcome_gate`` below to skip persisting cassettes for
+    failed test runs (incl. failed retries that pytest-rerunfailures will
+    re-attempt) so a "bad luck" recording can't poison future replays.
+    """
+    outcome = yield
+    rep = outcome.get_result()
+    setattr(item, f"rep_{rep.when}", rep)
+
+
+@pytest.fixture(autouse=True)
+def _vcr_outcome_gate(request, vcr):
+    """Tell the persister whether the test that owns this cassette passed.
+
+    Runs after ``vcr`` (which yields the active Cassette). At teardown time
+    the call-phase report is attached to the item by the makereport hook
+    above, so we can mark the cassette key passed/failed before vcrpy's
+    Cassette.__exit__ triggers persister.save_cassette.
+
+    Also prints a per-test hit/miss verdict when LITELLM_VCR_VERBOSE=1.
+    """
     yield
+    cassette = vcr  # name kept for the verbose-output line
+    rep_call = getattr(request.node, "rep_call", None)
+    test_passed = bool(rep_call and rep_call.passed)
+    cassette_path = getattr(cassette, "_path", None) if cassette is not None else None
+    if cassette_path:
+        mark_test_outcome_for_cassette(cassette_path, test_passed)
+
     if not vcr_verbose_enabled():
         return
-    verdict = format_vcr_verdict(vcr)
+    verdict = format_vcr_verdict(cassette)
     reporter = request.config.pluginmanager.get_plugin("terminalreporter")
     line = f"{verdict} :: {request.node.nodeid}"
     if reporter is not None:

tests/llm_translation/conftest.py

@@ -22,6 +22,7 @@ from tests._vcr_redis_persister import (  # noqa: E402
     filter_non_2xx_response,
     format_vcr_verdict,
     make_redis_persister,
+    mark_test_outcome_for_cassette,
     patch_vcrpy_aiohttp_record_path,
     vcr_verbose_enabled,
 )
@@ -128,16 +129,41 @@ def pytest_recording_configure(config, vcr):
     patch_vcrpy_aiohttp_record_path()
 
 
-@pytest.fixture(autouse=True)
-def _vcr_hit_miss_report(request, vcr):
-    """When LITELLM_VCR_VERBOSE=1, print a one-line cassette verdict per test.
+@pytest.hookimpl(hookwrapper=True)
+def pytest_runtest_makereport(item, call):
+    """Attach each phase's report to the item so fixture teardown can read it.
 
-    Runs after the `vcr` fixture (which yields the active Cassette), so we can
-    inspect play_count / dirty / len in teardown."""
+    Used by ``_vcr_outcome_gate`` below to skip persisting cassettes for
+    failed test runs (incl. failed retries that pytest-rerunfailures will
+    re-attempt) so a "bad luck" recording can't poison future replays.
+    """
+    outcome = yield
+    rep = outcome.get_result()
+    setattr(item, f"rep_{rep.when}", rep)
+
+
+@pytest.fixture(autouse=True)
+def _vcr_outcome_gate(request, vcr):
+    """Tell the persister whether the test that owns this cassette passed.
+
+    Runs after ``vcr`` (which yields the active Cassette). At teardown time
+    the call-phase report is attached to the item by the makereport hook
+    above, so we can mark the cassette key passed/failed before vcrpy's
+    Cassette.__exit__ triggers persister.save_cassette.
+
+    Also prints a per-test hit/miss verdict when LITELLM_VCR_VERBOSE=1.
+    """
     yield
+    cassette = vcr  # name kept for the verbose-output line
+    rep_call = getattr(request.node, "rep_call", None)
+    test_passed = bool(rep_call and rep_call.passed)
+    cassette_path = getattr(cassette, "_path", None) if cassette is not None else None
+    if cassette_path:
+        mark_test_outcome_for_cassette(cassette_path, test_passed)
+
     if not vcr_verbose_enabled():
         return
-    verdict = format_vcr_verdict(vcr)
+    verdict = format_vcr_verdict(cassette)
     reporter = request.config.pluginmanager.get_plugin("terminalreporter")
     line = f"{verdict} :: {request.node.nodeid}"
     if reporter is not None:

tests/llm_translation/test_vcr_redis_persister.py

@@ -16,6 +16,7 @@ from tests._vcr_redis_persister import (  # noqa: E402
     CASSETTE_TTL_SECONDS,
     filter_non_2xx_response,
     make_redis_persister,
+    mark_test_outcome_for_cassette,
     redis_key_for,
 )
 
@@ -113,6 +114,55 @@ def test_save_swallows_connection_errors_so_teardown_does_not_fail():
     )
 
 
+def test_save_skipped_when_test_marked_failed_and_prior_cassette_preserved():
+    # A flaky test that fails should NOT overwrite a previously-good cassette.
+    fake, persister = _persister_with_fake_redis()
+    cassette_id = "tests/llm_translation/test_x/test_flaky"
+    key = redis_key_for(cassette_id)
+
+    # Seed a "known-good" recording from a prior successful run.
+    good = _sample_cassette_dict()
+    persister.save_cassette(cassette_id, good, yamlserializer)
+    good_payload = fake.get(key)
+    assert good_payload is not None
+
+    # Simulate a failed run: the hook records "did not pass" before save.
+    mark_test_outcome_for_cassette(cassette_id, passed=False)
+    bad_response = {
+        "status": {"code": 200, "message": "OK"},
+        "headers": {},
+        "body": {"string": b'{"id":"BAD","type":"message"}'},
+    }
+    bad = {"requests": good["requests"], "responses": [bad_response]}
+    persister.save_cassette(cassette_id, bad, yamlserializer)
+
+    # Prior good payload is still there — the bad save was suppressed.
+    assert fake.get(key) == good_payload
+
+
+def test_save_proceeds_when_test_marked_passed():
+    fake, persister = _persister_with_fake_redis()
+    cassette_id = "tests/llm_translation/test_x/test_passed"
+    key = redis_key_for(cassette_id)
+
+    mark_test_outcome_for_cassette(cassette_id, passed=True)
+    persister.save_cassette(cassette_id, _sample_cassette_dict(), yamlserializer)
+
+    assert fake.get(key) is not None
+
+
+def test_save_proceeds_when_outcome_unknown():
+    # Used outside a pytest run (e.g. ad-hoc scripts), the outcome gate is
+    # bypassed so the persister still works.
+    fake, persister = _persister_with_fake_redis()
+    cassette_id = "tests/llm_translation/test_x/test_no_marker"
+    key = redis_key_for(cassette_id)
+
+    persister.save_cassette(cassette_id, _sample_cassette_dict(), yamlserializer)
+
+    assert fake.get(key) is not None
+
+
 def test_load_treats_connection_errors_as_cassette_miss():
     # An outage on read should fall through to a live call (CassetteNotFound),
     # not surface a redis exception in the test setup.