test(cli): cover credential lock failures without mutating stored tokens

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)

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)
 

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"}