scriptoriumadmin/litellm
1
0
Fork 0
mirror of https://github.com/BerriAI/litellm.git synced 2026-10-26 06:03:19 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 35
litellm/tests/unit/proxy/client/cli/test_agents.py
Yassin Kortam 1525e4bce5
feat(cli): keep the claude.ai login with lite claude --subscription and save pkce in config (#47385) 
`lite claude --subscription` sends the LiteLLM key in the x-litellm-api-key header through
ANTHROPIC_CUSTOM_HEADERS and leaves ANTHROPIC_AUTH_TOKEN and ANTHROPIC_API_KEY unset, so Claude Code
stays on its claude.ai login and its connectors keep working. The pre-launch key check sends the key
the same way.

`claude_subscription` and `pkce` are now `lite config` keys that default the new
`--subscription/--no-subscription` flag on `lite claude` and the `--pkce/--no-pkce` flag on
`lite login`. The flags still win for a single run.

Co-authored-by: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>
2026-10-19 15:33:29 -07:00

2210 lines
88 KiB
Python
Raw Blame History

import inspect
import json
import os
import stat
import subprocess
import sys
from pathlib import Path
from unittest.mock import patch
import click
import pytest
import requests
from click.testing import CliRunner
from litellm.proxy.client.cli.commands.agents import (
AgentRunError,
ModelSyncArgs,
ModelSyncSkipped,
_hand_off,
_replace_process,
_spawn_and_wait,
agent_commands,
agent_launch_args,
agent_model_sync_env,
agent_profile,
build_agent_env,
codex_model_sync_args,
opencode_model_sync_env,
run_agent,
verify_proxy_key,
verify_subscription_key,
)
from litellm.proxy.client.cli.commands.claude_settings import ClaudeSettingsError
from litellm.proxy.client.cli.main import cli as lite
AGENTS_MODULE = "litellm.proxy.client.cli.commands.agents"
CLAUDE_SETTINGS_MODULE = "litellm.proxy.client.cli.commands.claude_settings"
def _agent_command(name):
return next(c for c in agent_commands() if c.name == name)
def _default_of(func, param):
return inspect.signature(func).parameters[param].default
class _FakeResponse:
def __init__(self, status_code, body=None):
self.status_code = status_code
self.content = json.dumps(body).encode() if body is not None else b""
class _Recorder:
def __init__(self, returns=None):
self.returns = returns
self.calls = []
def __call__(self, *args):
self.calls.append(args)
return self.returns
_STOCK_REASONING_LEVELS = [
{"effort": "low", "description": "Fast responses with lighter reasoning"},
{"effort": "medium", "description": "Balances speed and reasoning depth for everyday tasks"},
{"effort": "high", "description": "Greater reasoning depth for complex problems"},
]
_STOCK_MODELS = {
"gpt-5.6-terra": {
"slug": "gpt-5.6-terra",
"display_name": "GPT-5.6 Terra",
"description": "Balanced agentic coding model for everyday work.",
"default_reasoning_level": "medium",
"supported_reasoning_levels": _STOCK_REASONING_LEVELS,
"shell_type": "unified_exec",
"visibility": "list",
"supported_in_api": True,
"priority": 7,
"availability_nux": None,
"upgrade": None,
"base_instructions": "You are Codex, a coding agent based on GPT-5.6.",
"apply_patch_tool_type": "freeform",
"supports_parallel_tool_calls": True,
"context_window": 272000,
"comp_hash": "terra-hash",
},
"gpt-5.5": {
"slug": "gpt-5.5",
"display_name": "GPT-5.5",
"description": "Frontier model for complex coding, research, and real-world work.",
"default_reasoning_level": "medium",
"supported_reasoning_levels": _STOCK_REASONING_LEVELS,
"shell_type": "unified_exec",
"visibility": "list",
"supported_in_api": True,
"priority": 12,
"availability_nux": None,
"upgrade": None,
"base_instructions": "You are Codex, a coding agent based on GPT-5.",
"apply_patch_tool_type": "freeform",
"supports_parallel_tool_calls": True,
"context_window": 272000,
"comp_hash": "gpt-5.5-hash",
},
"gpt-5.4": {
"slug": "gpt-5.4",
"display_name": "GPT-5.4",
"description": "Strong model for everyday coding.",
"default_reasoning_level": "medium",
"supported_reasoning_levels": _STOCK_REASONING_LEVELS,
"shell_type": "unified_exec",
"visibility": "hide",
"supported_in_api": True,
"priority": 16,
"availability_nux": None,
"upgrade": {
"model": "gpt-5.6-terra",
"migration_markdown": "GPT-5.4 is no longer available. Switch to GPT-5.6 Terra to continue.",
"retirement_at": "2026-08-31T19:00:00Z",
},
"base_instructions": "You are Codex, a coding agent based on GPT-5.",
"apply_patch_tool_type": "freeform",
"supports_parallel_tool_calls": True,
"context_window": 272000,
"comp_hash": "gpt-5.4-hash",
},
"codex-auto-review": {
"slug": "codex-auto-review",
"display_name": "Codex Auto Review",
"description": None,
"supported_reasoning_levels": [],
"shell_type": "unified_exec",
"visibility": "hide",
"supported_in_api": False,
"priority": 43,
"availability_nux": None,
"upgrade": None,
"base_instructions": "You are Codex, reviewing a change.",
"apply_patch_tool_type": None,
"supports_parallel_tool_calls": True,
"context_window": 272000,
"comp_hash": "review-hash",
},
}
_STOCK_CATALOG = json.dumps({"models": list(_STOCK_MODELS.values())})
class _FakeRun:
"""A `codex` that prints `stock` from a bare `debug models` and answers a catalog override with `returncode`.
`stock=None` is a Codex with no `debug models` at all: every call answers with `returncode` and `stderr`.
"""
def __init__(self, returncode=0, stderr="", stock=_STOCK_CATALOG):
self.returncode = returncode
self.stderr = stderr
self.stock = stock
self.calls = []
def __call__(self, args, **kwargs):
self.calls.append((args, kwargs))
if self.stock is not None and "model_catalog_json=" not in str(args):
return subprocess.CompletedProcess(args, 0, self.stock, "")
return subprocess.CompletedProcess(args, self.returncode, "", self.stderr)
class _FakeJsonResponse:
def __init__(self, status_code, payload=None):
self.status_code = status_code
self._payload = payload
def json(self):
return self._payload
def _overlay_writer(tmp_path):
def write(text, environ):
target = tmp_path / "overlay.json"
target.write_text(text)
return target
return write
def _overlay(args):
return json.loads(Path(args[2]).read_text())
class TestAgentProfile:
def test_claude_is_anthropic(self):
name, profiles = agent_profile("claude")
assert name == "Claude Code"
assert profiles == frozenset({"anthropic"})
def test_claude_full_path_uses_basename(self):
name, profiles = agent_profile("/usr/local/bin/claude")
assert name == "Claude Code"
assert profiles == frozenset({"anthropic"})
def test_codex_and_opencode_are_openai(self):
assert agent_profile("codex") == ("Codex", frozenset({"openai"}))
assert agent_profile("opencode") == ("OpenCode", frozenset({"openai"}))
def test_pi_is_litellm(self):
assert agent_profile("pi") == ("pi", frozenset({"litellm"}))
def test_unknown_command_gets_both_profiles(self):
name, profiles = agent_profile("mytool")
assert name == "mytool"
assert profiles == frozenset({"anthropic", "openai"})
class TestBuildAgentEnv:
def test_anthropic_profile_uses_bare_root_and_bearer(self):
env = build_agent_env({}, "http://localhost:4000/", "sk-key", frozenset({"anthropic"}))
assert env["ANTHROPIC_BASE_URL"] == "http://localhost:4000"
assert env["ANTHROPIC_AUTH_TOKEN"] == "sk-key"
assert env["ENABLE_TOOL_SEARCH"] == "true"
assert env["CLAUDE_CODE_ENABLE_GATEWAY_MODEL_DISCOVERY"] == "1"
assert "OPENAI_BASE_URL" not in env
assert "OPENAI_API_KEY" not in env
def test_anthropic_profile_preserves_existing_gateway_model_discovery(self):
env = build_agent_env(
{"CLAUDE_CODE_ENABLE_GATEWAY_MODEL_DISCOVERY": "0"},
"http://localhost:4000",
"sk-key",
frozenset({"anthropic"}),
)
assert env["CLAUDE_CODE_ENABLE_GATEWAY_MODEL_DISCOVERY"] == "0"
def test_anthropic_profile_preserves_existing_tool_search(self):
env = build_agent_env(
{"ENABLE_TOOL_SEARCH": "false"},
"http://localhost:4000",
"sk-key",
frozenset({"anthropic"}),
)
assert env["ENABLE_TOOL_SEARCH"] == "false"
def test_anthropic_profile_drops_existing_api_key(self):
env = build_agent_env(
{"ANTHROPIC_API_KEY": "real-key"},
"http://localhost:4000",
"sk-key",
frozenset({"anthropic"}),
)
assert "ANTHROPIC_API_KEY" not in env
def test_openai_profile_appends_v1(self):
env = build_agent_env({}, "http://localhost:4000/", "sk-key", frozenset({"openai"}))
assert env["OPENAI_BASE_URL"] == "http://localhost:4000/v1"
assert env["OPENAI_API_KEY"] == "sk-key"
assert "ANTHROPIC_BASE_URL" not in env
assert "ENABLE_TOOL_SEARCH" not in env
assert "CLAUDE_CODE_ENABLE_GATEWAY_MODEL_DISCOVERY" not in env
def test_both_profiles_set_everything(self):
env = build_agent_env({}, "http://localhost:4000", "sk-key", frozenset({"anthropic", "openai"}))
assert env["ANTHROPIC_BASE_URL"] == "http://localhost:4000"
assert env["OPENAI_BASE_URL"] == "http://localhost:4000/v1"
assert env["ANTHROPIC_AUTH_TOKEN"] == "sk-key"
assert env["OPENAI_API_KEY"] == "sk-key"
assert env["ENABLE_TOOL_SEARCH"] == "true"
def test_litellm_profile_exports_only_the_proxy_key(self):
env = build_agent_env({}, "http://localhost:4000/", "sk-key", frozenset({"litellm"}))
assert env["LITELLM_PROXY_API_KEY"] == "sk-key"
assert "ANTHROPIC_BASE_URL" not in env
assert "OPENAI_BASE_URL" not in env
assert "OPENAI_API_KEY" not in env
def test_preserves_unrelated_env_and_does_not_mutate_input(self):
base = {"PATH": "/usr/bin", "ANTHROPIC_API_KEY": "real-key"}
env = build_agent_env(base, "http://localhost:4000", "sk-key", frozenset({"anthropic"}))
assert env["PATH"] == "/usr/bin"
assert base == {"PATH": "/usr/bin", "ANTHROPIC_API_KEY": "real-key"}
def test_subscription_sends_the_key_in_a_header_and_sets_no_credential(self):
env = build_agent_env({}, "http://localhost:4000/", "sk-key", frozenset({"anthropic"}), subscription=True)
assert env["ANTHROPIC_BASE_URL"] == "http://localhost:4000"
assert env["ANTHROPIC_CUSTOM_HEADERS"] == "x-litellm-api-key: Bearer sk-key"
assert "ANTHROPIC_AUTH_TOKEN" not in env
assert "ANTHROPIC_API_KEY" not in env
def test_subscription_clears_inherited_credentials(self):
base = {"ANTHROPIC_AUTH_TOKEN": "stale-token", "ANTHROPIC_API_KEY": "real-key"}
env = build_agent_env(base, "http://localhost:4000", "sk-key", frozenset({"anthropic"}), subscription=True)
assert "ANTHROPIC_AUTH_TOKEN" not in env
assert "ANTHROPIC_API_KEY" not in env
assert base == {"ANTHROPIC_AUTH_TOKEN": "stale-token", "ANTHROPIC_API_KEY": "real-key"}
def test_subscription_keeps_the_users_other_custom_headers(self):
env = build_agent_env(
{"ANTHROPIC_CUSTOM_HEADERS": "x-team: blue\n\nx-trace: 1\n"},
"http://localhost:4000",
"sk-key",
frozenset({"anthropic"}),
subscription=True,
)
assert env["ANTHROPIC_CUSTOM_HEADERS"].splitlines() == [
"x-team: blue",
"x-trace: 1",
"x-litellm-api-key: Bearer sk-key",
]
def test_subscription_replaces_a_stale_litellm_key_header_whatever_its_case(self):
env = build_agent_env(
{"ANTHROPIC_CUSTOM_HEADERS": "X-LiteLLM-Api-Key: Bearer old\nx-team: blue"},
"http://localhost:4000",
"sk-new",
frozenset({"anthropic"}),
subscription=True,
)
assert env["ANTHROPIC_CUSTOM_HEADERS"].splitlines() == ["x-team: blue", "x-litellm-api-key: Bearer sk-new"]
def test_subscription_leaves_openai_agents_on_the_bearer_key(self):
env = build_agent_env({}, "http://localhost:4000", "sk-key", frozenset({"openai"}), subscription=True)
assert env["OPENAI_API_KEY"] == "sk-key"
assert "ANTHROPIC_CUSTOM_HEADERS" not in env
def test_default_mode_does_not_touch_custom_headers(self):
env = build_agent_env(
{"ANTHROPIC_CUSTOM_HEADERS": "x-team: blue"}, "http://localhost:4000", "sk-key", frozenset({"anthropic"})
)
assert env["ANTHROPIC_CUSTOM_HEADERS"] == "x-team: blue"
class TestAgentLaunchArgs:
def test_claude_and_opencode_get_no_extra_args(self):
assert agent_launch_args("claude", "http://localhost:4000") == []
assert agent_launch_args("opencode", "http://localhost:4000") == []
def test_unknown_agent_gets_no_extra_args(self):
assert agent_launch_args("mytool", "http://localhost:4000") == []
def test_codex_points_provider_at_proxy_over_http(self):
args = agent_launch_args("codex", "http://localhost:4000/")
joined = " ".join(args)
assert 'model_provider="litellm"' in args
assert 'model_providers.litellm.base_url="http://localhost:4000/v1"' in args
assert 'model_providers.litellm.env_key="OPENAI_API_KEY"' in args
assert 'model_providers.litellm.wire_api="responses"' in args
assert "model_providers.litellm.supports_websockets=false" in args
assert "model_providers.litellm.requires_openai_auth=false" in args
assert "model_providers.litellm.http_headers={}" in args
assert joined.count("-c") == 8
def test_codex_uses_basename(self):
assert agent_launch_args("/usr/local/bin/codex", "http://localhost:4000") == (
agent_launch_args("codex", "http://localhost:4000")
)
def test_pi_gets_no_static_args(self):
assert agent_launch_args("pi", "http://localhost:4000") == []
class TestVerifyProxyKey:
def test_ok_status_passes_and_uses_models_endpoint(self):
captured = {}
def fake_get(url, headers, timeout):
captured["url"] = url
captured["headers"] = headers
return _FakeResponse(200)
verify_proxy_key("http://localhost:4000/", "sk-key", get=fake_get)
assert captured["url"] == "http://localhost:4000/v1/models"
assert captured["headers"] == {"Authorization": "Bearer sk-key"}
@pytest.mark.parametrize("status", [401, 403])
def test_rejected_key_raises(self, status):
with pytest.raises(AgentRunError, match="rejected your key"):
verify_proxy_key(
"http://localhost:4000",
"sk-key",
get=lambda *a, **k: _FakeResponse(status),
)
def test_unreachable_proxy_raises(self):
def boom(*a, **k):
raise requests.ConnectionError("refused")
with pytest.raises(AgentRunError, match="Could not reach"):
verify_proxy_key("http://localhost:4000", "sk-key", get=boom)
def test_other_non_2xx_is_tolerated(self):
verify_proxy_key(
"http://localhost:4000",
"sk-key",
get=lambda *a, **k: _FakeResponse(500),
)
def test_key_header_moves_the_bearer_off_authorization(self):
captured = {}
def fake_get(url, headers, timeout):
captured["headers"] = headers
return _FakeResponse(200)
verify_proxy_key("http://localhost:4000", "sk-key", get=fake_get, key_header="x-litellm-api-key")
assert captured["headers"] == {"x-litellm-api-key": "Bearer sk-key"}
def test_subscription_probe_sends_the_key_only_in_the_litellm_header(self):
captured = {}
def fake_get(url, headers, timeout):
captured["headers"] = headers
return _FakeResponse(200)
verify_subscription_key("http://localhost:4000", "sk-key", get=fake_get)
assert captured["headers"] == {"x-litellm-api-key": "Bearer sk-key"}
def test_subscription_probe_reports_a_rejected_key(self):
with pytest.raises(AgentRunError, match="rejected your key"):
verify_subscription_key("http://localhost:4000", "sk-key", get=lambda *a, **k: _FakeResponse(401))
class TestOpencodeModelSync:
@staticmethod
def _listing(*models):
return {"object": "list", "data": list(models)}
def _sync(self, listing, base_env=None, base_url="http://localhost:4000/"):
captured = {}
def fake_get(url, headers, timeout):
captured["url"] = url
captured["headers"] = headers
return _FakeResponse(200, listing)
env = opencode_model_sync_env(base_env or {}, base_url, "sk-key", get=fake_get)
return captured, env
def test_declares_proxy_as_litellm_provider_with_listed_models(self):
listing = self._listing(
{"id": "gpt-5.5", "object": "model", "created": 1, "owned_by": "openai", "mode": "chat"},
{"id": "claude-opus-4-7", "object": "model", "created": 1, "owned_by": "openai"},
)
captured, env = self._sync(listing)
assert captured["url"] == "http://localhost:4000/v1/models"
assert captured["headers"] == {"Authorization": "Bearer sk-key"}
config = json.loads(env["OPENCODE_CONFIG_CONTENT"])
provider = config["provider"]["litellm"]
assert provider["npm"] == "@ai-sdk/openai-compatible"
assert provider["name"] == "LiteLLM"
assert provider["options"] == {
"baseURL": "http://localhost:4000/v1",
"apiKey": "{env:OPENAI_API_KEY}",
}
assert provider["models"] == {
"gpt-5.5": {"name": "gpt-5.5"},
"claude-opus-4-7": {"name": "claude-opus-4-7"},
}
assert "sk-key" not in env["OPENCODE_CONFIG_CONTENT"]
def test_token_limits_become_opencode_limits(self):
listing = self._listing(
{
"id": "gpt-5.5",
"object": "model",
"created": 1,
"owned_by": "openai",
"max_input_tokens": 400000,
"max_output_tokens": 128000,
},
{"id": "half", "object": "model", "created": 1, "owned_by": "openai", "max_input_tokens": 8192},
)
_, env = self._sync(listing)
models = json.loads(env["OPENCODE_CONFIG_CONTENT"])["provider"]["litellm"]["models"]
assert models["gpt-5.5"]["limit"] == {"context": 400000, "output": 128000}
assert "limit" not in models["half"]
def test_non_chat_models_are_left_out(self):
listing = self._listing(
{"id": "chat", "object": "model", "created": 1, "owned_by": "openai", "mode": "chat"},
{"id": "resp", "object": "model", "created": 1, "owned_by": "openai", "mode": "responses"},
{"id": "embed", "object": "model", "created": 1, "owned_by": "openai", "mode": "embedding"},
{"id": "img", "object": "model", "created": 1, "owned_by": "openai", "mode": "image_generation"},
)
_, env = self._sync(listing)
models = json.loads(env["OPENCODE_CONFIG_CONTENT"])["provider"]["litellm"]["models"]
assert set(models) == {"chat", "resp"}
def test_existing_config_content_is_left_alone(self):
calls = []
def fake_get(*a, **k):
calls.append(a)
return _FakeResponse(200, self._listing())
result = opencode_model_sync_env(
{"OPENCODE_CONFIG_CONTENT": "{}"}, "http://localhost:4000", "sk-key", get=fake_get
)
assert isinstance(result, ModelSyncSkipped)
assert "OPENCODE_CONFIG_CONTENT" in result.reason
assert calls == []
def test_unreachable_proxy_is_reported_not_raised(self):
def boom(*a, **k):
raise requests.ConnectionError("refused")
result = opencode_model_sync_env({}, "http://localhost:4000", "sk-key", get=boom)
assert isinstance(result, ModelSyncSkipped)
assert "refused" in result.reason
def test_non_200_is_reported(self):
result = opencode_model_sync_env({}, "http://localhost:4000", "sk-key", get=lambda *a, **k: _FakeResponse(500))
assert isinstance(result, ModelSyncSkipped)
assert "HTTP 500" in result.reason
def test_unexpected_body_is_reported(self):
result = opencode_model_sync_env(
{}, "http://localhost:4000", "sk-key", get=lambda *a, **k: _FakeResponse(200, {"data": "nope"})
)
assert isinstance(result, ModelSyncSkipped)
assert "unexpected body" in result.reason
@pytest.mark.parametrize("command", ["claude", "pi", "/usr/bin/claude"])
def test_only_opencode_and_codex_sync(self, command):
def boom(*a, **k):
raise AssertionError("no agent other than opencode or codex should call the proxy")
assert agent_model_sync_env(command, {}, "http://localhost:4000", "sk-key", False, get=boom) == {}
def test_skip_verify_keeps_the_launch_offline(self):
def boom(*a, **k):
raise AssertionError("--skip-verify must not touch the proxy")
result = agent_model_sync_env("opencode", {}, "http://localhost:4000", "sk-key", True, get=boom)
assert isinstance(result, ModelSyncSkipped)
assert "--skip-verify" in result.reason
def test_full_path_opencode_syncs(self):
listing = self._listing({"id": "m", "object": "model", "created": 1, "owned_by": "x"})
env = agent_model_sync_env(
"/opt/bin/opencode",
{},
"http://localhost:4000",
"sk-key",
False,
get=lambda *a, **k: _FakeResponse(200, listing),
)
assert "m" in json.loads(env["OPENCODE_CONFIG_CONTENT"])["provider"]["litellm"]["models"]
def test_default_http_client_is_requests_get(self):
assert _default_of(agent_model_sync_env, "get") is requests.get
assert _default_of(opencode_model_sync_env, "get") is requests.get
class TestCodexModelSync:
@staticmethod
def _listing(*models):
return {"object": "list", "data": list(models)}
@staticmethod
def _row(model_id, **extra):
return {"id": model_id, "object": "model", "created": 1, "owned_by": "openai", **extra}
def _sync(self, listing, codex_home, base_url="http://localhost:4000/", run=None):
captured = {}
def fake_get(url, headers, timeout):
captured["url"] = url
captured["headers"] = headers
return _FakeResponse(200, listing)
result = codex_model_sync_args(
{"CODEX_HOME": str(codex_home)},
base_url,
"sk-key",
get=fake_get,
run=_FakeRun() if run is None else run,
)
return captured, result
@staticmethod
def _catalog_path(result):
assert isinstance(result, ModelSyncArgs)
flag, override = result.args
assert flag == "-c"
key, _, value = override.partition("=")
assert key == "model_catalog_json"
return json.loads(value)
def test_writes_catalog_under_codex_home_and_points_codex_at_it(self, tmp_path):
listing = self._listing(self._row("gpt-5.5", mode="chat"), self._row("claude-opus-4-7"))
captured, result = self._sync(listing, tmp_path / "codex")
assert captured["url"] == "http://localhost:4000/v1/models"
assert captured["headers"] == {"Authorization": "Bearer sk-key"}
path = self._catalog_path(result)
assert path == str(tmp_path / "codex" / "litellm-models.json")
text = (tmp_path / "codex" / "litellm-models.json").read_text()
assert "sk-key" not in text
catalog = json.loads(text)
assert [m["slug"] for m in catalog["models"]] == ["gpt-5.5", "claude-opus-4-7"]
assert [m["display_name"] for m in catalog["models"]] == ["GPT-5.5", "claude-opus-4-7"]
assert [m["priority"] for m in catalog["models"]] == [0, 1]
def _entries(self, codex_home):
return {m["slug"]: m for m in json.loads((codex_home / "litellm-models.json").read_text())["models"]}
def test_known_model_keeps_the_installed_codex_entry(self, tmp_path):
self._sync(self._listing(self._row("gpt-5.5", mode="chat")), tmp_path)
assert self._entries(tmp_path)["gpt-5.5"] == {**_STOCK_MODELS["gpt-5.5"], "priority": 0}
def test_hidden_stock_model_is_listed_when_the_proxy_serves_it(self, tmp_path):
self._sync(self._listing(self._row("gpt-5.4")), tmp_path)
entry = self._entries(tmp_path)["gpt-5.4"]
assert entry["visibility"] == "list"
assert entry["upgrade"] is None
assert entry["supported_reasoning_levels"] == _STOCK_REASONING_LEVELS
def test_api_disabled_stock_model_is_selectable_when_the_proxy_serves_it(self, tmp_path):
self._sync(self._listing(self._row("codex-auto-review")), tmp_path)
entry = self._entries(tmp_path)["codex-auto-review"]
assert entry["supported_in_api"] is True
assert entry["visibility"] == "list"
assert entry["base_instructions"] == _STOCK_MODELS["codex-auto-review"]["base_instructions"]
def test_stock_upgrade_nudge_survives_when_its_target_is_listed(self, tmp_path):
self._sync(self._listing(self._row("gpt-5.4"), self._row("gpt-5.6-terra")), tmp_path)
entries = self._entries(tmp_path)
assert entries["gpt-5.4"]["upgrade"] == _STOCK_MODELS["gpt-5.4"]["upgrade"]
assert [entries["gpt-5.4"]["priority"], entries["gpt-5.6-terra"]["priority"]] == [0, 1]
def test_stock_catalog_is_decoded_as_utf8_regardless_of_locale(self, tmp_path):
description = "Modelo equilibrado para el trabajo diario, con acentos y ñ."
catalog = {"models": [{**_STOCK_MODELS["gpt-5.5"], "description": description}]}
stock = json.dumps(catalog, ensure_ascii=False).encode("utf-8")
def locale_bound_run(args, **kwargs):
if "model_catalog_json=" in str(args):
return subprocess.CompletedProcess(args, 0, "", "")
return subprocess.CompletedProcess(args, 0, stock.decode(kwargs.get("encoding") or "ascii"), "")
_, result = self._sync(self._listing(self._row("gpt-5.5")), tmp_path, run=locale_bound_run)
assert isinstance(result, ModelSyncArgs)
written = json.loads((tmp_path / "litellm-models.json").read_text(encoding="utf-8"))["models"]
assert [m["description"] for m in written] == [description]
def test_unparseable_stock_catalog_is_reported(self, tmp_path):
_, result = self._sync(self._listing(self._row("m")), tmp_path, run=_FakeRun(stock="not json"))
assert isinstance(result, ModelSyncSkipped)
assert result.reason.startswith("`codex debug models` printed no model catalog: ")
assert not (tmp_path / "litellm-models.json").exists()
def test_unknown_model_gets_the_fields_codex_requires(self, tmp_path):
_, result = self._sync(self._listing(self._row("m")), tmp_path)
entry = json.loads((tmp_path / "litellm-models.json").read_text())["models"][0]
assert entry["visibility"] == "list"
assert entry["supported_in_api"] is True
assert entry["shell_type"] == "unified_exec"
assert entry["supported_reasoning_levels"] == []
assert entry["truncation_policy"] == {"mode": "bytes", "limit": 10000}
assert entry["experimental_supported_tools"] == []
assert entry["support_verbosity"] is False
assert entry["supports_reasoning_summaries"] is False
assert entry["supports_parallel_tool_calls"] is False
for nullable in ("description", "availability_nux", "upgrade", "default_verbosity", "apply_patch_tool_type"):
assert nullable in entry and entry[nullable] is None
assert entry["base_instructions"].startswith("You are a coding agent running in the Codex CLI")
def test_context_window_comes_from_max_input_tokens_for_unknown_models_only(self, tmp_path):
listing = self._listing(
self._row("big", max_input_tokens=400000),
self._row("unknown"),
self._row("gpt-5.5", max_input_tokens=400000),
)
self._sync(listing, tmp_path)
models = self._entries(tmp_path)
assert models["big"]["context_window"] == 400000
assert models["unknown"]["context_window"] is None
assert models["gpt-5.5"]["context_window"] == 272000
def test_non_chat_models_are_left_out(self, tmp_path):
listing = self._listing(
self._row("chat", mode="chat"),
self._row("resp", mode="responses"),
self._row("embed", mode="embedding"),
self._row("img", mode="image_generation"),
)
self._sync(listing, tmp_path)
slugs = {m["slug"] for m in json.loads((tmp_path / "litellm-models.json").read_text())["models"]}
assert slugs == {"chat", "resp"}
def test_listing_without_chat_models_is_skipped_and_writes_nothing(self, tmp_path):
_, result = self._sync(self._listing(self._row("embed", mode="embedding")), tmp_path)
assert isinstance(result, ModelSyncSkipped)
assert "no chat models" in result.reason
assert not (tmp_path / "litellm-models.json").exists()
def test_catalog_is_rewritten_on_every_launch(self, tmp_path):
self._sync(self._listing(self._row("old")), tmp_path)
self._sync(self._listing(self._row("new")), tmp_path)
slugs = [m["slug"] for m in json.loads((tmp_path / "litellm-models.json").read_text())["models"]]
assert slugs == ["new"]
def test_catalog_is_replaced_whole_and_leaves_no_temp_files(self, tmp_path):
self._sync(self._listing(*(self._row(f"m{i}") for i in range(50))), tmp_path)
self._sync(self._listing(self._row("new")), tmp_path)
assert [p.name for p in tmp_path.iterdir()] == ["litellm-models.json"]
assert json.loads((tmp_path / "litellm-models.json").read_text())["models"][0]["slug"] == "new"
def test_defaults_to_dot_codex_in_home(self, tmp_path):
result = codex_model_sync_args(
{},
"http://localhost:4000",
"sk-key",
get=lambda *a, **k: _FakeResponse(200, self._listing(self._row("m"))),
run=_FakeRun(),
home=lambda: tmp_path,
)
assert self._catalog_path(result) == str(tmp_path / ".codex" / "litellm-models.json")
def test_default_home_is_the_users(self):
assert _default_of(codex_model_sync_args, "home") == Path.home
def test_missing_base_instructions_is_reported_not_raised(self, tmp_path):
result = codex_model_sync_args(
{"CODEX_HOME": str(tmp_path)},
"http://localhost:4000",
"sk-key",
get=lambda *a, **k: _FakeResponse(200, self._listing(self._row("m"))),
instructions_path=tmp_path / "missing.md",
)
assert isinstance(result, ModelSyncSkipped)
assert "could not read" in result.reason
assert not (tmp_path / "litellm-models.json").exists()
def test_unwritable_catalog_path_is_reported_not_raised(self, tmp_path):
blocker = tmp_path / "file"
blocker.write_text("")
_, result = self._sync(self._listing(self._row("m")), blocker / "codex")
assert isinstance(result, ModelSyncSkipped)
assert "could not write" in result.reason
def test_failed_replace_is_reported_and_leaves_no_temp_file(self, tmp_path):
(tmp_path / "litellm-models.json").mkdir()
_, result = self._sync(self._listing(self._row("m")), tmp_path)
assert isinstance(result, ModelSyncSkipped)
assert "could not write" in result.reason
assert [p.name for p in tmp_path.iterdir()] == ["litellm-models.json"]
def test_unreachable_proxy_is_reported_not_raised(self, tmp_path):
def boom(*a, **k):
raise requests.ConnectionError("refused")
result = codex_model_sync_args({"CODEX_HOME": str(tmp_path)}, "http://localhost:4000", "sk-key", get=boom)
assert isinstance(result, ModelSyncSkipped)
assert "refused" in result.reason
assert not (tmp_path / "litellm-models.json").exists()
@pytest.mark.parametrize(
("response", "reason"),
[(_FakeResponse(500), "HTTP 500"), (_FakeResponse(200, {"data": "nope"}), "unexpected body")],
)
def test_bad_response_is_reported(self, tmp_path, response, reason):
result = codex_model_sync_args(
{"CODEX_HOME": str(tmp_path)}, "http://localhost:4000", "sk-key", get=lambda *a, **k: response
)
assert isinstance(result, ModelSyncSkipped)
assert reason in result.reason
@pytest.mark.parametrize("binary", ["codex", "/opt/bin/codex", "codex.cmd", "/c/npm/codex.CMD"])
def test_codex_syncs_through_the_agent_dispatch_with_the_binary_it_will_run(self, tmp_path, binary):
run = _FakeRun()
result = agent_model_sync_env(
binary,
{"CODEX_HOME": str(tmp_path)},
"http://localhost:4000",
"sk-key",
False,
get=lambda *a, **k: _FakeResponse(200, self._listing(self._row("m"))),
run=run,
)
assert self._catalog_path(result) == str(tmp_path / "litellm-models.json")
assert len(run.calls) == 2
assert all(binary in command for command, _ in run.calls)
def test_opencode_dispatch_never_runs_codex(self):
def boom(*a, **k):
raise AssertionError("only the Codex sync reads its catalog back")
result = agent_model_sync_env(
"opencode",
{},
"http://localhost:4000",
"sk-key",
False,
get=lambda *a, **k: _FakeResponse(200, self._listing(self._row("m"))),
run=boom,
)
assert "OPENCODE_CONFIG_CONTENT" in result
def test_codex_lists_its_own_models_then_reads_the_catalog_back_before_launch(self, tmp_path):
run = _FakeRun()
_, result = self._sync(self._listing(self._row("m")), tmp_path, run=run)
path = self._catalog_path(result)
assert [command for command, _ in run.calls] == [
("codex", "debug", "models"),
("codex", "-c", f"model_catalog_json={json.dumps(path)}", "debug", "models"),
]
for _, options in run.calls:
assert options["env"] == {"CODEX_HOME": str(tmp_path)}
assert options["stdin"] is subprocess.DEVNULL
assert options["capture_output"] is True
assert options["encoding"] == "utf-8"
assert options["timeout"] == 10
def test_codex_rejecting_the_catalog_skips_the_sync_and_keeps_the_file(self, tmp_path):
stderr = (
"Error: failed to parse model_catalog_json path `/home/me/.codex/litellm-models.json` as JSON: "
"missing field `supports_parallel_tool_calls` at line 1 column 21648\n"
)
_, result = self._sync(self._listing(self._row("m")), tmp_path, run=_FakeRun(1, stderr))
assert isinstance(result, ModelSyncSkipped)
assert result.reason == (
"`codex debug models` exited 1: Error: failed to parse model_catalog_json path "
"`/home/me/.codex/litellm-models.json` as JSON: missing field `supports_parallel_tool_calls` "
"at line 1 column 21648"
)
assert (tmp_path / "litellm-models.json").exists()
def test_codex_without_debug_models_skips_the_sync(self, tmp_path):
stderr = "error: unrecognized subcommand 'models'\n\nUsage: codex debug [OPTIONS] <COMMAND>\n"
run = _FakeRun(2, stderr, stock=None)
_, result = self._sync(self._listing(self._row("m")), tmp_path, run=run)
assert isinstance(result, ModelSyncSkipped)
assert result.reason == "`codex debug models` exited 2: error: unrecognized subcommand 'models'"
assert len(run.calls) == 1
assert not (tmp_path / "litellm-models.json").exists()
def test_codex_failing_silently_is_reported(self, tmp_path):
_, result = self._sync(self._listing(self._row("m")), tmp_path, run=_FakeRun(1))
assert isinstance(result, ModelSyncSkipped)
assert result.reason == "`codex debug models` exited 1: no output"
@pytest.mark.parametrize(
"error", [OSError("codex vanished"), subprocess.TimeoutExpired("codex", 10)], ids=["oserror", "timeout"]
)
def test_unrunnable_preflight_is_reported_not_raised(self, tmp_path, error):
def failing_run(*a, **k):
raise error
_, result = self._sync(self._listing(self._row("m")), tmp_path, run=failing_run)
assert isinstance(result, ModelSyncSkipped)
assert result.reason.startswith("`codex debug models` failed: ")
assert str(error) in result.reason
def test_windows_shim_preflight_goes_through_cmd_exe(self, tmp_path):
shim = _WINDOWS_CLAUDE_CMD.replace("claude", "codex")
run = _FakeRun()
result = codex_model_sync_args(
{"CODEX_HOME": str(tmp_path)},
"http://localhost:4000",
"sk-key",
binary=shim,
get=lambda *a, **k: _FakeResponse(200, self._listing(self._row("m"))),
run=run,
)
override = f"model_catalog_json={json.dumps(self._catalog_path(result))}"
doubled = override.replace('"', '""')
assert [command for command, _ in run.calls] == [
f'{_CMD_PREFIX}""{shim}" "debug" "models""',
f'{_CMD_PREFIX}""{shim}" "-c" "{doubled}" "debug" "models""',
]
def test_default_binary_is_codex_on_path(self):
assert _default_of(codex_model_sync_args, "binary") == "codex"
def test_default_runner_is_subprocess_run(self):
assert _default_of(codex_model_sync_args, "run") is subprocess.run
assert _default_of(agent_model_sync_env, "run") is subprocess.run
def test_skip_verify_keeps_the_launch_offline(self):
def boom(*a, **k):
raise AssertionError("--skip-verify must not touch the proxy")
result = agent_model_sync_env("codex", {}, "http://localhost:4000", "sk-key", True, get=boom)
assert isinstance(result, ModelSyncSkipped)
assert "--skip-verify" in result.reason
def test_default_http_client_is_requests_get(self):
assert _default_of(codex_model_sync_args, "get") is requests.get
class TestRunAgent:
def test_synced_args_precede_user_args_and_follow_provider_overrides(self):
calls = {}
run_agent(
"http://localhost:4000",
"sk-key",
["codex", "exec", "hi"],
base_env={},
sync_models=lambda *a: ModelSyncArgs(("-c", 'model_catalog_json="/tmp/c.json"')),
which=lambda name: "/usr/local/bin/codex",
verify=lambda *a: None,
launcher=lambda p, a, e: calls.update(args=tuple(a), env=dict(e)),
)
args = calls["args"]
assert args[-2:] == ("exec", "hi")
assert args[args.index('model_catalog_json="/tmp/c.json"') - 1] == "-c"
assert (
args.index('model_provider="litellm"') < args.index('model_catalog_json="/tmp/c.json"') < args.index("exec")
)
assert calls["env"]["OPENAI_API_KEY"] == "sk-key"
assert "model_catalog_json" not in json.dumps(calls["env"])
def test_synced_model_config_reaches_the_agent_alongside_profile_env(self):
calls = {}
run_agent(
"http://localhost:4000",
"sk-key",
["opencode"],
base_env={"HOME": "/home/me"},
sync_models=lambda *a: {"OPENCODE_CONFIG_CONTENT": '{"provider":{}}'},
which=lambda name: "/usr/local/bin/opencode",
verify=lambda *a: None,
launcher=lambda p, a, e: calls.update(env=dict(e)),
)
assert calls["env"]["OPENCODE_CONFIG_CONTENT"] == '{"provider":{}}'
assert calls["env"]["OPENAI_BASE_URL"] == "http://localhost:4000/v1"
assert calls["env"]["OPENAI_API_KEY"] == "sk-key"
assert calls["env"]["HOME"] == "/home/me"
def test_sync_gets_the_launch_inputs_and_runs_after_verify(self):
order = []
calls = {}
def fake_sync(command, base_env, base_url, api_key, skip_verify):
order.append("sync")
calls["args"] = (command, dict(base_env), base_url, api_key, skip_verify)
return {"OPENCODE_CONFIG_CONTENT": '{"provider":{"litellm":{}}}'}
run_agent(
"http://localhost:4000",
"sk-key",
["opencode"],
base_env={"HOME": "/home/me"},
sync_models=fake_sync,
which=lambda name: "/usr/local/bin/opencode",
verify=lambda *a: order.append("verify"),
launcher=lambda p, a, e: order.append("launch"),
)
assert order == ["verify", "sync", "launch"]
assert calls["args"] == (
"/usr/local/bin/opencode",
{"HOME": "/home/me"},
"http://localhost:4000",
"sk-key",
False,
)
def test_unreachable_proxy_is_not_asked_for_models(self):
def failing_verify(*a):
raise AgentRunError("Could not reach the LiteLLM proxy")
def boom(*a):
raise AssertionError("a failed key check must not be followed by a model fetch")
with pytest.raises(AgentRunError):
run_agent(
"http://localhost:4000",
"sk-key",
["opencode"],
base_env={},
sync_models=boom,
which=lambda name: "/usr/local/bin/opencode",
verify=failing_verify,
launcher=lambda *a: None,
)
def test_skip_verify_reaches_the_sync_which_reports_the_skip(self):
warnings = []
calls = {}
def fake_sync(command, base_env, base_url, api_key, skip_verify):
calls["skip_verify"] = skip_verify
return ModelSyncSkipped("offline")
run_agent(
"http://localhost:4000",
"sk-key",
["opencode"],
skip_verify=True,
base_env={},
sync_models=fake_sync,
warn=warnings.append,
which=lambda name: "/usr/local/bin/opencode",
verify=lambda *a: pytest.fail("--skip-verify must not verify"),
launcher=lambda p, a, e: calls.update(env=dict(e)),
)
assert calls["skip_verify"] is True
assert "OPENCODE_CONFIG_CONTENT" not in calls["env"]
assert warnings == ["litellm: not syncing OpenCode models from the proxy: offline"]
def test_skipped_sync_still_launches_with_plain_openai_env(self):
calls = {}
run_agent(
"http://localhost:4000",
"sk-key",
["opencode"],
base_env={},
sync_models=lambda *a: ModelSyncSkipped("proxy said no"),
warn=lambda message: calls.setdefault("warned", message),
which=lambda name: "/usr/local/bin/opencode",
verify=lambda *a: None,
launcher=lambda p, a, e: calls.update(env=dict(e)),
)
assert calls["env"]["OPENAI_BASE_URL"] == "http://localhost:4000/v1"
assert "OPENCODE_CONFIG_CONTENT" not in calls["env"]
assert "proxy said no" in calls["warned"]
def test_non_opencode_agent_is_not_warned_about_model_sync(self):
warnings = []
run_agent(
"http://localhost:4000",
"sk-key",
["claude"],
base_env={},
warn=warnings.append,
which=lambda name: "/usr/local/bin/claude",
verify=lambda *a: None,
launcher=lambda *a: None,
sync_models=agent_model_sync_env,
)
assert warnings == []
def test_default_sync_is_the_agent_model_sync(self):
assert _default_of(run_agent, "sync_models") is agent_model_sync_env
def test_wires_env_and_launches_resolved_binary(self):
calls = {}
def fake_launcher(path, args, env):
calls["path"] = path
calls["args"] = tuple(args)
calls["env"] = dict(env)
run_agent(
"http://localhost:4000",
"sk-key",
["claude", "--resume"],
base_env={"PATH": "/usr/bin", "ANTHROPIC_API_KEY": "leaked"},
which=lambda name: "/usr/local/bin/claude",
verify=lambda *a: None,
launcher=fake_launcher,
)
assert calls["path"] == "/usr/local/bin/claude"
assert calls["args"] == ("claude", "--resume")
env = calls["env"]
assert env["ANTHROPIC_BASE_URL"] == "http://localhost:4000"
assert env["ANTHROPIC_AUTH_TOKEN"] == "sk-key"
assert env["ENABLE_TOOL_SEARCH"] == "true"
assert "ANTHROPIC_API_KEY" not in env
assert "OPENAI_BASE_URL" not in env
def test_codex_gets_openai_env(self):
calls = {}
run_agent(
"http://localhost:4000",
"sk-key",
["codex"],
base_env={},
which=lambda name: "/usr/local/bin/codex",
verify=lambda *a: None,
launcher=lambda p, a, e: calls.update(env=dict(e)),
)
assert calls["env"]["OPENAI_BASE_URL"] == "http://localhost:4000/v1"
assert calls["env"]["OPENAI_API_KEY"] == "sk-key"
assert "ANTHROPIC_BASE_URL" not in calls["env"]
assert "ENABLE_TOOL_SEARCH" not in calls["env"]
def test_codex_injects_proxy_provider_args_before_user_args(self):
calls = {}
run_agent(
"http://localhost:4000",
"sk-key",
["codex", "exec", "do a thing"],
base_env={},
which=lambda name: "/usr/local/bin/codex",
verify=lambda *a: None,
launcher=lambda p, a, e: calls.update(args=tuple(a)),
)
args = calls["args"]
assert args[0] == "codex"
assert args[-2:] == ("exec", "do a thing")
assert 'model_provider="litellm"' in args
assert 'model_providers.litellm.base_url="http://localhost:4000/v1"' in args
# overrides must precede the codex subcommand so codex parses them
assert args.index('model_provider="litellm"') < args.index("exec")
def test_pi_preparer_runs_after_verify_and_before_launch(self):
order = []
captured = {}
def fake_prepare(base_url, api_key, base_env):
order.append("prepare")
captured["args"] = (base_url, api_key, dict(base_env))
return []
run_agent(
"http://localhost:4000",
"sk-key",
["pi"],
base_env={"HOME": "/home/u"},
which=lambda name: "/usr/local/bin/pi",
verify=lambda *a: order.append("verify"),
launcher=lambda *a: order.append("launch"),
preparers={"pi": fake_prepare},
)
assert order == ["verify", "prepare", "launch"]
assert captured["args"] == (
"http://localhost:4000",
"sk-key",
{"HOME": "/home/u"},
)
def test_pi_prepared_args_precede_user_args_and_env_has_proxy_key(self):
calls = {}
run_agent(
"http://localhost:4000",
"sk-key",
["pi", "-p", "hello"],
base_env={},
which=lambda name: "/usr/local/bin/pi",
verify=lambda *a: None,
launcher=lambda p, a, e: calls.update(args=tuple(a), env=dict(e)),
preparers={"pi": lambda *a: ["--model", "litellm/m-1"]},
)
assert calls["args"] == ("pi", "--model", "litellm/m-1", "-p", "hello")
assert calls["env"]["LITELLM_PROXY_API_KEY"] == "sk-key"
assert "OPENAI_API_KEY" not in calls["env"]
assert "ANTHROPIC_BASE_URL" not in calls["env"]
def test_failed_preparer_aborts_before_launch(self):
launched = []
def boom(*a):
raise AgentRunError("sync failed")
with pytest.raises(AgentRunError, match="sync failed"):
run_agent(
"http://localhost:4000",
"sk-key",
["pi"],
base_env={},
which=lambda name: "/usr/local/bin/pi",
verify=lambda *a: None,
launcher=lambda *a: launched.append(a),
preparers={"pi": boom},
)
assert launched == []
def test_prepare_pi_syncs_models_json_and_pins_first_model(self, tmp_path):
from litellm.proxy.client.cli.commands.agents import prepare_pi
def fake_get(url, headers, timeout):
if url.endswith("/model_group/info"):
return _FakeJsonResponse(
200,
{"data": [{"model_group": "m-first", "max_input_tokens": 131072, "max_output_tokens": 8192}]},
)
return _FakeJsonResponse(200, {"data": [{"id": "m-first"}, {"id": "m-second"}]})
pin = prepare_pi(
"http://localhost:4000",
"sk-key",
{"PI_CODING_AGENT_DIR": str(tmp_path)},
get=fake_get,
)
assert pin == ("--model", "litellm/m-first")
import json
written = json.loads((tmp_path / "models.json").read_text())
assert written["providers"]["litellm"]["apiKey"] == "$LITELLM_PROXY_API_KEY"
assert written["providers"]["litellm"]["models"] == [
{"id": "m-first", "contextWindow": 131072, "maxTokens": 8192},
{"id": "m-second"},
]
def test_prepare_pi_surfaces_fetch_failure_as_agent_error(self, tmp_path):
from litellm.proxy.client.cli.commands.agents import prepare_pi
with pytest.raises(AgentRunError, match="HTTP 500"):
prepare_pi(
"http://localhost:4000",
"sk-key",
{"PI_CODING_AGENT_DIR": str(tmp_path)},
get=lambda *a, **k: _FakeJsonResponse(500),
)
def test_claude_has_no_preparer(self):
prepared = []
def fake_prepare(*a):
prepared.append(a)
return []
run_agent(
"http://localhost:4000",
"sk-key",
["claude"],
base_env={},
which=lambda name: "/usr/local/bin/claude",
verify=lambda *a: None,
launcher=lambda *a: None,
preparers={"pi": fake_prepare},
)
assert prepared == []
def test_claude_launches_without_injected_args(self):
calls = {}
run_agent(
"http://localhost:4000",
"sk-key",
["claude", "--resume"],
base_env={},
which=lambda name: "/usr/local/bin/claude",
verify=lambda *a: None,
launcher=lambda p, a, e: calls.update(args=tuple(a)),
)
assert calls["args"] == ("claude", "--resume")
def test_missing_binary_raises_with_install_hint(self):
with pytest.raises(AgentRunError, match=r"claude.*Install it first"):
run_agent(
"http://localhost:4000",
"sk-key",
["claude"],
base_env={},
which=lambda name: None,
verify=lambda *a: None,
launcher=lambda *a: None,
)
def test_skip_verify_does_not_call_verify(self):
verified = []
launched = []
run_agent(
"http://localhost:4000",
"sk-key",
["claude"],
skip_verify=True,
base_env={},
which=lambda name: "/usr/local/bin/claude",
verify=lambda *a: verified.append(a),
launcher=lambda *a: launched.append(a),
)
assert verified == []
assert len(launched) == 1
def test_verify_failure_aborts_before_launch(self):
launched = []
def boom(*a):
raise AgentRunError("rejected")
with pytest.raises(AgentRunError):
run_agent(
"http://localhost:4000",
"sk-key",
["claude"],
base_env={},
which=lambda name: "/usr/local/bin/claude",
verify=boom,
launcher=lambda *a: launched.append(a),
)
assert launched == []
def test_empty_command_raises(self):
with pytest.raises(AgentRunError):
run_agent("http://localhost:4000", "sk-key", [])
def test_subscription_launch_probes_with_the_header_and_exports_it(self):
probed = []
launched = {}
run_agent(
"http://localhost:4000",
"sk-key",
["claude"],
subscription=True,
base_env={"ANTHROPIC_AUTH_TOKEN": "stale"},
which=lambda name: "/usr/local/bin/claude",
verify=lambda *a: probed.append(("bearer", a)),
verify_subscription=lambda *a: probed.append(("header", a)),
launcher=lambda p, a, e: launched.update(env=dict(e)),
read_settings=lambda path: {},
)
assert probed == [("header", ("http://localhost:4000", "sk-key"))]
assert launched["env"]["ANTHROPIC_CUSTOM_HEADERS"] == "x-litellm-api-key: Bearer sk-key"
assert "ANTHROPIC_AUTH_TOKEN" not in launched["env"]
def test_subscription_launch_blanks_the_credentials_claude_code_would_load_from_settings_json(self, tmp_path):
launched = {}
run_agent(
"http://localhost:4000",
"sk-key",
["claude", "--resume"],
subscription=True,
base_env={},
which=lambda name: "/usr/local/bin/claude",
verify_subscription=lambda *a: None,
launcher=lambda p, a, e: launched.update(args=tuple(a)),
read_settings=lambda path: {},
write_overlay=_overlay_writer(tmp_path),
)
assert launched["args"][:2] == ("claude", "--settings")
assert _overlay(launched["args"]) == {
"env": {
"ANTHROPIC_AUTH_TOKEN": "",
"ANTHROPIC_API_KEY": "",
"ANTHROPIC_CUSTOM_HEADERS": "x-litellm-api-key: Bearer sk-key",
}
}
assert launched["args"][3:] == ("--resume",)
def test_subscription_launch_keeps_the_key_out_of_argv_and_in_one_private_file_next_to_settings_json(
self, tmp_path
):
config_dir = tmp_path / "cfg"
config_dir.mkdir()
stale = config_dir / "litellm-subscription-overlay.json"
stale.write_text('{"env": {"ANTHROPIC_CUSTOM_HEADERS": "x-litellm-api-key: Bearer old-key"}}')
stale.chmod(0o644)
launched = {}
for _ in range(2):
run_agent(
"http://localhost:4000",
"sk-key",
["claude"],
subscription=True,
base_env={"CLAUDE_CONFIG_DIR": str(config_dir)},
which=lambda name: "/usr/local/bin/claude",
verify_subscription=lambda *a: None,
launcher=lambda p, a, e: launched.update(args=list(a)),
read_settings=lambda path: {},
)
assert "sk-key" not in " ".join(launched["args"])
assert Path(launched["args"][2]) == stale
assert sorted(path.name for path in config_dir.iterdir()) == ["litellm-subscription-overlay.json"]
assert json.loads(stale.read_text())["env"]["ANTHROPIC_CUSTOM_HEADERS"] == "x-litellm-api-key: Bearer sk-key"
if os.name == "posix":
assert stat.S_IMODE(stale.stat().st_mode) == 0o600
def test_subscription_launch_accepts_inline_settings_json_longer_than_a_file_name(self, tmp_path):
long_value = "a" * 400
launched = {}
run_agent(
"http://localhost:4000",
"sk-key",
["claude", "--settings", json.dumps({"env": {"FOO": long_value}})],
subscription=True,
base_env={},
which=lambda name: "/usr/local/bin/claude",
verify_subscription=lambda *a: None,
launcher=lambda p, a, e: launched.update(args=list(a)),
read_settings=lambda path: {},
write_overlay=_overlay_writer(tmp_path),
)
assert _overlay(launched["args"])["env"]["FOO"] == long_value
def test_subscription_launch_reports_a_missing_settings_file_as_a_clean_cli_error(self, tmp_path):
with pytest.raises(AgentRunError, match="file not found"):
run_agent(
"http://localhost:4000",
"sk-key",
["claude", "--settings", str(tmp_path / "nope.json")],
subscription=True,
base_env={},
which=lambda name: "/usr/local/bin/claude",
verify_subscription=lambda *a: None,
launcher=lambda *a: None,
read_settings=lambda path: {},
write_overlay=_overlay_writer(tmp_path),
)
def test_subscription_launch_reads_the_settings_files_claude_code_merges_below_the_flag(self, tmp_path):
seen = []
run_agent(
"http://localhost:4000",
"sk-key",
["claude"],
subscription=True,
base_env={"CLAUDE_CONFIG_DIR": str(tmp_path / "cfg")},
which=lambda name: "/usr/local/bin/claude",
verify_subscription=lambda *a: None,
launcher=lambda *a: None,
read_settings=lambda path: seen.append(path) or {},
write_overlay=_overlay_writer(tmp_path),
project_dir=tmp_path / "repo",
)
assert seen == [
tmp_path / "repo" / ".claude" / "settings.local.json",
tmp_path / "repo" / ".claude" / "settings.json",
tmp_path / "cfg" / "settings.json",
]
@pytest.mark.parametrize("separate_flag", [True, False])
def test_subscription_launch_merges_the_blanks_into_the_users_settings_flag(self, tmp_path, separate_flag):
settings_file = tmp_path / "mine.json"
settings_file.write_text('{"model": "opus", "env": {"ANTHROPIC_AUTH_TOKEN": "mine", "FOO": "bar"}}')
user_flag = ["--settings", str(settings_file)] if separate_flag else [f"--settings={settings_file}"]
launched = {}
run_agent(
"http://localhost:4000",
"sk-key",
["claude", "--resume", *user_flag, "-p", "hi"],
subscription=True,
base_env={},
which=lambda name: "/usr/local/bin/claude",
verify_subscription=lambda *a: None,
launcher=lambda p, a, e: launched.update(args=list(a)),
read_settings=lambda path: {},
write_overlay=_overlay_writer(tmp_path),
)
assert launched["args"][:2] == ["claude", "--settings"]
assert _overlay(launched["args"]) == {
"model": "opus",
"env": {
"ANTHROPIC_AUTH_TOKEN": "",
"FOO": "bar",
"ANTHROPIC_API_KEY": "",
"ANTHROPIC_CUSTOM_HEADERS": "x-litellm-api-key: Bearer sk-key",
},
}
assert launched["args"][3:] == ["--resume", "-p", "hi"]
def test_subscription_launch_merges_the_blanks_into_inline_settings_json(self, tmp_path):
launched = {}
run_agent(
"http://localhost:4000",
"sk-key",
["claude", "--settings", '{"env": {"ANTHROPIC_API_KEY": "mine"}, "theme": "dark"}'],
subscription=True,
base_env={},
which=lambda name: "/usr/local/bin/claude",
verify_subscription=lambda *a: None,
launcher=lambda p, a, e: launched.update(args=list(a)),
read_settings=lambda path: {},
write_overlay=_overlay_writer(tmp_path),
)
assert launched["args"][:2] == ["claude", "--settings"]
assert _overlay(launched["args"]) == {
"theme": "dark",
"env": {
"ANTHROPIC_API_KEY": "",
"ANTHROPIC_AUTH_TOKEN": "",
"ANTHROPIC_CUSTOM_HEADERS": "x-litellm-api-key: Bearer sk-key",
},
}
assert launched["args"][3:] == []
@pytest.mark.parametrize(
("flag_headers", "local_headers", "project_headers", "user_headers", "expected"),
[
("X-Flag: f", "X-Local: l", "X-Proj: p", "X-User: u", "X-Flag: f\nx-litellm-api-key: Bearer sk-key"),
(None, "X-Local: l", "X-Proj: p", "X-User: u", "X-Local: l\nx-litellm-api-key: Bearer sk-key"),
(
None,
None,
"X-Proj: p\nx-litellm-api-key: Bearer stale",
"X-User: u",
"X-Proj: p\nx-litellm-api-key: Bearer sk-key",
),
(None, None, None, "X-User: u", "X-User: u\nx-litellm-api-key: Bearer sk-key"),
(None, None, None, None, "X-Proc: h\nx-litellm-api-key: Bearer sk-key"),
],
)
def test_subscription_overlay_keeps_the_custom_headers_claude_code_would_apply_and_adds_the_key(
self, tmp_path, flag_headers, local_headers, project_headers, user_headers, expected
):
"""Claude Code applies settings `env` over the process env, in precedence order, so the overlay must too."""
user_flag = (
["--settings", json.dumps({"env": {"ANTHROPIC_CUSTOM_HEADERS": flag_headers}})] if flag_headers else []
)
files = {
tmp_path / "repo" / ".claude" / "settings.local.json": local_headers,
tmp_path / "repo" / ".claude" / "settings.json": project_headers,
tmp_path / "cfg" / "settings.json": user_headers,
}
launched = {}
run_agent(
"http://localhost:4000",
"sk-key",
["claude", *user_flag],
subscription=True,
base_env={"ANTHROPIC_CUSTOM_HEADERS": "X-Proc: h", "CLAUDE_CONFIG_DIR": str(tmp_path / "cfg")},
which=lambda name: "/usr/local/bin/claude",
verify_subscription=lambda *a: None,
launcher=lambda p, a, e: launched.update(args=list(a), env=dict(e)),
read_settings=lambda path: {"env": {"ANTHROPIC_CUSTOM_HEADERS": files[path]}} if files[path] else {},
write_overlay=_overlay_writer(tmp_path),
project_dir=tmp_path / "repo",
)
assert _overlay(launched["args"])["env"]["ANTHROPIC_CUSTOM_HEADERS"] == expected
assert launched["env"]["ANTHROPIC_CUSTOM_HEADERS"] == "X-Proc: h\nx-litellm-api-key: Bearer sk-key"
def test_subscription_launch_strips_every_settings_flag_and_merges_the_last_one_claude_code_would_keep(
self, tmp_path
):
launched = {}
run_agent(
"http://localhost:4000",
"sk-key",
[
"claude",
"--settings",
'{"env": {"ANTHROPIC_AUTH_TOKEN": "first"}, "theme": "light"}',
"--resume",
'--settings={"theme": "dark"}',
],
subscription=True,
base_env={},
which=lambda name: "/usr/local/bin/claude",
verify_subscription=lambda *a: None,
launcher=lambda p, a, e: launched.update(args=list(a)),
read_settings=lambda path: {},
write_overlay=_overlay_writer(tmp_path),
)
assert launched["args"][:2] == ["claude", "--settings"]
assert _overlay(launched["args"]) == {
"theme": "dark",
"env": {
"ANTHROPIC_AUTH_TOKEN": "",
"ANTHROPIC_API_KEY": "",
"ANTHROPIC_CUSTOM_HEADERS": "x-litellm-api-key: Bearer sk-key",
},
}
assert launched["args"][3:] == ["--resume"]
def test_subscription_launch_reports_a_malformed_settings_file_as_a_clean_cli_error(self):
def broken(path):
raise ClaudeSettingsError(f"{path} contains invalid JSON")
with pytest.raises(AgentRunError, match="contains invalid JSON"):
run_agent(
"http://localhost:4000",
"sk-key",
["claude"],
subscription=True,
base_env={},
which=lambda name: "/usr/local/bin/claude",
verify_subscription=lambda *a: None,
launcher=lambda *a: None,
read_settings=broken,
)
def test_subscription_launch_rejects_a_settings_flag_that_is_not_a_json_object(self):
with pytest.raises(AgentRunError, match="is not a JSON object"):
run_agent(
"http://localhost:4000",
"sk-key",
["claude", "--settings", '{"env": 1, '],
subscription=True,
base_env={},
which=lambda name: "/usr/local/bin/claude",
verify_subscription=lambda *a: None,
launcher=lambda *a: None,
read_settings=lambda path: {},
)
def test_default_launch_still_probes_with_the_bearer(self):
probed = []
run_agent(
"http://localhost:4000",
"sk-key",
["claude"],
base_env={},
which=lambda name: "/usr/local/bin/claude",
verify=lambda *a: probed.append(("bearer", a)),
verify_subscription=lambda *a: probed.append(("header", a)),
launcher=lambda *a: None,
)
assert probed == [("bearer", ("http://localhost:4000", "sk-key"))]
def test_subscription_rejection_aborts_before_launch(self):
launched = []
def boom(*a):
raise AgentRunError("rejected")
with pytest.raises(AgentRunError):
run_agent(
"http://localhost:4000",
"sk-key",
["claude"],
subscription=True,
base_env={},
which=lambda name: "/usr/local/bin/claude",
verify_subscription=boom,
launcher=lambda *a: launched.append(a),
)
assert launched == []
def test_reattach_terminal_runs_just_before_launch(self):
order = []
run_agent(
"http://localhost:4000",
"sk-key",
["claude"],
skip_verify=True,
base_env={},
which=lambda name: "/usr/local/bin/claude",
launcher=lambda *a: order.append("launch"),
reattach_terminal=lambda: order.append("reattach"),
)
assert order == ["reattach", "launch"]
def test_no_reattach_terminal_by_default(self):
order = []
run_agent(
"http://localhost:4000",
"sk-key",
["claude"],
skip_verify=True,
base_env={},
which=lambda name: "/usr/local/bin/claude",
launcher=lambda *a: order.append("launch"),
)
assert order == ["launch"]
_WINDOWS_CLAUDE_EXE = "C:\\Program Files\\Claude\\claude.exe"
_WINDOWS_CLAUDE_CMD = "C:\\Users\\dev\\AppData\\Roaming\\npm\\claude.cmd"
_AGENT_ENV = {"ANTHROPIC_BASE_URL": "http://localhost:4000"}
_CMD_PREFIX = "cmd.exe /d /e:on /v:off /s /c "
def _shim_command_line(*args):
spawn = _Recorder(returns=0)
with pytest.raises(SystemExit):
_hand_off(
_WINDOWS_CLAUDE_CMD,
["claude", *args],
_AGENT_ENV,
platform="win32",
replace=_Recorder(),
spawn=spawn,
)
return spawn.calls[0][0]
class TestHandOff:
def test_windows_spawns_child_instead_of_exec(self):
replace = _Recorder()
spawn = _Recorder(returns=0)
with pytest.raises(SystemExit) as excinfo:
_hand_off(
_WINDOWS_CLAUDE_EXE,
["claude", "--resume"],
_AGENT_ENV,
platform="win32",
replace=replace,
spawn=spawn,
)
assert excinfo.value.code == 0
assert replace.calls == []
assert spawn.calls == [
((_WINDOWS_CLAUDE_EXE, "--resume"), _AGENT_ENV),
]
@pytest.mark.parametrize("code", [1, 42, 130])
def test_windows_propagates_child_exit_code(self, code):
with pytest.raises(SystemExit) as excinfo:
_hand_off(
_WINDOWS_CLAUDE_EXE,
["claude"],
_AGENT_ENV,
platform="win32",
replace=_Recorder(),
spawn=_Recorder(returns=code),
)
assert excinfo.value.code == code
@pytest.mark.parametrize(
"path",
[
_WINDOWS_CLAUDE_CMD,
"C:\\shims\\claude.CMD",
"C:\\shims\\claude.bat",
],
)
def test_windows_batch_shim_goes_through_cmd_exe(self, path):
spawn = _Recorder(returns=0)
with pytest.raises(SystemExit):
_hand_off(
path,
["claude", "--resume"],
_AGENT_ENV,
platform="win32",
replace=_Recorder(),
spawn=spawn,
)
assert spawn.calls[0][0] == f'{_CMD_PREFIX}""{path}" "--resume""'
def test_windows_shim_quotes_a_path_containing_spaces(self):
spawn = _Recorder(returns=0)
path = "C:\\Program Files\\npm\\claude.cmd"
with pytest.raises(SystemExit):
_hand_off(
path,
["claude", "-p", "hello world"],
_AGENT_ENV,
platform="win32",
replace=_Recorder(),
spawn=spawn,
)
expected = f'{_CMD_PREFIX}""C:\\Program Files\\npm\\claude.cmd" "-p" "hello world""'
assert spawn.calls[0][0] == expected
@pytest.mark.parametrize("payload", ["a&calc", "a|calc", "a>out", "a^b", "a&&calc"])
def test_windows_shim_never_leaves_a_metacharacter_unquoted(self, payload):
expected = f'{_CMD_PREFIX}""{_WINDOWS_CLAUDE_CMD}" "-p" "{payload}""'
assert _shim_command_line("-p", payload) == expected
def test_windows_shim_doubles_an_embedded_quote(self):
assert _shim_command_line("-p", 'say "hi"').endswith('"-p" "say ""hi""""')
@pytest.mark.parametrize(
"payload, quoted",
[
("%PATH%", "%%cd:~,%PATH%%cd:~,%"),
("100%", "100%%cd:~,%"),
("%OS%%CD%", "%%cd:~,%OS%%cd:~,%%%cd:~,%CD%%cd:~,%"),
],
)
def test_windows_shim_stops_cmd_expanding_a_percent_variable(self, payload, quoted):
assert _shim_command_line("-p", payload).endswith(f'"-p" "{quoted}""')
def test_windows_shim_guards_a_percent_in_the_shim_path(self):
spawn = _Recorder(returns=0)
path = "C:\\dev%HOME%\\claude.cmd"
with pytest.raises(SystemExit):
_hand_off(
path,
["claude"],
_AGENT_ENV,
platform="win32",
replace=_Recorder(),
spawn=spawn,
)
assert spawn.calls[0][0] == f'{_CMD_PREFIX}""C:\\dev%%cd:~,%HOME%%cd:~,%\\claude.cmd""'
@pytest.mark.parametrize(
"payload, quoted",
[
("C:\\dir\\", "C:\\dir\\\\"),
('say \\"hi', 'say \\\\""hi'),
('a\\\\"b', 'a\\\\\\\\""b'),
],
)
def test_windows_shim_doubles_backslashes_that_precede_a_quote(self, payload, quoted):
assert _shim_command_line("-p", payload).endswith(f'"-p" "{quoted}""')
@pytest.mark.parametrize("payload", ["one\ntwo", "one\r\ntwo", "trailing\r"])
def test_windows_shim_refuses_an_argument_holding_a_line_break(self, payload):
with pytest.raises(AgentRunError, match="line break"):
_hand_off(
_WINDOWS_CLAUDE_CMD,
["claude", "-p", payload],
_AGENT_ENV,
platform="win32",
replace=_Recorder(),
spawn=_Recorder(returns=0),
)
def test_windows_shim_keeps_the_switches_the_quoting_depends_on(self):
command = _shim_command_line("-p", "hi")
assert command.startswith("cmd.exe ")
switches = command.split(" /c ")[0].split()[1:]
assert switches == ["/d", "/e:on", "/v:off", "/s"]
def test_windows_exe_is_not_wrapped_in_cmd_exe(self):
spawn = _Recorder(returns=0)
with pytest.raises(SystemExit):
_hand_off(
_WINDOWS_CLAUDE_EXE,
["claude"],
_AGENT_ENV,
platform="win32",
replace=_Recorder(),
spawn=spawn,
)
assert spawn.calls[0][0] == (_WINDOWS_CLAUDE_EXE,)
@pytest.mark.parametrize("platform", ["darwin", "linux", "freebsd8"])
def test_posix_still_replaces_the_process(self, platform):
replace = _Recorder()
spawn = _Recorder(returns=0)
_hand_off(
"/usr/local/bin/claude",
["claude", "--resume"],
_AGENT_ENV,
platform=platform,
replace=replace,
spawn=spawn,
)
assert spawn.calls == []
assert replace.calls == [
("/usr/local/bin/claude", ["claude", "--resume"], _AGENT_ENV),
]
path, args, env = replace.calls[0]
assert isinstance(args, list)
assert isinstance(env, dict)
def test_replace_process_calls_execvpe_with_argv_and_env(self):
execvpe = _Recorder()
_replace_process(
"/usr/local/bin/claude",
("claude", "--resume"),
_AGENT_ENV,
execvpe=execvpe,
)
assert execvpe.calls == [
("/usr/local/bin/claude", ["claude", "--resume"], _AGENT_ENV),
]
_path, argv, env = execvpe.calls[0]
assert isinstance(argv, list)
assert isinstance(env, dict)
def test_posix_default_replacement_is_execvpe(self):
assert _default_of(run_agent, "launcher") is _hand_off
assert _default_of(_hand_off, "replace") is _replace_process
assert _default_of(_replace_process, "execvpe") is os.execvpe
assert _default_of(_hand_off, "spawn") is _spawn_and_wait
assert _default_of(_hand_off, "platform") == sys.platform
def test_spawn_and_wait_blocks_until_the_child_is_done(self, tmp_path):
marker = tmp_path / "child-finished"
script = (
"import os, pathlib, time; time.sleep(0.5); "
"pathlib.Path(os.environ['MARKER']).write_text('done'); "
"raise SystemExit(int(os.environ['RC']))"
)
code = _spawn_and_wait(
[sys.executable, "-c", script],
{"RC": "7", "MARKER": str(marker), "PATH": os.environ.get("PATH", "")},
)
assert marker.read_text() == "done"
assert code == 7
def test_windows_run_agent_spawns_resolved_binary_with_proxy_args(self):
spawn = _Recorder(returns=3)
replace = _Recorder()
def launcher(path, args, env):
_hand_off(path, args, env, platform="win32", replace=replace, spawn=spawn)
with pytest.raises(SystemExit) as excinfo:
run_agent(
"http://localhost:4000",
"sk-key",
["codex", "exec", "do a thing"],
skip_verify=True,
base_env={},
which=lambda name: _WINDOWS_CLAUDE_CMD.replace("claude", "codex"),
launcher=launcher,
)
assert excinfo.value.code == 3
assert replace.calls == []
command, env = spawn.calls[0]
shim = _WINDOWS_CLAUDE_CMD.replace("claude", "codex")
assert command.startswith(f'{_CMD_PREFIX}""{shim}" ')
assert command.endswith('"exec" "do a thing""')
assert '"model_provider=""litellm"""' in command
assert env["OPENAI_API_KEY"] == "sk-key"
class TestAgentCommands:
def setup_method(self):
self.runner = CliRunner()
def test_one_command_per_known_agent(self):
assert {c.name for c in agent_commands()} == {"claude", "codex", "opencode", "pi"}
@pytest.mark.parametrize("agent", [c.name for c in agent_commands()])
def test_lite_runs_every_known_agent_command(self, agent):
result = self.runner.invoke(lite, [agent, "--help"], prog_name="lite")
assert result.exit_code == 0, result.output
assert result.output.splitlines()[0].startswith(f"Usage: lite {agent} ")
def test_pi_is_hidden_from_help_but_still_registered(self):
hidden_by_name = {c.name: c.hidden for c in agent_commands()}
assert hidden_by_name == {"claude": False, "codex": False, "opencode": False, "pi": True}
def test_claude_launches_with_stored_key_and_forwards_args(self):
captured = {}
def fake_run_agent(base_url, api_key, command, **kwargs):
captured["base_url"] = base_url
captured["api_key"] = api_key
captured["command"] = list(command)
captured["skip_verify"] = kwargs.get("skip_verify")
with patch(f"{AGENTS_MODULE}.run_agent", side_effect=fake_run_agent):
result = self.runner.invoke(
_agent_command("claude"),
["--resume", "-p", "hi"],
obj={"base_url": "http://localhost:4000", "api_key": "sk-key"},
)
assert result.exit_code == 0, result.output
assert captured["api_key"] == "sk-key"
assert captured["command"] == ["claude", "--resume", "-p", "hi"]
assert captured["skip_verify"] is False
assert "routing Claude Code through proxy at http://localhost:4000" in result.output
def test_codex_shows_friendly_name(self):
captured = {}
with patch(
f"{AGENTS_MODULE}.run_agent",
side_effect=lambda b, k, c, **kw: captured.update(command=list(c)),
):
result = self.runner.invoke(
_agent_command("codex"),
["exec", "do a thing"],
obj={"base_url": "http://localhost:4000", "api_key": "sk-key"},
)
assert result.exit_code == 0, result.output
assert captured["command"] == ["codex", "exec", "do a thing"]
assert "routing Codex through proxy" in result.output
def test_opencode_launches_through_the_proxy(self):
captured = {}
with patch(f"{AGENTS_MODULE}.run_agent", side_effect=lambda b, k, c, **kw: captured.update(command=list(c))):
result = self.runner.invoke(
_agent_command("opencode"),
[],
obj={"base_url": "http://localhost:4000", "api_key": "sk-key"},
)
assert result.exit_code == 0, result.output
assert captured["command"] == ["opencode"]
assert "routing OpenCode through proxy at http://localhost:4000" in result.output
def test_claude_subscription_flag_reaches_run_agent_and_is_not_forwarded(self):
captured = {}
def fake_run_agent(base_url, api_key, command, **kwargs):
captured["command"] = list(command)
captured["subscription"] = kwargs.get("subscription")
with patch(f"{AGENTS_MODULE}.run_agent", side_effect=fake_run_agent):
result = self.runner.invoke(
_agent_command("claude"),
["--subscription", "--resume"],
obj={"base_url": "http://localhost:4000", "api_key": "sk-key"},
)
assert result.exit_code == 0, result.output
assert captured["subscription"] is True
assert captured["command"] == ["claude", "--resume"]
assert "keeping your claude.ai login" in result.output
@pytest.mark.parametrize(
("configured", "args", "expected"),
[
(False, [], False),
(True, [], True),
(True, ["--no-subscription"], False),
(False, ["--subscription"], True),
],
)
def test_subscription_comes_from_the_flag_else_the_saved_setting(self, configured, args, expected):
captured = {}
with (
patch(f"{AGENTS_MODULE}.claude_subscription_default", return_value=configured),
patch(
f"{AGENTS_MODULE}.run_agent",
side_effect=lambda b, k, c, **kw: captured.update(sub=kw.get("subscription"), command=list(c)),
),
):
result = self.runner.invoke(
_agent_command("claude"), args, obj={"base_url": "http://localhost:4000", "api_key": "sk-key"}
)
assert result.exit_code == 0, result.output
assert captured["sub"] is expected
assert captured["command"] == ["claude"]
assert ("keeping your claude.ai login" in result.output) is expected
@pytest.mark.parametrize("agent", ["codex", "opencode"])
def test_the_saved_claude_setting_does_not_change_other_agents(self, agent):
captured = {}
with (
patch(f"{AGENTS_MODULE}.claude_subscription_default", return_value=True),
patch(
f"{AGENTS_MODULE}.run_agent",
side_effect=lambda b, k, c, **kw: captured.update(sub=kw.get("subscription")),
),
):
result = self.runner.invoke(
_agent_command(agent), [], obj={"base_url": "http://localhost:4000", "api_key": "sk-key"}
)
assert result.exit_code == 0, result.output
assert captured["sub"] is False
@pytest.mark.parametrize("agent", ["codex", "opencode", "pi"])
def test_only_claude_offers_the_subscription_flag(self, agent):
assert "--subscription" not in {opt for param in _agent_command(agent).params for opt in param.opts}
def test_skip_verify_is_consumed_not_forwarded(self):
captured = {}
def fake_run_agent(base_url, api_key, command, **kwargs):
captured["command"] = list(command)
captured["skip_verify"] = kwargs.get("skip_verify")
with patch(f"{AGENTS_MODULE}.run_agent", side_effect=fake_run_agent):
result = self.runner.invoke(
_agent_command("claude"),
["--skip-verify", "--resume"],
obj={"base_url": "http://localhost:4000", "api_key": "sk-key"},
)
assert result.exit_code == 0, result.output
assert captured["skip_verify"] is True
assert captured["command"] == ["claude", "--resume"]
def test_non_interactive_without_key_errors_clearly(self):
with (
patch(f"{AGENTS_MODULE}._is_interactive", return_value=False),
patch(f"{AGENTS_MODULE}.run_agent") as mock_run,
):
result = self.runner.invoke(
_agent_command("claude"),
[],
obj={"base_url": "http://localhost:4000", "api_key": None},
)
assert result.exit_code != 0
assert "LITELLM_PROXY_API_KEY" in result.output
mock_run.assert_not_called()
def test_interactive_without_key_logs_in_then_launches(self, secret_vault_factory):
captured = {}
vault = secret_vault_factory()
@click.command()
def fake_login():
pass
with (
patch(f"{AGENTS_MODULE}._is_interactive", return_value=True),
patch(f"{AGENTS_MODULE}.login", fake_login),
patch(f"{AGENTS_MODULE}.get_stored_api_key", return_value="sk-after-login") as mock_get,
patch(
f"{AGENTS_MODULE}.run_agent",
side_effect=lambda base_url, api_key, command, **k: captured.update(api_key=api_key),
),
):
result = self.runner.invoke(
_agent_command("claude"),
[],
obj={"base_url": "http://localhost:4000", "api_key": None, "secret_vault": vault},
)
assert result.exit_code == 0, result.output
assert captured["api_key"] == "sk-after-login"
mock_get.assert_called_once_with(expected_base_url="http://localhost:4000", vault=vault)
def test_child_exit_code_reaches_the_shell(self):
with patch(f"{AGENTS_MODULE}.run_agent", side_effect=SystemExit(42)):
result = self.runner.invoke(
_agent_command("claude"),
[],
obj={"base_url": "http://localhost:4000", "api_key": "sk-key"},
)
assert result.exit_code == 42
def test_agent_run_error_becomes_click_error(self):
with patch(
f"{AGENTS_MODULE}.run_agent",
side_effect=AgentRunError("could not reach proxy"),
):
result = self.runner.invoke(
_agent_command("claude"),
[],
obj={"base_url": "http://localhost:4000", "api_key": "sk-key"},
)
assert result.exit_code != 0
assert "could not reach proxy" in result.output
def test_interactive_session_reattaches_terminal_before_handoff(self):
from litellm.proxy.client.cli.commands.agents import (
_restore_controlling_terminal,
)
captured = {}
with (
patch(f"{AGENTS_MODULE}._is_interactive", return_value=True),
patch(
f"{AGENTS_MODULE}.run_agent",
side_effect=lambda b, k, c, **kw: captured.update(kw),
),
):
result = self.runner.invoke(
_agent_command("claude"),
[],
obj={"base_url": "http://localhost:4000", "api_key": "sk-key"},
)
assert result.exit_code == 0, result.output
assert captured["reattach_terminal"] is _restore_controlling_terminal
def test_non_interactive_agent_mode_leaves_stdin_alone(self):
captured = {}
with (
patch(f"{AGENTS_MODULE}._is_interactive", return_value=False),
patch(
f"{AGENTS_MODULE}.run_agent",
side_effect=lambda b, k, c, **kw: captured.update(kw),
),
):
result = self.runner.invoke(
_agent_command("claude"),
[],
obj={"base_url": "http://localhost:4000", "api_key": "sk-key"},
)
assert result.exit_code == 0, result.output
assert captured["reattach_terminal"] is None
class TestPrepareCodex:
def test_registers_the_installed_script_as_a_session_scoped_stop_hook(self):
from litellm.proxy.client.cli.commands.agents import prepare_codex
args = prepare_codex(
"http://localhost:4000", "sk-key", {}, install=lambda: "/py /home/me/.litellm/statusline.py"
)
assert args == (
"-c",
'hooks.Stop=[{hooks=[{type="command",command="/py /home/me/.litellm/statusline.py"}]}]',
)
def test_a_failed_install_is_an_agent_error_not_a_crash(self):
from litellm.proxy.client.cli.commands.agents import AgentRunError, prepare_codex
from litellm.proxy.client.cli.commands.claude_settings import ClaudeSettingsError
def boom():
raise ClaudeSettingsError("disk full")
with pytest.raises(AgentRunError, match="disk full"):
prepare_codex("http://localhost:4000", "sk-key", {}, install=boom)
def test_a_config_that_already_declares_hooks_keeps_them_and_skips_ours(self, tmp_path):
from litellm.proxy.client.cli.commands.agents import prepare_codex
warnings = []
env = {"CODEX_HOME": str(tmp_path)}
for body in (
'[[hooks.Stop]]\nhooks = [{ type = "command", command = "mine" }]\n',
"hooks.Stop = []\n",
"[hooks]\n",
):
(tmp_path / "config.toml").write_text(body)
assert (
prepare_codex("http://localhost:4000", "sk", env, install=lambda: "/py /s.py", warn=warnings.append)
== ()
)
(tmp_path / "config.toml").write_text('model = "gpt-5.6-sol"\n[projects."/x"]\ntrust_level = "trusted"\n')
assert (
prepare_codex("http://localhost:4000", "sk", env, install=lambda: "/py /s.py", warn=warnings.append) != ()
)
assert len(warnings) == 3 and "already declares hooks" in warnings[0]
def test_a_config_that_cannot_be_read_or_decoded_still_lets_codex_launch(self, tmp_path):
# A UTF-16 config.toml (a Windows Notepad save) is Codex's problem to report at launch, not a reason
# for the hook pre-check to abort `lite codex` with a traceback before Codex ever starts.
from litellm.proxy.client.cli.commands.agents import codex_declares_stop_hooks, prepare_codex
config = tmp_path / "config.toml"
config.write_bytes('[[hooks.Stop]]\nhooks = [{ type = "command", command = "mine" }]\n'.encode("utf-16"))
assert codex_declares_stop_hooks(config) is False
assert codex_declares_stop_hooks(tmp_path / "absent.toml") is False
args = prepare_codex("http://localhost:4000", "sk", {"CODEX_HOME": str(tmp_path)}, install=lambda: "/py /s.py")
assert args[0] == "-c" and "hooks.Stop=" in args[1]
def test_codex_is_wired_through_the_preparer_registry(self):
from litellm.proxy.client.cli.commands.agents import _PREPARERS, prepare_codex
assert _PREPARERS["codex"] is prepare_codex
Reference in a new issue View git blame Copy permalink