mirror of
https://github.com/BerriAI/litellm.git
synced 2026-09-14 23:21:35 +00:00
feat(cli): add lite up/lite down to ambiently route Claude Code through the proxy
Patches ~/.claude/settings.json in place (env.ANTHROPIC_BASE_URL + apiKeyHelper via `lite auth print-token`) so any `claude` session started afterward, from any terminal, routes through the local LiteLLM proxy with no wrapper command needed, unlike the existing `lite claude` subprocess-exec approach. Backs up the original file first and restores it on Ctrl-C/SIGTERM, or via `lite down` after an unclean exit. Cursor is not supported: no equivalent file-based config to patch.
This commit is contained in:
parent
53aaabba5e
commit
28975f6ee0
5 changed files with 509 additions and 2 deletions
|
|
@ -471,6 +471,24 @@ The token minted by `lite login` is a short-lived, per-session agent credential,
|
|||
|
||||
The credential is short-lived by design (default 24h, configurable via `LITELLM_CLI_JWT_EXPIRATION_HOURS`); run `lite login` again to refresh it, which also re-reads your latest team and user settings. It does not appear in the Keys UI and cannot be rotated or revoked mid-session. `lite auth print-token` (usable as Claude Code's `apiKeyHelper`) prints it while it's still fresh and fails once it expires -- there is no silent renewal, so a long-running session needs a fresh `lite login` once a day. `lite claude`, `lite codex`, and `lite opencode` work with it on a default deployment; `EXPERIMENTAL_UI_LOGIN` is not required. If you need a long-lived, rotatable key that shows up in the Keys UI, create a dedicated virtual key in the dashboard and pass it via `--api-key` or `LITELLM_PROXY_API_KEY` instead.
|
||||
|
||||
### Route Every Claude Code Session Through the Proxy
|
||||
|
||||
`lite claude` wraps a single invocation, but `lite up` goes further: it patches `~/.claude/settings.json`, Claude Code's own config file, so that every Claude Code session started afterward -- from any terminal, launched normally with just `claude`, no wrapper needed -- routes through your LiteLLM proxy. It sets `env.ANTHROPIC_BASE_URL` to the proxy URL and `apiKeyHelper` to a `lite auth print-token` invocation, drops any stray static `ANTHROPIC_API_KEY` so the helper-issued token wins, and leaves every other setting in the file untouched. It backs up the original file before patching it.
|
||||
|
||||
Two things need to already be true: you've run `lite login`, since the apiKeyHelper depends on that stored token, and the proxy is already reachable, since `lite up` does not start one for you.
|
||||
|
||||
```bash
|
||||
lite login
|
||||
litellm --config litellm/proxy/dev_config.yaml &
|
||||
lite up
|
||||
```
|
||||
|
||||
`lite up` runs in the foreground and blocks. Press Ctrl-C to stop it, which restores the original settings file and exits. If the process is ever killed uncleanly instead -- `kill -9`, a crash -- the settings file is left patched, and `lite down` is the manual recovery path: run it at any later point to restore from the same backup.
|
||||
|
||||
This is a one-time file patch and restore, not a live traffic interceptor. A Claude Code session already running before `lite up` started keeps whatever `ANTHROPIC_BASE_URL` and token it loaded at its own startup, and a session still running when `lite up` stops keeps routing through the proxy until it exits; only sessions *started* while the patch is in effect are affected, and only *new* sessions after a restore go back to Anthropic directly.
|
||||
|
||||
Cursor is not supported: it has no equivalent file-based config to hot-patch this way, since its model routing lives in its own app storage and is configured through its GUI.
|
||||
|
||||
## Environment Variables
|
||||
|
||||
The CLI respects the following environment variables:
|
||||
|
|
|
|||
|
|
@ -212,7 +212,7 @@ def _is_interactive() -> bool:
|
|||
return sys.stdin.isatty()
|
||||
|
||||
|
||||
def _resolve_api_key(ctx: click.Context) -> str:
|
||||
def resolve_api_key(ctx: click.Context) -> str:
|
||||
base_url = ctx.obj["base_url"]
|
||||
api_key = ctx.obj.get("api_key")
|
||||
if api_key:
|
||||
|
|
@ -238,7 +238,7 @@ _SKIP_VERIFY_HELP = "Skip the pre-launch key check against the proxy."
|
|||
def _launch(ctx: click.Context, binary: str, args: Sequence[str], *, skip_verify: bool) -> None:
|
||||
base_url = ctx.obj["base_url"]
|
||||
started_interactive = _is_interactive()
|
||||
api_key = _resolve_api_key(ctx)
|
||||
api_key = resolve_api_key(ctx)
|
||||
|
||||
display_name, _ = agent_profile(binary)
|
||||
click.echo(f"litellm: routing {display_name} through proxy at {base_url.rstrip('/')}")
|
||||
|
|
@ -288,5 +288,6 @@ __all__ = [
|
|||
"agent_launch_args",
|
||||
"verify_proxy_key",
|
||||
"agent_profile",
|
||||
"resolve_api_key",
|
||||
"AgentRunError",
|
||||
]
|
||||
|
|
|
|||
231
litellm/proxy/client/cli/commands/up.py
Normal file
231
litellm/proxy/client/cli/commands/up.py
Normal file
|
|
@ -0,0 +1,231 @@
|
|||
import atexit
|
||||
import json
|
||||
import os
|
||||
import shlex
|
||||
import shutil
|
||||
import signal
|
||||
import sys
|
||||
import threading
|
||||
from dataclasses import dataclass
|
||||
from pathlib import Path
|
||||
from types import FrameType
|
||||
from typing import Dict, Mapping, Optional
|
||||
|
||||
import click
|
||||
from pydantic import JsonValue, TypeAdapter
|
||||
|
||||
from litellm.litellm_core_utils.cli_token_utils import is_cli_token_fresh
|
||||
|
||||
from .agents import AgentRunError, resolve_api_key, verify_proxy_key
|
||||
from .auth import load_token, login
|
||||
|
||||
ENV_KEY = "env"
|
||||
API_KEY_HELPER_KEY = "apiKeyHelper"
|
||||
ANTHROPIC_BASE_URL_KEY = "ANTHROPIC_BASE_URL"
|
||||
ANTHROPIC_API_KEY_KEY = "ANTHROPIC_API_KEY"
|
||||
|
||||
CLAUDE_SETTINGS_PATH = Path.home() / ".claude" / "settings.json"
|
||||
BACKUP_PATH = Path.home() / ".litellm" / "claude_settings_backup.json"
|
||||
|
||||
|
||||
class UpError(Exception):
|
||||
"""Raised for any user-actionable failure while starting/stopping interception."""
|
||||
|
||||
|
||||
@dataclass(frozen=True, slots=True)
|
||||
class BackupRecord:
|
||||
"""Snapshot of ~/.claude/settings.json taken right before `lite up` patches it."""
|
||||
|
||||
existed: bool
|
||||
content: Optional[Dict[str, JsonValue]]
|
||||
|
||||
|
||||
_SETTINGS_ADAPTER = TypeAdapter(Dict[str, JsonValue])
|
||||
_BACKUP_RECORD_ADAPTER = TypeAdapter(BackupRecord)
|
||||
|
||||
|
||||
def load_json_or_empty(path: Path) -> Dict[str, JsonValue]:
|
||||
if not path.exists():
|
||||
return {}
|
||||
with open(path, "r") as f:
|
||||
return _SETTINGS_ADAPTER.validate_json(f.read())
|
||||
|
||||
|
||||
def merge_claude_settings(
|
||||
settings: Mapping[str, JsonValue], base_url: str, api_key_helper: str
|
||||
) -> Dict[str, JsonValue]:
|
||||
"""Return a new settings dict wired to route Claude Code through the proxy.
|
||||
|
||||
Only env.ANTHROPIC_BASE_URL and the top-level apiKeyHelper are overridden; a
|
||||
stray env.ANTHROPIC_API_KEY is dropped so it cannot outrank the helper-issued
|
||||
token (same reasoning as build_agent_env in agents.py). Every other key is
|
||||
preserved untouched.
|
||||
"""
|
||||
raw_env = settings.get(ENV_KEY, {})
|
||||
base_env = raw_env if isinstance(raw_env, dict) else {}
|
||||
env = {**base_env, ANTHROPIC_BASE_URL_KEY: base_url.rstrip("/")}
|
||||
env.pop(ANTHROPIC_API_KEY_KEY, None)
|
||||
return {**settings, ENV_KEY: env, API_KEY_HELPER_KEY: api_key_helper}
|
||||
|
||||
|
||||
def write_backup(record: BackupRecord) -> None:
|
||||
BACKUP_PATH.parent.mkdir(exist_ok=True)
|
||||
with open(BACKUP_PATH, "w") as f:
|
||||
json.dump({"existed": record.existed, "content": record.content}, f, indent=2)
|
||||
os.chmod(BACKUP_PATH, 0o600)
|
||||
|
||||
|
||||
def read_backup() -> Optional[BackupRecord]:
|
||||
if not BACKUP_PATH.exists():
|
||||
return None
|
||||
with open(BACKUP_PATH, "r") as f:
|
||||
return _BACKUP_RECORD_ADAPTER.validate_json(f.read())
|
||||
|
||||
|
||||
def restore_claude_settings() -> Optional[BackupRecord]:
|
||||
"""Restore ~/.claude/settings.json from the backup, then delete the backup.
|
||||
|
||||
Returns the restored record, or None if there was nothing to restore.
|
||||
"""
|
||||
record = read_backup()
|
||||
if record is None:
|
||||
return None
|
||||
if record.existed and record.content is not None:
|
||||
with open(CLAUDE_SETTINGS_PATH, "w") as f:
|
||||
json.dump(record.content, f, indent=2)
|
||||
elif CLAUDE_SETTINGS_PATH.exists():
|
||||
CLAUDE_SETTINGS_PATH.unlink()
|
||||
BACKUP_PATH.unlink()
|
||||
return record
|
||||
|
||||
|
||||
def resolve_api_key_helper() -> str:
|
||||
"""Build the shell command Claude Code should run for its apiKeyHelper.
|
||||
|
||||
Resolves `lite` to an absolute path so the helper works regardless of the
|
||||
PATH visible to whatever subprocess Claude Code spawns it from.
|
||||
"""
|
||||
lite_path = shutil.which("lite")
|
||||
if lite_path is None:
|
||||
raise UpError(
|
||||
"Could not find `lite` on your PATH. Claude Code's apiKeyHelper needs "
|
||||
"an absolute path to it, so `lite up` cannot continue."
|
||||
)
|
||||
return f"{shlex.quote(lite_path)} auth print-token"
|
||||
|
||||
|
||||
def _ensure_fresh_login(ctx: click.Context) -> None:
|
||||
token_data = load_token()
|
||||
if token_data and is_cli_token_fresh(token_data):
|
||||
return
|
||||
|
||||
if not sys.stdin.isatty():
|
||||
raise UpError(
|
||||
"No fresh LiteLLM login found. Run `lite login` first (apiKeyHelper "
|
||||
"reads this token on every Claude Code request)."
|
||||
)
|
||||
|
||||
click.echo("No fresh LiteLLM login found; starting login...")
|
||||
ctx.invoke(login)
|
||||
token_data = load_token()
|
||||
if not token_data or not is_cli_token_fresh(token_data):
|
||||
raise UpError("Login did not produce a usable token; cannot start `lite up`.")
|
||||
|
||||
|
||||
def _restore_and_report() -> None:
|
||||
record = restore_claude_settings()
|
||||
if record is None:
|
||||
click.echo("Nothing to restore.")
|
||||
return
|
||||
if record.existed:
|
||||
click.echo(f"Restored {CLAUDE_SETTINGS_PATH} to its original contents.")
|
||||
else:
|
||||
click.echo(f"Removed {CLAUDE_SETTINGS_PATH} (it did not exist before `lite up`).")
|
||||
|
||||
|
||||
@click.command(name="up")
|
||||
@click.pass_context
|
||||
def up(ctx: click.Context) -> None:
|
||||
"""Route every Claude Code session through your LiteLLM proxy until stopped.
|
||||
|
||||
Patches ~/.claude/settings.json so Claude Code picks up the proxy on its own
|
||||
next startup, from any terminal -- no need to launch it through `lite`.
|
||||
Press Ctrl-C to stop and restore your original settings. Assumes the proxy
|
||||
is already running (this does not start one for you). Cursor is not
|
||||
supported: it has no equivalent file-based config to patch.
|
||||
"""
|
||||
base_url = ctx.obj["base_url"]
|
||||
|
||||
try:
|
||||
_ensure_fresh_login(ctx)
|
||||
api_key = resolve_api_key(ctx)
|
||||
verify_proxy_key(base_url, api_key)
|
||||
|
||||
if BACKUP_PATH.exists():
|
||||
raise UpError(
|
||||
f"{BACKUP_PATH} already exists -- `lite up` looks like it's already "
|
||||
"running (or crashed without cleanup). Run `lite down` first."
|
||||
)
|
||||
|
||||
api_key_helper = resolve_api_key_helper()
|
||||
original_existed = CLAUDE_SETTINGS_PATH.exists()
|
||||
original_settings = load_json_or_empty(CLAUDE_SETTINGS_PATH)
|
||||
write_backup(
|
||||
BackupRecord(
|
||||
existed=original_existed,
|
||||
content=original_settings if original_existed else None,
|
||||
)
|
||||
)
|
||||
|
||||
CLAUDE_SETTINGS_PATH.parent.mkdir(exist_ok=True)
|
||||
merged = merge_claude_settings(original_settings, base_url, api_key_helper)
|
||||
with open(CLAUDE_SETTINGS_PATH, "w") as f:
|
||||
json.dump(merged, f, indent=2)
|
||||
except (AgentRunError, UpError) as e:
|
||||
raise click.ClickException(str(e))
|
||||
|
||||
click.echo(f"litellm: routing Claude Code through proxy at {base_url.rstrip('/')}")
|
||||
click.echo("Press Ctrl-C to stop and restore your original settings.")
|
||||
|
||||
stop_event = threading.Event()
|
||||
restored = threading.Lock()
|
||||
|
||||
def _handle_signal(_signum: int, _frame: Optional[FrameType]) -> None:
|
||||
stop_event.set()
|
||||
|
||||
def _restore_once() -> None:
|
||||
if restored.acquire(blocking=False):
|
||||
_restore_and_report()
|
||||
|
||||
signal.signal(signal.SIGINT, _handle_signal)
|
||||
signal.signal(signal.SIGTERM, _handle_signal)
|
||||
atexit.register(_restore_once)
|
||||
|
||||
stop_event.wait()
|
||||
_restore_once()
|
||||
|
||||
|
||||
@click.command(name="down")
|
||||
def down() -> None:
|
||||
"""Restore ~/.claude/settings.json if a `lite up` session left it patched.
|
||||
|
||||
Use this after a `lite up` process was killed uncleanly (e.g. `kill -9`)
|
||||
instead of stopped with Ctrl-C.
|
||||
"""
|
||||
_restore_and_report()
|
||||
|
||||
|
||||
__all__ = [
|
||||
"up",
|
||||
"down",
|
||||
"BackupRecord",
|
||||
"load_json_or_empty",
|
||||
"merge_claude_settings",
|
||||
"write_backup",
|
||||
"read_backup",
|
||||
"restore_claude_settings",
|
||||
"resolve_api_key_helper",
|
||||
"UpError",
|
||||
"CLAUDE_SETTINGS_PATH",
|
||||
"BACKUP_PATH",
|
||||
]
|
||||
|
|
@ -18,6 +18,7 @@ from .commands.keys import keys
|
|||
# local imports
|
||||
from .commands.models import models
|
||||
from .commands.teams import teams
|
||||
from .commands.up import down, up
|
||||
from .commands.users import users
|
||||
from .interface import interactive_shell
|
||||
|
||||
|
|
@ -131,6 +132,9 @@ cli.add_command(users)
|
|||
# Add a top-level command per coding agent (claude, codex, opencode, ...)
|
||||
for agent_command in agent_commands():
|
||||
cli.add_command(agent_command)
|
||||
# Add the up/down commands (route Claude Code through the local LiteLLM proxy)
|
||||
cli.add_command(up)
|
||||
cli.add_command(down)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
|
|
|
|||
253
tests/test_litellm/proxy/client/cli/test_up_commands.py
Normal file
253
tests/test_litellm/proxy/client/cli/test_up_commands.py
Normal file
|
|
@ -0,0 +1,253 @@
|
|||
import json
|
||||
import shutil
|
||||
import sys
|
||||
from unittest.mock import patch
|
||||
|
||||
import pytest
|
||||
from click.testing import CliRunner
|
||||
|
||||
from litellm.proxy.client.cli.commands import up as up_module
|
||||
from litellm.proxy.client.cli.commands.agents import AgentRunError
|
||||
from litellm.proxy.client.cli.commands.up import (
|
||||
BackupRecord,
|
||||
UpError,
|
||||
down,
|
||||
merge_claude_settings,
|
||||
read_backup,
|
||||
resolve_api_key_helper,
|
||||
restore_claude_settings,
|
||||
up,
|
||||
write_backup,
|
||||
)
|
||||
|
||||
UP_MODULE = "litellm.proxy.client.cli.commands.up"
|
||||
|
||||
|
||||
def _patch_paths(monkeypatch, tmp_path):
|
||||
settings_path = tmp_path / "claude_settings.json"
|
||||
backup_path = tmp_path / "backup.json"
|
||||
monkeypatch.setattr(up_module, "CLAUDE_SETTINGS_PATH", settings_path)
|
||||
monkeypatch.setattr(up_module, "BACKUP_PATH", backup_path)
|
||||
return settings_path, backup_path
|
||||
|
||||
|
||||
class TestMergeClaudeSettings:
|
||||
def test_preserves_unrelated_top_level_keys(self):
|
||||
merged = merge_claude_settings({"theme": "dark"}, "http://localhost:4000", "helper")
|
||||
assert merged["theme"] == "dark"
|
||||
|
||||
def test_preserves_unrelated_env_keys(self):
|
||||
settings = {"env": {"SOME_OTHER_VAR": "value"}}
|
||||
merged = merge_claude_settings(settings, "http://localhost:4000", "helper")
|
||||
assert merged["env"]["SOME_OTHER_VAR"] == "value"
|
||||
|
||||
def test_overrides_base_url_and_helper(self):
|
||||
settings = {
|
||||
"env": {"ANTHROPIC_BASE_URL": "https://old.example.com"},
|
||||
"apiKeyHelper": "old-helper",
|
||||
}
|
||||
merged = merge_claude_settings(settings, "http://localhost:4000/", "new-helper")
|
||||
assert merged["env"]["ANTHROPIC_BASE_URL"] == "http://localhost:4000"
|
||||
assert merged["apiKeyHelper"] == "new-helper"
|
||||
|
||||
def test_drops_stray_api_key(self):
|
||||
settings = {"env": {"ANTHROPIC_API_KEY": "leaked-key"}}
|
||||
merged = merge_claude_settings(settings, "http://localhost:4000", "helper")
|
||||
assert "ANTHROPIC_API_KEY" not in merged["env"]
|
||||
|
||||
def test_works_from_empty_settings(self):
|
||||
merged = merge_claude_settings({}, "http://localhost:4000", "helper")
|
||||
assert merged["env"] == {"ANTHROPIC_BASE_URL": "http://localhost:4000"}
|
||||
assert merged["apiKeyHelper"] == "helper"
|
||||
|
||||
def test_does_not_mutate_input(self):
|
||||
settings = {"env": {"FOO": "bar"}}
|
||||
merge_claude_settings(settings, "http://localhost:4000", "helper")
|
||||
assert settings == {"env": {"FOO": "bar"}}
|
||||
|
||||
|
||||
class TestBackupRoundTrip:
|
||||
def test_restores_original_content_when_file_existed(self, monkeypatch, tmp_path):
|
||||
settings_path, backup_path = _patch_paths(monkeypatch, tmp_path)
|
||||
original = {"apiKeyHelper": "old-helper", "theme": "dark"}
|
||||
settings_path.write_text(json.dumps(original))
|
||||
|
||||
write_backup(BackupRecord(existed=True, content=original))
|
||||
settings_path.write_text(json.dumps({"apiKeyHelper": "lite-helper"}))
|
||||
|
||||
restored = restore_claude_settings()
|
||||
|
||||
assert restored is not None
|
||||
assert restored.existed is True
|
||||
assert json.loads(settings_path.read_text()) == original
|
||||
assert not backup_path.exists()
|
||||
|
||||
def test_deletes_settings_file_when_it_did_not_exist_before(self, monkeypatch, tmp_path):
|
||||
settings_path, backup_path = _patch_paths(monkeypatch, tmp_path)
|
||||
write_backup(BackupRecord(existed=False, content=None))
|
||||
settings_path.write_text(json.dumps({"apiKeyHelper": "lite-helper"}))
|
||||
|
||||
restored = restore_claude_settings()
|
||||
|
||||
assert restored is not None
|
||||
assert restored.existed is False
|
||||
assert not settings_path.exists()
|
||||
assert not backup_path.exists()
|
||||
|
||||
def test_no_backup_is_a_no_op_returning_none(self, monkeypatch, tmp_path):
|
||||
settings_path, _backup_path = _patch_paths(monkeypatch, tmp_path)
|
||||
assert restore_claude_settings() is None
|
||||
assert not settings_path.exists()
|
||||
|
||||
def test_read_backup_round_trips_write_backup(self, monkeypatch, tmp_path):
|
||||
_patch_paths(monkeypatch, tmp_path)
|
||||
write_backup(BackupRecord(existed=True, content={"a": 1}))
|
||||
assert read_backup() == BackupRecord(existed=True, content={"a": 1})
|
||||
|
||||
def test_read_backup_missing_file_returns_none(self, monkeypatch, tmp_path):
|
||||
_patch_paths(monkeypatch, tmp_path)
|
||||
assert read_backup() is None
|
||||
|
||||
def test_backup_file_always_removed_after_restore(self, monkeypatch, tmp_path):
|
||||
_settings_path, backup_path = _patch_paths(monkeypatch, tmp_path)
|
||||
write_backup(BackupRecord(existed=False, content=None))
|
||||
assert backup_path.exists()
|
||||
|
||||
restore_claude_settings()
|
||||
|
||||
assert not backup_path.exists()
|
||||
|
||||
|
||||
class TestResolveApiKeyHelper:
|
||||
def test_returns_helper_command_when_lite_found(self, monkeypatch):
|
||||
monkeypatch.setattr(shutil, "which", lambda name: "/usr/local/bin/lite")
|
||||
helper = resolve_api_key_helper()
|
||||
assert helper == "/usr/local/bin/lite auth print-token"
|
||||
|
||||
def test_raises_when_lite_not_on_path(self, monkeypatch):
|
||||
monkeypatch.setattr(shutil, "which", lambda name: None)
|
||||
with pytest.raises(UpError, match="Could not find `lite`"):
|
||||
resolve_api_key_helper()
|
||||
|
||||
|
||||
class TestUpCommand:
|
||||
def setup_method(self):
|
||||
self.runner = CliRunner()
|
||||
|
||||
def test_refuses_double_start_without_touching_settings_file(self, monkeypatch, tmp_path):
|
||||
settings_path, backup_path = _patch_paths(monkeypatch, tmp_path)
|
||||
existing_backup = {"existed": False, "content": None}
|
||||
backup_path.write_text(json.dumps(existing_backup))
|
||||
|
||||
with (
|
||||
patch(f"{UP_MODULE}.load_token", return_value={"key": "sk-fresh"}),
|
||||
patch(f"{UP_MODULE}.is_cli_token_fresh", return_value=True),
|
||||
patch(f"{UP_MODULE}.resolve_api_key", return_value="sk-fresh"),
|
||||
patch(f"{UP_MODULE}.verify_proxy_key"),
|
||||
):
|
||||
result = self.runner.invoke(up, obj={"base_url": "http://localhost:4000"})
|
||||
|
||||
assert result.exit_code != 0
|
||||
assert "already" in result.output
|
||||
assert "lite down" in result.output
|
||||
assert not settings_path.exists()
|
||||
assert json.loads(backup_path.read_text()) == existing_backup
|
||||
|
||||
def test_no_fresh_login_non_interactive_fails_cleanly(self, monkeypatch, tmp_path):
|
||||
_patch_paths(monkeypatch, tmp_path)
|
||||
monkeypatch.setattr(sys.stdin, "isatty", lambda: False)
|
||||
|
||||
with patch(f"{UP_MODULE}.load_token", return_value=None):
|
||||
result = self.runner.invoke(up, obj={"base_url": "http://localhost:4000"})
|
||||
|
||||
assert result.exit_code != 0
|
||||
assert "lite login" in result.output
|
||||
|
||||
def test_unreachable_proxy_fails_cleanly(self, monkeypatch, tmp_path):
|
||||
_patch_paths(monkeypatch, tmp_path)
|
||||
|
||||
with (
|
||||
patch(f"{UP_MODULE}.load_token", return_value={"key": "sk-fresh"}),
|
||||
patch(f"{UP_MODULE}.is_cli_token_fresh", return_value=True),
|
||||
patch(f"{UP_MODULE}.resolve_api_key", return_value="sk-fresh"),
|
||||
patch(
|
||||
f"{UP_MODULE}.verify_proxy_key",
|
||||
side_effect=AgentRunError("Could not reach the LiteLLM proxy at http://localhost:4000"),
|
||||
),
|
||||
):
|
||||
result = self.runner.invoke(up, obj={"base_url": "http://localhost:4000"})
|
||||
|
||||
assert result.exit_code != 0
|
||||
assert "Could not reach the LiteLLM proxy" in result.output
|
||||
|
||||
def test_happy_path_writes_settings_and_backup_then_restores_on_stop(self, monkeypatch, tmp_path):
|
||||
settings_path, backup_path = _patch_paths(monkeypatch, tmp_path)
|
||||
original = {"theme": "dark"}
|
||||
settings_path.write_text(json.dumps(original))
|
||||
|
||||
captured = {}
|
||||
|
||||
def fake_wait(self, timeout=None):
|
||||
captured["settings"] = json.loads(settings_path.read_text())
|
||||
captured["backup_existed"] = backup_path.exists()
|
||||
return True
|
||||
|
||||
with (
|
||||
patch(f"{UP_MODULE}.load_token", return_value={"key": "sk-fresh"}),
|
||||
patch(f"{UP_MODULE}.is_cli_token_fresh", return_value=True),
|
||||
patch(f"{UP_MODULE}.resolve_api_key", return_value="sk-fresh"),
|
||||
patch(f"{UP_MODULE}.verify_proxy_key"),
|
||||
patch(
|
||||
f"{UP_MODULE}.resolve_api_key_helper",
|
||||
return_value="/usr/local/bin/lite auth print-token",
|
||||
),
|
||||
patch(f"{UP_MODULE}.signal.signal"),
|
||||
patch(f"{UP_MODULE}.atexit.register"),
|
||||
patch("threading.Event.wait", new=fake_wait),
|
||||
):
|
||||
result = self.runner.invoke(up, obj={"base_url": "http://localhost:4000"})
|
||||
|
||||
assert result.exit_code == 0, result.output
|
||||
assert captured["backup_existed"] is True
|
||||
assert captured["settings"]["theme"] == "dark"
|
||||
assert captured["settings"]["env"]["ANTHROPIC_BASE_URL"] == "http://localhost:4000"
|
||||
assert captured["settings"]["apiKeyHelper"] == "/usr/local/bin/lite auth print-token"
|
||||
assert json.loads(settings_path.read_text()) == original
|
||||
assert not backup_path.exists()
|
||||
|
||||
|
||||
class TestDownCommand:
|
||||
def setup_method(self):
|
||||
self.runner = CliRunner()
|
||||
|
||||
def test_restores_when_backup_exists(self, monkeypatch, tmp_path):
|
||||
settings_path, backup_path = _patch_paths(monkeypatch, tmp_path)
|
||||
original = {"apiKeyHelper": "old-helper"}
|
||||
write_backup(BackupRecord(existed=True, content=original))
|
||||
settings_path.write_text(json.dumps({"apiKeyHelper": "lite-helper"}))
|
||||
|
||||
result = self.runner.invoke(down)
|
||||
|
||||
assert result.exit_code == 0, result.output
|
||||
assert "Restored" in result.output
|
||||
assert json.loads(settings_path.read_text()) == original
|
||||
assert not backup_path.exists()
|
||||
|
||||
def test_removes_settings_file_when_it_did_not_exist_before(self, monkeypatch, tmp_path):
|
||||
settings_path, _backup_path = _patch_paths(monkeypatch, tmp_path)
|
||||
write_backup(BackupRecord(existed=False, content=None))
|
||||
settings_path.write_text(json.dumps({"apiKeyHelper": "lite-helper"}))
|
||||
|
||||
result = self.runner.invoke(down)
|
||||
|
||||
assert result.exit_code == 0, result.output
|
||||
assert "Removed" in result.output
|
||||
assert not settings_path.exists()
|
||||
|
||||
def test_prints_nothing_to_restore_when_no_backup(self, monkeypatch, tmp_path):
|
||||
_patch_paths(monkeypatch, tmp_path)
|
||||
|
||||
result = self.runner.invoke(down)
|
||||
|
||||
assert result.exit_code == 0, result.output
|
||||
assert "Nothing to restore." in result.output
|
||||
Loading…
Add table
Reference in a new issue