From 128930dcda0a3fe5b021fff644f6dc21e62f6728 Mon Sep 17 00:00:00 2001 From: Jon Tang Date: Fri, 23 Oct 2026 11:56:05 -0700 Subject: [PATCH] test(cli): cover credential lock failures without mutating stored tokens --- .../test_cli_token_utils.py | 57 +++++++++++++++++++ .../proxy/client/cli/test_auth_commands.py | 25 ++++++++ .../unit/proxy/client/cli/test_pkce_login.py | 31 +++++++++- 3 files changed, 112 insertions(+), 1 deletion(-) diff --git a/tests/unit/litellm_core_utils/test_cli_token_utils.py b/tests/unit/litellm_core_utils/test_cli_token_utils.py index 7e7eee5373f..1f398017817 100644 --- a/tests/unit/litellm_core_utils/test_cli_token_utils.py +++ b/tests/unit/litellm_core_utils/test_cli_token_utils.py @@ -6,6 +6,8 @@ import sys import tempfile import threading import time +from pathlib import Path +from typing import Final import pytest @@ -32,6 +34,7 @@ from litellm.litellm_core_utils.cli_token_utils import ( CredentialNotRecorded, CredentialNotSaved, clear_cli_token, + cli_token_lock, get_cli_token_file_path, get_litellm_gateway_api_key, is_cli_token_fresh, @@ -39,6 +42,8 @@ from litellm.litellm_core_utils.cli_token_utils import ( save_cli_token, ) +from tests.unit.litellm_core_utils.fake_secret_vault import FakeSecretVault + SERVER = "https://proxy.example.com" OTHER_SERVER = "https://other-proxy.example.com" @@ -389,6 +394,58 @@ class TestLoadCliToken: assert load_cli_token(vault=secret_vault_factory(blob=_blob())) is None +def test_load_does_not_migrate_the_credential_when_the_lock_cannot_be_opened( + isolated_home: Path, + secret_vault_factory: type[FakeSecretVault], +) -> None: + path: Final = _write_legacy_file(isolated_home, timestamp=1000.0) + original: Final = path.read_bytes() + vault: Final = secret_vault_factory() + Path(cli_token_lock().lock_file).mkdir() + + assert load_cli_token(vault=vault) is None + assert path.read_bytes() == original + assert vault.reads == 0 + assert vault.writes == [] + + +def test_save_preserves_both_stores_when_the_lock_cannot_be_opened( + isolated_home: Path, + secret_vault_factory: type[FakeSecretVault], +) -> None: + path: Final = _write_legacy_file(isolated_home, timestamp=1000.0) + original: Final = path.read_bytes() + vault: Final = secret_vault_factory(blob=_blob()) + original_secret: Final = vault.blob + Path(cli_token_lock().lock_file).mkdir() + + outcome: Final = save_cli_token( + CliTokenRecord(base_url=SERVER, key="sk-replacement", timestamp=2000.0), vault=vault + ) + assert isinstance(outcome, CredentialNotSaved) + assert path.read_bytes() == original + assert vault.blob == original_secret + assert vault.reads == 0 + assert vault.writes == [] + + +def test_clear_preserves_both_stores_when_the_lock_cannot_be_opened( + isolated_home: Path, + secret_vault_factory: type[FakeSecretVault], +) -> None: + path: Final = _write_legacy_file(isolated_home, timestamp=1000.0) + original: Final = path.read_bytes() + vault: Final = secret_vault_factory(blob=_blob()) + original_secret: Final = vault.blob + Path(cli_token_lock().lock_file).mkdir() + + outcome: Final = clear_cli_token(vault=vault) + assert isinstance(outcome, CredentialNotCleared) + assert path.read_bytes() == original + assert vault.blob == original_secret + assert vault.erases == 0 + + class TestGetLitellmGatewayApiKey: def test_returns_the_vault_secret_when_the_origin_matches(self, isolated_home, secret_vault_factory): _write_metadata_only_file(isolated_home) diff --git a/tests/unit/proxy/client/cli/test_auth_commands.py b/tests/unit/proxy/client/cli/test_auth_commands.py index 7979c2eaec1..f8db47f47f5 100644 --- a/tests/unit/proxy/client/cli/test_auth_commands.py +++ b/tests/unit/proxy/client/cli/test_auth_commands.py @@ -22,6 +22,7 @@ from litellm.litellm_core_utils.cli_token_utils import ( CliTokenRecord, CredentialNotRecorded, CredentialNotSaved, + cli_token_lock, save_cli_token, ) from litellm.proxy.client.cli import cli @@ -2181,6 +2182,30 @@ def test_print_token_uses_the_saved_key_when_the_initializer_did_not_get_one( assert result.output == f"{record.key}\n" +@responses.activate +def test_logout_reports_a_lock_failure_without_revoking_or_clearing_the_credential( + isolated_home: Path, + secret_vault_factory: type[FakeSecretVault], +) -> None: + _write_home_json(isolated_home, "token.json", _pkce_record(timestamp=1000.0, expires_at=0.0)) + responses.add(responses.POST, f"{PKCE_BASE_URL}/revoke", json={}, status=200) + path: Final = isolated_home / ".litellm" / "token.json" + original: Final = path.read_bytes() + vault: Final = secret_vault_factory(blob=_secret_blob("https://test.example.com", "sk-fixture-vault")) + original_secret: Final = vault.blob + Path(cli_token_lock().lock_file).mkdir() + + result: Final = CliRunner().invoke(logout, obj={"secret_vault": vault}) + assert result.exit_code == 1 + assert result.stdout == "" + assert "Could not lock the CLI credential store. No credential operation was started" in result.stderr + assert len(responses.calls) == 0 + assert path.read_bytes() == original + assert vault.blob == original_secret + assert vault.reads == 0 + assert vault.erases == 0 + + def test_print_token_without_a_context_object_reports_no_login_when_nothing_is_stored(isolated_home): result = CliRunner().invoke(print_token) diff --git a/tests/unit/proxy/client/cli/test_pkce_login.py b/tests/unit/proxy/client/cli/test_pkce_login.py index 7253fca1005..84f0580f64f 100644 --- a/tests/unit/proxy/client/cli/test_pkce_login.py +++ b/tests/unit/proxy/client/cli/test_pkce_login.py @@ -4,17 +4,20 @@ import hashlib import json import re import socket +import sys import threading import time import urllib.error import urllib.request from base64 import urlsafe_b64encode +from pathlib import Path +from typing import Final from urllib.parse import parse_qs, urlparse import pytest import requests -from litellm.litellm_core_utils.cli_token_utils import is_cli_token_fresh +from litellm.litellm_core_utils.cli_token_utils import cli_token_lock, is_cli_token_fresh from litellm.proxy.client.cli.commands.pkce_login import ( CallbackCode, CallbackDenied, @@ -708,6 +711,32 @@ def test_fresh_api_key_does_not_renew_a_cleared_or_replaced_login(current): assert http.calls == [] +@pytest.mark.parametrize(("now", "expected"), [(999_950.0, "sk-cli-old"), (1_000_001.0, None)]) +def test_fresh_api_key_reports_a_lock_failure_without_reading_or_spending_the_refresh_token( + capsys: pytest.CaptureFixture[str], + now: float, + expected: str | None, +) -> None: + Path(cli_token_lock().lock_file).mkdir() + http: Final = _FakeHttp({("POST", f"{BASE}/token"): _FakeResponse(200, TOKEN_JSON)}) + key: Final = _fresh( + STORED, + lambda _: pytest.fail("Unexpected credential save"), + http, + reload=lambda: pytest.fail("Credential read without the lock"), + now=lambda: now, + warn=lambda message: print(message, file=sys.stderr), + ) + assert key == expected + assert http.calls == [] + captured: Final = capsys.readouterr() + assert captured.out == "" + assert ( + captured.err + == "Could not renew the key: the CLI credential store could not be locked; no refresh was attempted\n" + ) + + def test_fresh_api_key_without_refresh_inputs_expires_like_a_classic_key(): http = _FakeHttp() no_refresh = {key: value for key, value in STORED.items() if key != "refresh_token"}