mirror of
https://github.com/BerriAI/litellm.git
synced 2026-09-16 23:41:43 +00:00
121 lines
4.1 KiB
Python
121 lines
4.1 KiB
Python
from __future__ import annotations
|
|
|
|
import importlib
|
|
from typing import Final
|
|
|
|
import pytest
|
|
|
|
models = importlib.import_module("tests.rust-python-harness.shared.reporting.models")
|
|
strategy_module = importlib.import_module("tests.rust-python-harness.shared.reporting.strategy")
|
|
ui = importlib.import_module("tests.rust-python-harness.shared.reporting.ui")
|
|
contracts = importlib.import_module("tests.rust-python-harness.shared.unit_runners.contracts")
|
|
cli = importlib.import_module("tests.rust-python-harness.cli")
|
|
|
|
UNIT_TEST_CONTRACTS = contracts.UNIT_TEST_CONTRACTS
|
|
CaseResult = models.CaseResult
|
|
Coverage = models.Coverage
|
|
HarnessCase = models.HarnessCase
|
|
HarnessRun = models.HarnessRun
|
|
RunStatus = models.RunStatus
|
|
ModuleCaseSpec = strategy_module.ModuleCaseSpec
|
|
NotImplementedCaseSpec = strategy_module.NotImplementedCaseSpec
|
|
SkippedCaseSpec = strategy_module.SkippedCaseSpec
|
|
_format_duration = ui._format_duration
|
|
_summary = ui._summary
|
|
|
|
|
|
def _case(module: str = "tests.example") -> HarnessCase:
|
|
return HarnessCase(
|
|
strategy_id="example",
|
|
strategy_label="Example",
|
|
sdk_function="messages",
|
|
spec=ModuleCaseSpec(coverage=Coverage.COMPLETE, module=module),
|
|
)
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"module",
|
|
[
|
|
"tests.rust-python-harness.strategies.e2e_parity.sdk.ocr.test_sdk_parity",
|
|
"tests.rust-python-harness.strategies.trace_parity.sdk.ocr.case",
|
|
"tests.rust-python-harness.strategies.trace_parity.sdk.messages.case",
|
|
"tests.rust-python-harness.strategies.trace_parity.sdk.chat_completions.case",
|
|
"tests.rust-python-harness.strategies.trace_parity.sdk.transcription.case",
|
|
],
|
|
)
|
|
def test_implemented_namespace_case_modules_remain_importable(module: str) -> None:
|
|
assert importlib.import_module(module)
|
|
|
|
|
|
def test_should_mark_not_implemented_and_skipped_cases_without_running() -> None:
|
|
not_implemented: Final = CaseResult(
|
|
case=HarnessCase(
|
|
strategy_id="example",
|
|
strategy_label="Example",
|
|
sdk_function="messages",
|
|
spec=NotImplementedCaseSpec(reason="No case is registered."),
|
|
)
|
|
)
|
|
skipped: Final = CaseResult(
|
|
case=HarnessCase(
|
|
strategy_id="example",
|
|
strategy_label="Example",
|
|
sdk_function="messages",
|
|
spec=SkippedCaseSpec(reason="The surface does not apply."),
|
|
)
|
|
)
|
|
|
|
not_implemented.set_initial_status()
|
|
skipped.set_initial_status()
|
|
|
|
assert not_implemented.status is RunStatus.NOT_IMPLEMENTED
|
|
assert skipped.status is RunStatus.SKIPPED
|
|
|
|
|
|
def test_should_finalize_a_fully_passing_case() -> None:
|
|
result = CaseResult(case=_case())
|
|
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())
|
|
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_format_developer_facing_run_context() -> None:
|
|
run = HarnessRun.from_cases((_case(),))
|
|
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"
|
|
|
|
|
|
def test_should_leave_functions_without_unit_test_contracts_unimplemented() -> None:
|
|
assert "messages" not in UNIT_TEST_CONTRACTS
|
|
|
|
|
|
def test_strategy_subcommand_accepts_function_filter(capsys: pytest.CaptureFixture[str]) -> None:
|
|
exit_code: Final = cli.main(["run", "unit_tests_rust", "--function", "messages"])
|
|
|
|
captured: Final = capsys.readouterr()
|
|
assert exit_code == 0
|
|
assert "- messages: not_implemented" in captured.out
|
|
assert "unit_tests_rust:messages: not_implemented" not in captured.out
|