mirror of
https://github.com/BerriAI/litellm.git
synced 2026-09-05 08:07:05 +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
45 lines
1.5 KiB
Python
45 lines
1.5 KiB
Python
from __future__ import annotations
|
|
|
|
from collections.abc import Sequence, Set
|
|
from dataclasses import replace
|
|
from pathlib import Path
|
|
from typing import Final
|
|
|
|
from ..shared.reporting.models import HarnessCase, SdkFunction, Strategy, Surface
|
|
from ..shared.reporting.orchestration import run_strategies
|
|
from ..shared.reporting.ui import make_dashboard
|
|
|
|
REPO_ROOT: Final = Path(__file__).resolve().parents[3]
|
|
|
|
|
|
def select_cases(
|
|
strategies: Sequence[Strategy],
|
|
sdk_functions: Set[SdkFunction],
|
|
surface: Surface | None = None,
|
|
) -> tuple[HarnessCase, ...]:
|
|
return tuple(
|
|
case
|
|
for strategy in strategies
|
|
for case in strategy.cases
|
|
if (not sdk_functions or case.sdk_function in sdk_functions)
|
|
and (surface is None or case.surface == surface)
|
|
)
|
|
|
|
|
|
def run_command(
|
|
strategies: Sequence[Strategy],
|
|
cases: Sequence[HarnessCase],
|
|
runner_args: Sequence[str] = (),
|
|
) -> int:
|
|
grouped: Final = {
|
|
strategy.id: tuple(case for case in cases if case.strategy_id == strategy.id)
|
|
for strategy in strategies
|
|
}
|
|
visible: Final = tuple(strategy for strategy in strategies if grouped[strategy.id])
|
|
runners: Final = tuple(replace(strategy, cases=grouped[strategy.id]) for strategy in visible)
|
|
dashboard: Final = make_dashboard(visible)
|
|
with dashboard:
|
|
exit_code, run = run_strategies(runners, REPO_ROOT, dashboard.update, runner_args)
|
|
if exit_code != 130:
|
|
dashboard.finish(run, exit_code)
|
|
return exit_code
|