mirror of
https://github.com/BerriAI/litellm.git
synced 2026-09-28 01:32:17 +00:00
f12f7b5a03
* test(integration): group /v1/messages contracts under tests/integration/messages Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> * test(integration): make ci coverage census collect nested test dirs Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> * test(integration): nest /v1/messages contracts under messages_endpoint/providers Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> --------- Co-authored-by: kerry <kerry@berri.ai> Co-authored-by: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>
99 lines
3.7 KiB
Python
99 lines
3.7 KiB
Python
from __future__ import annotations
|
|
|
|
import argparse
|
|
import json
|
|
import os
|
|
import subprocess
|
|
import sys
|
|
from pathlib import Path
|
|
from types import MappingProxyType
|
|
from typing import Final
|
|
|
|
GROUPS: Final = MappingProxyType(
|
|
{
|
|
"management": ("management", "authorization", "configuration"),
|
|
"accounting": ("pricing", "spend"),
|
|
"database": ("database",),
|
|
"providers": ("providers", "routing", "streaming", "messages_endpoint"),
|
|
"extensions": ("observability", "compatibility"),
|
|
"mcp": ("mcp",),
|
|
"sdk": ("sdk",),
|
|
"cost": ("cost_calculation",),
|
|
}
|
|
)
|
|
|
|
|
|
def main() -> int:
|
|
parser: Final = argparse.ArgumentParser()
|
|
parser.add_argument("group", choices=tuple(GROUPS))
|
|
parser.add_argument("--results", type=Path, default=Path("test-results/integration"))
|
|
parser.add_argument("--seed", type=int, default=int(os.environ.get("INTEGRATION_SEED", "4106601")))
|
|
parser.add_argument("--order-seed", type=int, default=int(os.environ.get("INTEGRATION_ORDER_SEED", "0")))
|
|
parser.add_argument("--workers", type=int, default=int(os.environ.get("INTEGRATION_WORKERS", "1")))
|
|
parser.add_argument("--list", action="store_true", help="print the group's test files and exit")
|
|
parser.add_argument("files", nargs="*", help="run only these files of the group")
|
|
options: Final = parser.parse_intermixed_args()
|
|
root: Final = Path(__file__).resolve().parents[2]
|
|
group_files: Final = tuple(
|
|
str(path.relative_to(root))
|
|
for folder in GROUPS[options.group]
|
|
for path in sorted((root / "tests/integration" / folder).rglob("test_*.py"))
|
|
)
|
|
if options.list:
|
|
print("\n".join(group_files))
|
|
return 0
|
|
foreign: Final = sorted(set(options.files) - set(group_files))
|
|
if foreign:
|
|
parser.error(f"Not in the {options.group} group: {', '.join(foreign)}")
|
|
selected: Final = tuple(options.files) or group_files
|
|
if not selected:
|
|
parser.error(f"No integration test files selected for {options.group}")
|
|
output: Final = options.results.resolve()
|
|
output.mkdir(parents=True, exist_ok=True)
|
|
environment: Final = {
|
|
**os.environ,
|
|
"PYTHONPATH": os.pathsep.join((str(root), str(root / "tests"), str(root / "tests/e2e"))),
|
|
"INTEGRATION_RESULTS_DIR": str(output),
|
|
"LITELLM_LOCAL_MODEL_COST_MAP": "True",
|
|
}
|
|
result: Final = subprocess.call(
|
|
[
|
|
sys.executable,
|
|
"-m",
|
|
"pytest",
|
|
*selected,
|
|
"-vv",
|
|
"-rs",
|
|
"--strict-markers",
|
|
"-p",
|
|
"no:pytest-retry",
|
|
"-p",
|
|
"no:rerunfailures",
|
|
"--timeout=90",
|
|
"--durations=15",
|
|
f"--hypothesis-seed={options.seed}",
|
|
f"--integration-order-seed={options.order_seed}",
|
|
f"--junitxml={output / 'junit.xml'}",
|
|
"-o",
|
|
"junit_family=xunit1",
|
|
*(("-n", str(options.workers)) if options.workers > 1 else ()),
|
|
],
|
|
cwd=root,
|
|
env=environment,
|
|
)
|
|
if result != 0:
|
|
return result
|
|
evidence: Final = json.loads((output / "execution.json").read_text())
|
|
collected_files: Final = {node.split("::", 1)[0] for node in evidence["collected"]}
|
|
empty: Final = tuple(path for path in selected if path not in collected_files)
|
|
if empty:
|
|
sys.stderr.write(f"Selected integration files collected zero tests: {', '.join(empty)}\n")
|
|
return 1
|
|
if not evidence["complete"]:
|
|
sys.stderr.write("Integration run did not complete: a collected node neither passed nor skipped\n")
|
|
return 1
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|