mirror of
https://github.com/BerriAI/litellm.git
synced 2026-09-12 23:01:41 +00:00
refactor(tests): clarify fixture recording pipeline
This commit is contained in:
Yujong Lee
parent
e7b88562e9
commit
65cc43728d
24 changed files with 931 additions and 694 deletions
|
|
@ -34,6 +34,25 @@
|
|||
- Provider responses are replayed unchanged, so the parity test does not fuzz or validate provider behavior
|
||||
- Because Hypothesis does not run the parity assertion directly, parity failures are not automatically shrunk
|
||||
|
||||
## Recording fixtures
|
||||
|
||||
The recording command runs four explicit stages:
|
||||
|
||||
1. Generate deterministic SDK inputs for every configured OCR target
|
||||
2. Build target-scoped, deduplicated recording jobs
|
||||
3. Record upstream responses through one globally bounded worker pool
|
||||
4. Persist each fixture and report whether it was recorded, cached, or failed
|
||||
|
||||
Run it with:
|
||||
|
||||
```shell
|
||||
uv run python -m tests.test_litellm.ocr.fixtures.record --examples 4 --concurrency 4
|
||||
```
|
||||
|
||||
`--concurrency` caps all provider calls across all targets. Each completed job is reported immediately, and the final
|
||||
summary reports recorded, cached, and failed totals. Independent jobs finish after a failure, then the command exits
|
||||
nonzero if any job failed
|
||||
|
||||
## OCR input boundaries
|
||||
|
||||
OCR strategies generate only public `litellm.ocr()` and `litellm.aocr()` inputs. Every case contains the normalized
|
||||
|
|
|
|||
|
|
@ -1,103 +0,0 @@
|
|||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import logging
|
||||
from collections.abc import Mapping, Sequence
|
||||
from dataclasses import dataclass
|
||||
from pathlib import Path
|
||||
from typing import Final, Generic, Protocol, TypeVar, cast
|
||||
|
||||
from hypothesis.strategies import SearchStrategy
|
||||
from pydantic import BaseModel
|
||||
|
||||
from tests.route_parity.fixture_models import SdkInputBase
|
||||
from tests.route_parity.fixture_recorder import ProviderSpec, generate_case_inputs, record_cases
|
||||
|
||||
LOGGER: Final = logging.getLogger(__name__)
|
||||
InputT = TypeVar("InputT", bound=SdkInputBase)
|
||||
InputT_contra = TypeVar("InputT_contra", bound=SdkInputBase, contravariant=True)
|
||||
DependencyT_contra = TypeVar("DependencyT_contra", contravariant=True)
|
||||
CaseT = TypeVar("CaseT", bound=BaseModel)
|
||||
|
||||
|
||||
@dataclass(frozen=True, slots=True)
|
||||
class GeneratorArgs:
|
||||
concurrency: int
|
||||
examples: int
|
||||
fixture_dir: Path | None
|
||||
|
||||
|
||||
@dataclass(frozen=True, slots=True)
|
||||
class FixtureTarget(Generic[InputT]):
|
||||
name: str
|
||||
provider_spec: ProviderSpec
|
||||
strategy: SearchStrategy[InputT]
|
||||
invocation: FixtureInvocation[InputT]
|
||||
required_inputs: tuple[InputT, ...] = ()
|
||||
|
||||
|
||||
class FixtureInvocation(Protocol[InputT_contra]):
|
||||
def execute(self, provider_url: str, case_input: InputT_contra) -> None: ...
|
||||
|
||||
|
||||
class FixtureSource(Protocol[InputT, DependencyT_contra]):
|
||||
def targets(
|
||||
self,
|
||||
environ: Mapping[str, str],
|
||||
dependency: DependencyT_contra,
|
||||
/,
|
||||
) -> tuple[FixtureTarget[InputT], ...]: ...
|
||||
|
||||
|
||||
def discover_fixture_targets(
|
||||
sources: tuple[FixtureSource[InputT, DependencyT_contra], ...],
|
||||
environ: Mapping[str, str],
|
||||
dependency: DependencyT_contra,
|
||||
) -> tuple[FixtureTarget[InputT], ...]:
|
||||
return tuple(target for source in sources for target in source.targets(environ, dependency))
|
||||
|
||||
|
||||
def generate_target_fixtures(
|
||||
target: FixtureTarget[InputT],
|
||||
root: Path,
|
||||
examples: int,
|
||||
concurrency: int,
|
||||
case_type: type[CaseT],
|
||||
) -> None:
|
||||
generated_inputs: Final = generate_case_inputs(target.strategy, examples)
|
||||
case_inputs: Final = (*target.required_inputs, *generated_inputs)
|
||||
results: Final = record_cases(
|
||||
target.provider_spec,
|
||||
root / target.name,
|
||||
case_inputs,
|
||||
target.invocation.execute,
|
||||
case_type,
|
||||
concurrency,
|
||||
)
|
||||
for result in results:
|
||||
LOGGER.info(
|
||||
"%s %s",
|
||||
"cached" if result.cache_hit else "recorded",
|
||||
target.name,
|
||||
)
|
||||
|
||||
|
||||
def require_targets(
|
||||
targets: tuple[FixtureTarget[InputT], ...], error_message: str
|
||||
) -> tuple[FixtureTarget[InputT], ...]:
|
||||
if targets:
|
||||
return targets
|
||||
raise SystemExit(error_message)
|
||||
|
||||
|
||||
def parse_generator_args(argv: Sequence[str] | None = None) -> GeneratorArgs:
|
||||
parser: Final = argparse.ArgumentParser()
|
||||
parser.add_argument("--concurrency", type=int, default=4)
|
||||
parser.add_argument("--examples", type=int, default=4)
|
||||
parser.add_argument("--fixture-dir", type=Path)
|
||||
namespace: Final = parser.parse_args(argv)
|
||||
return GeneratorArgs(
|
||||
concurrency=cast(int, namespace.concurrency),
|
||||
examples=cast(int, namespace.examples),
|
||||
fixture_dir=cast(Path | None, namespace.fixture_dir),
|
||||
)
|
||||
1
tests/route_parity/fixtures/__init__.py
Normal file
1
tests/route_parity/fixtures/__init__.py
Normal file
|
|
@ -0,0 +1 @@
|
|||
from __future__ import annotations
|
||||
22
tests/route_parity/fixtures/inputs.py
Normal file
22
tests/route_parity/fixtures/inputs.py
Normal file
|
|
@ -0,0 +1,22 @@
|
|||
from __future__ import annotations
|
||||
|
||||
import queue
|
||||
from typing import Final, TypeVar
|
||||
|
||||
from hypothesis import given, settings
|
||||
from hypothesis.strategies import SearchStrategy
|
||||
|
||||
InputT = TypeVar("InputT")
|
||||
|
||||
|
||||
def generate_case_inputs(strategy: SearchStrategy[InputT], examples: int) -> tuple[InputT, ...]:
|
||||
generated: Final[queue.SimpleQueue[InputT | None]] = queue.SimpleQueue()
|
||||
|
||||
@settings(max_examples=examples, deadline=None, derandomize=True)
|
||||
@given(case_input=strategy)
|
||||
def generate_case(case_input: InputT) -> None:
|
||||
generated.put(case_input)
|
||||
|
||||
generate_case()
|
||||
generated.put(None)
|
||||
return tuple(iter(generated.get, None))
|
||||
221
tests/route_parity/fixtures/pipeline.py
Normal file
221
tests/route_parity/fixtures/pipeline.py
Normal file
|
|
@ -0,0 +1,221 @@
|
|||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import logging
|
||||
from collections.abc import Sequence
|
||||
from concurrent.futures import Future, ThreadPoolExecutor, as_completed
|
||||
from dataclasses import dataclass, field
|
||||
from pathlib import Path
|
||||
from types import MappingProxyType
|
||||
from typing import Final, Generic, Literal, Protocol, TypeVar, cast
|
||||
|
||||
from hypothesis.strategies import SearchStrategy
|
||||
from pydantic import BaseModel
|
||||
|
||||
from tests.route_parity.fixtures.inputs import generate_case_inputs
|
||||
from tests.route_parity.fixtures.recording import ProviderSpec, record_upstream_responses
|
||||
from tests.route_parity.fixtures.store import (
|
||||
FixtureInput,
|
||||
canonical_json,
|
||||
fixture_cache_key,
|
||||
fixture_id,
|
||||
fixture_path,
|
||||
load_fixture,
|
||||
save_fixture,
|
||||
)
|
||||
|
||||
LOGGER: Final = logging.getLogger(__name__)
|
||||
InputT = TypeVar("InputT", bound=FixtureInput)
|
||||
InputT_contra = TypeVar("InputT_contra", bound=FixtureInput, contravariant=True)
|
||||
CaseT = TypeVar("CaseT", bound=BaseModel)
|
||||
|
||||
|
||||
@dataclass(frozen=True, slots=True)
|
||||
class RecordingArgs:
|
||||
concurrency: int
|
||||
examples: int
|
||||
fixture_dir: Path | None
|
||||
|
||||
|
||||
class RecordingInvocation(Protocol[InputT_contra]):
|
||||
def execute(self, provider_url: str, case_input: InputT_contra) -> None: ...
|
||||
|
||||
|
||||
@dataclass(frozen=True, slots=True)
|
||||
class RecordingTarget(Generic[InputT]):
|
||||
name: str
|
||||
provider_spec: ProviderSpec
|
||||
strategy: SearchStrategy[InputT]
|
||||
invocation: RecordingInvocation[InputT] = field(repr=False)
|
||||
required_inputs: tuple[InputT, ...] = ()
|
||||
|
||||
|
||||
@dataclass(frozen=True, slots=True)
|
||||
class RecordingJob(Generic[InputT]):
|
||||
target_name: str
|
||||
directory: Path
|
||||
provider_spec: ProviderSpec
|
||||
case_input: InputT
|
||||
invocation: RecordingInvocation[InputT] = field(repr=False)
|
||||
|
||||
@property
|
||||
def case_id(self) -> str:
|
||||
return fixture_id(self.case_input, self.target_name)
|
||||
|
||||
|
||||
@dataclass(frozen=True, slots=True)
|
||||
class RecordedFixture:
|
||||
target_name: str
|
||||
case_id: str
|
||||
path: Path
|
||||
kind: Literal["recorded"] = field(default="recorded", init=False)
|
||||
|
||||
|
||||
@dataclass(frozen=True, slots=True)
|
||||
class CachedFixture:
|
||||
target_name: str
|
||||
case_id: str
|
||||
path: Path
|
||||
kind: Literal["cached"] = field(default="cached", init=False)
|
||||
|
||||
|
||||
@dataclass(frozen=True, slots=True)
|
||||
class FailedFixture:
|
||||
target_name: str
|
||||
case_id: str
|
||||
error: Exception = field(repr=False)
|
||||
kind: Literal["failed"] = field(default="failed", init=False)
|
||||
|
||||
|
||||
RecordingOutcome = RecordedFixture | CachedFixture | FailedFixture
|
||||
|
||||
|
||||
@dataclass(frozen=True, slots=True)
|
||||
class RecordingSummary:
|
||||
recorded: tuple[RecordedFixture, ...]
|
||||
cached: tuple[CachedFixture, ...]
|
||||
failed: tuple[FailedFixture, ...]
|
||||
|
||||
@property
|
||||
def exit_code(self) -> int:
|
||||
return 1 if self.failed else 0
|
||||
|
||||
|
||||
def _unique_inputs(target: RecordingTarget[InputT], examples: int) -> tuple[InputT, ...]:
|
||||
generated_inputs: Final = generate_case_inputs(target.strategy, examples)
|
||||
case_inputs: Final = (*target.required_inputs, *generated_inputs)
|
||||
return tuple({canonical_json(fixture_cache_key(case_input)): case_input for case_input in case_inputs}.values())
|
||||
|
||||
|
||||
def build_recording_jobs(
|
||||
targets: tuple[RecordingTarget[InputT], ...],
|
||||
root: Path,
|
||||
examples: int,
|
||||
) -> tuple[RecordingJob[InputT], ...]:
|
||||
if examples < 1:
|
||||
raise ValueError("examples must be at least 1")
|
||||
return tuple(
|
||||
RecordingJob(
|
||||
target_name=target.name,
|
||||
directory=root / target.name,
|
||||
provider_spec=target.provider_spec,
|
||||
case_input=case_input,
|
||||
invocation=target.invocation,
|
||||
)
|
||||
for target in targets
|
||||
for case_input in _unique_inputs(target, examples)
|
||||
)
|
||||
|
||||
|
||||
def _record_job(job: RecordingJob[InputT], case_type: type[CaseT]) -> RecordedFixture | CachedFixture:
|
||||
cached: Final = load_fixture(job.directory, job.case_input, case_type)
|
||||
if cached is not None:
|
||||
return CachedFixture(
|
||||
target_name=job.target_name,
|
||||
case_id=job.case_id,
|
||||
path=fixture_path(job.directory, job.case_input),
|
||||
)
|
||||
provider_responses: Final = record_upstream_responses(
|
||||
job.provider_spec,
|
||||
job.case_input,
|
||||
job.invocation.execute,
|
||||
)
|
||||
case: Final = case_type.model_validate({"litellm_input": job.case_input, "provider_responses": provider_responses})
|
||||
path: Final = save_fixture(job.directory, job.case_input, case)
|
||||
return RecordedFixture(target_name=job.target_name, case_id=job.case_id, path=path)
|
||||
|
||||
|
||||
def _completed_outcome(
|
||||
completed: int,
|
||||
total: int,
|
||||
job: RecordingJob[InputT],
|
||||
future: Future[RecordedFixture | CachedFixture],
|
||||
) -> RecordingOutcome:
|
||||
try:
|
||||
outcome: Final = future.result()
|
||||
except Exception as error:
|
||||
failed: Final = FailedFixture(target_name=job.target_name, case_id=job.case_id, error=error)
|
||||
LOGGER.error(
|
||||
"[%d/%d] failed %s %s: %s",
|
||||
completed,
|
||||
total,
|
||||
failed.target_name,
|
||||
failed.case_id,
|
||||
type(error).__name__,
|
||||
)
|
||||
return failed
|
||||
LOGGER.info("[%d/%d] %s %s %s", completed, total, outcome.kind, outcome.target_name, outcome.case_id)
|
||||
return outcome
|
||||
|
||||
|
||||
def record_fixtures(
|
||||
targets: tuple[RecordingTarget[InputT], ...],
|
||||
root: Path,
|
||||
examples: int,
|
||||
concurrency: int,
|
||||
case_type: type[CaseT],
|
||||
) -> RecordingSummary:
|
||||
if concurrency < 1:
|
||||
raise ValueError("concurrency must be at least 1")
|
||||
jobs: Final = build_recording_jobs(targets, root, examples)
|
||||
total: Final = len(jobs)
|
||||
LOGGER.info("Recording %d fixtures across %d targets with concurrency %d", total, len(targets), concurrency)
|
||||
with ThreadPoolExecutor(max_workers=concurrency) as executor:
|
||||
future_jobs: Final = MappingProxyType({executor.submit(_record_job, job, case_type): job for job in jobs})
|
||||
outcomes: Final = tuple(
|
||||
_completed_outcome(completed, total, future_jobs[future], future)
|
||||
for completed, future in enumerate(as_completed(future_jobs), start=1)
|
||||
)
|
||||
summary: Final = RecordingSummary(
|
||||
recorded=tuple(outcome for outcome in outcomes if isinstance(outcome, RecordedFixture)),
|
||||
cached=tuple(outcome for outcome in outcomes if isinstance(outcome, CachedFixture)),
|
||||
failed=tuple(outcome for outcome in outcomes if isinstance(outcome, FailedFixture)),
|
||||
)
|
||||
LOGGER.info(
|
||||
"Finished %d fixtures: %d recorded, %d cached, %d failed",
|
||||
total,
|
||||
len(summary.recorded),
|
||||
len(summary.cached),
|
||||
len(summary.failed),
|
||||
)
|
||||
return summary
|
||||
|
||||
|
||||
def _positive_int(value: str) -> int:
|
||||
parsed: Final = int(value)
|
||||
if parsed < 1:
|
||||
raise argparse.ArgumentTypeError("must be at least 1")
|
||||
return parsed
|
||||
|
||||
|
||||
def parse_recording_args(argv: Sequence[str] | None = None) -> RecordingArgs:
|
||||
parser: Final = argparse.ArgumentParser()
|
||||
parser.add_argument("--concurrency", type=_positive_int, default=4)
|
||||
parser.add_argument("--examples", type=_positive_int, default=4)
|
||||
parser.add_argument("--fixture-dir", type=Path)
|
||||
namespace: Final = parser.parse_args(argv)
|
||||
return RecordingArgs(
|
||||
concurrency=cast(int, namespace.concurrency),
|
||||
examples=cast(int, namespace.examples),
|
||||
fixture_dir=cast(Path | None, namespace.fixture_dir),
|
||||
)
|
||||
|
|
@ -1,26 +1,16 @@
|
|||
from __future__ import annotations
|
||||
|
||||
import hashlib
|
||||
import os
|
||||
import queue
|
||||
import threading
|
||||
from collections.abc import Callable, Generator, Iterable
|
||||
from concurrent.futures import ThreadPoolExecutor
|
||||
from contextlib import contextmanager
|
||||
from dataclasses import dataclass
|
||||
from datetime import datetime, timezone
|
||||
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
|
||||
from pathlib import Path
|
||||
from typing import Final, Generic, Protocol, TypeVar, cast
|
||||
from typing import Final, TypeVar, cast
|
||||
from urllib.parse import urlsplit, urlunsplit
|
||||
|
||||
import httpx
|
||||
import pytest
|
||||
from hypothesis import given, settings
|
||||
from hypothesis.strategies import SearchStrategy
|
||||
from pydantic import AwareDatetime, BaseModel, ConfigDict, ValidationError
|
||||
|
||||
from tests.route_parity.json_file_cache import JsonFileCache, canonical_json
|
||||
from tests.route_parity.recorded_http import (
|
||||
HttpHeader,
|
||||
RecordedHttpResponse,
|
||||
|
|
@ -29,7 +19,6 @@ from tests.route_parity.recorded_http import (
|
|||
RecordedStreamChunk,
|
||||
)
|
||||
|
||||
FIXTURE_SCHEMA_VERSION: Final = 1
|
||||
_PARITY_PROVIDER_HOST: Final = "parity-provider.invalid"
|
||||
|
||||
_HOP_BY_HOP_HEADERS: Final = frozenset(
|
||||
|
|
@ -46,12 +35,7 @@ _HOP_BY_HOP_HEADERS: Final = frozenset(
|
|||
)
|
||||
|
||||
|
||||
class FixtureInput(Protocol):
|
||||
def canonical_input(self) -> dict[str, object]: ...
|
||||
|
||||
|
||||
InputT = TypeVar("InputT", bound=FixtureInput)
|
||||
CaseT = TypeVar("CaseT", bound=BaseModel)
|
||||
InputT = TypeVar("InputT")
|
||||
|
||||
|
||||
@dataclass(frozen=True, slots=True)
|
||||
|
|
@ -59,20 +43,6 @@ class ProviderSpec:
|
|||
upstream_base: str
|
||||
|
||||
|
||||
@dataclass(frozen=True, slots=True)
|
||||
class RecorderResult(Generic[CaseT]):
|
||||
case: CaseT
|
||||
cache_hit: bool
|
||||
|
||||
|
||||
class FixtureEnvelope(BaseModel):
|
||||
model_config = ConfigDict(frozen=True, extra="forbid")
|
||||
|
||||
schema_version: int
|
||||
recorded_at: AwareDatetime
|
||||
case: dict[str, object]
|
||||
|
||||
|
||||
def _excluded_headers(headers: tuple[tuple[str, str], ...]) -> frozenset[str]:
|
||||
connection_values: Final = tuple(value for name, value in headers if name.lower() == "connection")
|
||||
connection_headers: Final = frozenset(
|
||||
|
|
@ -100,7 +70,7 @@ def _normalized_response_header(name: str, value: str) -> str:
|
|||
return urlunsplit(("http", _PARITY_PROVIDER_HOST, parsed.path, parsed.query, parsed.fragment))
|
||||
|
||||
|
||||
def _local_response_header(name: str, value: str, provider_url: str) -> str:
|
||||
def local_response_header(name: str, value: str, provider_url: str) -> str:
|
||||
if name.lower() not in {"location", "operation-location"}:
|
||||
return value
|
||||
parsed: Final = urlsplit(value)
|
||||
|
|
@ -193,7 +163,7 @@ class _RecordingHandler(BaseHTTPRequestHandler):
|
|||
provider: Final = self.server
|
||||
assert isinstance(provider, _RecordingProvider)
|
||||
for header in headers:
|
||||
self.send_header(header.name, _local_response_header(header.name, header.value, provider.url))
|
||||
self.send_header(header.name, local_response_header(header.name, header.value, provider.url))
|
||||
self.send_header("transfer-encoding", "chunked")
|
||||
self.end_headers()
|
||||
chunks: Final = tuple(self._relay_chunks(upstream.iter_bytes()))
|
||||
|
|
@ -219,7 +189,7 @@ class _RecordingHandler(BaseHTTPRequestHandler):
|
|||
provider: Final = self.server
|
||||
assert isinstance(provider, _RecordingProvider)
|
||||
for header in headers:
|
||||
self.send_header(header.name, _local_response_header(header.name, header.value, provider.url))
|
||||
self.send_header(header.name, local_response_header(header.name, header.value, provider.url))
|
||||
self.send_header("content-length", str(len(body)))
|
||||
self.end_headers()
|
||||
self.wfile.write(body)
|
||||
|
|
@ -241,23 +211,6 @@ def _recording_provider(spec: ProviderSpec) -> Generator[_RecordingProvider]:
|
|||
thread.join(timeout=5)
|
||||
|
||||
|
||||
def generate_case_inputs(strategy: SearchStrategy[InputT], examples: int) -> tuple[InputT, ...]:
|
||||
generated: Final[queue.SimpleQueue[InputT | None]] = queue.SimpleQueue()
|
||||
|
||||
@settings(max_examples=examples, deadline=None, derandomize=True)
|
||||
@given(case_input=strategy)
|
||||
def generate_case(case_input: InputT) -> None:
|
||||
generated.put(case_input)
|
||||
|
||||
generate_case()
|
||||
generated.put(None)
|
||||
return tuple(iter(generated.get, None))
|
||||
|
||||
|
||||
def fixture_cache_key(case_input: FixtureInput) -> dict[str, object]:
|
||||
return case_input.canonical_input()
|
||||
|
||||
|
||||
def _invoke_and_take_responses(
|
||||
recorder: _RecordingProvider,
|
||||
case_input: InputT,
|
||||
|
|
@ -273,118 +226,10 @@ def _invoke_and_take_responses(
|
|||
return recorder.take_responses()
|
||||
|
||||
|
||||
def _load_fixture(raw_fixture: dict[str, object], path: Path, case_type: type[CaseT]) -> CaseT:
|
||||
schema_version: Final = raw_fixture.get("schema_version")
|
||||
if schema_version != FIXTURE_SCHEMA_VERSION:
|
||||
raise ValueError(
|
||||
f"fixture {path} has schema_version {schema_version!r}, expected {FIXTURE_SCHEMA_VERSION}; "
|
||||
"delete it and regenerate the fixture bundle"
|
||||
)
|
||||
try:
|
||||
envelope: Final = FixtureEnvelope.model_validate(raw_fixture)
|
||||
return case_type.model_validate(envelope.case)
|
||||
except ValidationError as error:
|
||||
raise ValueError(f"invalid parity fixture {path} ({len(error.errors())} validation errors)") from error
|
||||
|
||||
|
||||
def record_case(
|
||||
def record_upstream_responses(
|
||||
spec: ProviderSpec,
|
||||
root: Path,
|
||||
case_input: InputT,
|
||||
sdk_call: Callable[[str, InputT], object],
|
||||
case_type: type[CaseT],
|
||||
) -> RecorderResult[CaseT]:
|
||||
cache: Final = JsonFileCache(root)
|
||||
cache_key: Final = fixture_cache_key(case_input)
|
||||
cached: Final = cache.get(cache_key)
|
||||
if cached is not None:
|
||||
return RecorderResult(case=_load_fixture(cached, cache.path_for(cache_key), case_type), cache_hit=True)
|
||||
|
||||
) -> tuple[RecordedResponse, ...]:
|
||||
with _recording_provider(spec) as recorder:
|
||||
upstream_responses: Final = _invoke_and_take_responses(recorder, case_input, sdk_call)
|
||||
|
||||
case: Final = case_type.model_validate({"litellm_input": case_input, "provider_responses": upstream_responses})
|
||||
envelope: Final = FixtureEnvelope(
|
||||
schema_version=FIXTURE_SCHEMA_VERSION,
|
||||
recorded_at=datetime.now(timezone.utc),
|
||||
case=cast(dict[str, object], case.model_dump(mode="json", exclude_unset=True)),
|
||||
)
|
||||
cache.put(cache_key, cast(dict[str, object], envelope.model_dump(mode="json", exclude_unset=True)))
|
||||
return RecorderResult(case=case, cache_hit=False)
|
||||
|
||||
|
||||
def record_cases(
|
||||
spec: ProviderSpec,
|
||||
root: Path,
|
||||
case_inputs: tuple[InputT, ...],
|
||||
sdk_call: Callable[[str, InputT], object],
|
||||
case_type: type[CaseT],
|
||||
max_concurrency: int,
|
||||
) -> tuple[RecorderResult[CaseT], ...]:
|
||||
if max_concurrency < 1:
|
||||
raise ValueError("max_concurrency must be at least 1")
|
||||
unique_inputs: Final = tuple(
|
||||
{canonical_json(fixture_cache_key(case_input)): case_input for case_input in case_inputs}.values()
|
||||
)
|
||||
with ThreadPoolExecutor(max_workers=max_concurrency) as executor:
|
||||
futures: Final = tuple(
|
||||
executor.submit(record_case, spec, root, case_input, sdk_call, case_type) for case_input in unique_inputs
|
||||
)
|
||||
return tuple(future.result() for future in futures)
|
||||
|
||||
|
||||
def fixture_directory(configured: Path | None, env_value: str | None, default: Path) -> Path:
|
||||
return (configured or Path(env_value or default)).expanduser()
|
||||
|
||||
|
||||
def recorded_fixtures(directory: Path, case_type: type[CaseT]) -> tuple[CaseT, ...]:
|
||||
cache: Final = JsonFileCache(directory)
|
||||
return tuple(_load_fixture(raw_fixture, path, case_type) for path, raw_fixture in cache.values_with_paths())
|
||||
|
||||
|
||||
def fixture_id(case_input: FixtureInput, prefix: str) -> str:
|
||||
input_json: Final = canonical_json(case_input.canonical_input())
|
||||
digest: Final = hashlib.sha256(input_json.encode("utf-8")).hexdigest()[:8]
|
||||
return f"{prefix}-{digest}"
|
||||
|
||||
|
||||
def parametrize_recorded_fixtures(
|
||||
metafunc: pytest.Metafunc,
|
||||
*,
|
||||
fixture_name: str,
|
||||
case_type: type[CaseT],
|
||||
env_var: str,
|
||||
default_directory: Path,
|
||||
regeneration_command: str,
|
||||
id_builder: Callable[[CaseT], str],
|
||||
) -> None:
|
||||
if fixture_name not in metafunc.fixturenames:
|
||||
return
|
||||
configured: Final = os.environ.get(env_var)
|
||||
if configured == "":
|
||||
raise pytest.UsageError(f"{env_var} is set but empty")
|
||||
directory: Final = Path(configured).expanduser() if configured is not None else default_directory
|
||||
try:
|
||||
fixtures: Final = recorded_fixtures(directory, case_type)
|
||||
except (ValidationError, ValueError) as error:
|
||||
raise pytest.UsageError(
|
||||
f"Invalid parity fixture bundle at {directory}. "
|
||||
"Each fixture must use the current versioned envelope. "
|
||||
f"Record fresh fixtures in an empty directory with: `{regeneration_command}`. "
|
||||
f"Validation details: {error}"
|
||||
) from error
|
||||
if fixtures:
|
||||
metafunc.parametrize(fixture_name, fixtures, ids=tuple(id_builder(fixture) for fixture in fixtures))
|
||||
return
|
||||
if configured is not None:
|
||||
raise pytest.UsageError(f"no recorded fixtures in {directory}")
|
||||
metafunc.parametrize(
|
||||
fixture_name,
|
||||
(
|
||||
pytest.param(
|
||||
None,
|
||||
marks=pytest.mark.skip(reason=f"no recorded fixtures in {directory}"),
|
||||
id="no-recorded-fixtures",
|
||||
),
|
||||
),
|
||||
)
|
||||
return _invoke_and_take_responses(recorder, case_input, sdk_call)
|
||||
143
tests/route_parity/fixtures/store.py
Normal file
143
tests/route_parity/fixtures/store.py
Normal file
|
|
@ -0,0 +1,143 @@
|
|||
from __future__ import annotations
|
||||
|
||||
import hashlib
|
||||
import json
|
||||
import os
|
||||
from collections.abc import Callable, Mapping
|
||||
from datetime import datetime, timezone
|
||||
from pathlib import Path
|
||||
from typing import Final, Protocol, TypeVar, cast
|
||||
|
||||
import pytest
|
||||
from pydantic import AwareDatetime, BaseModel, ConfigDict, TypeAdapter, ValidationError
|
||||
|
||||
FIXTURE_SCHEMA_VERSION: Final = 1
|
||||
JSON_OBJECT: Final = TypeAdapter(dict[str, object])
|
||||
|
||||
|
||||
class FixtureInput(Protocol):
|
||||
def canonical_input(self) -> dict[str, object]: ...
|
||||
|
||||
|
||||
CaseT = TypeVar("CaseT", bound=BaseModel)
|
||||
|
||||
|
||||
class FixtureEnvelope(BaseModel):
|
||||
model_config = ConfigDict(frozen=True, extra="forbid")
|
||||
|
||||
schema_version: int
|
||||
recorded_at: AwareDatetime
|
||||
case: dict[str, object]
|
||||
|
||||
|
||||
def canonical_json(value: Mapping[str, object]) -> str:
|
||||
return json.dumps(value, sort_keys=True, separators=(",", ":"), ensure_ascii=True)
|
||||
|
||||
|
||||
def fixture_cache_key(case_input: FixtureInput) -> dict[str, object]:
|
||||
return case_input.canonical_input()
|
||||
|
||||
|
||||
def fixture_path(directory: Path, case_input: FixtureInput) -> Path:
|
||||
input_json: Final = canonical_json(fixture_cache_key(case_input))
|
||||
digest: Final = hashlib.sha256(input_json.encode("utf-8")).hexdigest()
|
||||
return directory / f"{digest}.json"
|
||||
|
||||
|
||||
def load_fixture(directory: Path, case_input: FixtureInput, case_type: type[CaseT]) -> CaseT | None:
|
||||
path: Final = fixture_path(directory, case_input)
|
||||
if not path.is_file():
|
||||
return None
|
||||
return _load_fixture(JSON_OBJECT.validate_json(path.read_text(encoding="utf-8")), path, case_type)
|
||||
|
||||
|
||||
def save_fixture(directory: Path, case_input: FixtureInput, case: BaseModel) -> Path:
|
||||
directory.mkdir(parents=True, exist_ok=True)
|
||||
path: Final = fixture_path(directory, case_input)
|
||||
temporary_path: Final = path.with_suffix(".tmp")
|
||||
envelope: Final = FixtureEnvelope(
|
||||
schema_version=FIXTURE_SCHEMA_VERSION,
|
||||
recorded_at=datetime.now(timezone.utc),
|
||||
case=cast(dict[str, object], case.model_dump(mode="json", exclude_unset=True)),
|
||||
)
|
||||
serialized: Final = (
|
||||
json.dumps(envelope.model_dump(mode="json", exclude_unset=True), indent=2, sort_keys=True) + "\n"
|
||||
)
|
||||
temporary_path.write_text(serialized, encoding="utf-8")
|
||||
temporary_path.replace(path)
|
||||
return path
|
||||
|
||||
|
||||
def _load_fixture(raw_fixture: dict[str, object], path: Path, case_type: type[CaseT]) -> CaseT:
|
||||
schema_version: Final = raw_fixture.get("schema_version")
|
||||
if schema_version != FIXTURE_SCHEMA_VERSION:
|
||||
raise ValueError(
|
||||
f"fixture {path} has schema_version {schema_version!r}, expected {FIXTURE_SCHEMA_VERSION}; "
|
||||
"delete it and regenerate the fixture bundle"
|
||||
)
|
||||
try:
|
||||
envelope: Final = FixtureEnvelope.model_validate(raw_fixture)
|
||||
return case_type.model_validate(envelope.case)
|
||||
except ValidationError as error:
|
||||
raise ValueError(f"invalid parity fixture {path} ({len(error.errors())} validation errors)") from error
|
||||
|
||||
|
||||
def recorded_fixtures(directory: Path, case_type: type[CaseT]) -> tuple[CaseT, ...]:
|
||||
if not directory.is_dir():
|
||||
return ()
|
||||
paths: Final = tuple(sorted(directory.rglob("*.json")))
|
||||
return tuple(
|
||||
_load_fixture(JSON_OBJECT.validate_json(path.read_text(encoding="utf-8")), path, case_type) for path in paths
|
||||
)
|
||||
|
||||
|
||||
def fixture_directory(configured: Path | None, env_value: str | None, default: Path) -> Path:
|
||||
return (configured or Path(env_value or default)).expanduser()
|
||||
|
||||
|
||||
def fixture_id(case_input: FixtureInput, prefix: str) -> str:
|
||||
input_json: Final = canonical_json(case_input.canonical_input())
|
||||
digest: Final = hashlib.sha256(input_json.encode("utf-8")).hexdigest()[:8]
|
||||
return f"{prefix}-{digest}"
|
||||
|
||||
|
||||
def parametrize_recorded_fixtures(
|
||||
metafunc: pytest.Metafunc,
|
||||
*,
|
||||
fixture_name: str,
|
||||
case_type: type[CaseT],
|
||||
env_var: str,
|
||||
default_directory: Path,
|
||||
regeneration_command: str,
|
||||
id_builder: Callable[[CaseT], str],
|
||||
) -> None:
|
||||
if fixture_name not in metafunc.fixturenames:
|
||||
return
|
||||
configured: Final = os.environ.get(env_var)
|
||||
if configured == "":
|
||||
raise pytest.UsageError(f"{env_var} is set but empty")
|
||||
directory: Final = Path(configured).expanduser() if configured is not None else default_directory
|
||||
try:
|
||||
fixtures: Final = recorded_fixtures(directory, case_type)
|
||||
except (ValidationError, ValueError) as error:
|
||||
raise pytest.UsageError(
|
||||
f"Invalid parity fixture bundle at {directory}. "
|
||||
"Each fixture must use the current versioned envelope. "
|
||||
f"Record fresh fixtures in an empty directory with: `{regeneration_command}`. "
|
||||
f"Validation details: {error}"
|
||||
) from error
|
||||
if fixtures:
|
||||
metafunc.parametrize(fixture_name, fixtures, ids=tuple(id_builder(fixture) for fixture in fixtures))
|
||||
return
|
||||
if configured is not None:
|
||||
raise pytest.UsageError(f"no recorded fixtures in {directory}")
|
||||
metafunc.parametrize(
|
||||
fixture_name,
|
||||
(
|
||||
pytest.param(
|
||||
None,
|
||||
marks=pytest.mark.skip(reason=f"no recorded fixtures in {directory}"),
|
||||
id="no-recorded-fixtures",
|
||||
),
|
||||
),
|
||||
)
|
||||
20
tests/route_parity/fixtures/test_inputs.py
Normal file
20
tests/route_parity/fixtures/test_inputs.py
Normal file
|
|
@ -0,0 +1,20 @@
|
|||
from __future__ import annotations
|
||||
|
||||
from typing import Final
|
||||
|
||||
from hypothesis import strategies as st
|
||||
from pydantic import BaseModel, ConfigDict
|
||||
|
||||
from tests.route_parity.fixtures.inputs import generate_case_inputs
|
||||
|
||||
|
||||
class _Input(BaseModel):
|
||||
model_config = ConfigDict(frozen=True, extra="forbid")
|
||||
|
||||
identifier: str
|
||||
|
||||
|
||||
def test_generate_case_inputs_is_deterministic() -> None:
|
||||
strategy: Final = st.builds(_Input, identifier=st.integers().map(str))
|
||||
|
||||
assert generate_case_inputs(strategy, examples=4) == generate_case_inputs(strategy, examples=4)
|
||||
189
tests/route_parity/fixtures/test_pipeline.py
Normal file
189
tests/route_parity/fixtures/test_pipeline.py
Normal file
|
|
@ -0,0 +1,189 @@
|
|||
from __future__ import annotations
|
||||
|
||||
import logging
|
||||
import threading
|
||||
from collections.abc import Generator
|
||||
from contextlib import contextmanager
|
||||
from dataclasses import dataclass
|
||||
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
|
||||
from pathlib import Path
|
||||
from typing import Final, Literal
|
||||
|
||||
import httpx
|
||||
import pytest
|
||||
from hypothesis import strategies as st
|
||||
from pydantic import BaseModel, ConfigDict
|
||||
|
||||
from tests.route_parity.fixtures.pipeline import (
|
||||
RecordingInvocation,
|
||||
RecordingTarget,
|
||||
build_recording_jobs,
|
||||
record_fixtures,
|
||||
)
|
||||
from tests.route_parity.fixtures.recording import ProviderSpec
|
||||
from tests.route_parity.fixtures.store import fixture_path
|
||||
from tests.route_parity.recorded_http import RecordedResponse
|
||||
|
||||
|
||||
class _FixtureInput(BaseModel):
|
||||
model_config = ConfigDict(frozen=True, extra="forbid")
|
||||
|
||||
identifier: str
|
||||
|
||||
def canonical_input(self) -> dict[str, object]:
|
||||
return {"identifier": self.identifier}
|
||||
|
||||
|
||||
class _ParityCase(BaseModel):
|
||||
model_config = ConfigDict(frozen=True, extra="forbid")
|
||||
|
||||
litellm_input: _FixtureInput
|
||||
provider_responses: tuple[RecordedResponse, ...]
|
||||
|
||||
|
||||
class _Upstream(ThreadingHTTPServer):
|
||||
daemon_threads = True
|
||||
|
||||
def __init__(self) -> None:
|
||||
super().__init__(("127.0.0.1", 0), _UpstreamHandler)
|
||||
|
||||
@property
|
||||
def url(self) -> str:
|
||||
return f"http://127.0.0.1:{self.server_address[1]}"
|
||||
|
||||
|
||||
class _UpstreamHandler(BaseHTTPRequestHandler):
|
||||
protocol_version = "HTTP/1.1"
|
||||
|
||||
def do_POST(self) -> None:
|
||||
length: Final = int(self.headers.get("content-length") or "0")
|
||||
self.rfile.read(length)
|
||||
body: Final = b"{}"
|
||||
self.send_response(200)
|
||||
self.send_header("content-type", "application/json")
|
||||
self.send_header("content-length", str(len(body)))
|
||||
self.end_headers()
|
||||
self.wfile.write(body)
|
||||
|
||||
def log_message(self, format: str, *args: object) -> None:
|
||||
return
|
||||
|
||||
|
||||
@contextmanager
|
||||
def _upstream() -> Generator[_Upstream]:
|
||||
server: Final = _Upstream()
|
||||
thread: Final = threading.Thread(target=server.serve_forever, daemon=True)
|
||||
thread.start()
|
||||
try:
|
||||
yield server
|
||||
finally:
|
||||
server.shutdown()
|
||||
server.server_close()
|
||||
thread.join(timeout=5)
|
||||
|
||||
|
||||
@dataclass(frozen=True, slots=True)
|
||||
class _OrderedInvocation:
|
||||
order: Literal["slow", "fast"]
|
||||
slow_started: threading.Event
|
||||
fast_finished: threading.Event
|
||||
|
||||
def execute(self, provider_url: str, case_input: _FixtureInput) -> None:
|
||||
if self.order == "slow":
|
||||
self.slow_started.set()
|
||||
if not self.fast_finished.wait(timeout=2):
|
||||
raise TimeoutError("fast recording did not finish")
|
||||
else:
|
||||
if not self.slow_started.wait(timeout=2):
|
||||
raise TimeoutError("slow recording did not start")
|
||||
response: Final = httpx.post(f"{provider_url}/record", json={"id": case_input.identifier}, timeout=5)
|
||||
response.raise_for_status()
|
||||
if self.order == "fast":
|
||||
self.fast_finished.set()
|
||||
|
||||
|
||||
@dataclass(frozen=True, slots=True)
|
||||
class _Invocation:
|
||||
def execute(self, provider_url: str, case_input: _FixtureInput) -> None:
|
||||
response: Final = httpx.post(f"{provider_url}/record", json={"id": case_input.identifier}, timeout=5)
|
||||
response.raise_for_status()
|
||||
|
||||
|
||||
def _target(
|
||||
name: str,
|
||||
upstream_url: str,
|
||||
case_input: _FixtureInput,
|
||||
invocation: RecordingInvocation[_FixtureInput],
|
||||
) -> RecordingTarget[_FixtureInput]:
|
||||
return RecordingTarget(
|
||||
name=name,
|
||||
provider_spec=ProviderSpec(upstream_base=upstream_url),
|
||||
strategy=st.just(case_input),
|
||||
invocation=invocation,
|
||||
required_inputs=(case_input,),
|
||||
)
|
||||
|
||||
|
||||
def test_build_jobs_keeps_required_inputs_before_generated_inputs_and_deduplicates(tmp_path: Path) -> None:
|
||||
required: Final = _FixtureInput(identifier="required")
|
||||
generated: Final = _FixtureInput(identifier="generated")
|
||||
target: Final = RecordingTarget(
|
||||
name="ordered",
|
||||
provider_spec=ProviderSpec(upstream_base="https://provider.invalid"),
|
||||
strategy=st.just(generated),
|
||||
invocation=_Invocation(),
|
||||
required_inputs=(required, required),
|
||||
)
|
||||
|
||||
jobs: Final = build_recording_jobs((target,), tmp_path, examples=1)
|
||||
|
||||
assert tuple(job.case_input.identifier for job in jobs) == ("required", "generated")
|
||||
|
||||
|
||||
def test_progress_follows_completion_order(tmp_path: Path, caplog: pytest.LogCaptureFixture) -> None:
|
||||
slow_started: Final = threading.Event()
|
||||
fast_finished: Final = threading.Event()
|
||||
with _upstream() as upstream:
|
||||
targets: Final = (
|
||||
_target(
|
||||
"slow",
|
||||
upstream.url,
|
||||
_FixtureInput(identifier="slow"),
|
||||
_OrderedInvocation("slow", slow_started, fast_finished),
|
||||
),
|
||||
_target(
|
||||
"fast",
|
||||
upstream.url,
|
||||
_FixtureInput(identifier="fast"),
|
||||
_OrderedInvocation("fast", slow_started, fast_finished),
|
||||
),
|
||||
)
|
||||
with caplog.at_level(logging.INFO, logger="tests.route_parity.fixtures.pipeline"):
|
||||
summary: Final = record_fixtures(targets, tmp_path, 1, 2, _ParityCase)
|
||||
|
||||
progress: Final = tuple(record.message for record in caplog.records if record.message.startswith("["))
|
||||
assert len(summary.recorded) == 2
|
||||
assert summary.exit_code == 0
|
||||
assert "recorded fast" in progress[0]
|
||||
assert "recorded slow" in progress[1]
|
||||
assert caplog.records[0].message == "Recording 2 fixtures across 2 targets with concurrency 2"
|
||||
assert caplog.records[-1].message == "Finished 2 fixtures: 2 recorded, 0 cached, 0 failed"
|
||||
|
||||
|
||||
def test_failure_does_not_stop_independent_recordings(tmp_path: Path) -> None:
|
||||
stale_input: Final = _FixtureInput(identifier="stale")
|
||||
stale_directory: Final = tmp_path / "stale"
|
||||
stale_directory.mkdir()
|
||||
fixture_path(stale_directory, stale_input).write_text('{"schema_version": 0}\n', encoding="utf-8")
|
||||
with _upstream() as upstream:
|
||||
targets: Final = (
|
||||
_target("stale", upstream.url, stale_input, _Invocation()),
|
||||
_target("valid", upstream.url, _FixtureInput(identifier="valid"), _Invocation()),
|
||||
)
|
||||
summary: Final = record_fixtures(targets, tmp_path, 1, 2, _ParityCase)
|
||||
|
||||
assert len(summary.recorded) == 1
|
||||
assert summary.recorded[0].target_name == "valid"
|
||||
assert len(summary.failed) == 1
|
||||
assert summary.failed[0].target_name == "stale"
|
||||
assert summary.exit_code == 1
|
||||
|
|
@ -3,6 +3,7 @@ from __future__ import annotations
|
|||
import threading
|
||||
from collections.abc import Callable, Generator
|
||||
from contextlib import contextmanager
|
||||
from dataclasses import dataclass
|
||||
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
|
||||
from pathlib import Path
|
||||
from typing import Final
|
||||
|
|
@ -12,16 +13,13 @@ import pytest
|
|||
from hypothesis import strategies as st
|
||||
from pydantic import BaseModel, ConfigDict
|
||||
|
||||
from tests.route_parity.fixture_recorder import (
|
||||
from tests.route_parity.fixtures.pipeline import RecordingTarget, record_fixtures
|
||||
from tests.route_parity.fixtures.recording import ProviderSpec, record_upstream_responses
|
||||
from tests.route_parity.fixtures.store import (
|
||||
FIXTURE_SCHEMA_VERSION,
|
||||
ProviderSpec,
|
||||
fixture_cache_key,
|
||||
generate_case_inputs,
|
||||
record_case,
|
||||
record_cases,
|
||||
fixture_path,
|
||||
recorded_fixtures,
|
||||
)
|
||||
from tests.route_parity.json_file_cache import JsonFileCache
|
||||
from tests.route_parity.recorded_http import (
|
||||
HttpHeader,
|
||||
RecordedHttpStreamResponse,
|
||||
|
|
@ -53,6 +51,14 @@ class _ParityCase(BaseModel):
|
|||
provider_responses: tuple[RecordedResponse, ...]
|
||||
|
||||
|
||||
@dataclass(frozen=True, slots=True)
|
||||
class _Invocation:
|
||||
sdk_call: Callable[[str, _FixtureInput], object]
|
||||
|
||||
def execute(self, provider_url: str, case_input: _FixtureInput) -> None:
|
||||
self.sdk_call(provider_url, case_input)
|
||||
|
||||
|
||||
class _ControlledUpstream(ThreadingHTTPServer):
|
||||
daemon_threads = True
|
||||
|
||||
|
|
@ -195,54 +201,97 @@ def _polling_sdk_call(api_base: str, case_input: _FixtureInput) -> object:
|
|||
return completed
|
||||
|
||||
|
||||
def test_generate_case_inputs_is_deterministic() -> None:
|
||||
strategy: Final = st.builds(_FixtureInput, identifier=st.integers().map(str))
|
||||
|
||||
assert generate_case_inputs(strategy, examples=4) == generate_case_inputs(strategy, examples=4)
|
||||
|
||||
|
||||
def test_record_cases_deduplicates_and_limits_concurrency(tmp_path: Path) -> None:
|
||||
case_inputs: Final = (_case("one"), _case("two"), _case("one"), _case("three"))
|
||||
def test_recording_deduplicates_per_target_and_caps_global_concurrency(tmp_path: Path) -> None:
|
||||
shared_input: Final = _case("shared")
|
||||
with _controlled_upstream() as upstream:
|
||||
spec: Final = ProviderSpec(upstream_base=upstream.url)
|
||||
results: Final = record_cases(spec, tmp_path, case_inputs, _sdk_call, _ParityCase, max_concurrency=2)
|
||||
targets: Final = (
|
||||
RecordingTarget(
|
||||
name="first",
|
||||
provider_spec=spec,
|
||||
strategy=st.just(shared_input),
|
||||
invocation=_Invocation(_sdk_call),
|
||||
required_inputs=(shared_input, shared_input),
|
||||
),
|
||||
RecordingTarget(
|
||||
name="second",
|
||||
provider_spec=spec,
|
||||
strategy=st.just(shared_input),
|
||||
invocation=_Invocation(_sdk_call),
|
||||
required_inputs=(shared_input,),
|
||||
),
|
||||
)
|
||||
summary: Final = record_fixtures(targets, tmp_path, examples=1, concurrency=2, case_type=_ParityCase)
|
||||
|
||||
assert len(results) == 3
|
||||
assert upstream.request_count == 3
|
||||
assert len(summary.recorded) == 2
|
||||
assert {result.target_name for result in summary.recorded} == {"first", "second"}
|
||||
assert summary.cached == ()
|
||||
assert summary.failed == ()
|
||||
assert upstream.request_count == 2
|
||||
assert upstream.max_active_requests == 2
|
||||
assert len(recorded_fixtures(tmp_path, _ParityCase)) == 3
|
||||
for fixture_path in tmp_path.glob("*.json"):
|
||||
contents = fixture_path.read_text(encoding="utf-8")
|
||||
assert len(recorded_fixtures(tmp_path, _ParityCase)) == 2
|
||||
for path in tmp_path.rglob("*.json"):
|
||||
contents = path.read_text(encoding="utf-8")
|
||||
assert f'"schema_version": {FIXTURE_SCHEMA_VERSION}' in contents
|
||||
assert '"recorded_at":' in contents
|
||||
|
||||
|
||||
def test_record_case_rejects_stale_fixture_before_provider_call(tmp_path: Path) -> None:
|
||||
def test_pipeline_rejects_stale_fixture_before_provider_call(tmp_path: Path) -> None:
|
||||
case_input: Final = _case("stale")
|
||||
cache: Final = JsonFileCache(tmp_path)
|
||||
fixture_path: Final = cache.put(fixture_cache_key(case_input), {"schema_version": 0})
|
||||
directory: Final = tmp_path / "stale-target"
|
||||
directory.mkdir()
|
||||
path: Final = fixture_path(directory, case_input)
|
||||
path.write_text('{"schema_version": 0}\n', encoding="utf-8")
|
||||
target: Final = RecordingTarget(
|
||||
name="stale-target",
|
||||
provider_spec=ProviderSpec(upstream_base="http://127.0.0.1:1"),
|
||||
strategy=st.just(case_input),
|
||||
invocation=_Invocation(_sdk_call),
|
||||
)
|
||||
|
||||
with pytest.raises(ValueError, match=f"{fixture_path} has schema_version 0, expected {FIXTURE_SCHEMA_VERSION}"):
|
||||
record_case(
|
||||
ProviderSpec(upstream_base="http://127.0.0.1:1"),
|
||||
tmp_path,
|
||||
case_input,
|
||||
_sdk_call,
|
||||
_ParityCase,
|
||||
)
|
||||
summary: Final = record_fixtures(
|
||||
(target,),
|
||||
tmp_path,
|
||||
examples=1,
|
||||
concurrency=1,
|
||||
case_type=_ParityCase,
|
||||
)
|
||||
|
||||
assert summary.recorded == ()
|
||||
assert summary.cached == ()
|
||||
assert len(summary.failed) == 1
|
||||
assert str(summary.failed[0].error) == (
|
||||
f"fixture {path} has schema_version 0, expected {FIXTURE_SCHEMA_VERSION}; "
|
||||
"delete it and regenerate the fixture bundle"
|
||||
)
|
||||
|
||||
|
||||
def test_streaming_response_records_and_replays_chunks(tmp_path: Path) -> None:
|
||||
def test_cached_fixture_is_reported_without_provider_call(tmp_path: Path) -> None:
|
||||
case_input: Final = _case("cached")
|
||||
with _controlled_upstream() as upstream:
|
||||
result: Final = record_case(
|
||||
target: Final = RecordingTarget(
|
||||
name="cached-target",
|
||||
provider_spec=ProviderSpec(upstream_base=upstream.url),
|
||||
strategy=st.just(case_input),
|
||||
invocation=_Invocation(_sdk_call),
|
||||
)
|
||||
first: Final = record_fixtures((target,), tmp_path, 1, 1, _ParityCase)
|
||||
second: Final = record_fixtures((target,), tmp_path, 1, 1, _ParityCase)
|
||||
|
||||
assert len(first.recorded) == 1
|
||||
assert len(second.cached) == 1
|
||||
assert upstream.request_count == 1
|
||||
|
||||
|
||||
def test_streaming_response_records_and_replays_chunks() -> None:
|
||||
with _controlled_upstream() as upstream:
|
||||
responses: Final = record_upstream_responses(
|
||||
ProviderSpec(upstream_base=upstream.url),
|
||||
tmp_path,
|
||||
_case("stream"),
|
||||
_stream_sdk_call,
|
||||
_ParityCase,
|
||||
)
|
||||
|
||||
response: Final = result.case.provider_responses[0]
|
||||
response: Final = responses[0]
|
||||
assert isinstance(response, RecordedHttpStreamResponse)
|
||||
assert tuple(chunk.data_bytes() for chunk in response.chunks) == _SSE_CHUNKS
|
||||
assert isinstance(response.model_dump(mode="json")["chunks"], list)
|
||||
|
|
@ -256,17 +305,15 @@ def test_streaming_response_records_and_replays_chunks(tmp_path: Path) -> None:
|
|||
assert replayed_chunks == _SSE_CHUNKS
|
||||
|
||||
|
||||
def test_non_successful_provider_response_is_recorded(tmp_path: Path) -> None:
|
||||
def test_non_successful_provider_response_is_recorded() -> None:
|
||||
with _controlled_upstream() as upstream:
|
||||
result: Final = record_case(
|
||||
responses: Final = record_upstream_responses(
|
||||
ProviderSpec(upstream_base=upstream.url),
|
||||
tmp_path,
|
||||
_case("provider-error"),
|
||||
_error_sdk_call,
|
||||
_ParityCase,
|
||||
)
|
||||
|
||||
response: Final = result.case.provider_responses[0]
|
||||
response: Final = responses[0]
|
||||
assert response.status_code == 429
|
||||
|
||||
|
||||
|
|
@ -285,21 +332,18 @@ def test_stream_response_model_rejects_buffered_body() -> None:
|
|||
|
||||
@pytest.mark.parametrize("sdk_call", (_multi_sdk_call, _polling_sdk_call))
|
||||
def test_multiple_provider_calls_record_and_replay_in_order(
|
||||
tmp_path: Path,
|
||||
sdk_call: Callable[[str, _FixtureInput], object],
|
||||
) -> None:
|
||||
with _controlled_upstream() as upstream:
|
||||
result: Final = record_case(
|
||||
responses: Final = record_upstream_responses(
|
||||
ProviderSpec(upstream_base=upstream.url),
|
||||
tmp_path,
|
||||
_case(sdk_call.__name__),
|
||||
sdk_call,
|
||||
_ParityCase,
|
||||
)
|
||||
|
||||
assert len(result.case.provider_responses) == 2
|
||||
assert len(responses) == 2
|
||||
with replay_server() as provider:
|
||||
for response in result.case.provider_responses:
|
||||
for response in responses:
|
||||
provider.enqueue_response(response)
|
||||
sdk_call(provider.url, _case(sdk_call.__name__))
|
||||
requests: Final = provider.take_requests(2)
|
||||
|
|
@ -1,51 +0,0 @@
|
|||
from __future__ import annotations
|
||||
|
||||
import hashlib
|
||||
import json
|
||||
from collections.abc import Mapping
|
||||
from dataclasses import dataclass
|
||||
from pathlib import Path
|
||||
from typing import Final
|
||||
|
||||
from pydantic import TypeAdapter
|
||||
|
||||
JSON_OBJECT: Final = TypeAdapter(dict[str, object])
|
||||
|
||||
|
||||
def canonical_json(value: Mapping[str, object]) -> str:
|
||||
return json.dumps(value, sort_keys=True, separators=(",", ":"), ensure_ascii=True)
|
||||
|
||||
|
||||
@dataclass(frozen=True, slots=True)
|
||||
class JsonFileCache:
|
||||
root: Path
|
||||
|
||||
def path_for(self, key: Mapping[str, object]) -> Path:
|
||||
digest: Final = hashlib.sha256(canonical_json(key).encode("utf-8")).hexdigest()
|
||||
return self.root / f"{digest}.json"
|
||||
|
||||
def get(self, key: Mapping[str, object]) -> dict[str, object] | None:
|
||||
path: Final = self.path_for(key)
|
||||
if not path.is_file():
|
||||
return None
|
||||
return JSON_OBJECT.validate_json(path.read_text(encoding="utf-8"))
|
||||
|
||||
def put(self, key: Mapping[str, object], value: Mapping[str, object]) -> Path:
|
||||
self.root.mkdir(parents=True, exist_ok=True)
|
||||
path: Final = self.path_for(key)
|
||||
temporary_path: Final = path.with_suffix(".tmp")
|
||||
temporary_path.write_text(json.dumps(value, indent=2, sort_keys=True) + "\n", encoding="utf-8")
|
||||
temporary_path.replace(path)
|
||||
return path
|
||||
|
||||
def values(self) -> tuple[dict[str, object], ...]:
|
||||
if not self.root.is_dir():
|
||||
return ()
|
||||
paths: Final = tuple(sorted(self.root.rglob("*.json")))
|
||||
return tuple(JSON_OBJECT.validate_json(path.read_text(encoding="utf-8")) for path in paths)
|
||||
|
||||
def values_with_paths(self) -> tuple[tuple[Path, dict[str, object]], ...]:
|
||||
if not self.root.is_dir():
|
||||
return ()
|
||||
paths: Final = tuple(sorted(self.root.rglob("*.json")))
|
||||
return tuple((path, JSON_OBJECT.validate_json(path.read_text(encoding="utf-8"))) for path in paths)
|
||||
|
|
@ -10,7 +10,7 @@ from typing import Final
|
|||
|
||||
from pydantic import JsonValue, TypeAdapter
|
||||
|
||||
from tests.route_parity.fixture_recorder import _local_response_header
|
||||
from tests.route_parity.fixtures.recording import local_response_header
|
||||
from tests.route_parity.models import CapturedRequest
|
||||
from tests.route_parity.recorded_http import RecordedHttpResponse, RecordedHttpStreamResponse, RecordedResponse
|
||||
|
||||
|
|
@ -102,7 +102,7 @@ class _ReplayHandler(BaseHTTPRequestHandler):
|
|||
self.send_response_only(response.status_code)
|
||||
for header in response.headers:
|
||||
if header.name.lower() not in EXCLUDED_RESPONSE_HEADERS:
|
||||
self.send_header(header.name, _local_response_header(header.name, header.value, provider.url))
|
||||
self.send_header(header.name, local_response_header(header.name, header.value, provider.url))
|
||||
if isinstance(response, RecordedHttpResponse):
|
||||
response_body: Final = response.body_bytes()
|
||||
self.send_header("content-length", str(len(response_body)))
|
||||
|
|
|
|||
|
|
@ -1,80 +0,0 @@
|
|||
from __future__ import annotations
|
||||
|
||||
import queue
|
||||
from collections.abc import Mapping
|
||||
from dataclasses import dataclass, field
|
||||
from typing import Final
|
||||
|
||||
from hypothesis import strategies as st
|
||||
|
||||
from tests.route_parity.fixture_generator import FixtureSource, FixtureTarget, discover_fixture_targets
|
||||
from tests.route_parity.fixture_models import SdkInputBase
|
||||
from tests.route_parity.fixture_recorder import ProviderSpec
|
||||
|
||||
|
||||
class ExampleSdkInput(SdkInputBase):
|
||||
model: str
|
||||
|
||||
|
||||
@dataclass(frozen=True, slots=True)
|
||||
class ExampleInvocation:
|
||||
calls: queue.SimpleQueue[dict[str, object]]
|
||||
api_key: str = field(repr=False)
|
||||
|
||||
def execute(self, provider_url: str, case_input: ExampleSdkInput) -> None:
|
||||
self.calls.put(
|
||||
{
|
||||
"api_base": provider_url,
|
||||
"api_key": self.api_key,
|
||||
**case_input.as_sdk_kwargs(),
|
||||
}
|
||||
)
|
||||
|
||||
|
||||
@dataclass(frozen=True, slots=True)
|
||||
class ExampleProvider:
|
||||
name: str
|
||||
key_name: str
|
||||
|
||||
def targets(
|
||||
self,
|
||||
environ: Mapping[str, str],
|
||||
calls: queue.SimpleQueue[dict[str, object]],
|
||||
) -> tuple[FixtureTarget[ExampleSdkInput], ...]:
|
||||
api_key: Final = environ.get(self.key_name)
|
||||
if not api_key:
|
||||
return ()
|
||||
|
||||
case_input: Final = ExampleSdkInput(model=f"{self.name}/model")
|
||||
return (
|
||||
FixtureTarget(
|
||||
name=self.name,
|
||||
provider_spec=ProviderSpec(upstream_base=f"https://{self.name}.example"),
|
||||
strategy=st.just(case_input),
|
||||
invocation=ExampleInvocation(calls=calls, api_key=api_key),
|
||||
required_inputs=(case_input,),
|
||||
),
|
||||
)
|
||||
|
||||
|
||||
def test_discover_fixture_targets_flattens_configured_providers_and_injects_sdk_call() -> None:
|
||||
calls: Final[queue.SimpleQueue[dict[str, object]]] = queue.SimpleQueue()
|
||||
|
||||
providers: Final[tuple[FixtureSource[ExampleSdkInput, queue.SimpleQueue[dict[str, object]]], ...]] = (
|
||||
ExampleProvider(name="first", key_name="FIRST_KEY"),
|
||||
ExampleProvider(name="skipped", key_name="SKIPPED_KEY"),
|
||||
ExampleProvider(name="second", key_name="SECOND_KEY"),
|
||||
)
|
||||
targets: Final = discover_fixture_targets(
|
||||
providers,
|
||||
{"FIRST_KEY": "first-secret", "SECOND_KEY": "second-secret"},
|
||||
calls,
|
||||
)
|
||||
|
||||
assert tuple(target.name for target in targets) == ("first", "second")
|
||||
targets[1].invocation.execute("http://127.0.0.1:1234", targets[1].required_inputs[0])
|
||||
assert calls.get_nowait() == {
|
||||
"api_base": "http://127.0.0.1:1234",
|
||||
"api_key": "second-secret",
|
||||
"model": "second/model",
|
||||
}
|
||||
|
|
@ -5,7 +5,7 @@ from typing import Final
|
|||
|
||||
import pytest
|
||||
|
||||
from tests.route_parity.fixture_recorder import fixture_id, parametrize_recorded_fixtures
|
||||
from tests.route_parity.fixtures.store import fixture_id, parametrize_recorded_fixtures
|
||||
from tests.test_litellm.ocr.fixtures.models import OcrParityCase
|
||||
|
||||
FIXTURE_DIR_ENV: Final = "LITELLM_OCR_FIXTURE_DIR"
|
||||
|
|
@ -27,7 +27,7 @@ def pytest_generate_tests(metafunc: pytest.Metafunc) -> None:
|
|||
env_var=FIXTURE_DIR_ENV,
|
||||
default_directory=default_directory,
|
||||
regeneration_command=(
|
||||
f"uv run python -m tests.test_litellm.ocr.fixtures.generate --fixture-dir {default_directory}"
|
||||
f"uv run python -m tests.test_litellm.ocr.fixtures.record --fixture-dir {default_directory}"
|
||||
),
|
||||
id_builder=_fixture_id,
|
||||
)
|
||||
|
|
|
|||
|
|
@ -6,10 +6,10 @@ from typing import Final, cast
|
|||
from hypothesis import strategies as st
|
||||
from hypothesis.strategies import DrawFn, SearchStrategy
|
||||
|
||||
from tests.route_parity.fixture_recorder import ProviderSpec
|
||||
from tests.route_parity.fixtures.recording import ProviderSpec
|
||||
from tests.test_litellm.ocr.fixtures.common import (
|
||||
OcrFixtureClient,
|
||||
OcrFixtureTarget,
|
||||
OcrRecordingTarget,
|
||||
invoke_with_api_key,
|
||||
pdf_document,
|
||||
public_document_strategy,
|
||||
|
|
@ -71,45 +71,45 @@ def azure_document_intelligence_input_strategy(draw: DrawFn) -> AzureDocumentInt
|
|||
)
|
||||
|
||||
|
||||
class AzureMistralFixtureSource:
|
||||
def targets(self, environ: Mapping[str, str], client: OcrFixtureClient) -> tuple[OcrFixtureTarget, ...]:
|
||||
api_key: Final = environ.get("AZURE_AI_API_KEY")
|
||||
upstream_base: Final = environ.get("AZURE_AI_API_BASE")
|
||||
configured_model: Final = environ.get("AZURE_AI_OCR_MODEL")
|
||||
if not api_key or not upstream_base or not configured_model:
|
||||
return ()
|
||||
model: Final = configured_model if configured_model.startswith("azure_ai/") else f"azure_ai/{configured_model}"
|
||||
return (
|
||||
OcrFixtureTarget(
|
||||
name="azure-mistral",
|
||||
provider_spec=ProviderSpec(upstream_base=upstream_base.rstrip("/")),
|
||||
strategy=cast(
|
||||
SearchStrategy[OcrSdkInputBase],
|
||||
mistral_input_strategy(MISTRAL_MODEL).map(lambda case_input: _as_azure_mistral(case_input, model)),
|
||||
),
|
||||
invocation=invoke_with_api_key(client, api_key),
|
||||
required_inputs=cast(
|
||||
tuple[OcrSdkInputBase, ...],
|
||||
tuple(
|
||||
_as_azure_mistral(case_input, model) for case_input in required_mistral_inputs(MISTRAL_MODEL)
|
||||
),
|
||||
),
|
||||
def azure_mistral_recording_targets(
|
||||
environ: Mapping[str, str], client: OcrFixtureClient
|
||||
) -> tuple[OcrRecordingTarget, ...]:
|
||||
api_key: Final = environ.get("AZURE_AI_API_KEY")
|
||||
upstream_base: Final = environ.get("AZURE_AI_API_BASE")
|
||||
configured_model: Final = environ.get("AZURE_AI_OCR_MODEL")
|
||||
if not api_key or not upstream_base or not configured_model:
|
||||
return ()
|
||||
model: Final = configured_model if configured_model.startswith("azure_ai/") else f"azure_ai/{configured_model}"
|
||||
return (
|
||||
OcrRecordingTarget(
|
||||
name="azure-mistral",
|
||||
provider_spec=ProviderSpec(upstream_base=upstream_base.rstrip("/")),
|
||||
strategy=cast(
|
||||
SearchStrategy[OcrSdkInputBase],
|
||||
mistral_input_strategy(MISTRAL_MODEL).map(lambda case_input: _as_azure_mistral(case_input, model)),
|
||||
),
|
||||
)
|
||||
invocation=invoke_with_api_key(client, api_key),
|
||||
required_inputs=cast(
|
||||
tuple[OcrSdkInputBase, ...],
|
||||
tuple(_as_azure_mistral(case_input, model) for case_input in required_mistral_inputs(MISTRAL_MODEL)),
|
||||
),
|
||||
),
|
||||
)
|
||||
|
||||
|
||||
class AzureDocumentIntelligenceFixtureSource:
|
||||
def targets(self, environ: Mapping[str, str], client: OcrFixtureClient) -> tuple[OcrFixtureTarget, ...]:
|
||||
api_key: Final = environ.get("AZURE_DOCUMENT_INTELLIGENCE_API_KEY")
|
||||
upstream_base: Final = environ.get("AZURE_DOCUMENT_INTELLIGENCE_ENDPOINT")
|
||||
if not api_key or not upstream_base:
|
||||
return ()
|
||||
return (
|
||||
OcrFixtureTarget(
|
||||
name="azure-document-intelligence",
|
||||
provider_spec=ProviderSpec(upstream_base=upstream_base.rstrip("/")),
|
||||
strategy=cast(SearchStrategy[OcrSdkInputBase], azure_document_intelligence_input_strategy()),
|
||||
invocation=invoke_with_api_key(client, api_key),
|
||||
required_inputs=cast(tuple[OcrSdkInputBase, ...], _required_document_intelligence_inputs()),
|
||||
),
|
||||
)
|
||||
def azure_document_intelligence_recording_targets(
|
||||
environ: Mapping[str, str], client: OcrFixtureClient
|
||||
) -> tuple[OcrRecordingTarget, ...]:
|
||||
api_key: Final = environ.get("AZURE_DOCUMENT_INTELLIGENCE_API_KEY")
|
||||
upstream_base: Final = environ.get("AZURE_DOCUMENT_INTELLIGENCE_ENDPOINT")
|
||||
if not api_key or not upstream_base:
|
||||
return ()
|
||||
return (
|
||||
OcrRecordingTarget(
|
||||
name="azure-document-intelligence",
|
||||
provider_spec=ProviderSpec(upstream_base=upstream_base.rstrip("/")),
|
||||
strategy=cast(SearchStrategy[OcrSdkInputBase], azure_document_intelligence_input_strategy()),
|
||||
invocation=invoke_with_api_key(client, api_key),
|
||||
required_inputs=cast(tuple[OcrSdkInputBase, ...], _required_document_intelligence_inputs()),
|
||||
),
|
||||
)
|
||||
|
|
|
|||
|
|
@ -9,7 +9,7 @@ from urllib.parse import quote
|
|||
from hypothesis import strategies as st
|
||||
from hypothesis.strategies import SearchStrategy
|
||||
|
||||
from tests.route_parity.fixture_generator import FixtureTarget
|
||||
from tests.route_parity.fixtures.pipeline import RecordingTarget
|
||||
from tests.test_litellm.ocr.fixtures.models import (
|
||||
JsonSchemaDefinition,
|
||||
JsonSchemaResponseFormat,
|
||||
|
|
@ -18,7 +18,7 @@ from tests.test_litellm.ocr.fixtures.models import (
|
|||
OcrSdkInputBase,
|
||||
)
|
||||
|
||||
OcrFixtureTarget = FixtureTarget[OcrSdkInputBase]
|
||||
OcrRecordingTarget = RecordingTarget[OcrSdkInputBase]
|
||||
|
||||
|
||||
class OcrFixtureClient(Protocol):
|
||||
|
|
|
|||
|
|
@ -1,99 +0,0 @@
|
|||
from __future__ import annotations
|
||||
|
||||
import logging
|
||||
import os
|
||||
from collections.abc import Mapping
|
||||
from pathlib import Path
|
||||
from typing import Final, cast
|
||||
|
||||
from dotenv import load_dotenv
|
||||
|
||||
import litellm
|
||||
from litellm.rust_bridge.ocr import use_litellm_rust
|
||||
from tests.route_parity.fixture_generator import (
|
||||
FixtureSource,
|
||||
discover_fixture_targets,
|
||||
generate_target_fixtures,
|
||||
parse_generator_args,
|
||||
)
|
||||
from tests.route_parity.fixture_generator import require_targets as require_fixture_targets
|
||||
from tests.route_parity.fixture_recorder import fixture_directory
|
||||
from tests.test_litellm.ocr.fixtures.azure import (
|
||||
AzureDocumentIntelligenceFixtureSource,
|
||||
AzureMistralFixtureSource,
|
||||
azure_document_intelligence_input_strategy,
|
||||
)
|
||||
from tests.test_litellm.ocr.fixtures.common import OcrFixtureClient, OcrFixtureTarget, OcrSdkCall
|
||||
from tests.test_litellm.ocr.fixtures.mistral import (
|
||||
MistralFixtureSource,
|
||||
mistral_input_strategy,
|
||||
)
|
||||
from tests.test_litellm.ocr.fixtures.models import OcrParityCase, OcrSdkInputBase
|
||||
from tests.test_litellm.ocr.fixtures.reducto import (
|
||||
ReductoFixtureSource,
|
||||
reducto_legacy_input_strategy,
|
||||
reducto_v3_input_strategy,
|
||||
)
|
||||
from tests.test_litellm.ocr.fixtures.vertex import (
|
||||
VertexFixtureSource,
|
||||
vertex_deepseek_input_strategy,
|
||||
)
|
||||
|
||||
__all__ = (
|
||||
"azure_document_intelligence_input_strategy",
|
||||
"mistral_input_strategy",
|
||||
"reducto_legacy_input_strategy",
|
||||
"reducto_v3_input_strategy",
|
||||
"vertex_deepseek_input_strategy",
|
||||
)
|
||||
|
||||
FIXTURE_DIR_ENV: Final = "LITELLM_OCR_FIXTURE_DIR"
|
||||
OCR_FIXTURE_SOURCES: Final[tuple[FixtureSource[OcrSdkInputBase, OcrFixtureClient], ...]] = (
|
||||
MistralFixtureSource(),
|
||||
AzureMistralFixtureSource(),
|
||||
AzureDocumentIntelligenceFixtureSource(),
|
||||
VertexFixtureSource(),
|
||||
ReductoFixtureSource(),
|
||||
)
|
||||
|
||||
|
||||
class LiteLLMOcrFixtureClient:
|
||||
def __init__(self, sdk_call: OcrSdkCall) -> None:
|
||||
self.sdk_call: Final = sdk_call
|
||||
|
||||
def execute(self, api_base: str, api_key: str, case_input: OcrSdkInputBase) -> None:
|
||||
self.sdk_call(api_base=api_base, api_key=api_key, **case_input.as_sdk_kwargs())
|
||||
|
||||
|
||||
def discover_targets(
|
||||
environ: Mapping[str, str],
|
||||
client: OcrFixtureClient,
|
||||
) -> tuple[OcrFixtureTarget, ...]:
|
||||
return discover_fixture_targets(OCR_FIXTURE_SOURCES, environ, client)
|
||||
|
||||
|
||||
def require_targets(targets: tuple[OcrFixtureTarget, ...]) -> tuple[OcrFixtureTarget, ...]:
|
||||
return require_fixture_targets(
|
||||
targets,
|
||||
"No OCR fixture providers are configured. Set a supported provider API key and endpoint",
|
||||
)
|
||||
|
||||
|
||||
def main() -> None:
|
||||
logging.basicConfig(level=logging.INFO, format="%(message)s")
|
||||
load_dotenv()
|
||||
args: Final = parse_generator_args()
|
||||
client: Final = LiteLLMOcrFixtureClient(cast(OcrSdkCall, litellm.ocr))
|
||||
targets: Final = require_targets(discover_targets(os.environ, client))
|
||||
root: Final = fixture_directory(
|
||||
args.fixture_dir,
|
||||
os.environ.get(FIXTURE_DIR_ENV),
|
||||
Path(__file__).with_name("data"),
|
||||
)
|
||||
use_litellm_rust(False, ocr=None, aocr=None)
|
||||
for target in targets:
|
||||
generate_target_fixtures(target, root, args.examples, args.concurrency, OcrParityCase)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
|
|
@ -6,10 +6,10 @@ from typing import Final, cast
|
|||
from hypothesis import strategies as st
|
||||
from hypothesis.strategies import DrawFn, SearchStrategy
|
||||
|
||||
from tests.route_parity.fixture_recorder import ProviderSpec
|
||||
from tests.route_parity.fixtures.recording import ProviderSpec
|
||||
from tests.test_litellm.ocr.fixtures.common import (
|
||||
OcrFixtureClient,
|
||||
OcrFixtureTarget,
|
||||
OcrRecordingTarget,
|
||||
annotation_format,
|
||||
image_document,
|
||||
invoke_with_api_key,
|
||||
|
|
@ -80,19 +80,18 @@ def required_mistral_inputs(model: str) -> tuple[MistralOcrSdkInput, ...]:
|
|||
return tuple(MistralOcrSdkInput.model_validate({"model": model, "document": document, **case}) for case in cases)
|
||||
|
||||
|
||||
class MistralFixtureSource:
|
||||
def targets(self, environ: Mapping[str, str], client: OcrFixtureClient) -> tuple[OcrFixtureTarget, ...]:
|
||||
api_key: Final = environ.get("MISTRAL_API_KEY")
|
||||
if not api_key:
|
||||
return ()
|
||||
configured: Final = environ.get("MISTRAL_API_BASE", "https://api.mistral.ai").rstrip("/")
|
||||
upstream_base: Final = configured.removesuffix("/v1")
|
||||
return (
|
||||
OcrFixtureTarget(
|
||||
name="mistral-ocr",
|
||||
provider_spec=ProviderSpec(upstream_base=upstream_base),
|
||||
strategy=cast(SearchStrategy[OcrSdkInputBase], mistral_input_strategy(MISTRAL_MODEL)),
|
||||
invocation=invoke_with_api_key(client, api_key),
|
||||
required_inputs=cast(tuple[OcrSdkInputBase, ...], required_mistral_inputs(MISTRAL_MODEL)),
|
||||
),
|
||||
)
|
||||
def mistral_recording_targets(environ: Mapping[str, str], client: OcrFixtureClient) -> tuple[OcrRecordingTarget, ...]:
|
||||
api_key: Final = environ.get("MISTRAL_API_KEY")
|
||||
if not api_key:
|
||||
return ()
|
||||
configured: Final = environ.get("MISTRAL_API_BASE", "https://api.mistral.ai").rstrip("/")
|
||||
upstream_base: Final = configured.removesuffix("/v1")
|
||||
return (
|
||||
OcrRecordingTarget(
|
||||
name="mistral-ocr",
|
||||
provider_spec=ProviderSpec(upstream_base=upstream_base),
|
||||
strategy=cast(SearchStrategy[OcrSdkInputBase], mistral_input_strategy(MISTRAL_MODEL)),
|
||||
invocation=invoke_with_api_key(client, api_key),
|
||||
required_inputs=cast(tuple[OcrSdkInputBase, ...], required_mistral_inputs(MISTRAL_MODEL)),
|
||||
),
|
||||
)
|
||||
|
|
|
|||
72
tests/test_litellm/ocr/fixtures/record.py
Normal file
72
tests/test_litellm/ocr/fixtures/record.py
Normal file
|
|
@ -0,0 +1,72 @@
|
|||
from __future__ import annotations
|
||||
|
||||
import logging
|
||||
import os
|
||||
from collections.abc import Mapping
|
||||
from pathlib import Path
|
||||
from typing import Final, cast
|
||||
|
||||
from dotenv import load_dotenv
|
||||
|
||||
import litellm
|
||||
from litellm.rust_bridge.ocr import use_litellm_rust
|
||||
from tests.route_parity.fixtures.pipeline import parse_recording_args, record_fixtures
|
||||
from tests.route_parity.fixtures.store import fixture_directory
|
||||
from tests.test_litellm.ocr.fixtures.azure import (
|
||||
azure_document_intelligence_recording_targets,
|
||||
azure_mistral_recording_targets,
|
||||
)
|
||||
from tests.test_litellm.ocr.fixtures.common import OcrFixtureClient, OcrRecordingTarget, OcrSdkCall
|
||||
from tests.test_litellm.ocr.fixtures.mistral import mistral_recording_targets
|
||||
from tests.test_litellm.ocr.fixtures.models import OcrParityCase, OcrSdkInputBase
|
||||
from tests.test_litellm.ocr.fixtures.reducto import reducto_recording_targets
|
||||
from tests.test_litellm.ocr.fixtures.vertex import vertex_recording_targets
|
||||
|
||||
FIXTURE_DIR_ENV: Final = "LITELLM_OCR_FIXTURE_DIR"
|
||||
|
||||
|
||||
class LiteLLMOcrFixtureClient:
|
||||
def __init__(self, sdk_call: OcrSdkCall) -> None:
|
||||
self.sdk_call: Final = sdk_call
|
||||
|
||||
def execute(self, api_base: str, api_key: str, case_input: OcrSdkInputBase) -> None:
|
||||
self.sdk_call(api_base=api_base, api_key=api_key, **case_input.as_sdk_kwargs())
|
||||
|
||||
|
||||
def discover_targets(
|
||||
environ: Mapping[str, str],
|
||||
client: OcrFixtureClient,
|
||||
) -> tuple[OcrRecordingTarget, ...]:
|
||||
return (
|
||||
*mistral_recording_targets(environ, client),
|
||||
*azure_mistral_recording_targets(environ, client),
|
||||
*azure_document_intelligence_recording_targets(environ, client),
|
||||
*vertex_recording_targets(environ, client),
|
||||
*reducto_recording_targets(environ, client),
|
||||
)
|
||||
|
||||
|
||||
def require_targets(targets: tuple[OcrRecordingTarget, ...]) -> tuple[OcrRecordingTarget, ...]:
|
||||
if targets:
|
||||
return targets
|
||||
raise SystemExit("No OCR fixture providers are configured. Set a supported provider API key and endpoint")
|
||||
|
||||
|
||||
def main() -> int:
|
||||
logging.basicConfig(level=logging.INFO, format="%(message)s")
|
||||
load_dotenv()
|
||||
args: Final = parse_recording_args()
|
||||
client: Final = LiteLLMOcrFixtureClient(cast(OcrSdkCall, litellm.ocr))
|
||||
targets: Final = require_targets(discover_targets(os.environ, client))
|
||||
root: Final = fixture_directory(
|
||||
args.fixture_dir,
|
||||
os.environ.get(FIXTURE_DIR_ENV),
|
||||
Path(__file__).with_name("data"),
|
||||
)
|
||||
use_litellm_rust(False, ocr=None, aocr=None)
|
||||
summary: Final = record_fixtures(targets, root, args.examples, args.concurrency, OcrParityCase)
|
||||
return summary.exit_code
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
raise SystemExit(main())
|
||||
|
|
@ -6,10 +6,10 @@ from typing import Final, cast
|
|||
from hypothesis import strategies as st
|
||||
from hypothesis.strategies import DrawFn, SearchStrategy
|
||||
|
||||
from tests.route_parity.fixture_recorder import ProviderSpec
|
||||
from tests.route_parity.fixtures.recording import ProviderSpec
|
||||
from tests.test_litellm.ocr.fixtures.common import (
|
||||
OcrFixtureClient,
|
||||
OcrFixtureTarget,
|
||||
OcrRecordingTarget,
|
||||
fixture_pdf_data_uri,
|
||||
invoke_with_api_key,
|
||||
)
|
||||
|
|
@ -155,33 +155,32 @@ def _required_v3_inputs(document: ReductoDocumentUrlDocument) -> tuple[ReductoPa
|
|||
)
|
||||
|
||||
|
||||
class ReductoFixtureSource:
|
||||
def targets(self, environ: Mapping[str, str], client: OcrFixtureClient) -> tuple[OcrFixtureTarget, ...]:
|
||||
api_key: Final = environ.get("REDUCTO_API_KEY")
|
||||
if not api_key:
|
||||
return ()
|
||||
upstream_base: Final = environ.get("REDUCTO_API_BASE", _REDUCTO_API_BASE).rstrip("/")
|
||||
document: Final = ReductoDocumentUrlDocument(type="document_url", document_url=fixture_pdf_data_uri())
|
||||
invocation: Final = invoke_with_api_key(client, api_key)
|
||||
return (
|
||||
OcrFixtureTarget(
|
||||
name="reducto-v3",
|
||||
provider_spec=ProviderSpec(upstream_base=upstream_base),
|
||||
strategy=cast(SearchStrategy[OcrSdkInputBase], reducto_v3_input_strategy(document)),
|
||||
invocation=invocation,
|
||||
required_inputs=cast(tuple[OcrSdkInputBase, ...], _required_v3_inputs(document)),
|
||||
),
|
||||
OcrFixtureTarget(
|
||||
name="reducto-legacy",
|
||||
provider_spec=ProviderSpec(upstream_base=upstream_base),
|
||||
strategy=cast(SearchStrategy[OcrSdkInputBase], reducto_legacy_input_strategy(document)),
|
||||
invocation=invocation,
|
||||
required_inputs=cast(
|
||||
tuple[OcrSdkInputBase, ...],
|
||||
(
|
||||
ReductoParseLegacySdkInput(model="reducto/parse-legacy", document=document),
|
||||
ReductoParseLegacySdkInput(model="reducto/parse-legacy", document=document, enhance={}),
|
||||
),
|
||||
def reducto_recording_targets(environ: Mapping[str, str], client: OcrFixtureClient) -> tuple[OcrRecordingTarget, ...]:
|
||||
api_key: Final = environ.get("REDUCTO_API_KEY")
|
||||
if not api_key:
|
||||
return ()
|
||||
upstream_base: Final = environ.get("REDUCTO_API_BASE", _REDUCTO_API_BASE).rstrip("/")
|
||||
document: Final = ReductoDocumentUrlDocument(type="document_url", document_url=fixture_pdf_data_uri())
|
||||
invocation: Final = invoke_with_api_key(client, api_key)
|
||||
return (
|
||||
OcrRecordingTarget(
|
||||
name="reducto-v3",
|
||||
provider_spec=ProviderSpec(upstream_base=upstream_base),
|
||||
strategy=cast(SearchStrategy[OcrSdkInputBase], reducto_v3_input_strategy(document)),
|
||||
invocation=invocation,
|
||||
required_inputs=cast(tuple[OcrSdkInputBase, ...], _required_v3_inputs(document)),
|
||||
),
|
||||
OcrRecordingTarget(
|
||||
name="reducto-legacy",
|
||||
provider_spec=ProviderSpec(upstream_base=upstream_base),
|
||||
strategy=cast(SearchStrategy[OcrSdkInputBase], reducto_legacy_input_strategy(document)),
|
||||
invocation=invocation,
|
||||
required_inputs=cast(
|
||||
tuple[OcrSdkInputBase, ...],
|
||||
(
|
||||
ReductoParseLegacySdkInput(model="reducto/parse-legacy", document=document),
|
||||
ReductoParseLegacySdkInput(model="reducto/parse-legacy", document=document, enhance={}),
|
||||
),
|
||||
),
|
||||
)
|
||||
),
|
||||
)
|
||||
|
|
|
|||
|
|
@ -6,10 +6,10 @@ from typing import Final, cast
|
|||
from hypothesis import strategies as st
|
||||
from hypothesis.strategies import DrawFn, SearchStrategy
|
||||
|
||||
from tests.route_parity.fixture_recorder import ProviderSpec
|
||||
from tests.route_parity.fixtures.recording import ProviderSpec
|
||||
from tests.test_litellm.ocr.fixtures.common import (
|
||||
OcrFixtureClient,
|
||||
OcrFixtureTarget,
|
||||
OcrRecordingTarget,
|
||||
image_document,
|
||||
invoke_with_api_key,
|
||||
public_document_strategy,
|
||||
|
|
@ -53,42 +53,41 @@ def vertex_deepseek_input_strategy(draw: DrawFn, project: str, location: str) ->
|
|||
)
|
||||
|
||||
|
||||
class VertexFixtureSource:
|
||||
def targets(self, environ: Mapping[str, str], client: OcrFixtureClient) -> tuple[OcrFixtureTarget, ...]:
|
||||
api_key: Final = environ.get("VERTEX_AI_API_KEY")
|
||||
project: Final = environ.get("VERTEXAI_PROJECT") or environ.get("VERTEX_PROJECT")
|
||||
location: Final = environ.get("VERTEXAI_LOCATION") or environ.get("VERTEX_LOCATION") or "us-central1"
|
||||
if not api_key or not project:
|
||||
return ()
|
||||
upstream_base: Final = environ.get("VERTEX_AI_API_BASE") or f"https://{location}-aiplatform.googleapis.com"
|
||||
invocation: Final = invoke_with_api_key(client, api_key)
|
||||
return (
|
||||
OcrFixtureTarget(
|
||||
name="vertex-mistral",
|
||||
provider_spec=ProviderSpec(upstream_base=upstream_base.rstrip("/")),
|
||||
strategy=cast(
|
||||
SearchStrategy[OcrSdkInputBase],
|
||||
st.builds(
|
||||
_as_vertex_mistral,
|
||||
case_input=mistral_input_strategy(MISTRAL_MODEL),
|
||||
project=st.just(project),
|
||||
location=st.just(location),
|
||||
),
|
||||
),
|
||||
invocation=invocation,
|
||||
required_inputs=cast(
|
||||
tuple[OcrSdkInputBase, ...],
|
||||
tuple(
|
||||
_as_vertex_mistral(case_input, project, location)
|
||||
for case_input in required_mistral_inputs(MISTRAL_MODEL)
|
||||
),
|
||||
def vertex_recording_targets(environ: Mapping[str, str], client: OcrFixtureClient) -> tuple[OcrRecordingTarget, ...]:
|
||||
api_key: Final = environ.get("VERTEX_AI_API_KEY")
|
||||
project: Final = environ.get("VERTEXAI_PROJECT") or environ.get("VERTEX_PROJECT")
|
||||
location: Final = environ.get("VERTEXAI_LOCATION") or environ.get("VERTEX_LOCATION") or "us-central1"
|
||||
if not api_key or not project:
|
||||
return ()
|
||||
upstream_base: Final = environ.get("VERTEX_AI_API_BASE") or f"https://{location}-aiplatform.googleapis.com"
|
||||
invocation: Final = invoke_with_api_key(client, api_key)
|
||||
return (
|
||||
OcrRecordingTarget(
|
||||
name="vertex-mistral",
|
||||
provider_spec=ProviderSpec(upstream_base=upstream_base.rstrip("/")),
|
||||
strategy=cast(
|
||||
SearchStrategy[OcrSdkInputBase],
|
||||
st.builds(
|
||||
_as_vertex_mistral,
|
||||
case_input=mistral_input_strategy(MISTRAL_MODEL),
|
||||
project=st.just(project),
|
||||
location=st.just(location),
|
||||
),
|
||||
),
|
||||
OcrFixtureTarget(
|
||||
name="vertex-deepseek",
|
||||
provider_spec=ProviderSpec(upstream_base=upstream_base.rstrip("/")),
|
||||
strategy=cast(SearchStrategy[OcrSdkInputBase], vertex_deepseek_input_strategy(project, location)),
|
||||
invocation=invocation,
|
||||
required_inputs=cast(tuple[OcrSdkInputBase, ...], _required_deepseek_inputs(project, location)),
|
||||
invocation=invocation,
|
||||
required_inputs=cast(
|
||||
tuple[OcrSdkInputBase, ...],
|
||||
tuple(
|
||||
_as_vertex_mistral(case_input, project, location)
|
||||
for case_input in required_mistral_inputs(MISTRAL_MODEL)
|
||||
),
|
||||
),
|
||||
)
|
||||
),
|
||||
OcrRecordingTarget(
|
||||
name="vertex-deepseek",
|
||||
provider_spec=ProviderSpec(upstream_base=upstream_base.rstrip("/")),
|
||||
strategy=cast(SearchStrategy[OcrSdkInputBase], vertex_deepseek_input_strategy(project, location)),
|
||||
invocation=invocation,
|
||||
required_inputs=cast(tuple[OcrSdkInputBase, ...], _required_deepseek_inputs(project, location)),
|
||||
),
|
||||
)
|
||||
|
|
|
|||
|
|
@ -11,13 +11,8 @@ from litellm.llms.base_llm.ocr.transformation import BaseOCRConfig
|
|||
from litellm.llms.mistral.ocr.transformation import MistralOCRConfig
|
||||
from litellm.llms.reducto.ocr.transformation import ReductoParseLegacyConfig, ReductoParseV3Config
|
||||
from litellm.llms.vertex_ai.ocr.deepseek_transformation import VertexAIDeepSeekOCRConfig
|
||||
from tests.test_litellm.ocr.fixtures.generate import (
|
||||
azure_document_intelligence_input_strategy,
|
||||
mistral_input_strategy,
|
||||
reducto_legacy_input_strategy,
|
||||
reducto_v3_input_strategy,
|
||||
vertex_deepseek_input_strategy,
|
||||
)
|
||||
from tests.test_litellm.ocr.fixtures.azure import azure_document_intelligence_input_strategy
|
||||
from tests.test_litellm.ocr.fixtures.mistral import mistral_input_strategy
|
||||
from tests.test_litellm.ocr.fixtures.models import (
|
||||
AzureDocumentIntelligenceOcrSdkInput,
|
||||
AzureMistralOcrSdkInput,
|
||||
|
|
@ -39,6 +34,8 @@ from tests.test_litellm.ocr.fixtures.models import (
|
|||
VertexDeepSeekOcrSdkInput,
|
||||
VertexMistralOcrSdkInput,
|
||||
)
|
||||
from tests.test_litellm.ocr.fixtures.reducto import reducto_legacy_input_strategy, reducto_v3_input_strategy
|
||||
from tests.test_litellm.ocr.fixtures.vertex import vertex_deepseek_input_strategy
|
||||
|
||||
COMMON_FIELDS: Final = frozenset(
|
||||
{"boundary", "model", "document", "custom_llm_provider", "vertex_project", "vertex_location"}
|
||||
|
|
|
|||
|
|
@ -7,13 +7,13 @@ from typing import Final
|
|||
|
||||
import pytest
|
||||
|
||||
from tests.route_parity.fixture_recorder import generate_case_inputs
|
||||
from tests.test_litellm.ocr.fixtures.generate import (
|
||||
from tests.route_parity.fixtures.inputs import generate_case_inputs
|
||||
from tests.route_parity.fixtures.pipeline import parse_recording_args
|
||||
from tests.test_litellm.ocr.fixtures.models import OcrSdkInputBase
|
||||
from tests.test_litellm.ocr.fixtures.record import (
|
||||
discover_targets,
|
||||
parse_generator_args,
|
||||
require_targets,
|
||||
)
|
||||
from tests.test_litellm.ocr.fixtures.models import OcrSdkInputBase
|
||||
|
||||
|
||||
class _UnusedOcrClient:
|
||||
|
|
@ -33,13 +33,13 @@ _UNUSED_OCR_CLIENT: Final = _UnusedOcrClient()
|
|||
|
||||
|
||||
def test_parse_args_has_no_model_selection() -> None:
|
||||
args: Final = parse_generator_args(["--examples", "2", "--concurrency", "3", "--fixture-dir", "/tmp/ocr"])
|
||||
args: Final = parse_recording_args(["--examples", "2", "--concurrency", "3", "--fixture-dir", "/tmp/ocr"])
|
||||
|
||||
assert args.examples == 2
|
||||
assert args.concurrency == 3
|
||||
assert args.fixture_dir == Path("/tmp/ocr")
|
||||
with pytest.raises(SystemExit):
|
||||
parse_generator_args(["--model", "mistral/mistral-ocr-latest"])
|
||||
parse_recording_args(["--model", "mistral/mistral-ocr-latest"])
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
|
|
@ -94,9 +94,9 @@ def test_azure_mistral_discovery_requires_and_normalizes_deployment_model() -> N
|
|||
}
|
||||
assert discover_targets(incomplete, _UNUSED_OCR_CLIENT) == ()
|
||||
|
||||
target: Final = discover_targets({**incomplete, "AZURE_AI_OCR_MODEL": "mistral-ocr-deployment"}, _UNUSED_OCR_CLIENT)[
|
||||
0
|
||||
]
|
||||
target: Final = discover_targets(
|
||||
{**incomplete, "AZURE_AI_OCR_MODEL": "mistral-ocr-deployment"}, _UNUSED_OCR_CLIENT
|
||||
)[0]
|
||||
assert target.required_inputs[0].canonical_input()["model"] == "azure_ai/mistral-ocr-deployment"
|
||||
|
||||
|
||||
|
|
@ -18,7 +18,7 @@ from litellm.rust_bridge import get_native_bridge
|
|||
from litellm.rust_bridge import ocr as rust_ocr_bridge
|
||||
from litellm.rust_bridge.ocr import RustAocr, RustOcr
|
||||
from tests.route_parity.compare import assert_model_parity, assert_parity, assert_request_parity
|
||||
from tests.route_parity.fixture_recorder import recorded_fixtures
|
||||
from tests.route_parity.fixtures.store import recorded_fixtures
|
||||
from tests.route_parity.inprocess import run_in_process
|
||||
from tests.route_parity.models import (
|
||||
SDKCommand,
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue