litellm/tests/sdk_function_trace/compare.py
yujonglee 198906495f
refactor(python-bridge): split routes and add shared function tracing (#39031)
* refactor(python-bridge): split non-streaming bridge modules

* refactor(python-bridge): bring shared function tracing into route layer

* feat(dev): list Python route functions and call sites

* feat(dev): list Rust route functions and call sites

* docs(dev): record OCR parity gaps across Python and Rust

* feat(dev): list executed SDK calls with runtime tracing

* feat(dev): report Python vs Rust SDK pipeline steps in one CLI

* feat(dev): side-by-side pipeline step report in compare CLI

* fix(dev): drop invalid Final annotations in compare cell loop

* feat(dev): blue python-only and yellow rust-only steps in compare CLI

* feat(dev): vertical layout with section spacing in compare CLI

* fix(dev): validate SDK trace stages across sync and async routes

* refactor(rust): align SDK route call structure with Python

* refactor(python-bridge): share sync and async route call wrappers

* refactor(dev): split compare CLI into fixtures, runtime, and report modules

* fix(ci): run SDK trace tests and satisfy test lint
2026-09-02 16:26:35 -07:00

45 lines
1.8 KiB
Python

from __future__ import annotations
import argparse
import os
import sys
from typing import Final
from tests.sdk_function_trace.fixtures import ROUTES
from tests.sdk_function_trace.report import compare, render
def _run(route: str, asynchronous: bool, *, full: bool, colorize: bool) -> bool:
comparison: Final = compare(route, asynchronous=asynchronous)
sys.stdout.write(render(comparison, full=full, colorize=colorize))
return comparison.passed
def main() -> None:
parser: Final = argparse.ArgumentParser(description="Compare Python and Rust SDK pipeline steps per route")
parser.add_argument("--route", choices=("all", *ROUTES), default="all")
mode: Final = parser.add_mutually_exclusive_group()
mode.add_argument("--async", dest="asynchronous", action="store_true", default=True)
mode.add_argument("--sync", dest="asynchronous", action="store_false")
mode.add_argument("--both", action="store_true", help="run async and sync for every selected route")
parser.add_argument(
"--check", action="store_true", help="exit nonzero for missing, extra, or reordered comparable steps"
)
parser.add_argument(
"--full", action="store_true", help="print every captured runtime event instead of pipeline steps"
)
args: Final = parser.parse_args()
os.environ.setdefault("LITELLM_LOCAL_MODEL_COST_MAP", "True")
colorize: Final = sys.stdout.isatty() and "NO_COLOR" not in os.environ
results: Final = tuple(
_run(selected, selected_mode, full=args.full, colorize=colorize)
for selected in ROUTES
if args.route in ("all", selected)
for selected_mode in ((True, False) if args.both else (args.asynchronous,))
)
if args.check and not all(results):
raise SystemExit(1)
if __name__ == "__main__":
main()