scriptoriumadmin/litellm
1
0
Fork 0
mirror of https://github.com/BerriAI/litellm.git synced 2026-09-23 00:41:40 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 81

fix(claude_code): harden parallel runner + de-dup basic_messaging cells

Browse source 
- run_claude_models_parallel: catch all exceptions in the per-model
  worker and wrap unexpected ones into a ClaudeCLIError so the
  documented 'errors as values' contract holds for OSError, ValueError,
  etc., not just ClaudeCLIError. Without this, an unexpected raise in
  any layer (rate limiter file I/O, infer_provider, etc.) abandons the
  remaining models' results and crashes the calling test.
- test_run_claude_places_extra_args_before_prompt: drop the dead first
  branch of the 'or' assertion — cmd[-3:] never matches that shape, so
  the alternative was misleading dead code.
- basic_messaging_{non_streaming,streaming}/test_*.py: extract the
  shared cell body into tests/claude_code/_basic_messaging.py.
  Each per-provider file now declares its model list and calls
  run_basic_messaging_cell(), eliminating ~700 lines of copy-paste
  across 10 files. Updated _builder_unit_tests/test_v0_layout.py to
  accept the helper-based pattern alongside direct run_claude() calls.

Co-authored-by: Yassin Kortam <yassin@berri.ai>
This commit is contained in:
Cursor Agent 2026-05-17 06:35:15 +00:00
parent 9a4eb2dea5
commit f41d3f91a3
No known key found for this signature in database
14 changed files with 197 additions and 634 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

109
tests/claude_code/_basic_messaging.py Normal file
View file

@ -0,0 +1,109 @@
"""Shared body for the `basic_messaging_*` × <provider> compat cells.
Every basic_messaging cell follows the same skeleton:
1. Read the proxy base URL + API key from env, fail-early if missing.
2. Fan the three Claude tiers out via `run_claude_models_parallel`.
3. Inspect each model's outcome and report one `compat_result` row per
model — `ClaudeCLIError`, non-zero exit, missing stream events
(streaming variant only), and empty assistant text are all per-model
fails; everything else is a per-model pass.
4. Surface a joined failure message via `pytest.fail(...)` so the
pytest run also goes red.
The conftest infers `(feature_id, provider)` purely from the test file
path, so each per-provider file just declares its model list and calls
`run_basic_messaging_cell(...)`. This keeps all cell logic in one place
— a future tweak to the env-missing guard, the failure-loop shape, or
the stream-events check now propagates to every cell automatically.
The leading underscore in the filename is what keeps pytest from
collecting this module as a test file.
"""
from __future__ import annotations
import os
from typing import Sequence
import pytest
from tests.claude_code.cli_driver import (
ClaudeCLIError,
failure_diagnostic,
run_claude_models_parallel,
)
PROXY_BASE_URL_ENV = "LITELLM_PROXY_BASE_URL"
PROXY_API_KEY_ENV = "LITELLM_PROXY_API_KEY"
def run_basic_messaging_cell(
*,
compat_result,
models: Sequence[str],
prompt: str,
require_stream_events: bool = False,
) -> None:
"""Run the shared `basic_messaging_*` × <provider> cell body.
`require_stream_events=True` adds the streaming-variant assertion
that at least one stream-json event is observed for each model —
the regression check that catches a proxy buffering the full
response before flushing.
"""
base_url = os.environ.get(PROXY_BASE_URL_ENV)
api_key = os.environ.get(PROXY_API_KEY_ENV)
if not base_url or not api_key:
compat_result.set(
{
"status": "fail",
"error": (
f"missing required env: set {PROXY_BASE_URL_ENV} and "
f"{PROXY_API_KEY_ENV} to point at a running LiteLLM proxy"
),
}
)
pytest.fail(
f"{PROXY_BASE_URL_ENV} / {PROXY_API_KEY_ENV} not configured",
pytrace=False,
)
outcomes = run_claude_models_parallel(
models=models,
prompt=prompt,
base_url=base_url,
api_key=api_key,
)
failures = []
for model in models:
outcome = outcomes[model]
if isinstance(outcome, ClaudeCLIError):
error = f"[{model}] {outcome}"
compat_result.add({"status": "fail", "error": error})
failures.append(error)
continue
if outcome.exit_code != 0:
error = f"[{model}] claude CLI failed: {failure_diagnostic(outcome)}"
compat_result.add({"status": "fail", "error": error})
failures.append(error)
continue
if require_stream_events and not outcome.events:
error = f"[{model}] no stream-json events emitted; streaming wire silent"
compat_result.add({"status": "fail", "error": error})
failures.append(error)
continue
if not outcome.text.strip():
error = f"[{model}] claude returned empty assistant text"
compat_result.add({"status": "fail", "error": error})
failures.append(error)
continue
compat_result.add({"status": "pass"})
if failures:
pytest.fail("; ".join(failures), pytrace=False)

