From 4c51ca72d6b27fdfd1ae1e4ec118ef9e4ad82c4b Mon Sep 17 00:00:00 2001 From: ishaan-berri <155045088+ishaan-berri@users.noreply.github.com> Date: Wed, 2 Sep 2026 15:51:08 -0700 Subject: [PATCH] test: add interactive Rust Python parity harness (#39419) * test: add interactive Rust Python parity harness * test: simplify Rust Python parity harness structure * test: show parity confidence by SDK section --------- Co-authored-by: yuneng-jiang --- tests/rust-python-harness/README.md | 145 +++++++++ tests/rust-python-harness/__init__.py | 5 + tests/rust-python-harness/__main__.py | 4 + tests/rust-python-harness/catalog.py | 93 ++++++ tests/rust-python-harness/cli.py | 180 +++++++++++ .../e2e_fuzz_tests/README.md | 3 + .../e2e_fuzz_tests/strategy.json | 12 + tests/rust-python-harness/models.py | 219 +++++++++++++ tests/rust-python-harness/runner.py | 160 ++++++++++ tests/rust-python-harness/ui.py | 295 ++++++++++++++++++ .../unit_tests_rust/README.md | 3 + .../unit_tests_rust/strategy.json | 12 + .../validate_sub_methods/README.md | 3 + .../validate_sub_methods/strategy.json | 12 + tests/test_rust_python_harness.py | 236 ++++++++++++++ 15 files changed, 1382 insertions(+) create mode 100644 tests/rust-python-harness/README.md create mode 100644 tests/rust-python-harness/__init__.py create mode 100644 tests/rust-python-harness/__main__.py create mode 100644 tests/rust-python-harness/catalog.py create mode 100644 tests/rust-python-harness/cli.py create mode 100644 tests/rust-python-harness/e2e_fuzz_tests/README.md create mode 100644 tests/rust-python-harness/e2e_fuzz_tests/strategy.json create mode 100644 tests/rust-python-harness/models.py create mode 100644 tests/rust-python-harness/runner.py create mode 100644 tests/rust-python-harness/ui.py create mode 100644 tests/rust-python-harness/unit_tests_rust/README.md create mode 100644 tests/rust-python-harness/unit_tests_rust/strategy.json create mode 100644 tests/rust-python-harness/validate_sub_methods/README.md create mode 100644 tests/rust-python-harness/validate_sub_methods/strategy.json create mode 100644 tests/test_rust_python_harness.py diff --git a/tests/rust-python-harness/README.md b/tests/rust-python-harness/README.md new file mode 100644 index 00000000000..e94ac87c3b3 --- /dev/null +++ b/tests/rust-python-harness/README.md @@ -0,0 +1,145 @@ +# Rust ↔ Python SDK parity harness + +This folder is the operator-facing harness for the Rust migration test plan. It runs pytest normally, listens to test events in-process, and redraws a live matrix grouped by testing strategy and SDK-level function. + +The matrix always has these SDK columns: + +- `ocr / aocr` +- `messages / amessages` +- `responses / aresponses` +- `count_tokens` + +The harness has three deliberately broad test-strategy folders: + +| Strategy | Folder | +| --- | --- | +| Public SDK parity over generated and recorded inputs | [`e2e_fuzz_tests/`](e2e_fuzz_tests/) | +| Focused tests of Rust-owned behavior | [`unit_tests_rust/`](unit_tests_rust/) | +| Isolated transform and Python-to-Rust helper coverage | [`validate_sub_methods/`](validate_sub_methods/) | + +## Run it + +From the repository root: + +```bash +poetry run python -m tests.rust-python-harness +``` + +The default runs every configured test once and updates all matching cells in real time. Narrow a run by strategy, SDK function, or both: + +```bash +poetry run python -m tests.rust-python-harness --strategy e2e_fuzz_tests +poetry run python -m tests.rust-python-harness --function messages +poetry run python -m tests.rust-python-harness --strategy validate_sub_methods --function ocr +``` + +For a guided run, use the interactive picker. It asks which strategy rows and SDK +function columns to include, then hands the terminal to the live dashboard. It never +captures keys while tests are running, so Ctrl-C and pytest debugging remain safe. + +```bash +poetry run python -m tests.rust-python-harness --interactive +``` + +Useful operator options: + +```bash +# Inspect coverage and pytest selectors without running anything. +poetry run python -m tests.rust-python-harness --list + +# Stable line-oriented output for CI logs or redirected output. +poetry run python -m tests.rust-python-harness --plain + +# Measure Python reference lines exercised by this parity run and build an HTML heatmap. +poetry run python -m tests.rust-python-harness --coverage + +# Forward pytest options. Use the equals form when the value begins with a dash. +poetry run python -m tests.rust-python-harness --pytest-arg=-x +``` + +The process returns pytest's exit code. A configured selector that collects no test is also a failure. A planned cell has no selector yet and does not fail the run. + +The dashboard adapts to narrow terminals, shows elapsed time and unique-test progress, +and prints the three slowest tests when the run ends. Each failure includes a focused +`poetry run pytest ... -q` command. Redirected output and CI automatically use the +line-oriented plain renderer; `--plain` lets you opt into it locally. + +The final screen includes a confidence score for every SDK section. It is the direct +ratio of required strategy rows with passing evidence, such as `1/3 = 33%`; High means +all required strategies passed, Medium means some passed, and Low means none passed. +This behavioral score is intentionally shown separately from Python and Rust LOC. + +Coverage reports are written outside the three strategy folders at +`target/rust-python-harness/`. Open `python-html/index.html` to inspect executed and +missing Python lines; `python.json` and `python.xml` are available for automation. +Coverage is finalized after pytest exits, because worker processes must flush their +data first. + +## Port coverage and confidence + +Treat these as separate signals instead of one ambiguous coverage percentage: + +| Signal | Tool | What it proves | +| --- | --- | --- | +| Python reference LOC | `coverage.py` / `pytest-cov` via `--coverage` | The mapped Python behavior ran | +| Rust port LOC | `cargo-llvm-cov` | The mapped Rust implementation ran | +| Parity contracts | This harness matrix | Python and Rust had the same observable behavior | + +`validate_sub_methods/` owns the future source-section inventory that maps a stable +Python qualified symbol to its Rust symbol. That inventory is the denominator for +per-function rollups; raw coverage for the entire LiteLLM repository would obscure +the port's real gaps. `unit_tests_rust/` owns direct `cargo-llvm-cov` runs, while +`e2e_fuzz_tests/` owns behavioral parity and fuzz-case counts. Keep Python, Rust, and +parity percentages visible side by side and label section confidence High only when +the mapped implementation exists, every required strategy passes, and both sides meet +their LOC thresholds. Generated Rust LCOV/HTML and the combined index also belong in +`target/rust-python-harness/`, not in a fourth strategy folder. + +## Read the matrix + +| Mark | Meaning | +| --- | --- | +| `✓` | All collected tests passed | +| `✗` | At least one test failed | +| `!` | Test setup or teardown failed | +| `↷` | All collected tests skipped | +| `?` | A configured selector did not collect a test | +| `—` | Strategy is planned but has no test yet | +| `n/a` | Strategy does not apply to this SDK function | +| `◐` | The configured tests cover only part of the TDD's parity contract | + +The initial end-to-end entries deliberately show `◐`: the repository has Rust bridge tests for OCR, Messages, and Responses websocket plumbing, but those are not yet frozen-Python-oracle comparisons. The remaining TDD cells stay visible as planned work instead of disappearing from a green summary. + +## Attach parity tests + +Each of the three folders contains a concise `README.md` and a `strategy.json`. Add a pytest file or node ID to the appropriate SDK function's `selectors` list: + +```json +{ + "coverage": "complete", + "selectors": [ + "tests/rust-python-harness/validate_sub_methods/test_messages.py" + ] +} +``` + +Selectors use the same syntax as pytest. A file selector aggregates every test in the file; a node selector can target one test or parametrized family. The runner deduplicates selectors, so one test may intentionally prove more than one cell without executing twice. + +Use these coverage values: + +- `complete`: implements the full strategy contract for that SDK function. +- `partial`: useful coverage exists, but the TDD contract is not fully proven. +- `planned`: no runnable parity test exists yet. +- `not_applicable`: the strategy cannot apply, such as streaming for OCR. + +Keep comparison mechanics in shared harness modules and provider/function facts in the owning strategy folder. A Python/Rust mismatch is a test failure; do not normalize away observable return types, exception classes, private response fields, chunk ordering, or callback payload differences merely to make a cell green. + +## Architecture + +- `catalog.py` validates and loads every strategy manifest. +- `models.py` owns typed strategy, case, coverage, and run-state models. +- `runner.py` maps live pytest events back to one or more matrix cells. +- `ui.py` renders the interactive Rich dashboard and a dependency-free plain fallback. +- `cli.py` handles filtering and preserves pytest exit semantics. + +The harness is driven from Python, matching the SDK surface and existing test tooling. Rust remains responsible for the implementation under comparison; the harness does not move provider semantics into the PyO3 bridge. diff --git a/tests/rust-python-harness/__init__.py b/tests/rust-python-harness/__init__.py new file mode 100644 index 00000000000..70362674d2b --- /dev/null +++ b/tests/rust-python-harness/__init__.py @@ -0,0 +1,5 @@ +"""Interactive Rust/Python SDK parity test harness.""" + +from .catalog import load_catalog + +__all__ = ["load_catalog"] diff --git a/tests/rust-python-harness/__main__.py b/tests/rust-python-harness/__main__.py new file mode 100644 index 00000000000..bfdcd0c1158 --- /dev/null +++ b/tests/rust-python-harness/__main__.py @@ -0,0 +1,4 @@ +from .cli import main + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/tests/rust-python-harness/catalog.py b/tests/rust-python-harness/catalog.py new file mode 100644 index 00000000000..e23b9b125f0 --- /dev/null +++ b/tests/rust-python-harness/catalog.py @@ -0,0 +1,93 @@ +from __future__ import annotations + +import json +from pathlib import Path +from typing import Any + +from .models import Coverage, HarnessCase, SDK_FUNCTIONS, Strategy + +STRATEGIES_ROOT = Path(__file__).parent + + +def _require_string(value: Any, field: str, source: Path) -> str: + if not isinstance(value, str) or not value.strip(): + raise ValueError(f"{source}: {field} must be a non-empty string") + return value + + +def _load_strategy(source: Path) -> Strategy: + with source.open(encoding="utf-8") as stream: + data = json.load(stream) + + strategy_id = _require_string(data.get("id"), "id", source) + label = _require_string(data.get("label"), "label", source) + description = _require_string(data.get("description"), "description", source) + order = data.get("order") + if not isinstance(order, int): + raise ValueError(f"{source}: order must be an integer") + function_data = data.get("functions") + if not isinstance(function_data, dict): + raise ValueError(f"{source}: functions must be an object") + + missing = set(SDK_FUNCTIONS) - set(function_data) + extra = set(function_data) - set(SDK_FUNCTIONS) + if missing or extra: + raise ValueError( + f"{source}: functions must exactly match {SDK_FUNCTIONS}; missing={missing}, extra={extra}" + ) + + cases: list[HarnessCase] = [] + for sdk_function in SDK_FUNCTIONS: + case_data = function_data[sdk_function] + if not isinstance(case_data, dict): + raise ValueError(f"{source}: functions.{sdk_function} must be an object") + try: + coverage = Coverage(case_data.get("coverage")) + except ValueError as exc: + raise ValueError(f"{source}: invalid coverage for {sdk_function}") from exc + selectors = case_data.get("selectors", []) + if not isinstance(selectors, list) or not all( + isinstance(item, str) and item for item in selectors + ): + raise ValueError( + f"{source}: selectors for {sdk_function} must be a list of strings" + ) + if coverage is Coverage.NOT_APPLICABLE and selectors: + raise ValueError( + f"{source}: not_applicable case {sdk_function} cannot have selectors" + ) + cases.append( + HarnessCase( + strategy_id=strategy_id, + strategy_label=label, + sdk_function=sdk_function, + coverage=coverage, + selectors=tuple(selectors), + note=str(case_data.get("note", "")), + ) + ) + + return Strategy( + order=order, + id=strategy_id, + label=label, + description=description, + directory=source.parent, + cases=tuple(cases), + ) + + +def load_catalog(root: Path = STRATEGIES_ROOT) -> tuple[Strategy, ...]: + sources = sorted(root.glob("*/strategy.json")) + if not sources: + raise ValueError(f"No strategy manifests found below {root}") + strategies = tuple( + sorted( + (_load_strategy(source) for source in sources), + key=lambda strategy: strategy.order, + ) + ) + ids = [strategy.id for strategy in strategies] + if len(ids) != len(set(ids)): + raise ValueError(f"Duplicate strategy id in {root}") + return strategies diff --git a/tests/rust-python-harness/cli.py b/tests/rust-python-harness/cli.py new file mode 100644 index 00000000000..41c46f9613a --- /dev/null +++ b/tests/rust-python-harness/cli.py @@ -0,0 +1,180 @@ +from __future__ import annotations + +import argparse +import importlib.util +from collections.abc import Sequence +from pathlib import Path + +from .catalog import load_catalog +from .models import HarnessCase, Strategy +from .runner import run_pytest +from .ui import make_dashboard + +REPO_ROOT = Path(__file__).resolve().parents[2] +COVERAGE_ROOT = REPO_ROOT / "target" / "rust-python-harness" + + +def _parser() -> argparse.ArgumentParser: + parser = argparse.ArgumentParser( + prog="rust-python-harness", + description="Run Rust/Python parity tests with a live strategy-by-SDK-function dashboard.", + ) + parser.add_argument( + "-i", + "--interactive", + action="store_true", + help="pick strategies and SDK functions in a guided terminal menu", + ) + parser.add_argument( + "--list", action="store_true", help="show the catalog without running tests" + ) + parser.add_argument( + "--strategy", + action="append", + default=[], + metavar="ID", + help="run only this strategy", + ) + parser.add_argument( + "--function", + action="append", + default=[], + dest="sdk_functions", + choices=("ocr", "messages", "responses", "count_tokens"), + help="run only this SDK function", + ) + parser.add_argument( + "--plain", + action="store_true", + help="disable the interactive terminal dashboard", + ) + parser.add_argument( + "--coverage", + action="store_true", + help="write Python reference LOC reports (HTML, JSON, and XML)", + ) + parser.add_argument( + "--pytest-arg", + action="append", + default=[], + metavar="ARG", + help="append an argument to pytest (repeatable, for example --pytest-arg=-x)", + ) + return parser + + +def _coverage_pytest_args(output_root: Path = COVERAGE_ROOT) -> tuple[str, ...]: + output_root.mkdir(parents=True, exist_ok=True) + return ( + "--cov=litellm", + "--cov-context=test", + f"--cov-report=json:{output_root / 'python.json'}", + f"--cov-report=xml:{output_root / 'python.xml'}", + f"--cov-report=html:{output_root / 'python-html'}", + ) + + +def _pick_values( + title: str, options: Sequence[tuple[str, str]], input_fn=input +) -> set[str]: + print(f"\n{title} (Enter = all)") + for index, (value, label) in enumerate(options, start=1): + print(f" {index:>2}. {label} [{value}]") + while True: + answer = input_fn("Choose numbers, comma-separated: ").strip() + if not answer: + return set() + try: + indexes = {int(part.strip()) for part in answer.split(",")} + except ValueError: + print("Please enter numbers separated by commas.") + continue + if indexes and all(1 <= index <= len(options) for index in indexes): + return {options[index - 1][0] for index in indexes} + print(f"Choose values from 1 to {len(options)}.") + + +def _interactive_filters(strategies: Sequence[Strategy]) -> tuple[set[str], set[str]]: + strategy_ids = _pick_values( + "Testing strategies", [(strategy.id, strategy.label) for strategy in strategies] + ) + sdk_functions = _pick_values( + "SDK functions", + [(name, name) for name in ("ocr", "messages", "responses", "count_tokens")], + ) + return strategy_ids, sdk_functions + + +def _select( + strategies: Sequence[Strategy], strategy_ids: set[str], sdk_functions: set[str] +) -> tuple[HarnessCase, ...]: + known_ids = {strategy.id for strategy in strategies} + unknown = strategy_ids - known_ids + if unknown: + raise ValueError(f"Unknown strategy: {', '.join(sorted(unknown))}") + return tuple( + case + for strategy in strategies + if not strategy_ids or strategy.id in strategy_ids + for case in strategy.cases + if not sdk_functions or case.sdk_function in sdk_functions + ) + + +def _print_catalog(strategies: Sequence[Strategy]) -> None: + for strategy in strategies: + print(f"{strategy.id:20} {strategy.label}") + for case in strategy.cases: + selectors = ( + ", ".join(case.selectors) if case.selectors else "no test configured" + ) + print(f" {case.sdk_function:12} {case.coverage.value:14} {selectors}") + + +def main(argv: Sequence[str] | None = None) -> int: + args = _parser().parse_args(argv) + if args.coverage and importlib.util.find_spec("pytest_cov") is None: + _parser().error( + "--coverage requires the project's pytest-cov dependency; run with " + "`poetry run python -m tests.rust-python-harness --coverage`" + ) + strategies = load_catalog() + if args.list: + _print_catalog(strategies) + return 0 + + strategy_ids = set(args.strategy) + sdk_functions = set(args.sdk_functions) + if args.interactive: + picked_strategies, picked_functions = _interactive_filters(strategies) + strategy_ids = strategy_ids or picked_strategies + sdk_functions = sdk_functions or picked_functions + + try: + cases = _select(strategies, strategy_ids, sdk_functions) + except ValueError as exc: + _parser().error(str(exc)) + selected_strategy_ids = {case.strategy_id for case in cases} + visible_strategies = tuple( + strategy for strategy in strategies if strategy.id in selected_strategy_ids + ) + dashboard = make_dashboard( + visible_strategies, + plain=args.plain, + confidence_strategies=strategies, + ) + pytest_args = [*args.pytest_arg] + if args.coverage: + pytest_args.extend(_coverage_pytest_args()) + with dashboard: + exit_code, run = run_pytest( + cases=cases, + repo_root=REPO_ROOT, + on_update=dashboard.update, + pytest_args=pytest_args, + ) + dashboard.finish(run, exit_code) + if args.coverage and (COVERAGE_ROOT / "python.json").exists(): + print(f"Python LOC heatmap: {COVERAGE_ROOT / 'python-html' / 'index.html'}") + print(f"Machine-readable coverage: {COVERAGE_ROOT / 'python.json'}") + return exit_code diff --git a/tests/rust-python-harness/e2e_fuzz_tests/README.md b/tests/rust-python-harness/e2e_fuzz_tests/README.md new file mode 100644 index 00000000000..34b12050ff9 --- /dev/null +++ b/tests/rust-python-harness/e2e_fuzz_tests/README.md @@ -0,0 +1,3 @@ +# End-to-end fuzz tests + +Runs the same SDK call through the Python and Rust paths using generated inputs and recorded provider responses. It compares public results, streams, callbacks, and exceptions to catch behavior differences a unit test can miss. diff --git a/tests/rust-python-harness/e2e_fuzz_tests/strategy.json b/tests/rust-python-harness/e2e_fuzz_tests/strategy.json new file mode 100644 index 00000000000..abeea01d9b5 --- /dev/null +++ b/tests/rust-python-harness/e2e_fuzz_tests/strategy.json @@ -0,0 +1,12 @@ +{ + "order": 10, + "id": "e2e_fuzz_tests", + "label": "End-to-end fuzz tests", + "description": "Compare observable Python and Rust SDK behavior over generated and recorded inputs.", + "functions": { + "ocr": {"coverage": "partial", "selectors": ["tests/test_litellm/ocr/test_rust_bridge.py"], "note": "Bridge coverage exists; frozen-oracle fuzz parity is still being added."}, + "messages": {"coverage": "partial", "selectors": ["tests/test_litellm/anthropic_interface/test_rust_bridge_messages.py"], "note": "Bridge coverage exists; frozen-oracle fuzz parity is still being added."}, + "responses": {"coverage": "partial", "selectors": ["tests/test_litellm/responses/test_rust_bridge_websocket.py"], "note": "Covers the websocket bridge; full responses parity is still being added."}, + "count_tokens": {"coverage": "planned", "selectors": [], "note": "No Rust count_tokens parity test is present yet."} + } +} diff --git a/tests/rust-python-harness/models.py b/tests/rust-python-harness/models.py new file mode 100644 index 00000000000..21097e0f7d0 --- /dev/null +++ b/tests/rust-python-harness/models.py @@ -0,0 +1,219 @@ +from __future__ import annotations + +from dataclasses import dataclass, field +from enum import Enum +from pathlib import Path +from time import monotonic +from typing import Iterable + + +class Coverage(str, Enum): + COMPLETE = "complete" + PARTIAL = "partial" + PLANNED = "planned" + NOT_APPLICABLE = "not_applicable" + + +class RunStatus(str, Enum): + NOT_RUN = "not_run" + QUEUED = "queued" + RUNNING = "running" + PASSED = "passed" + FAILED = "failed" + SKIPPED = "skipped" + ERROR = "error" + MISSING = "missing" + PLANNED = "planned" + NOT_APPLICABLE = "not_applicable" + + +class ConfidenceLevel(str, Enum): + HIGH = "HIGH" + MEDIUM = "MEDIUM" + LOW = "LOW" + + +SDK_FUNCTIONS = ("ocr", "messages", "responses", "count_tokens") + + +@dataclass(frozen=True) +class HarnessCase: + strategy_id: str + strategy_label: str + sdk_function: str + coverage: Coverage + selectors: tuple[str, ...] + note: str = "" + + @property + def key(self) -> str: + return f"{self.strategy_id}:{self.sdk_function}" + + +@dataclass(frozen=True) +class Strategy: + order: int + id: str + label: str + description: str + directory: Path + cases: tuple[HarnessCase, ...] + + +@dataclass +class CaseResult: + case: HarnessCase + status: RunStatus = RunStatus.NOT_RUN + collected: set[str] = field(default_factory=set) + completed: set[str] = field(default_factory=set) + passed: int = 0 + failed: int = 0 + skipped: int = 0 + errors: int = 0 + outcomes: dict[str, RunStatus] = field(default_factory=dict) + durations: dict[str, float] = field(default_factory=dict) + + @property + def total(self) -> int: + return len(self.collected) + + @property + def duration(self) -> float: + return sum(self.durations.values()) + + def record(self, nodeid: str, status: RunStatus, duration: float = 0.0) -> None: + """Record a terminal outcome, allowing teardown errors to replace a pass.""" + self.outcomes[nodeid] = status + self.durations[nodeid] = self.durations.get(nodeid, 0.0) + duration + self.completed = set(self.outcomes) + values = tuple(self.outcomes.values()) + self.passed = values.count(RunStatus.PASSED) + self.failed = values.count(RunStatus.FAILED) + self.skipped = values.count(RunStatus.SKIPPED) + self.errors = values.count(RunStatus.ERROR) + self.finalize() + + def set_initial_status(self) -> None: + if self.case.coverage is Coverage.NOT_APPLICABLE: + self.status = RunStatus.NOT_APPLICABLE + elif not self.case.selectors: + self.status = RunStatus.PLANNED + else: + self.status = RunStatus.QUEUED + + def finalize(self) -> None: + if self.status in {RunStatus.NOT_APPLICABLE, RunStatus.PLANNED}: + return + if not self.collected: + self.status = RunStatus.MISSING + elif self.errors: + self.status = RunStatus.ERROR + elif self.failed: + self.status = RunStatus.FAILED + elif self.passed and len(self.completed) == len(self.collected): + self.status = RunStatus.PASSED + elif self.skipped and len(self.completed) == len(self.collected): + self.status = RunStatus.SKIPPED + + +@dataclass +class HarnessRun: + results: dict[str, CaseResult] + current_nodeid: str | None = None + failures: list[tuple[str, str]] = field(default_factory=list) + started_at: float = field(default_factory=monotonic) + finished_at: float | None = None + + @property + def duration(self) -> float: + return (self.finished_at or monotonic()) - self.started_at + + @property + def unique_tests(self) -> int: + return len( + {nodeid for result in self.results.values() for nodeid in result.collected} + ) + + @property + def completed_tests(self) -> int: + return len( + {nodeid for result in self.results.values() for nodeid in result.completed} + ) + + @classmethod + def from_cases(cls, cases: Iterable[HarnessCase]) -> "HarnessRun": + results = {case.key: CaseResult(case=case) for case in cases} + for result in results.values(): + result.set_initial_status() + return cls(results=results) + + +@dataclass(frozen=True) +class SectionConfidence: + sdk_function: str + verified_strategies: int + required_strategies: int + level: ConfidenceLevel + details: tuple[str, ...] + + @property + def percentage(self) -> int: + if not self.required_strategies: + return 0 + return round(100 * self.verified_strategies / self.required_strategies) + + +def section_confidence( + run: HarnessRun, strategies: Iterable[Strategy] +) -> tuple[SectionConfidence, ...]: + strategy_list = tuple(strategies) + scores: list[SectionConfidence] = [] + for sdk_function in SDK_FUNCTIONS: + cases = tuple( + case + for strategy in strategy_list + for case in strategy.cases + if case.sdk_function == sdk_function + and case.coverage is not Coverage.NOT_APPLICABLE + ) + verified = 0 + details: list[str] = [] + for case in cases: + result = run.results.get(case.key) + status = result.status if result is not None else RunStatus.NOT_RUN + if status is RunStatus.PASSED: + verified += 1 + details.append( + f"{STATUS_LABELS[status]} {case.strategy_id} ({case.coverage.value})" + ) + required = len(cases) + if required and verified == required: + level = ConfidenceLevel.HIGH + elif verified: + level = ConfidenceLevel.MEDIUM + else: + level = ConfidenceLevel.LOW + scores.append( + SectionConfidence( + sdk_function=sdk_function, + verified_strategies=verified, + required_strategies=required, + level=level, + details=tuple(details), + ) + ) + return tuple(scores) + + +STATUS_LABELS = { + RunStatus.NOT_RUN: "·", + RunStatus.QUEUED: "○", + RunStatus.RUNNING: "◉", + RunStatus.PASSED: "✓", + RunStatus.FAILED: "✗", + RunStatus.SKIPPED: "↷", + RunStatus.ERROR: "!", + RunStatus.MISSING: "?", + RunStatus.PLANNED: "—", + RunStatus.NOT_APPLICABLE: "n/a", +} diff --git a/tests/rust-python-harness/runner.py b/tests/rust-python-harness/runner.py new file mode 100644 index 00000000000..82393ef234e --- /dev/null +++ b/tests/rust-python-harness/runner.py @@ -0,0 +1,160 @@ +from __future__ import annotations + +import os +from collections.abc import Callable, Sequence +from pathlib import Path +from time import monotonic + +import pytest + +from .models import CaseResult, HarnessCase, HarnessRun, RunStatus + +UpdateCallback = Callable[[HarnessRun], None] + + +def selector_matches_node(selector: str, nodeid: str) -> bool: + normalized_selector = selector.replace("\\", "/") + normalized_nodeid = nodeid.replace("\\", "/") + if "::" in normalized_selector: + return normalized_nodeid == normalized_selector or normalized_nodeid.startswith( + f"{normalized_selector}[" + ) + return normalized_nodeid == normalized_selector or normalized_nodeid.startswith( + f"{normalized_selector}::" + ) + + +def selector_path(selector: str) -> Path: + return Path(selector.split("::", 1)[0]) + + +def runnable_selectors( + cases: Sequence[HarnessCase], repo_root: Path +) -> tuple[str, ...]: + selectors = { + selector + for case in cases + for selector in case.selectors + if (repo_root / selector_path(selector)).exists() + } + return tuple(sorted(selectors)) + + +class HarnessPytestPlugin: + def __init__(self, run: HarnessRun, on_update: UpdateCallback) -> None: + self.run = run + self.on_update = on_update + self.node_to_results: dict[str, list[CaseResult]] = {} + + def _notify(self) -> None: + self.on_update(self.run) + + def pytest_collection_modifyitems(self, items: list[pytest.Item]) -> None: + for item in items: + matched_results: list[CaseResult] = [] + for result in self.run.results.values(): + if any( + selector_matches_node(selector, item.nodeid) + for selector in result.case.selectors + ): + result.collected.add(item.nodeid) + matched_results.append(result) + if matched_results: + self.node_to_results[item.nodeid] = matched_results + for result in self.run.results.values(): + if result.status is RunStatus.QUEUED and not result.collected: + result.status = RunStatus.MISSING + self._notify() + + def pytest_runtest_logstart( + self, nodeid: str, location: tuple[str, int | None, str] + ) -> None: + del location + self.run.current_nodeid = nodeid + for result in self.node_to_results.get(nodeid, []): + if result.status not in {RunStatus.FAILED, RunStatus.ERROR}: + result.status = RunStatus.RUNNING + self._notify() + + def pytest_runtest_logreport(self, report: pytest.TestReport) -> None: + if report.when not in {"setup", "call", "teardown"}: + return + results = self.node_to_results.get(report.nodeid, []) + if not results: + return + + terminal = report.when == "call" or report.failed or report.skipped + if not terminal: + for result in results: + result.durations[report.nodeid] = ( + result.durations.get(report.nodeid, 0.0) + report.duration + ) + return + for result in results: + if report.when == "teardown" and not report.failed: + result.durations[report.nodeid] = ( + result.durations.get(report.nodeid, 0.0) + report.duration + ) + continue + if report.skipped: + status = RunStatus.SKIPPED + elif report.failed and report.when in {"setup", "teardown"}: + status = RunStatus.ERROR + elif report.failed: + status = RunStatus.FAILED + else: + status = RunStatus.PASSED + result.record(report.nodeid, status, report.duration) + if report.failed: + failure = (report.nodeid, str(report.longrepr)) + if failure not in self.run.failures: + self.run.failures.append(failure) + self._notify() + + def pytest_sessionfinish( + self, session: pytest.Session, exitstatus: int | pytest.ExitCode + ) -> None: + del session, exitstatus + self.run.current_nodeid = None + self.run.finished_at = monotonic() + for result in self.run.results.values(): + result.finalize() + self._notify() + + +def run_pytest( + cases: Sequence[HarnessCase], + repo_root: Path, + on_update: UpdateCallback, + pytest_args: Sequence[str] = (), +) -> tuple[int, HarnessRun]: + run = HarnessRun.from_cases(cases) + selectors = runnable_selectors(cases, repo_root) + if not selectors: + for result in run.results.values(): + result.finalize() + run.finished_at = monotonic() + on_update(run) + has_missing_test = any( + result.status is RunStatus.MISSING for result in run.results.values() + ) + exit_code = ( + int(pytest.ExitCode.TESTS_FAILED) + if has_missing_test + else int(pytest.ExitCode.OK) + ) + return exit_code, run + + plugin = HarnessPytestPlugin(run=run, on_update=on_update) + args = [*selectors, "-p", "no:terminal", *pytest_args] + previous_directory = Path.cwd() + try: + os.chdir(repo_root) + exit_code = int(pytest.main(args, plugins=[plugin])) + finally: + os.chdir(previous_directory) + if exit_code == 0 and any( + result.status is RunStatus.MISSING for result in run.results.values() + ): + exit_code = int(pytest.ExitCode.TESTS_FAILED) + return exit_code, run diff --git a/tests/rust-python-harness/ui.py b/tests/rust-python-harness/ui.py new file mode 100644 index 00000000000..57fedd17fa6 --- /dev/null +++ b/tests/rust-python-harness/ui.py @@ -0,0 +1,295 @@ +from __future__ import annotations + +import os +import shlex +import sys +from collections.abc import Sequence +from contextlib import AbstractContextManager +from pathlib import Path +from typing import Any + +from .models import ( + Coverage, + HarnessRun, + RunStatus, + SDK_FUNCTIONS, + Strategy, + section_confidence, +) + +STATUS_GLYPHS = { + RunStatus.NOT_RUN: "·", + RunStatus.QUEUED: "○", + RunStatus.RUNNING: "◉", + RunStatus.PASSED: "✓", + RunStatus.FAILED: "✗", + RunStatus.SKIPPED: "↷", + RunStatus.ERROR: "!", + RunStatus.MISSING: "?", + RunStatus.PLANNED: "—", + RunStatus.NOT_APPLICABLE: "n/a", +} + +STATUS_STYLES = { + RunStatus.QUEUED: "dim", + RunStatus.RUNNING: "bold cyan", + RunStatus.PASSED: "bold green", + RunStatus.FAILED: "bold red", + RunStatus.SKIPPED: "yellow", + RunStatus.ERROR: "bold red", + RunStatus.MISSING: "magenta", + RunStatus.PLANNED: "dim", + RunStatus.NOT_APPLICABLE: "dim", +} + + +def _format_duration(seconds: float) -> str: + if seconds < 1: + return f"{seconds * 1000:.0f}ms" + if seconds < 60: + return f"{seconds:.1f}s" + return f"{int(seconds // 60)}m {seconds % 60:.0f}s" + + +def _rerun_command(nodeid: str) -> str: + return f"poetry run pytest {shlex.quote(nodeid)} -q" + + +def _summary(run: HarnessRun) -> tuple[int, int, int, int]: + outcomes: dict[str, RunStatus] = {} + for result in run.results.values(): + outcomes.update(result.outcomes) + return ( + list(outcomes.values()).count(RunStatus.PASSED), + list(outcomes.values()).count(RunStatus.FAILED), + list(outcomes.values()).count(RunStatus.ERROR), + list(outcomes.values()).count(RunStatus.SKIPPED), + ) + + +def _cell_text(run: HarnessRun, strategy_id: str, sdk_function: str) -> tuple[str, str]: + result = run.results.get(f"{strategy_id}:{sdk_function}") + if result is None: + return "", "" + counts = "" + if result.total: + counts = f" {len(result.completed)}/{result.total}" + coverage = " ◐" if result.case.coverage is Coverage.PARTIAL else "" + return f"{STATUS_GLYPHS[result.status]}{counts}{coverage}", STATUS_STYLES.get( + result.status, "" + ) + + +class RichDashboard(AbstractContextManager["RichDashboard"]): + def __init__( + self, + strategies: Sequence[Strategy], + confidence_strategies: Sequence[Strategy], + ) -> None: + from rich.console import Console + from rich.live import Live + + self.strategies = strategies + self.confidence_strategies = confidence_strategies + self.console = Console() + self.live: Any = Live( + console=self.console, refresh_per_second=12, transient=False + ) + + def _table(self, run: HarnessRun) -> Any: + from rich import box + from rich.table import Table + from rich.text import Text + + narrow = self.console.width < 96 + if narrow: + table = Table(box=box.SIMPLE_HEAVY, expand=True, show_header=False) + table.add_column("Strategy", ratio=3) + table.add_column("Results", ratio=5) + for strategy in self.strategies: + values = [] + for sdk_function in SDK_FUNCTIONS: + value, style = _cell_text(run, strategy.id, sdk_function) + if value: + values.append( + Text.assemble((f"{sdk_function} ", "dim"), (value, style)) + ) + table.add_row(strategy.label, Text(" ").join(values)) + return table + + table = Table(box=box.ROUNDED, expand=True, title="Strategy × SDK function") + table.add_column("Strategy", ratio=3) + for label in ("ocr/aocr", "messages", "responses", "count_tokens"): + table.add_column(label, justify="center", ratio=1) + for strategy in self.strategies: + cells = [] + for sdk_function in SDK_FUNCTIONS: + value, style = _cell_text(run, strategy.id, sdk_function) + cells.append(Text(value, style=style)) + table.add_row(strategy.label, *cells) + return table + + def __enter__(self) -> "RichDashboard": + self.live.__enter__() + return self + + def __exit__(self, *args: object) -> None: + self.live.__exit__(*args) + + def update(self, run: HarnessRun) -> None: + from rich.markup import escape + from rich.panel import Panel + + active = run.current_nodeid or "Waiting for test events…" + if len(active) > max(40, self.console.width - 16): + active = f"…{active[-(self.console.width - 17):]}" + passed, failed, errors, skipped = _summary(run) + progress = ( + f"[bold]{run.completed_tests}/{run.unique_tests}[/bold] tests " + f"[green]{passed} passed[/green] [red]{failed + errors} failed[/red] " + f"[yellow]{skipped} skipped[/yellow] [dim]{_format_duration(run.duration)}[/dim]" + ) + legend = "✓ pass ✗ fail ! error ↷ skip\n? configured test missing — planned ◐ partial coverage" + self.live.update( + Panel( + self._table(run), + title="⚡ Rust ↔ Python parity lab", + subtitle=f"{progress}\n[dim]{escape(active)}[/dim]\n{legend}", + border_style="cyan", + ) + ) + + def finish(self, run: HarnessRun, exit_code: int) -> None: + self.update(run) + if run.failures: + from rich.markup import escape + from rich.panel import Panel + + for nodeid, detail in run.failures[:5]: + rerun = _rerun_command(nodeid) + self.console.print( + Panel( + f"{escape(detail)}\n\n[bold]Rerun just this test[/bold]\n" + f"[cyan]{escape(rerun)}[/cyan]", + title=f"✗ {escape(nodeid)}", + border_style="red", + ) + ) + durations: dict[str, float] = {} + for result in run.results.values(): + for nodeid, duration in result.durations.items(): + durations[nodeid] = max(duration, durations.get(nodeid, 0.0)) + if durations: + slow = sorted(durations.items(), key=lambda item: item[1], reverse=True)[:3] + self.console.print( + "[bold]Slowest tests[/bold] " + + " • ".join( + f"{Path(nodeid).name} [dim]{_format_duration(duration)}[/dim]" + for nodeid, duration in slow + ) + ) + from rich import box + from rich.table import Table + + confidence_table = Table( + title="Port confidence by SDK section", box=box.ROUNDED, expand=True + ) + confidence_table.add_column("SDK section") + confidence_table.add_column("Score", justify="right") + confidence_table.add_column("Confidence") + confidence_table.add_column("Strategy evidence", ratio=4) + confidence_styles = {"HIGH": "green", "MEDIUM": "yellow", "LOW": "red"} + for score in section_confidence(run, self.confidence_strategies): + confidence_table.add_row( + score.sdk_function, + f"{score.verified_strategies}/{score.required_strategies} {score.percentage}%", + f"[{confidence_styles[score.level.value]}]{score.level.value}[/]", + " ".join(score.details), + ) + self.console.print(confidence_table) + self.console.print( + "[dim]Score = required strategies with passing evidence. " + "LOC coverage remains a separate report.[/dim]" + ) + style = "green" if exit_code == 0 else "red" + self.console.print( + f"[{style}]Harness finished in {_format_duration(run.duration)} " + f"(exit {exit_code})[/{style}]" + ) + + +class PlainDashboard(AbstractContextManager["PlainDashboard"]): + def __init__( + self, + strategies: Sequence[Strategy], + confidence_strategies: Sequence[Strategy], + ) -> None: + self.strategies = strategies + self.confidence_strategies = confidence_strategies + self._seen: dict[str, tuple[RunStatus, int]] = {} + + def __enter__(self) -> "PlainDashboard": + print("Rust <-> Python SDK parity harness", flush=True) + return self + + def __exit__(self, *args: object) -> None: + return None + + def update(self, run: HarnessRun) -> None: + for key, result in run.results.items(): + state = (result.status, len(result.completed)) + if self._seen.get(key) != state: + self._seen[key] = state + progress = ( + f" {len(result.completed)}/{result.total}" if result.total else "" + ) + print( + f"{STATUS_GLYPHS[result.status]} {key}: {result.status.value}{progress}", + flush=True, + ) + + def finish(self, run: HarnessRun, exit_code: int) -> None: + self.update(run) + passed, failed, errors, skipped = _summary(run) + print( + f"Summary: {passed} passed, {failed} failed, {errors} errors, " + f"{skipped} skipped in {_format_duration(run.duration)}", + flush=True, + ) + for nodeid, _ in run.failures[:5]: + print(f"Rerun: {_rerun_command(nodeid)}", flush=True) + print("Port confidence by SDK section", flush=True) + for score in section_confidence(run, self.confidence_strategies): + print( + f" {score.sdk_function:12} " + f"{score.verified_strategies}/{score.required_strategies} " + f"{score.percentage:3}% {score.level.value:6} " + f"{' | '.join(score.details)}", + flush=True, + ) + print( + " Score = required strategies with passing evidence; LOC is reported separately.", + flush=True, + ) + print(f"Harness finished with exit code {exit_code}", flush=True) + + +def make_dashboard( + strategies: Sequence[Strategy], + plain: bool = False, + confidence_strategies: Sequence[Strategy] | None = None, +) -> RichDashboard | PlainDashboard: + confidence_strategies = confidence_strategies or strategies + interactive_terminal = ( + sys.stdout.isatty() + and not os.environ.get("CI") + and os.environ.get("TERM") != "dumb" + ) + if not plain and interactive_terminal: + try: + import rich # noqa: F401 + + return RichDashboard(strategies, confidence_strategies) + except ImportError: + pass + return PlainDashboard(strategies, confidence_strategies) diff --git a/tests/rust-python-harness/unit_tests_rust/README.md b/tests/rust-python-harness/unit_tests_rust/README.md new file mode 100644 index 00000000000..12c7eb0089c --- /dev/null +++ b/tests/rust-python-harness/unit_tests_rust/README.md @@ -0,0 +1,3 @@ +# Rust unit tests + +Holds focused Cargo tests for Rust-owned parsing, transforms, errors, and streaming behavior. These tests make failures fast to diagnose before the Python bridge or full SDK path is involved. diff --git a/tests/rust-python-harness/unit_tests_rust/strategy.json b/tests/rust-python-harness/unit_tests_rust/strategy.json new file mode 100644 index 00000000000..89e897c872d --- /dev/null +++ b/tests/rust-python-harness/unit_tests_rust/strategy.json @@ -0,0 +1,12 @@ +{ + "order": 20, + "id": "unit_tests_rust", + "label": "Rust unit tests", + "description": "Exercise Rust-owned behavior directly with focused unit tests.", + "functions": { + "ocr": {"coverage": "planned", "selectors": []}, + "messages": {"coverage": "planned", "selectors": []}, + "responses": {"coverage": "planned", "selectors": []}, + "count_tokens": {"coverage": "planned", "selectors": []} + } +} diff --git a/tests/rust-python-harness/validate_sub_methods/README.md b/tests/rust-python-harness/validate_sub_methods/README.md new file mode 100644 index 00000000000..24894366f23 --- /dev/null +++ b/tests/rust-python-harness/validate_sub_methods/README.md @@ -0,0 +1,3 @@ +# Validate sub-methods + +Checks each request, response, stream, and error-mapping sub-method independently across Python and Rust. It also validates that traced Python helpers have an explicit Rust implementation and parity test. diff --git a/tests/rust-python-harness/validate_sub_methods/strategy.json b/tests/rust-python-harness/validate_sub_methods/strategy.json new file mode 100644 index 00000000000..6e6381678e0 --- /dev/null +++ b/tests/rust-python-harness/validate_sub_methods/strategy.json @@ -0,0 +1,12 @@ +{ + "order": 30, + "id": "validate_sub_methods", + "label": "Validate sub-methods", + "description": "Compare isolated transforms and verify Python-to-Rust helper coverage.", + "functions": { + "ocr": {"coverage": "planned", "selectors": []}, + "messages": {"coverage": "planned", "selectors": []}, + "responses": {"coverage": "planned", "selectors": []}, + "count_tokens": {"coverage": "planned", "selectors": []} + } +} diff --git a/tests/test_rust_python_harness.py b/tests/test_rust_python_harness.py new file mode 100644 index 00000000000..90dc38663dd --- /dev/null +++ b/tests/test_rust_python_harness.py @@ -0,0 +1,236 @@ +from __future__ import annotations + +import importlib +import json +from pathlib import Path + +import pytest + +catalog = importlib.import_module("tests.rust-python-harness.catalog") +cli = importlib.import_module("tests.rust-python-harness.cli") +models = importlib.import_module("tests.rust-python-harness.models") +runner = importlib.import_module("tests.rust-python-harness.runner") +ui = importlib.import_module("tests.rust-python-harness.ui") + +load_catalog = catalog.load_catalog +_pick_values = cli._pick_values +_coverage_pytest_args = cli._coverage_pytest_args +_select = cli._select +CaseResult = models.CaseResult +Coverage = models.Coverage +HarnessCase = models.HarnessCase +HarnessRun = models.HarnessRun +RunStatus = models.RunStatus +SDK_FUNCTIONS = models.SDK_FUNCTIONS +section_confidence = models.section_confidence +run_pytest = runner.run_pytest +runnable_selectors = runner.runnable_selectors +selector_matches_node = runner.selector_matches_node +_format_duration = ui._format_duration +_rerun_command = ui._rerun_command +_summary = ui._summary + + +def _case( + *, selectors: tuple[str, ...] = (), coverage: Coverage = Coverage.COMPLETE +) -> HarnessCase: + return HarnessCase( + strategy_id="example", + strategy_label="Example", + sdk_function="messages", + coverage=coverage, + selectors=selectors, + ) + + +def _manifest() -> dict[str, object]: + return { + "order": 1, + "id": "example", + "label": "Example strategy", + "description": "Example description", + "functions": { + function: {"coverage": "planned", "selectors": []} + for function in SDK_FUNCTIONS + }, + } + + +def test_should_load_the_three_harness_strategies_in_order() -> None: + strategies = load_catalog() + + assert [strategy.id for strategy in strategies] == [ + "e2e_fuzz_tests", + "unit_tests_rust", + "validate_sub_methods", + ] + assert all( + tuple(case.sdk_function for case in strategy.cases) == SDK_FUNCTIONS + for strategy in strategies + ) + + +def test_should_reject_a_manifest_missing_an_sdk_function(tmp_path: Path) -> None: + strategy_directory = tmp_path / "example" + strategy_directory.mkdir() + manifest = _manifest() + del manifest["functions"]["count_tokens"] # type: ignore[index] + (strategy_directory / "strategy.json").write_text( + json.dumps(manifest), encoding="utf-8" + ) + + with pytest.raises(ValueError, match="functions must exactly match"): + load_catalog(tmp_path) + + +@pytest.mark.parametrize( + ("selector", "nodeid", "matches"), + [ + ("tests/test_parity.py", "tests/test_parity.py::test_one", True), + ("tests/test_parity.py::test_one", "tests/test_parity.py::test_one", True), + ( + "tests/test_parity.py::test_one", + "tests/test_parity.py::test_one[value]", + True, + ), + ("tests/test_parity.py::test_one", "tests/test_parity.py::test_two", False), + ], +) +def test_should_match_pytest_file_and_node_selectors( + selector: str, nodeid: str, matches: bool +) -> None: + assert selector_matches_node(selector, nodeid) is matches + + +def test_should_only_return_selectors_whose_files_exist(tmp_path: Path) -> None: + existing = tmp_path / "tests" / "test_parity.py" + existing.parent.mkdir() + existing.write_text("", encoding="utf-8") + case = _case( + selectors=("tests/test_parity.py", "tests/test_missing.py::test_missing") + ) + + assert runnable_selectors((case,), tmp_path) == ("tests/test_parity.py",) + + +def test_should_mark_planned_and_not_applicable_cases_without_running() -> None: + planned = CaseResult(case=_case(coverage=Coverage.PLANNED)) + not_applicable = CaseResult(case=_case(coverage=Coverage.NOT_APPLICABLE)) + + planned.set_initial_status() + not_applicable.set_initial_status() + + assert planned.status is RunStatus.PLANNED + assert not_applicable.status is RunStatus.NOT_APPLICABLE + + +def test_should_treat_an_all_planned_filtered_run_as_success(tmp_path: Path) -> None: + exit_code, run = run_pytest( + cases=(_case(coverage=Coverage.PLANNED),), + repo_root=tmp_path, + on_update=lambda _: None, + ) + + assert exit_code == 0 + assert next(iter(run.results.values())).status is RunStatus.PLANNED + + +def test_should_finalize_a_fully_passing_case() -> None: + result = CaseResult(case=_case(selectors=("tests/test_parity.py",))) + result.set_initial_status() + result.collected.update({"one", "two"}) + result.completed.update({"one", "two"}) + result.passed = 2 + + result.finalize() + + assert result.status is RunStatus.PASSED + + +def test_should_replace_a_pass_with_a_teardown_error() -> None: + result = CaseResult(case=_case(selectors=("tests/test_parity.py",))) + result.set_initial_status() + result.collected.add("one") + + result.record("one", RunStatus.PASSED, 0.1) + result.record("one", RunStatus.ERROR, 0.2) + + assert result.status is RunStatus.ERROR + assert result.passed == 0 + assert result.errors == 1 + assert result.duration == pytest.approx(0.3) + + +def test_should_filter_the_catalog_by_strategy_and_sdk_function() -> None: + strategies = load_catalog() + + cases = _select(strategies, {"e2e_fuzz_tests"}, {"messages"}) + + assert len(cases) == 1 + assert cases[0].key == "e2e_fuzz_tests:messages" + + +def test_should_reject_an_unknown_strategy() -> None: + with pytest.raises(ValueError, match="Unknown strategy"): + _select(load_catalog(), {"not-real"}, set()) + + +def test_should_pick_multiple_interactive_filters() -> None: + answers = iter(["nope", "1, 3"]) + + selected = _pick_values( + "Examples", + (("one", "One"), ("two", "Two"), ("three", "Three")), + input_fn=lambda _: next(answers), + ) + + assert selected == {"one", "three"} + + +def test_should_format_developer_facing_run_context() -> None: + run = HarnessRun.from_cases((_case(selectors=("tests/test_parity.py",)),)) + result = next(iter(run.results.values())) + result.collected.add("tests/test_parity.py::test_one") + result.record("tests/test_parity.py::test_one", RunStatus.PASSED, 1.25) + + assert _summary(run) == (1, 0, 0, 0) + assert _format_duration(1.25) == "1.2s" + assert _rerun_command("tests/test_parity.py::test_one") == ( + "poetry run pytest tests/test_parity.py::test_one -q" + ) + assert _rerun_command("tests/test_parity.py::test_one[value with spaces]") == ( + "poetry run pytest 'tests/test_parity.py::test_one[value with spaces]' -q" + ) + + +def test_should_build_python_coverage_reports_below_the_target_directory( + tmp_path: Path, +) -> None: + args = _coverage_pytest_args(tmp_path) + + assert tmp_path.is_dir() + assert "--cov=litellm" in args + assert "--cov-context=test" in args + assert f"--cov-report=json:{tmp_path / 'python.json'}" in args + assert f"--cov-report=xml:{tmp_path / 'python.xml'}" in args + assert f"--cov-report=html:{tmp_path / 'python-html'}" in args + + +def test_should_report_confidence_for_each_sdk_section() -> None: + strategies = load_catalog() + cases = tuple(case for strategy in strategies for case in strategy.cases) + run = HarnessRun.from_cases(cases) + passing = run.results["e2e_fuzz_tests:responses"] + passing.collected.add("tests/test_parity.py::test_one") + passing.record("tests/test_parity.py::test_one", RunStatus.PASSED) + + scores = { + score.sdk_function: score for score in section_confidence(run, strategies) + } + + assert scores["responses"].verified_strategies == 1 + assert scores["responses"].required_strategies == 3 + assert scores["responses"].percentage == 33 + assert scores["responses"].level.value == "MEDIUM" + assert scores["count_tokens"].percentage == 0 + assert scores["count_tokens"].level.value == "LOW"