litellm/tests/e2e/test_fixture_bundle.py

"""Harness coverage for the on-disk fixture bundle format (LIT-5729).

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
lossless Result round-trips - so replay can never silently drift from what
record wrote.
"""

from __future__ import annotations

from datetime import datetime, timedelta, timezone
from pathlib import Path

import pytest
from pydantic import BaseModel

from e2e_http import (
    NetworkError,
    RateLimitedError,
    Result,
    Success,
    UnauthorizedError,
    UnknownApiError,
    ValidationError,
)
from fixture_bundle import (
    BUNDLE_FORMAT_VERSION,
    MANIFEST_FILENAME,
    MAX_BUNDLE_AGE,
    BundleRecorder,
    FreshBundle,
    LoadedBundle,
    Manifest,
    RecordedRequest,
    RecordedResult,
    StaleBundle,
    UnreadableBundle,
    UnsafeBundleDir,
    check_freshness,
    format_age,
    from_result,
    interaction_filename,
    load_bundle,
    prepare_bundle,
    slug_for_test,
    to_result,
)

NOW = datetime(2026, 8, 18, 12, 0, 0, tzinfo=timezone.utc)


class Payload(BaseModel):
    value: str


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={})


class TestResultRoundTrip:
    @pytest.mark.parametrize(
        "result",
        [
            Success(status_code=201, data=Payload(value="ok")),
            NetworkError(message="connection refused"),
            UnauthorizedError(),
            RateLimitedError(retry_after_seconds=7, body="slow down"),
            ValidationError(message="bad shape"),
            UnknownApiError(status_code=502, body="upstream exploded"),
        ],
    )
    def test_every_result_kind_survives_disk_and_back(self, result: Result[Payload]) -> None:
        assert to_result(from_result(result), Payload) == result


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=RecordedResult(kind="unauthorized"),
        )
        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=RecordedResult(kind="unauthorized"),
            )
        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=RecordedResult(kind="unauthorized"),
            )
        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"),
        }