mirror of
https://github.com/BerriAI/litellm.git
synced 2026-09-10 22:41:41 +00:00
ee08c36fc0
* wip * refactor(tests): move sdk function tracing into rust python harness * dead code * fix: handle harness keyboard interrupts * refactor(tests): deduplicate rust python harness helpers * fix(harness): expose validated strategy choices * wip * refactor(harness): let strategies own parity reports * docs(harness): update strategy structure * refactor(harness): localize strategy report views * wip * fix(harness): satisfy mapping runner type checks * fix(harness): clarify trace parity output * wip * fix(harness): clarify unit mapping report * fix(harness): finalize trace parity contracts * refactor(harness): structure parity contracts * feat: derive unit test mapping from traces * feat(harness): map rstest test families * feat(ocr): port Azure document intelligence tests * feat(harness): enforce complete unit mappings * feat(ocr): add reducto core transforms * feat(harness): classify host-only unit tests * fix(ocr): complete Rust provider plumbing * fix(harness): reuse OCR parity workers
87 lines
3.4 KiB
Python
87 lines
3.4 KiB
Python
from __future__ import annotations
|
|
|
|
from typing import Final
|
|
|
|
from ...shared.reporting.models import Coverage, HarnessCase, HarnessRun, RunStatus, SdkFunction, Surface
|
|
from ...shared.reporting.strategy import ModuleCaseSpec
|
|
from ...shared.tracing.steps import Engine
|
|
from .models import GatewayRouteSpec, RouteFixture, RouteSpec, TraceScenario, TraceSuite
|
|
from .runner import run_trace_mode, scenario_nodeids, validate_trace_suite
|
|
|
|
|
|
def _fixture(_engine: Engine, _base_url: str) -> RouteFixture:
|
|
return RouteFixture(kwargs={}, provider_responses=())
|
|
|
|
|
|
def _case(*, surface: Surface = "sdk", function: SdkFunction = "ocr") -> HarnessCase:
|
|
return HarnessCase(
|
|
strategy_id="trace_parity",
|
|
strategy_label="Trace parity",
|
|
sdk_function=function,
|
|
spec=ModuleCaseSpec(coverage=Coverage.PARTIAL, module="example"),
|
|
surface=surface,
|
|
)
|
|
|
|
|
|
def test_scenario_filtering_and_occurrence_node_ids() -> None:
|
|
suite: Final = TraceSuite(
|
|
route=RouteSpec("ocr", ("ocr", "aocr"), ("ocr", "aocr"), _fixture),
|
|
scenarios=(
|
|
TraceScenario("one", _fixture, (), modes=("sync", "async")),
|
|
TraceScenario("two", _fixture, (), modes=("async",)),
|
|
),
|
|
)
|
|
case: Final = _case()
|
|
|
|
nodes: Final = scenario_nodeids(suite, case, frozenset({"two"}))
|
|
|
|
assert tuple(nodeid for _, _, nodeid in nodes) == ("trace:sdk:ocr:two:async",)
|
|
|
|
|
|
def test_scenario_validation_rejects_duplicate_and_unsafe_names() -> None:
|
|
route: Final = RouteSpec("ocr", ("ocr", "aocr"), ("ocr", "aocr"), _fixture)
|
|
duplicate: Final = TraceSuite(
|
|
route=route,
|
|
scenarios=(TraceScenario("same", _fixture, ()), TraceScenario("same", _fixture, ())),
|
|
)
|
|
unsafe: Final = TraceSuite(route=route, scenarios=(TraceScenario("bad:name", _fixture, ()),))
|
|
case: Final = _case()
|
|
|
|
assert validate_trace_suite(duplicate, case) is not None
|
|
assert validate_trace_suite(unsafe, case) is not None
|
|
|
|
|
|
def test_scenario_validation_rejects_invalid_modes_and_route_registration() -> None:
|
|
invalid_modes: Final = TraceSuite(
|
|
route=RouteSpec("ocr", ("ocr", "aocr"), ("ocr", "aocr"), _fixture),
|
|
scenarios=(TraceScenario("invalid", _fixture, (), modes=("sync", "sync")),),
|
|
)
|
|
wrong_function: Final = TraceSuite(
|
|
route=RouteSpec("messages", ("create", "acreate"), ("messages", "amessages"), _fixture),
|
|
scenarios=(TraceScenario("one", _fixture, ()),),
|
|
)
|
|
wrong_surface: Final = TraceSuite(
|
|
route=GatewayRouteSpec("ocr"),
|
|
scenarios=(TraceScenario("one", _fixture, ()),),
|
|
)
|
|
case: Final = _case()
|
|
|
|
assert "unique sync/async modes" in (validate_trace_suite(invalid_modes, case) or "")
|
|
assert "does not match case function" in (validate_trace_suite(wrong_function, case) or "")
|
|
assert "must use RouteSpec" in (validate_trace_suite(wrong_surface, case) or "")
|
|
|
|
|
|
def test_invalid_route_dispatch_records_harness_error() -> None:
|
|
case: Final = _case()
|
|
run: Final = HarnessRun.from_cases((case,))
|
|
result: Final = run.results[case.key]
|
|
suite: Final = TraceSuite(
|
|
route=GatewayRouteSpec("ocr"),
|
|
scenarios=(TraceScenario("one", _fixture, (), modes=("sync",)),),
|
|
)
|
|
nodeid: Final = "trace:sdk:ocr:one:sync"
|
|
|
|
run_trace_mode(run, result, suite, suite.scenarios[0], "sync", "sdk", nodeid, lambda _: None)
|
|
|
|
assert result.outcomes[nodeid] is RunStatus.ERROR
|
|
assert run.failures == [(nodeid, "gateway route cannot run on the sdk surface")]
|