mirror of
https://github.com/BerriAI/litellm.git
synced 2026-10-15 04:19:08 +00:00
fix(cli): run group subcommands in the lite shell and default to the logged-in gateway (#46511)
* fix(cli): run group subcommands in the lite shell and default to the logged-in gateway Typing a command group alone in the lite shell printed 'Usage: lite lite', the next subcommand was unknown, quoted arguments were split on spaces, and a bare lite connected to localhost:4000 even after lite login to a gateway. The shell now names the group in usage text, keeps the group as the prompt's scope so its subcommands run next, accepts a leading 'lite', parses lines with shlex, and a bare lite falls back to the server the stored lite login credential was issued for before localhost:4000. * refactor(cli): walk the shell's group path without recursion * fix(cli): keep backslashes and compact JSON in the lite shell and pair the login server with its key * fix(cli): keep the login server to the bare lite shell and accept lite options and = quoting at its prompt * fix(cli): keep the shell's server for a lite options line and name the right login in the keychain notice * fix(cli): keep compact JSON with a trailing = intact in the shell and pin options lines to the shell's server over LITELLM_PROXY_URL
This commit is contained in:
Mateo Wang
GitHub
parent
0ececab7a1
commit
95edbd3773
6 changed files with 404 additions and 40 deletions
|
|
@ -36,6 +36,8 @@ The base URL is resolved in this order of precedence:
|
|||
3. `base_url` from `~/.litellm/config.json`
|
||||
4. `http://localhost:4000`
|
||||
|
||||
The interactive shell you get from a bare `lite` puts one more step before `http://localhost:4000`: the server you last signed in to with `lite login`, unless you pass `--api-key` or set `LITELLM_PROXY_API_KEY`, since that key was not issued for it. One-shot commands such as `lite keys delete` never pick up that server on their own. Inside the shell, a `lite ...` line runs on the shell's server, with any key it names, unless it also names `--base-url`.
|
||||
|
||||
### Hiding commands from the listings
|
||||
|
||||
Deployments that hand `lite` to end users often want to advertise only part of it. Store the commands to keep out of the listings, comma separated:
|
||||
|
|
|
|||
|
|
@ -191,26 +191,26 @@ def storage_notice(outcome: SecretSave) -> str:
|
|||
)
|
||||
|
||||
|
||||
def keychain_unreadable_notice(vault: SecretVault) -> str:
|
||||
def keychain_unreadable_notice(vault: SecretVault, login_command: str = "lite login") -> str:
|
||||
"""Explain why the secret half of a stored login cannot be produced, and what fixes it"""
|
||||
match vault.read():
|
||||
case KeyringNotInstalled():
|
||||
return (
|
||||
"Your credential is in your OS keychain, which this install cannot read without the "
|
||||
f"keyring package. Install it with: {KEYRING_INSTALL_HINT}, or run 'lite login' to start over."
|
||||
f"keyring package. Install it with: {KEYRING_INSTALL_HINT}, or run '{login_command}' to start over."
|
||||
)
|
||||
case KeyringDisabled():
|
||||
return (
|
||||
f"Your credential is in your OS keychain, which {DISABLE_KEYRING_ENV_VAR} is blocking. "
|
||||
"Unset it, or run 'lite login' to start over."
|
||||
f"Unset it, or run '{login_command}' to start over."
|
||||
)
|
||||
case KeyringUnreachable():
|
||||
return (
|
||||
"Your credential is in your OS keychain, which could not be read. Unlock it, or run "
|
||||
"'lite login' to start over."
|
||||
f"'{login_command}' to start over."
|
||||
)
|
||||
case SecretFound() | SecretMissing():
|
||||
return "Your credential could not be read from your OS keychain. Run 'lite login' to start over."
|
||||
return f"Your credential could not be read from your OS keychain. Run '{login_command}' to start over."
|
||||
|
||||
|
||||
def context_secret_vault(ctx: click.Context) -> SecretVault:
|
||||
|
|
|
|||
|
|
@ -1,16 +1,23 @@
|
|||
# stdlib imports
|
||||
import functools
|
||||
import os
|
||||
import sys
|
||||
from typing import Final
|
||||
import re
|
||||
from typing import Final, TypeAlias
|
||||
|
||||
# third party imports
|
||||
import click
|
||||
|
||||
from litellm._logging import verbose_logger
|
||||
|
||||
from .commands.auth import context_secret_vault
|
||||
|
||||
def styled_prompt():
|
||||
"""Create a styled blue box prompt for user input."""
|
||||
ShellScope: TypeAlias = tuple[str, ...]
|
||||
SHELL_WORD: Final = re.compile(r"""(?P<prefix>[\w-]+=)?(?P<quoted>"[^"]*"|'[^']*')(?P<rest>\S*)|\S+""")
|
||||
OPENING_QUOTE: Final = re.compile(r"""(?:[\w-]+=)?["']""")
|
||||
|
||||
|
||||
def styled_prompt(scope: ShellScope = ()):
|
||||
"""Create a styled blue box prompt for user input, labelled with the group the shell is in."""
|
||||
|
||||
# Get terminal height to ensure we have enough space
|
||||
try:
|
||||
|
|
@ -40,7 +47,7 @@ def styled_prompt():
|
|||
# Create styled elements
|
||||
left_border: Final = click.style(vertical, fg="blue", bold=True)
|
||||
right_border: Final = click.style(vertical, fg="blue", bold=True)
|
||||
prompt_text: Final = click.style("> ", fg="cyan", bold=True)
|
||||
prompt_text: Final = click.style(" ".join((*scope, "> ")), fg="cyan", bold=True)
|
||||
|
||||
# Display the complete box structure first to reserve space
|
||||
click.echo(click.style(top_line, fg="blue", bold=True))
|
||||
|
|
@ -139,35 +146,60 @@ def handle_special_commands(user_input: str) -> bool:
|
|||
return False
|
||||
|
||||
|
||||
def execute_command(user_input: str, ctx: click.Context):
|
||||
"""Parse and execute a command."""
|
||||
# Parse command and arguments
|
||||
parts: Final = user_input.split()
|
||||
command: Final = parts[0]
|
||||
args: Final = parts[1:] if len(parts) > 1 else []
|
||||
def _subgroup(group: click.Group | None, name: str) -> click.Group | None:
|
||||
if group is None:
|
||||
return None
|
||||
child: Final = group.commands.get(name)
|
||||
return child if isinstance(child, click.Group) else None
|
||||
|
||||
# Import cli here to avoid circular import
|
||||
from . import main
|
||||
|
||||
cli: Final = main.cli
|
||||
def _group_at(group: click.Group, path: ShellScope) -> click.Group | None:
|
||||
"""The group `path` names under `group`, or None when it names a plain command or nothing"""
|
||||
return functools.reduce(_subgroup, path, group)
|
||||
|
||||
# Check if command exists
|
||||
if command not in cli.commands:
|
||||
click.echo(f"Unknown command: {command}")
|
||||
click.echo("Type 'help' to see available commands.")
|
||||
return
|
||||
|
||||
# Execute the command
|
||||
def split_shell_line(line: str) -> ShellScope:
|
||||
"""Split on spaces and unwrap a quote that opens a word or follows its leading `name=`, keeping the rest as typed"""
|
||||
words: Final = tuple(SHELL_WORD.finditer(line))
|
||||
if any(word["quoted"] is None and OPENING_QUOTE.match(word.group()) for word in words):
|
||||
raise ValueError("No closing quotation")
|
||||
return tuple(
|
||||
word.group() if word["quoted"] is None else f"{word['prefix'] or ''}{word['quoted'][1:-1]}{word['rest']}"
|
||||
for word in words
|
||||
)
|
||||
|
||||
|
||||
def resolve_command_path(words: ShellScope, scope: ShellScope, cli: click.Group) -> ShellScope:
|
||||
"""The full command path a line names: from the top after a leading `lite`, else inside the shell's group"""
|
||||
if words[0] == "lite":
|
||||
return words[1:]
|
||||
scope_group: Final = _group_at(cli, scope)
|
||||
if scope and scope_group is not None and (words[0] in scope_group.commands or words[0].startswith("-")):
|
||||
return (*scope, *words)
|
||||
return words
|
||||
|
||||
|
||||
def scope_after(path: ShellScope, scope: ShellScope, cli: click.Group) -> ShellScope:
|
||||
"""A group typed alone becomes the scope, a command run inside the scope keeps it, anything else leaves it"""
|
||||
group: Final = _group_at(cli, path)
|
||||
if path and group is not None and not group.invoke_without_command:
|
||||
return path
|
||||
return scope if path[: len(scope)] == scope else ()
|
||||
|
||||
|
||||
def execute_command(path: ShellScope, ctx: click.Context, cli: click.Group) -> None:
|
||||
"""Run a resolved command path as `lite` would, so its usage and errors name `lite <command>`."""
|
||||
try:
|
||||
# Create a new argument list for click to parse
|
||||
sys.argv = ["lite"] + [command] + args
|
||||
|
||||
# Get the command object and invoke it
|
||||
cmd: Final = cli.commands[command]
|
||||
|
||||
# Create a new context for the subcommand
|
||||
if path[0].startswith("-"):
|
||||
cli.main(
|
||||
["--base-url", ctx.obj["base_url"], *path],
|
||||
prog_name="lite",
|
||||
obj={"secret_vault": context_secret_vault(ctx)},
|
||||
standalone_mode=False,
|
||||
)
|
||||
return
|
||||
with ctx.scope():
|
||||
cmd.main(args, parent=ctx, standalone_mode=False)
|
||||
cli.commands[path[0]].main(list(path[1:]), prog_name=path[0], parent=ctx, standalone_mode=False)
|
||||
|
||||
except click.ClickException as e:
|
||||
e.show()
|
||||
|
|
@ -180,9 +212,33 @@ def execute_command(user_input: str, ctx: click.Context):
|
|||
click.echo(f"Error executing command: {e}")
|
||||
|
||||
|
||||
def interactive_shell(ctx: click.Context):
|
||||
def run_shell_line(user_input: str, scope: ShellScope, ctx: click.Context, cli: click.Group) -> ShellScope:
|
||||
"""Run one line typed at the prompt and return the scope the next prompt starts in."""
|
||||
try:
|
||||
words: Final = split_shell_line(user_input)
|
||||
except ValueError as e:
|
||||
click.echo(f"Could not read that command: {e}")
|
||||
return scope
|
||||
if not words:
|
||||
return scope
|
||||
|
||||
path: Final = resolve_command_path(words, scope, cli)
|
||||
if not path:
|
||||
show_commands()
|
||||
return ()
|
||||
if path[0] not in cli.commands and not path[0].startswith("-"):
|
||||
click.echo(f"Unknown command: {path[0]}")
|
||||
click.echo("Type 'help' to see available commands.")
|
||||
return scope
|
||||
|
||||
execute_command(path, ctx, cli)
|
||||
return scope_after(path, scope, cli)
|
||||
|
||||
|
||||
def interactive_shell(ctx: click.Context, cli: click.Group):
|
||||
"""Run the interactive shell."""
|
||||
setup_shell(ctx)
|
||||
scope: ShellScope = () # rebind-ok: a group typed alone stays the scope for the prompts after it
|
||||
|
||||
while True:
|
||||
try:
|
||||
|
|
@ -190,7 +246,7 @@ def interactive_shell(ctx: click.Context):
|
|||
click.echo("\n") # Extra spacing
|
||||
|
||||
# Show styled prompt
|
||||
user_input = styled_prompt()
|
||||
user_input = styled_prompt(scope)
|
||||
|
||||
if not user_input:
|
||||
continue
|
||||
|
|
@ -199,10 +255,10 @@ def interactive_shell(ctx: click.Context):
|
|||
if handle_special_commands(user_input):
|
||||
if user_input.lower() in ["exit", "quit"]:
|
||||
break
|
||||
scope = ()
|
||||
continue
|
||||
|
||||
# Execute regular commands
|
||||
execute_command(user_input, ctx)
|
||||
scope = run_shell_line(user_input, scope, ctx, cli)
|
||||
|
||||
except (KeyboardInterrupt, EOFError):
|
||||
click.echo("\nGoodbye!")
|
||||
|
|
|
|||
|
|
@ -6,6 +6,7 @@ from typing import Final, Protocol
|
|||
import click
|
||||
|
||||
from litellm._version import version as litellm_version
|
||||
from litellm.litellm_core_utils.cli_token_utils import load_cli_token
|
||||
from litellm.proxy.client.health import HealthManagementClient
|
||||
|
||||
from .commands.agents import agent_commands
|
||||
|
|
@ -14,6 +15,7 @@ from .commands.auth import (
|
|||
auth_group,
|
||||
context_secret_vault,
|
||||
get_stored_api_key,
|
||||
keychain_unreadable_notice,
|
||||
login,
|
||||
logout,
|
||||
whoami,
|
||||
|
|
@ -89,7 +91,7 @@ class HideConfiguredCommandsGroup(click.Group):
|
|||
envvar="LITELLM_PROXY_URL",
|
||||
show_envvar=True,
|
||||
default=None,
|
||||
show_default="base_url from `lite config`, else http://localhost:4000",
|
||||
show_default="base_url from `lite config`, else the `lite login` server for a bare `lite` with no key, else http://localhost:4000",
|
||||
help="Base URL of the LiteLLM proxy server",
|
||||
)
|
||||
@click.option(
|
||||
|
|
@ -105,10 +107,27 @@ def cli(ctx: click.Context, show_version: bool, base_url: str | None, api_key: s
|
|||
|
||||
stored_base_url: Final = get_config_value("base_url")
|
||||
base_url_provided: Final = base_url is not None
|
||||
shell: Final = ctx.invoked_subcommand is None and not show_version
|
||||
login: Final = (
|
||||
load_cli_token(vault=context_secret_vault(ctx))
|
||||
if shell and base_url is None and not stored_base_url and api_key is None
|
||||
else None
|
||||
)
|
||||
login_base_url: Final = login.base_url if login is not None and login.key else None
|
||||
if login is not None and login.base_url and not login.key:
|
||||
click.secho(
|
||||
keychain_unreadable_notice(
|
||||
context_secret_vault(ctx), login_command=f"lite --base-url {login.base_url} login"
|
||||
),
|
||||
fg="yellow",
|
||||
err=True,
|
||||
)
|
||||
|
||||
# Normalize once here so every downstream command (login, agents, http, ...) can safely
|
||||
# do f"{base_url}/some/path" without producing a double slash.
|
||||
base_url = ((stored_base_url or "http://localhost:4000") if base_url is None else base_url).rstrip("/")
|
||||
base_url = (
|
||||
(stored_base_url or login_base_url or "http://localhost:4000") if base_url is None else base_url
|
||||
).rstrip("/")
|
||||
|
||||
# If no API key provided via flag or environment variable, try to load from saved token.
|
||||
# Pass base_url so we only use the stored key when it was issued for this server.
|
||||
|
|
@ -137,7 +156,7 @@ def cli(ctx: click.Context, show_version: bool, base_url: str | None, api_key: s
|
|||
|
||||
# If no subcommand was invoked, start interactive mode
|
||||
if ctx.invoked_subcommand is None:
|
||||
interactive_shell(ctx)
|
||||
interactive_shell(ctx, cli)
|
||||
|
||||
|
||||
@cli.command()
|
||||
|
|
|
|||
|
|
@ -151,6 +151,110 @@ def test_default_base_url_unchanged_without_config_file(cli_runner, isolated_hom
|
|||
assert "LiteLLM Proxy Server URL: http://localhost:4000" in result.output
|
||||
|
||||
|
||||
def _write_login_token(home: Path, base_url: str, key: str | None) -> None:
|
||||
token_dir = home / ".litellm"
|
||||
token_dir.mkdir(exist_ok=True)
|
||||
secret: Final = {} if key is None else {"key": key}
|
||||
(token_dir / "token.json").write_text(json.dumps({"base_url": base_url, "user_id": "u-1", **secret}))
|
||||
|
||||
|
||||
LOGIN_SERVER: Final = "https://login-gateway.example.com"
|
||||
SHELL_VERSION_THEN_QUIT: Final = "version\nquit\n"
|
||||
|
||||
|
||||
def test_bare_lite_connects_to_the_server_it_logged_in_to(cli_runner, isolated_home, requests_mock, secret_vault_factory):
|
||||
_write_login_token(isolated_home, LOGIN_SERVER, "sk-from-login")
|
||||
requests_mock.get(f"{LOGIN_SERVER}/health/readiness", json={"litellm_version": "1.2.3"})
|
||||
context: Final[dict[str, object]] = {"secret_vault": secret_vault_factory()}
|
||||
|
||||
result = cli_runner.invoke(cli, [], obj=context, input=SHELL_VERSION_THEN_QUIT)
|
||||
|
||||
assert f"Connected to LiteLLM server: {LOGIN_SERVER}" in result.output
|
||||
assert f"LiteLLM Proxy Server URL: {LOGIN_SERVER}" in result.output
|
||||
assert requests_mock.last_request.headers.get("Authorization") == "Bearer sk-from-login"
|
||||
assert context["base_url_explicit"] is False
|
||||
|
||||
|
||||
def test_a_one_shot_command_never_goes_to_the_logged_in_server(
|
||||
cli_runner, isolated_home, requests_mock, secret_vault_factory
|
||||
):
|
||||
_write_login_token(isolated_home, LOGIN_SERVER, "sk-from-login")
|
||||
requests_mock.get("http://localhost:4000/health/readiness", json={"litellm_version": "1.2.3"})
|
||||
|
||||
result = cli_runner.invoke(cli, ["version"], obj={"secret_vault": secret_vault_factory()})
|
||||
|
||||
assert result.exit_code == 0
|
||||
assert "LiteLLM Proxy Server URL: http://localhost:4000" in result.output
|
||||
assert [request.url for request in requests_mock.request_history] == ["http://localhost:4000/health/readiness"]
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
("args", "env", "config", "expected_base_url"),
|
||||
[
|
||||
(["--base-url", "https://flag-proxy.example.com"], {}, {}, "https://flag-proxy.example.com"),
|
||||
([], {"LITELLM_PROXY_URL": "https://env-proxy.example.com"}, {}, "https://env-proxy.example.com"),
|
||||
([], {}, {"base_url": "https://config-proxy.example.com"}, "https://config-proxy.example.com"),
|
||||
],
|
||||
)
|
||||
def test_a_named_server_beats_the_logged_in_one_and_never_gets_its_key(
|
||||
cli_runner, isolated_home, monkeypatch, requests_mock, secret_vault_factory, args, env, config, expected_base_url
|
||||
):
|
||||
_write_login_token(isolated_home, LOGIN_SERVER, "sk-from-login")
|
||||
_write_config_file(isolated_home, config)
|
||||
for name, value in env.items():
|
||||
monkeypatch.setenv(name, value)
|
||||
requests_mock.get(f"{expected_base_url}/health/readiness", json={"litellm_version": "1.2.3"})
|
||||
|
||||
result = cli_runner.invoke(cli, args, obj={"secret_vault": secret_vault_factory()}, input=SHELL_VERSION_THEN_QUIT)
|
||||
|
||||
assert f"Connected to LiteLLM server: {expected_base_url}" in result.output
|
||||
assert f"LiteLLM Proxy Server URL: {expected_base_url}" in result.output
|
||||
assert requests_mock.last_request.headers.get("Authorization") is None
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
("args", "env"),
|
||||
[
|
||||
(["--api-key", "sk-named-key"], {}),
|
||||
([], {"LITELLM_PROXY_API_KEY": "sk-named-key"}),
|
||||
],
|
||||
)
|
||||
def test_a_key_given_without_a_server_never_goes_to_the_logged_in_one(
|
||||
cli_runner, isolated_home, monkeypatch, requests_mock, secret_vault_factory, args, env
|
||||
):
|
||||
_write_login_token(isolated_home, LOGIN_SERVER, "sk-from-login")
|
||||
for name, value in env.items():
|
||||
monkeypatch.setenv(name, value)
|
||||
requests_mock.get("http://localhost:4000/health/readiness", json={"litellm_version": "1.2.3"})
|
||||
|
||||
result = cli_runner.invoke(cli, args, obj={"secret_vault": secret_vault_factory()}, input=SHELL_VERSION_THEN_QUIT)
|
||||
|
||||
assert "Connected to LiteLLM server: http://localhost:4000" in result.output
|
||||
assert [request.url for request in requests_mock.request_history] == ["http://localhost:4000/health/readiness"]
|
||||
assert requests_mock.last_request.headers.get("Authorization") == "Bearer sk-named-key"
|
||||
|
||||
|
||||
def test_bare_lite_says_why_it_stays_on_localhost_when_the_keychain_is_locked(
|
||||
cli_runner, isolated_home, secret_vault_factory
|
||||
):
|
||||
_write_login_token(isolated_home, LOGIN_SERVER, None)
|
||||
|
||||
result = cli_runner.invoke(cli, [], obj={"secret_vault": secret_vault_factory(available=False)}, input="quit\n")
|
||||
|
||||
assert "Connected to LiteLLM server: http://localhost:4000" in result.output
|
||||
assert f"could not be read. Unlock it, or run 'lite --base-url {LOGIN_SERVER} login'" in result.output
|
||||
|
||||
|
||||
def test_lite_goes_back_to_localhost_after_logout_without_a_keychain(cli_runner, isolated_home, secret_vault_factory):
|
||||
_write_login_token(isolated_home, LOGIN_SERVER, "sk-from-login")
|
||||
no_keychain: Final = secret_vault_factory(available=False)
|
||||
cli_runner.invoke(cli, ["logout"], obj={"secret_vault": no_keychain})
|
||||
|
||||
result = cli_runner.invoke(cli, [], obj={"secret_vault": no_keychain}, input="quit\n")
|
||||
|
||||
assert "Connected to LiteLLM server: http://localhost:4000" in result.output
|
||||
|
||||
|
||||
def test_corrupt_config_file_falls_back_to_default(cli_runner, isolated_home):
|
||||
"""A corrupt config file must never crash the CLI. Exactly one warning proves
|
||||
the config file is read once per invocation, not once per lookup."""
|
||||
|
|
|
|||
183
tests/unit/proxy/client/cli/test_interface.py
Normal file
183
tests/unit/proxy/client/cli/test_interface.py
Normal file
|
|
@ -0,0 +1,183 @@
|
|||
from collections.abc import Mapping
|
||||
from typing import Final
|
||||
|
||||
import click
|
||||
import pytest
|
||||
from click.testing import CliRunner, Result
|
||||
|
||||
from litellm.proxy.client.cli import cli
|
||||
from litellm.proxy.client.cli.interface import run_shell_line
|
||||
|
||||
GATEWAY: Final = "https://gateway.example.com"
|
||||
|
||||
|
||||
def _shell(*lines: str, env: Mapping[str, str] | None = None) -> Result:
|
||||
"""Type each line into a bare `lite` shell against the gateway, then quit"""
|
||||
typed: Final = "".join(f"{line}\n" for line in (*lines, "quit"))
|
||||
return CliRunner(env=env).invoke(
|
||||
cli, ["--base-url", GATEWAY, "--api-key", "sk-shell"], input=typed, prog_name="lite"
|
||||
)
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def credentials_endpoint(requests_mock):
|
||||
return requests_mock.get(
|
||||
f"{GATEWAY}/credentials",
|
||||
json={"credentials": [{"credential_name": "shared-openai", "credential_info": {}}]},
|
||||
)
|
||||
|
||||
|
||||
def test_a_group_typed_alone_prints_its_own_usage_and_opens_a_prompt_inside_it(credentials_endpoint):
|
||||
result: Final = _shell("credentials", "list")
|
||||
|
||||
assert "Usage: lite credentials [OPTIONS] COMMAND [ARGS]..." in result.output
|
||||
assert "lite lite" not in result.output
|
||||
assert "credentials > " in result.output
|
||||
assert "Unknown command" not in result.output
|
||||
assert credentials_endpoint.call_count == 1
|
||||
assert credentials_endpoint.last_request.headers["Authorization"] == "Bearer sk-shell"
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"lines",
|
||||
[
|
||||
("credentials list",),
|
||||
("lite credentials list",),
|
||||
("credentials", "list"),
|
||||
("credentials", "credentials list"),
|
||||
("credentials", "lite credentials list"),
|
||||
],
|
||||
)
|
||||
def test_every_spelling_of_a_group_subcommand_runs_it(lines, credentials_endpoint):
|
||||
result: Final = _shell(*lines)
|
||||
|
||||
assert "Unknown command" not in result.output
|
||||
assert "shared-openai" in result.output
|
||||
assert credentials_endpoint.call_count == 1
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"create_line",
|
||||
[
|
||||
"""create shell-made --info '{"custom_llm_provider": "openai"}' --values '{"api_key": "YWJjZA==", "organization": "b3Jn"}'""",
|
||||
"""create shell-made --info {"custom_llm_provider":"openai"} --values {"api_key":"YWJjZA==","organization":"b3Jn"}""",
|
||||
"""create shell-made --info={"custom_llm_provider":"openai"} --values={"api_key":"YWJjZA==","organization":"b3Jn"}""",
|
||||
"""create shell-made --info='{"custom_llm_provider": "openai"}' --values='{"api_key": "YWJjZA==", "organization": "b3Jn"}'""",
|
||||
],
|
||||
)
|
||||
def test_quoted_and_compact_json_arguments_reach_the_gateway_as_typed(requests_mock, create_line):
|
||||
create: Final = requests_mock.post(f"{GATEWAY}/credentials", json={"credential_name": "shell-made"})
|
||||
|
||||
_shell("credentials", create_line)
|
||||
|
||||
assert create.call_count == 1
|
||||
assert create.last_request.json() == {
|
||||
"credential_name": "shell-made",
|
||||
"credential_info": {"custom_llm_provider": "openai"},
|
||||
"credential_values": {"api_key": "YWJjZA==", "organization": "b3Jn"},
|
||||
}
|
||||
|
||||
|
||||
def test_an_unclosed_quote_runs_nothing_and_keeps_the_group(requests_mock, credentials_endpoint):
|
||||
create: Final = requests_mock.post(f"{GATEWAY}/credentials", json={})
|
||||
|
||||
result: Final = _shell("credentials", "create shell-made --info '{", "list")
|
||||
|
||||
assert "Could not read that command" in result.output
|
||||
assert create.call_count == 0
|
||||
assert credentials_endpoint.call_count == 1
|
||||
|
||||
|
||||
def test_an_unknown_command_inside_a_group_keeps_the_group(credentials_endpoint):
|
||||
result: Final = _shell("credentials", "frobnicate", "list")
|
||||
|
||||
assert "Unknown command: frobnicate" in result.output
|
||||
assert credentials_endpoint.call_count == 1
|
||||
|
||||
|
||||
@pytest.mark.parametrize("leaving_line", ["models list", "help", "lite"])
|
||||
def test_a_top_level_line_inside_a_group_leaves_the_group(leaving_line, requests_mock, credentials_endpoint):
|
||||
requests_mock.get(f"{GATEWAY}/models", json={"data": []})
|
||||
|
||||
result: Final = _shell("credentials", leaving_line, "list")
|
||||
|
||||
assert "Unknown command: list" in result.output
|
||||
assert credentials_endpoint.call_count == 0
|
||||
|
||||
|
||||
def test_help_inside_a_group_shows_that_group_and_keeps_it(credentials_endpoint):
|
||||
result: Final = _shell("credentials", "--help", "list")
|
||||
|
||||
assert "Unknown command" not in result.output
|
||||
assert result.output.count("Usage: lite credentials [OPTIONS] COMMAND [ARGS]...") == 2
|
||||
assert credentials_endpoint.call_count == 1
|
||||
|
||||
|
||||
def test_lite_with_its_own_options_inside_the_shell_runs_against_the_server_it_names(
|
||||
requests_mock, credentials_endpoint
|
||||
):
|
||||
other: Final = requests_mock.get(
|
||||
"https://other-gateway.example.com/credentials", json={"credentials": [{"credential_name": "other-openai"}]}
|
||||
)
|
||||
|
||||
result: Final = _shell("lite --base-url https://other-gateway.example.com --api-key sk-other credentials list")
|
||||
|
||||
assert "other-openai" in result.output
|
||||
assert other.last_request.headers["Authorization"] == "Bearer sk-other"
|
||||
assert credentials_endpoint.call_count == 0
|
||||
|
||||
|
||||
def test_a_lite_line_that_names_only_a_key_stays_on_the_shell_server_over_litellm_proxy_url(credentials_endpoint):
|
||||
result: Final = _shell(
|
||||
"lite --api-key sk-typed credentials list", env={"LITELLM_PROXY_URL": "https://elsewhere.example.com"}
|
||||
)
|
||||
|
||||
assert "shared-openai" in result.output
|
||||
assert credentials_endpoint.last_request.headers["Authorization"] == "Bearer sk-typed"
|
||||
|
||||
|
||||
def test_a_group_that_runs_on_its_own_does_not_become_the_scope(capsys):
|
||||
@click.group()
|
||||
def root() -> None:
|
||||
pass
|
||||
|
||||
@root.group(invoke_without_command=True)
|
||||
def wizard() -> None:
|
||||
click.echo("wizard ran")
|
||||
|
||||
@wizard.command()
|
||||
def undo() -> None:
|
||||
pass
|
||||
|
||||
scope: Final = run_shell_line("wizard", (), click.Context(root, info_name="lite", obj={}), root)
|
||||
|
||||
assert scope == ()
|
||||
assert "wizard ran" in capsys.readouterr().out
|
||||
|
||||
|
||||
def test_a_line_splits_on_spaces_and_unwraps_quotes_that_open_a_word_or_follow_an_equals_sign(capsys):
|
||||
@click.group()
|
||||
def root() -> None:
|
||||
pass
|
||||
|
||||
@root.command()
|
||||
@click.argument("words", nargs=-1)
|
||||
def echo(words: tuple[str, ...]) -> None:
|
||||
click.echo("\n".join(words))
|
||||
|
||||
run_shell_line(
|
||||
r"""echo C:\Users\me\notes.txt "two words" '{"a": 1}' info='{"a": 1}' {"b":2} it's issue#1""",
|
||||
(),
|
||||
click.Context(root, info_name="lite", obj={}),
|
||||
root,
|
||||
)
|
||||
|
||||
assert capsys.readouterr().out.splitlines() == [
|
||||
r"C:\Users\me\notes.txt",
|
||||
"two words",
|
||||
'{"a": 1}',
|
||||
'info={"a": 1}',
|
||||
'{"b":2}',
|
||||
"it's",
|
||||
"issue#1",
|
||||
]
|
||||
Loading…
Add table
Reference in a new issue