mirror of
https://github.com/BerriAI/litellm.git
synced 2026-08-28 05:25:59 +00:00
ec47bbaaaa
The record/replay harness stored a streamed provider response as one buffered body, so a replayed stream arrived coalesced and the /v1/messages streaming test could not be edge-wired. Keep each SSE transfer chunk in the bundle in the order the provider sent it (a new streamed response shape at BUNDLE_FORMAT_VERSION 4) so replay reproduces the provider's split points, the recorded usage chunk keeps its position, and a mid-stream upstream error replays as the same mid-stream error rather than a clean body. Resolves LIT-5742
231 lines
9.3 KiB
Python
231 lines
9.3 KiB
Python
"""Harness coverage for the on-disk fixture bundle format (LIT-5729/LIT-5745).
|
|
|
|
No proxy and no ``e2e`` marker: these pin the bundle CONTRACT - the seven-day
|
|
freshness gate that names the bundle's age, record mode's wipe safety (never
|
|
delete a directory that is not a bundle), collision-free per-test slugs, and
|
|
grouped-in-order loading - so replay can never silently drift from what
|
|
record wrote.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from datetime import datetime, timedelta, timezone
|
|
from pathlib import Path
|
|
|
|
from fixture_bundle import (
|
|
BUNDLE_FORMAT_VERSION,
|
|
MANIFEST_FILENAME,
|
|
MAX_BUNDLE_AGE,
|
|
BundleRecorder,
|
|
FreshBundle,
|
|
LoadedBundle,
|
|
Manifest,
|
|
RecordedHttpResponse,
|
|
RecordedRequest,
|
|
RecordedStreamedResponse,
|
|
StaleBundle,
|
|
UnreadableBundle,
|
|
UnsafeBundleDir,
|
|
check_freshness,
|
|
format_age,
|
|
interaction_filename,
|
|
load_bundle,
|
|
prepare_bundle,
|
|
slug_for_test,
|
|
)
|
|
|
|
NOW = datetime(2026, 8, 18, 12, 0, 0, tzinfo=timezone.utc)
|
|
|
|
|
|
def write_manifest(
|
|
root: Path, recorded_at: datetime, *, format_version: int = BUNDLE_FORMAT_VERSION
|
|
) -> None:
|
|
root.mkdir(parents=True, exist_ok=True)
|
|
manifest = Manifest(
|
|
format_version=format_version, recorded_at=recorded_at, harness_version="abc1234"
|
|
)
|
|
(root / MANIFEST_FILENAME).write_text(manifest.model_dump_json(), encoding="utf-8")
|
|
|
|
|
|
def prepared(root: Path) -> BundleRecorder:
|
|
recorder = prepare_bundle(root)
|
|
assert isinstance(recorder, BundleRecorder)
|
|
return recorder
|
|
|
|
|
|
def plain_request(path: str) -> RecordedRequest:
|
|
return RecordedRequest(method="post", path=path, headers={})
|
|
|
|
|
|
def plain_response() -> RecordedHttpResponse:
|
|
return RecordedHttpResponse(status_code=401, headers={}, body_b64="")
|
|
|
|
|
|
class TestFreshness:
|
|
def test_bundle_at_the_limit_is_still_fresh(self, tmp_path: Path) -> None:
|
|
root = tmp_path / "bundle"
|
|
write_manifest(root, NOW - MAX_BUNDLE_AGE)
|
|
assert isinstance(check_freshness(root, now=NOW), FreshBundle)
|
|
|
|
def test_stale_bundle_reports_age_and_limit(self, tmp_path: Path) -> None:
|
|
root = tmp_path / "bundle"
|
|
write_manifest(root, NOW - timedelta(days=8, hours=3))
|
|
freshness = check_freshness(root, now=NOW)
|
|
assert isinstance(freshness, StaleBundle)
|
|
assert freshness.age == timedelta(days=8, hours=3)
|
|
assert format_age(freshness.age) == "8d3h"
|
|
assert freshness.limit == MAX_BUNDLE_AGE
|
|
|
|
def test_naive_recorded_at_is_read_as_utc(self, tmp_path: Path) -> None:
|
|
root = tmp_path / "bundle"
|
|
write_manifest(root, (NOW - timedelta(days=1)).replace(tzinfo=None))
|
|
assert isinstance(check_freshness(root, now=NOW), FreshBundle)
|
|
|
|
def test_missing_manifest_is_unreadable_with_recording_hint(self, tmp_path: Path) -> None:
|
|
freshness = check_freshness(tmp_path / "absent", now=NOW)
|
|
assert isinstance(freshness, UnreadableBundle)
|
|
assert MANIFEST_FILENAME in freshness.reason
|
|
assert "E2E_FIXTURE_MODE=record" in freshness.reason
|
|
|
|
def test_corrupt_manifest_is_unreadable(self, tmp_path: Path) -> None:
|
|
root = tmp_path / "bundle"
|
|
root.mkdir()
|
|
(root / MANIFEST_FILENAME).write_text("{not json", encoding="utf-8")
|
|
assert isinstance(check_freshness(root, now=NOW), UnreadableBundle)
|
|
|
|
def test_unknown_format_version_is_unreadable(self, tmp_path: Path) -> None:
|
|
root = tmp_path / "bundle"
|
|
write_manifest(root, NOW, format_version=BUNDLE_FORMAT_VERSION + 1)
|
|
freshness = check_freshness(root, now=NOW)
|
|
assert isinstance(freshness, UnreadableBundle)
|
|
assert f"format_version {BUNDLE_FORMAT_VERSION + 1}" in freshness.reason
|
|
|
|
|
|
class TestPrepareBundle:
|
|
def test_fresh_directory_gets_a_fresh_manifest(self, tmp_path: Path) -> None:
|
|
root = tmp_path / "bundle"
|
|
prepared(root)
|
|
freshness = check_freshness(root, now=datetime.now(timezone.utc))
|
|
assert isinstance(freshness, FreshBundle)
|
|
assert freshness.manifest.format_version == BUNDLE_FORMAT_VERSION
|
|
assert freshness.manifest.harness_version
|
|
|
|
def test_record_wipes_the_previous_bundle_instead_of_reading_it(self, tmp_path: Path) -> None:
|
|
root = tmp_path / "bundle"
|
|
prepared(root).record(
|
|
test_key="old.py::test_old",
|
|
request=plain_request("/stale"),
|
|
response=plain_response(),
|
|
)
|
|
assert any(entry.is_dir() for entry in root.iterdir())
|
|
prepared(root)
|
|
assert {entry.name for entry in root.iterdir()} == {MANIFEST_FILENAME}
|
|
|
|
def test_refuses_to_wipe_a_directory_that_is_not_a_bundle(self, tmp_path: Path) -> None:
|
|
root = tmp_path / "precious"
|
|
root.mkdir()
|
|
(root / "notes.txt").write_text("keep me", encoding="utf-8")
|
|
outcome = prepare_bundle(root)
|
|
assert isinstance(outcome, UnsafeBundleDir)
|
|
assert MANIFEST_FILENAME in outcome.reason
|
|
assert (root / "notes.txt").read_text(encoding="utf-8") == "keep me"
|
|
|
|
def test_refuses_a_path_that_is_a_file(self, tmp_path: Path) -> None:
|
|
target = tmp_path / "not-a-dir"
|
|
target.write_text("x", encoding="utf-8")
|
|
outcome = prepare_bundle(target)
|
|
assert isinstance(outcome, UnsafeBundleDir)
|
|
assert "not a directory" in outcome.reason
|
|
|
|
|
|
class TestSlugs:
|
|
def test_slug_for_test_is_deterministic(self) -> None:
|
|
key = "tests/e2e/suite/test_mod.py::TestX::test_case"
|
|
assert slug_for_test(key) == slug_for_test(key)
|
|
|
|
def test_same_tail_in_different_files_never_collides(self) -> None:
|
|
first = slug_for_test("tests/e2e/a/test_a.py::test_case")
|
|
second = slug_for_test("tests/e2e/b/test_b.py::test_case")
|
|
assert first != second
|
|
assert first.startswith("test_case-")
|
|
assert second.startswith("test_case-")
|
|
|
|
def test_interaction_filename_orders_and_slugs(self) -> None:
|
|
request = RecordedRequest(method="post", path="/chat/completions", headers={})
|
|
assert interaction_filename(3, request) == "0003-post-chat-completions.json"
|
|
|
|
|
|
class TestRecordAndLoad:
|
|
def test_load_returns_interactions_in_recorded_order(self, tmp_path: Path) -> None:
|
|
root = tmp_path / "bundle"
|
|
recorder = prepared(root)
|
|
key = "suite/test_mod.py::test_ordered"
|
|
for path in ("/first", "/second", "/third"):
|
|
recorder.record(
|
|
test_key=key,
|
|
request=plain_request(path),
|
|
response=plain_response(),
|
|
)
|
|
loaded = load_bundle(root)
|
|
assert isinstance(loaded, LoadedBundle)
|
|
assert [
|
|
interaction.request.path for interaction in loaded.interactions[slug_for_test(key)]
|
|
] == ["/first", "/second", "/third"]
|
|
|
|
def test_interactions_group_per_test(self, tmp_path: Path) -> None:
|
|
root = tmp_path / "bundle"
|
|
recorder = prepared(root)
|
|
for key in ("suite/test_a.py::test_one", "suite/test_b.py::test_two"):
|
|
recorder.record(
|
|
test_key=key,
|
|
request=plain_request(f"/{key[-3:]}"),
|
|
response=plain_response(),
|
|
)
|
|
loaded = load_bundle(root)
|
|
assert isinstance(loaded, LoadedBundle)
|
|
assert set(loaded.interactions) == {
|
|
slug_for_test("suite/test_a.py::test_one"),
|
|
slug_for_test("suite/test_b.py::test_two"),
|
|
}
|
|
|
|
def test_a_streamed_response_round_trips_through_the_bundle(self, tmp_path: Path) -> None:
|
|
"""LIT-5742: the two response shapes share one file format and are told apart
|
|
by their ``kind`` tag, so a streamed recording comes back with its chunk list
|
|
intact rather than as a buffered response with an empty body."""
|
|
root = tmp_path / "bundle"
|
|
recorder = prepared(root)
|
|
key = "suite/test_mod.py::test_streamed"
|
|
recorder.record(
|
|
test_key=key,
|
|
request=plain_request("/messages"),
|
|
response=RecordedStreamedResponse(
|
|
status_code=200,
|
|
headers={"content-type": "text/event-stream"},
|
|
chunks_b64=["Zmly", "c3Q="],
|
|
truncated="upstream: hung up",
|
|
),
|
|
)
|
|
loaded = load_bundle(root)
|
|
assert isinstance(loaded, LoadedBundle)
|
|
(interaction,) = loaded.interactions[slug_for_test(key)]
|
|
response = interaction.response
|
|
assert isinstance(response, RecordedStreamedResponse)
|
|
assert response.chunks_b64 == ["Zmly", "c3Q="]
|
|
assert response.truncated == "upstream: hung up"
|
|
|
|
def test_load_bundle_rejects_a_foreign_format_version(self, tmp_path: Path) -> None:
|
|
"""A bundle is written atomically, so a manifest from another format version
|
|
means every response inside it may have a shape this code cannot read. Loading
|
|
has to refuse it by name, the way the freshness gate does, rather than parse
|
|
what it happens to understand."""
|
|
root = tmp_path / "bundle"
|
|
prepared(root).record(
|
|
test_key="suite/test_mod.py::test_old",
|
|
request=plain_request("/chat"),
|
|
response=plain_response(),
|
|
)
|
|
write_manifest(root, NOW, format_version=BUNDLE_FORMAT_VERSION - 1)
|
|
loaded = load_bundle(root)
|
|
assert isinstance(loaded, UnreadableBundle)
|
|
assert f"format_version {BUNDLE_FORMAT_VERSION - 1}" in loaded.reason
|
|
assert "E2E_FIXTURE_MODE=record" in loaded.reason
|