mirror of
https://github.com/BerriAI/litellm.git
synced 2026-09-17 23:51:30 +00:00
1714 lines
67 KiB
Python
1714 lines
67 KiB
Python
import inspect
|
|
import json
|
|
import os
|
|
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,
|
|
)
|
|
|
|
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
|
|
|
|
|
|
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"}
|
|
|
|
|
|
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),
|
|
)
|
|
|
|
|
|
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_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"}
|
|
|
|
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_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
|