15
tests/claude_code/_builder_unit_tests/test_v0_layout.py
View file

@ -116,11 +116,18 @@ def test_azure_test_file_drives_the_proxy(feature_id):
so every Azure cell in the v0 matrix exercises a real route through
the LiteLLM proxy — same shape as the other provider columns. Pin
that here so a future regression doesn't silently revert these
cells to the old `not_applicable` boilerplate."""
cells to the old `not_applicable` boilerplate.
We accept either the direct `run_claude(...)` family of entrypoints
or a per-feature shared helper (e.g. `run_basic_messaging_cell`)
that wraps them — both shapes drive the proxy, and we don't want
this layout pin to block legitimate de-duplication of test bodies.
"""
text = (REPO_ROOT / feature_id / "test_azure.py").read_text()
assert "run_claude" in text, (
f"{feature_id}/test_azure.py must drive the claude CLI via run_claude(); "
"the not_applicable stub was removed when Foundry started hosting Claude."
assert "run_claude" in text or "run_basic_messaging_cell" in text, (
f"{feature_id}/test_azure.py must drive the claude CLI via run_claude() "
"or a shared helper that wraps it; the not_applicable stub was removed "
"when Foundry started hosting Claude."
)
assert '"status": "not_applicable"' not in text, (
f"{feature_id}/test_azure.py still reports not_applicable; Microsoft Foundry "

8
tests/claude_code/_driver_unit_tests/test_cli_driver.py
View file

@ -86,12 +86,8 @@ def test_run_claude_places_extra_args_before_prompt():
runner=runner,
)
cmd = captured["cmd"]
assert cmd[-3:] == ["--allowed-tools", "Bash", "--"] or cmd[-2:] == [
"--",
"say hi",
]
# Stronger: prompt is last, `--` immediately precedes it, and the
# caller's extra_args sit somewhere earlier in the command.
# Prompt is last, `--` immediately precedes it, and the caller's
# extra_args sit somewhere earlier in the command.
assert cmd[-2:] == ["--", "say hi"]
assert "--allowed-tools" in cmd
assert cmd.index("--allowed-tools") < cmd.index("--")

69
tests/claude_code/basic_messaging_non_streaming/test_anthropic.py
View file

@ -12,27 +12,15 @@ The (feature, provider) for this cell is inferred from the file path by
feature_id provider
Per the PRD, every cell exercises Claude Haiku 4.5, Sonnet 4.6, and Opus
4.7; the cell only goes green if all three pass. We fan the three model
runs out in parallel inside this single test so the per-cell wall time is
bounded by the slowest model rather than the sum, and report one
`compat_result.add(...)` entry per model so the matrix builder still sees
three rows for this (feature, provider).
4.7; the cell only goes green if all three pass. The shared
`run_basic_messaging_cell` helper fans the three model runs out in
parallel and reports one `compat_result.add(...)` entry per model so
the matrix builder still sees three rows for this (feature, provider).
"""
from __future__ import annotations
import os
import pytest
from tests.claude_code.cli_driver import (
ClaudeCLIError,
failure_diagnostic,
run_claude_models_parallel,
)
PROXY_BASE_URL_ENV = "LITELLM_PROXY_BASE_URL"
PROXY_API_KEY_ENV = "LITELLM_PROXY_API_KEY"
from tests.claude_code._basic_messaging import run_basic_messaging_cell
# Per the PRD: each cell is exercised against three Claude tiers via the
# Anthropic provider. Aliases are configured in the LiteLLM proxy's
@ -52,51 +40,8 @@ def test_basic_messaging_non_streaming_anthropic(compat_result):
The whole point of this slice is to prove the path works at all —
so the assertion is intentionally lenient on the reply contents.
"""
base_url = os.environ.get(PROXY_BASE_URL_ENV)
api_key = os.environ.get(PROXY_API_KEY_ENV)
if not base_url or not api_key:
compat_result.set(
{
"status": "fail",
"error": (
f"missing required env: set {PROXY_BASE_URL_ENV} and "
f"{PROXY_API_KEY_ENV} to point at a running LiteLLM proxy"
),
}
)
pytest.fail(
f"{PROXY_BASE_URL_ENV} / {PROXY_API_KEY_ENV} not configured", pytrace=False
)
outcomes = run_claude_models_parallel(
run_basic_messaging_cell(
compat_result=compat_result,
models=ANTHROPIC_MODELS,
prompt="Reply with the single word 'pong' and nothing else.",
base_url=base_url,
api_key=api_key,
)
failures = []
for model in ANTHROPIC_MODELS:
outcome = outcomes[model]
if isinstance(outcome, ClaudeCLIError):
error = f"[{model}] {outcome}"
compat_result.add({"status": "fail", "error": error})
failures.append(error)
continue
if outcome.exit_code != 0:
error = f"[{model}] claude CLI failed: {failure_diagnostic(outcome)}"
compat_result.add({"status": "fail", "error": error})
failures.append(error)
continue
if not outcome.text.strip():
error = f"[{model}] claude returned empty assistant text"
compat_result.add({"status": "fail", "error": error})
failures.append(error)
continue
compat_result.add({"status": "pass"})
if failures:
pytest.fail("; ".join(failures), pytrace=False)

68
tests/claude_code/basic_messaging_non_streaming/test_azure.py
View file

@ -17,26 +17,15 @@ The (feature, provider) for this cell is inferred from the file path by
feature_id provider
Per the PRD, every cell exercises Claude Haiku 4.5, Sonnet 4.6, and Opus
4.7; the cell only goes green if all three pass. We fan the three model
runs out in parallel inside this single test and report one
`compat_result.add(...)` entry per model so the matrix builder still sees
three rows for this (feature, provider).
4.7; the cell only goes green if all three pass. The shared
`run_basic_messaging_cell` helper fans the three model runs out in
parallel and reports one `compat_result.add(...)` entry per model so
the matrix builder still sees three rows for this (feature, provider).
"""
from __future__ import annotations
import os
import pytest
from tests.claude_code.cli_driver import (
ClaudeCLIError,
failure_diagnostic,
run_claude_models_parallel,
)
PROXY_BASE_URL_ENV = "LITELLM_PROXY_BASE_URL"
PROXY_API_KEY_ENV = "LITELLM_PROXY_API_KEY"
from tests.claude_code._basic_messaging import run_basic_messaging_cell
# Per-model aliases registered in the LiteLLM proxy's routing config to
# point at Microsoft Foundry's Anthropic deployments. The driver only
@ -57,51 +46,8 @@ def test_basic_messaging_non_streaming_azure(compat_result):
The whole point of this slice is to prove the path works at all —
so the assertion is intentionally lenient on the reply contents.
"""
base_url = os.environ.get(PROXY_BASE_URL_ENV)
api_key = os.environ.get(PROXY_API_KEY_ENV)
if not base_url or not api_key:
compat_result.set(
{
"status": "fail",
"error": (
f"missing required env: set {PROXY_BASE_URL_ENV} and "
f"{PROXY_API_KEY_ENV} to point at a running LiteLLM proxy"
),
}
)
pytest.fail(
f"{PROXY_BASE_URL_ENV} / {PROXY_API_KEY_ENV} not configured", pytrace=False
)
outcomes = run_claude_models_parallel(
run_basic_messaging_cell(
compat_result=compat_result,
models=AZURE_MODELS,
prompt="Reply with the single word 'pong' and nothing else.",
base_url=base_url,
api_key=api_key,
)
failures = []
for model in AZURE_MODELS:
outcome = outcomes[model]
if isinstance(outcome, ClaudeCLIError):
error = f"[{model}] {outcome}"
compat_result.add({"status": "fail", "error": error})
failures.append(error)
continue
if outcome.exit_code != 0:
error = f"[{model}] claude CLI failed: {failure_diagnostic(outcome)}"
compat_result.add({"status": "fail", "error": error})
failures.append(error)
continue
if not outcome.text.strip():
error = f"[{model}] claude returned empty assistant text"
compat_result.add({"status": "fail", "error": error})
failures.append(error)
continue
compat_result.add({"status": "pass"})
if failures:
pytest.fail("; ".join(failures), pytrace=False)

68
tests/claude_code/basic_messaging_non_streaming/test_bedrock_converse.py
View file

@ -12,26 +12,15 @@ The (feature, provider) for this cell is inferred from the file path by
feature_id provider
Per the PRD, every cell exercises Claude Haiku 4.5, Sonnet 4.6, and Opus
4.7; the cell only goes green if all three pass. We fan the three model
runs out in parallel inside this single test and report one
`compat_result.add(...)` entry per model so the matrix builder still sees
three rows for this (feature, provider).
4.7; the cell only goes green if all three pass. The shared
`run_basic_messaging_cell` helper fans the three model runs out in
parallel and reports one `compat_result.add(...)` entry per model so
the matrix builder still sees three rows for this (feature, provider).
"""
from __future__ import annotations
import os
import pytest
from tests.claude_code.cli_driver import (
ClaudeCLIError,
failure_diagnostic,
run_claude_models_parallel,
)
PROXY_BASE_URL_ENV = "LITELLM_PROXY_BASE_URL"
PROXY_API_KEY_ENV = "LITELLM_PROXY_API_KEY"
from tests.claude_code._basic_messaging import run_basic_messaging_cell
# Per-model aliases registered in the LiteLLM proxy's routing config to
# point at Bedrock's Converse endpoint. The driver only sends the alias;
@ -46,51 +35,8 @@ BEDROCK_CONVERSE_MODELS = [
def test_basic_messaging_non_streaming_bedrock_converse(compat_result):
"""Drive the `claude` CLI against the LiteLLM proxy and assert a reply."""
base_url = os.environ.get(PROXY_BASE_URL_ENV)
api_key = os.environ.get(PROXY_API_KEY_ENV)
if not base_url or not api_key:
compat_result.set(
{
"status": "fail",
"error": (
f"missing required env: set {PROXY_BASE_URL_ENV} and "
f"{PROXY_API_KEY_ENV} to point at a running LiteLLM proxy"
),
}
)
pytest.fail(
f"{PROXY_BASE_URL_ENV} / {PROXY_API_KEY_ENV} not configured", pytrace=False
)
outcomes = run_claude_models_parallel(
run_basic_messaging_cell(
compat_result=compat_result,
models=BEDROCK_CONVERSE_MODELS,
prompt="Reply with the single word 'pong' and nothing else.",
base_url=base_url,
api_key=api_key,
)
failures = []
for model in BEDROCK_CONVERSE_MODELS:
outcome = outcomes[model]
if isinstance(outcome, ClaudeCLIError):
error = f"[{model}] {outcome}"
compat_result.add({"status": "fail", "error": error})
failures.append(error)
continue
if outcome.exit_code != 0:
error = f"[{model}] claude CLI failed: {failure_diagnostic(outcome)}"
compat_result.add({"status": "fail", "error": error})
failures.append(error)
continue
if not outcome.text.strip():
error = f"[{model}] claude returned empty assistant text"
compat_result.add({"status": "fail", "error": error})
failures.append(error)
continue
compat_result.add({"status": "pass"})
if failures:
pytest.fail("; ".join(failures), pytrace=False)

68
tests/claude_code/basic_messaging_non_streaming/test_bedrock_invoke.py
View file

@ -12,26 +12,15 @@ The (feature, provider) for this cell is inferred from the file path by
feature_id provider
Per the PRD, every cell exercises Claude Haiku 4.5, Sonnet 4.6, and Opus
4.7; the cell only goes green if all three pass. We fan the three model
runs out in parallel inside this single test and report one
`compat_result.add(...)` entry per model so the matrix builder still sees
three rows for this (feature, provider).
4.7; the cell only goes green if all three pass. The shared
`run_basic_messaging_cell` helper fans the three model runs out in
parallel and reports one `compat_result.add(...)` entry per model so
the matrix builder still sees three rows for this (feature, provider).
"""
from __future__ import annotations
import os
import pytest
from tests.claude_code.cli_driver import (
ClaudeCLIError,
failure_diagnostic,
run_claude_models_parallel,
)
PROXY_BASE_URL_ENV = "LITELLM_PROXY_BASE_URL"
PROXY_API_KEY_ENV = "LITELLM_PROXY_API_KEY"
from tests.claude_code._basic_messaging import run_basic_messaging_cell
# Per-model aliases registered in the LiteLLM proxy's routing config to
# point at Bedrock's legacy InvokeModel endpoint. The driver only sends
@ -46,51 +35,8 @@ BEDROCK_INVOKE_MODELS = [
def test_basic_messaging_non_streaming_bedrock_invoke(compat_result):
"""Drive the `claude` CLI against the LiteLLM proxy and assert a reply."""
base_url = os.environ.get(PROXY_BASE_URL_ENV)
api_key = os.environ.get(PROXY_API_KEY_ENV)
if not base_url or not api_key:
compat_result.set(
{
"status": "fail",
"error": (
f"missing required env: set {PROXY_BASE_URL_ENV} and "
f"{PROXY_API_KEY_ENV} to point at a running LiteLLM proxy"
),
}
)
pytest.fail(
f"{PROXY_BASE_URL_ENV} / {PROXY_API_KEY_ENV} not configured", pytrace=False
)
outcomes = run_claude_models_parallel(
run_basic_messaging_cell(
compat_result=compat_result,
models=BEDROCK_INVOKE_MODELS,
prompt="Reply with the single word 'pong' and nothing else.",
base_url=base_url,
api_key=api_key,
)
failures = []
for model in BEDROCK_INVOKE_MODELS:
outcome = outcomes[model]
if isinstance(outcome, ClaudeCLIError):
error = f"[{model}] {outcome}"
compat_result.add({"status": "fail", "error": error})
failures.append(error)
continue
if outcome.exit_code != 0:
error = f"[{model}] claude CLI failed: {failure_diagnostic(outcome)}"
compat_result.add({"status": "fail", "error": error})
failures.append(error)
continue
if not outcome.text.strip():
error = f"[{model}] claude returned empty assistant text"
compat_result.add({"status": "fail", "error": error})
failures.append(error)
continue
compat_result.add({"status": "pass"})
if failures:
pytest.fail("; ".join(failures), pytrace=False)

68
tests/claude_code/basic_messaging_non_streaming/test_vertex_ai.py
View file

@ -12,26 +12,15 @@ The (feature, provider) for this cell is inferred from the file path by
feature_id provider
Per the PRD, every cell exercises Claude Haiku 4.5, Sonnet 4.6, and Opus
4.7; the cell only goes green if all three pass. We fan the three model
runs out in parallel inside this single test and report one
`compat_result.add(...)` entry per model so the matrix builder still sees
three rows for this (feature, provider).
4.7; the cell only goes green if all three pass. The shared
`run_basic_messaging_cell` helper fans the three model runs out in
parallel and reports one `compat_result.add(...)` entry per model so
the matrix builder still sees three rows for this (feature, provider).
"""
from __future__ import annotations
import os
import pytest
from tests.claude_code.cli_driver import (
ClaudeCLIError,
failure_diagnostic,
run_claude_models_parallel,
)
PROXY_BASE_URL_ENV = "LITELLM_PROXY_BASE_URL"
PROXY_API_KEY_ENV = "LITELLM_PROXY_API_KEY"
from tests.claude_code._basic_messaging import run_basic_messaging_cell
# Per-model aliases registered in the LiteLLM proxy's routing config to
# point at Vertex AI's Anthropic model endpoints. The driver only sends
@ -46,51 +35,8 @@ VERTEX_AI_MODELS = [
def test_basic_messaging_non_streaming_vertex_ai(compat_result):
"""Drive the `claude` CLI against the LiteLLM proxy and assert a reply."""
base_url = os.environ.get(PROXY_BASE_URL_ENV)
api_key = os.environ.get(PROXY_API_KEY_ENV)
if not base_url or not api_key:
compat_result.set(
{
"status": "fail",
"error": (
f"missing required env: set {PROXY_BASE_URL_ENV} and "
f"{PROXY_API_KEY_ENV} to point at a running LiteLLM proxy"
),
}
)
pytest.fail(
f"{PROXY_BASE_URL_ENV} / {PROXY_API_KEY_ENV} not configured", pytrace=False
)
outcomes = run_claude_models_parallel(
run_basic_messaging_cell(
compat_result=compat_result,
models=VERTEX_AI_MODELS,
prompt="Reply with the single word 'pong' and nothing else.",
base_url=base_url,
api_key=api_key,
)
failures = []
for model in VERTEX_AI_MODELS:
outcome = outcomes[model]
if isinstance(outcome, ClaudeCLIError):
error = f"[{model}] {outcome}"
compat_result.add({"status": "fail", "error": error})
failures.append(error)
continue
if outcome.exit_code != 0:
error = f"[{model}] claude CLI failed: {failure_diagnostic(outcome)}"
compat_result.add({"status": "fail", "error": error})
failures.append(error)
continue
if not outcome.text.strip():
error = f"[{model}] claude returned empty assistant text"
compat_result.add({"status": "fail", "error": error})
failures.append(error)
continue
compat_result.add({"status": "pass"})
if failures:
pytest.fail("; ".join(failures), pytrace=False)

76
tests/claude_code/basic_messaging_streaming/test_anthropic.py
View file

@ -18,25 +18,17 @@ The (feature, provider) for this cell is inferred from the file path by
^^^^^^^^^^^^^^^^^^^^^^^^^ ^^^^^^^^^
feature_id provider
The three Claude tiers run in parallel inside this single test, with
one `compat_result.add(...)` entry per model so the matrix builder
still sees three rows for this (feature, provider).
The shared `run_basic_messaging_cell` helper fans the three Claude tiers
out in parallel inside this single test, with one
`compat_result.add(...)` entry per model so the matrix builder still
sees three rows for this (feature, provider). The `require_stream_events`
flag adds the streaming-only assertion that at least one stream-json
event was observed per model.
"""
from __future__ import annotations
import os
import pytest
from tests.claude_code.cli_driver import (
ClaudeCLIError,
failure_diagnostic,
run_claude_models_parallel,
)
PROXY_BASE_URL_ENV = "LITELLM_PROXY_BASE_URL"
PROXY_API_KEY_ENV = "LITELLM_PROXY_API_KEY"
from tests.claude_code._basic_messaging import run_basic_messaging_cell
ANTHROPIC_MODELS = [
"claude-haiku-4-5",
@ -49,57 +41,9 @@ def test_basic_messaging_streaming_anthropic(compat_result):
"""Drive the `claude` CLI against the LiteLLM proxy and assert a
non-empty streamed reply (at least one stream-json event observed).
"""
base_url = os.environ.get(PROXY_BASE_URL_ENV)
api_key = os.environ.get(PROXY_API_KEY_ENV)
if not base_url or not api_key:
compat_result.set(
{
"status": "fail",
"error": (
f"missing required env: set {PROXY_BASE_URL_ENV} and "
f"{PROXY_API_KEY_ENV} to point at a running LiteLLM proxy"
),
}
)
pytest.fail(
f"{PROXY_BASE_URL_ENV} / {PROXY_API_KEY_ENV} not configured", pytrace=False
)
outcomes = run_claude_models_parallel(
run_basic_messaging_cell(
compat_result=compat_result,
models=ANTHROPIC_MODELS,
prompt="Count from 1 to 5, one number per line.",
base_url=base_url,
api_key=api_key,
require_stream_events=True,
)
failures = []
for model in ANTHROPIC_MODELS:
outcome = outcomes[model]
if isinstance(outcome, ClaudeCLIError):
error = f"[{model}] {outcome}"
compat_result.add({"status": "fail", "error": error})
failures.append(error)
continue
if outcome.exit_code != 0:
error = f"[{model}] claude CLI failed: {failure_diagnostic(outcome)}"
compat_result.add({"status": "fail", "error": error})
failures.append(error)
continue
if not outcome.events:
error = f"[{model}] no stream-json events emitted; streaming wire silent"
compat_result.add({"status": "fail", "error": error})
failures.append(error)
continue
if not outcome.text.strip():
error = f"[{model}] claude returned empty assistant text"
compat_result.add({"status": "fail", "error": error})
failures.append(error)
continue
compat_result.add({"status": "pass"})
if failures:
pytest.fail("; ".join(failures), pytrace=False)

67
tests/claude_code/basic_messaging_streaming/test_azure.py
View file

@ -19,18 +19,7 @@ The (feature, provider) for this cell is inferred from the file path by
from __future__ import annotations
import os
import pytest
from tests.claude_code.cli_driver import (
ClaudeCLIError,
failure_diagnostic,
run_claude_models_parallel,
)
PROXY_BASE_URL_ENV = "LITELLM_PROXY_BASE_URL"
PROXY_API_KEY_ENV = "LITELLM_PROXY_API_KEY"
from tests.claude_code._basic_messaging import run_basic_messaging_cell
AZURE_MODELS = [
"claude-haiku-4-5-azure",
@ -43,57 +32,9 @@ def test_basic_messaging_streaming_azure(compat_result):
"""Drive the `claude` CLI against the LiteLLM proxy and assert a
non-empty streamed reply (at least one stream-json event observed).
"""
base_url = os.environ.get(PROXY_BASE_URL_ENV)
api_key = os.environ.get(PROXY_API_KEY_ENV)
if not base_url or not api_key:
compat_result.set(
{
"status": "fail",
"error": (
f"missing required env: set {PROXY_BASE_URL_ENV} and "
f"{PROXY_API_KEY_ENV} to point at a running LiteLLM proxy"
),
}
)
pytest.fail(
f"{PROXY_BASE_URL_ENV} / {PROXY_API_KEY_ENV} not configured", pytrace=False
)
outcomes = run_claude_models_parallel(
run_basic_messaging_cell(
compat_result=compat_result,
models=AZURE_MODELS,
prompt="Count from 1 to 5, one number per line.",
base_url=base_url,
api_key=api_key,
require_stream_events=True,
)
failures = []
for model in AZURE_MODELS:
outcome = outcomes[model]
if isinstance(outcome, ClaudeCLIError):
error = f"[{model}] {outcome}"
compat_result.add({"status": "fail", "error": error})
failures.append(error)
continue
if outcome.exit_code != 0:
error = f"[{model}] claude CLI failed: {failure_diagnostic(outcome)}"
compat_result.add({"status": "fail", "error": error})
failures.append(error)
continue
if not outcome.events:
error = f"[{model}] no stream-json events emitted; streaming wire silent"
compat_result.add({"status": "fail", "error": error})
failures.append(error)
continue
if not outcome.text.strip():
error = f"[{model}] claude returned empty assistant text"
compat_result.add({"status": "fail", "error": error})
failures.append(error)
continue
compat_result.add({"status": "pass"})
if failures:
pytest.fail("; ".join(failures), pytrace=False)

67
tests/claude_code/basic_messaging_streaming/test_bedrock_converse.py
View file

@ -15,18 +15,7 @@ The (feature, provider) for this cell is inferred from the file path by
from __future__ import annotations
import os
import pytest
from tests.claude_code.cli_driver import (
ClaudeCLIError,
failure_diagnostic,
run_claude_models_parallel,
)
PROXY_BASE_URL_ENV = "LITELLM_PROXY_BASE_URL"
PROXY_API_KEY_ENV = "LITELLM_PROXY_API_KEY"
from tests.claude_code._basic_messaging import run_basic_messaging_cell
BEDROCK_CONVERSE_MODELS = [
"claude-haiku-4-5-bedrock-converse",
@ -39,57 +28,9 @@ def test_basic_messaging_streaming_bedrock_converse(compat_result):
"""Drive the `claude` CLI against the LiteLLM proxy and assert a
non-empty streamed reply (at least one stream-json event observed).
"""
base_url = os.environ.get(PROXY_BASE_URL_ENV)
api_key = os.environ.get(PROXY_API_KEY_ENV)
if not base_url or not api_key:
compat_result.set(
{
"status": "fail",
"error": (
f"missing required env: set {PROXY_BASE_URL_ENV} and "
f"{PROXY_API_KEY_ENV} to point at a running LiteLLM proxy"
),
}
)
pytest.fail(
f"{PROXY_BASE_URL_ENV} / {PROXY_API_KEY_ENV} not configured", pytrace=False
)
outcomes = run_claude_models_parallel(
run_basic_messaging_cell(
compat_result=compat_result,
models=BEDROCK_CONVERSE_MODELS,
prompt="Count from 1 to 5, one number per line.",
base_url=base_url,
api_key=api_key,
require_stream_events=True,
)
failures = []
for model in BEDROCK_CONVERSE_MODELS:
outcome = outcomes[model]
if isinstance(outcome, ClaudeCLIError):
error = f"[{model}] {outcome}"
compat_result.add({"status": "fail", "error": error})
failures.append(error)
continue
if outcome.exit_code != 0:
error = f"[{model}] claude CLI failed: {failure_diagnostic(outcome)}"
compat_result.add({"status": "fail", "error": error})
failures.append(error)
continue
if not outcome.events:
error = f"[{model}] no stream-json events emitted; streaming wire silent"
compat_result.add({"status": "fail", "error": error})
failures.append(error)
continue
if not outcome.text.strip():
error = f"[{model}] claude returned empty assistant text"
compat_result.add({"status": "fail", "error": error})
failures.append(error)
continue
compat_result.add({"status": "pass"})
if failures:
pytest.fail("; ".join(failures), pytrace=False)

67
tests/claude_code/basic_messaging_streaming/test_bedrock_invoke.py
View file

@ -15,18 +15,7 @@ The (feature, provider) for this cell is inferred from the file path by
from __future__ import annotations
import os
import pytest
from tests.claude_code.cli_driver import (
ClaudeCLIError,
failure_diagnostic,
run_claude_models_parallel,
)
PROXY_BASE_URL_ENV = "LITELLM_PROXY_BASE_URL"
PROXY_API_KEY_ENV = "LITELLM_PROXY_API_KEY"
from tests.claude_code._basic_messaging import run_basic_messaging_cell
BEDROCK_INVOKE_MODELS = [
"claude-haiku-4-5-bedrock-invoke",
@ -39,57 +28,9 @@ def test_basic_messaging_streaming_bedrock_invoke(compat_result):
"""Drive the `claude` CLI against the LiteLLM proxy and assert a
non-empty streamed reply (at least one stream-json event observed).
"""
base_url = os.environ.get(PROXY_BASE_URL_ENV)
api_key = os.environ.get(PROXY_API_KEY_ENV)
if not base_url or not api_key:
compat_result.set(
{
"status": "fail",
"error": (
f"missing required env: set {PROXY_BASE_URL_ENV} and "
f"{PROXY_API_KEY_ENV} to point at a running LiteLLM proxy"
),
}
)
pytest.fail(
f"{PROXY_BASE_URL_ENV} / {PROXY_API_KEY_ENV} not configured", pytrace=False
)
outcomes = run_claude_models_parallel(
run_basic_messaging_cell(
compat_result=compat_result,
models=BEDROCK_INVOKE_MODELS,
prompt="Count from 1 to 5, one number per line.",
base_url=base_url,
api_key=api_key,
require_stream_events=True,
)
failures = []
for model in BEDROCK_INVOKE_MODELS:
outcome = outcomes[model]
if isinstance(outcome, ClaudeCLIError):
error = f"[{model}] {outcome}"
compat_result.add({"status": "fail", "error": error})
failures.append(error)
continue
if outcome.exit_code != 0:
error = f"[{model}] claude CLI failed: {failure_diagnostic(outcome)}"
compat_result.add({"status": "fail", "error": error})
failures.append(error)
continue
if not outcome.events:
error = f"[{model}] no stream-json events emitted; streaming wire silent"
compat_result.add({"status": "fail", "error": error})
failures.append(error)
continue
if not outcome.text.strip():
error = f"[{model}] claude returned empty assistant text"
compat_result.add({"status": "fail", "error": error})
failures.append(error)
continue
compat_result.add({"status": "pass"})
if failures:
pytest.fail("; ".join(failures), pytrace=False)

67
tests/claude_code/basic_messaging_streaming/test_vertex_ai.py
View file

@ -15,18 +15,7 @@ The (feature, provider) for this cell is inferred from the file path by
from __future__ import annotations
import os
import pytest
from tests.claude_code.cli_driver import (
ClaudeCLIError,
failure_diagnostic,
run_claude_models_parallel,
)
PROXY_BASE_URL_ENV = "LITELLM_PROXY_BASE_URL"
PROXY_API_KEY_ENV = "LITELLM_PROXY_API_KEY"
from tests.claude_code._basic_messaging import run_basic_messaging_cell
VERTEX_AI_MODELS = [
"claude-haiku-4-5-vertex",
@ -39,57 +28,9 @@ def test_basic_messaging_streaming_vertex_ai(compat_result):
"""Drive the `claude` CLI against the LiteLLM proxy and assert a
non-empty streamed reply (at least one stream-json event observed).
"""
base_url = os.environ.get(PROXY_BASE_URL_ENV)
api_key = os.environ.get(PROXY_API_KEY_ENV)
if not base_url or not api_key:
compat_result.set(
{
"status": "fail",
"error": (
f"missing required env: set {PROXY_BASE_URL_ENV} and "
f"{PROXY_API_KEY_ENV} to point at a running LiteLLM proxy"
),
}
)
pytest.fail(
f"{PROXY_BASE_URL_ENV} / {PROXY_API_KEY_ENV} not configured", pytrace=False
)
outcomes = run_claude_models_parallel(
run_basic_messaging_cell(
compat_result=compat_result,
models=VERTEX_AI_MODELS,
prompt="Count from 1 to 5, one number per line.",
base_url=base_url,
api_key=api_key,
require_stream_events=True,
)
failures = []
for model in VERTEX_AI_MODELS:
outcome = outcomes[model]
if isinstance(outcome, ClaudeCLIError):
error = f"[{model}] {outcome}"
compat_result.add({"status": "fail", "error": error})
failures.append(error)
continue
if outcome.exit_code != 0:
error = f"[{model}] claude CLI failed: {failure_diagnostic(outcome)}"
compat_result.add({"status": "fail", "error": error})
failures.append(error)
continue
if not outcome.events:
error = f"[{model}] no stream-json events emitted; streaming wire silent"
compat_result.add({"status": "fail", "error": error})
failures.append(error)
continue
if not outcome.text.strip():
error = f"[{model}] claude returned empty assistant text"
compat_result.add({"status": "fail", "error": error})
failures.append(error)
continue
compat_result.add({"status": "pass"})
if failures:
pytest.fail("; ".join(failures), pytrace=False)

14
tests/claude_code/cli_driver.py
View file

@ -279,6 +279,20 @@ def run_claude_models_parallel(
except ClaudeCLIError as exc:
elapsed = time.monotonic() - started
return model, exc, elapsed
except Exception as exc:
# Honor the documented "errors as values" contract for any
# exception type — not just ClaudeCLIError. The rate
# limiter does file I/O (OSError), `infer_provider` can
# raise ValueError on edge-case model strings, and a future
# bug elsewhere in the call stack must not abort the entire
# parallel batch and lose the other models' outcomes.
elapsed = time.monotonic() - started
wrapped = ClaudeCLIError(
f"unexpected error running model {model!r}: "
f"{type(exc).__name__}: {exc}"
)
wrapped.__cause__ = exc
return model, wrapped, elapsed
outcomes: Dict[str, ModelResult] = {}
durations: Dict[str, float] = {}