refactor(vertex): fold _creds_cache_key into _sa_snapshot (simplify pass)

Final /simplify-code pass on the full diff: after the content-digest
rework, _creds_cache_key had become a wrapper with zero production
callers — get_vertex_credentials inlined the same read/except dance —
kept alive only by its own tests. One helper now owns the
(bytes, cache_key) resolution for all three cases (ADC sentinel,
readable file, unreadable fallback); prod calls it, tests target it.
No behavior change: 10/10 green, same mutation-check results.
This commit is contained in:
kshitijk4poor
2026-08-29 12:24:16 +05:30
committed by kshitij
parent 91608eb20e
commit c03c72a19e
2 changed files with 24 additions and 30 deletions
+16 -24
View File
@@ -132,24 +132,23 @@ def _read_sa_file(resolved_path: str) -> Tuple[bytes, Tuple[Any, ...]]:
return raw, (resolved_path, digest)
def _creds_cache_key(resolved_path: Optional[str]) -> Tuple[Any, ...]:
"""Cache key for the Credentials object backing *resolved_path*.
def _sa_snapshot(resolved_path: Optional[str]) -> Tuple[Optional[bytes], Tuple[Any, ...]]:
"""Resolve (bytes-or-None, cache key) for one credential attempt.
Content-digest keyed via _read_sa_file (see there for why stat
signatures are not enough for credential identity). ADC has no file
to fingerprint; it keeps a plain sentinel key and its existing
refresh/expiry handling.
A read failure falls back to the bare path: worst case we serve the
cached credentials exactly as the pre-signature code did.
- No path (ADC): (None, ("__adc__",)) — sentinel key, existing
refresh/expiry handling.
- Readable file: (bytes, (path, sha256)) via _read_sa_file — the
caller builds credentials from the SAME bytes the key fingerprints.
- Unreadable file: (None, (path,)) — bare-path key, and the caller
falls back to the SDK's own file read: byte-for-byte the
pre-signature behavior.
"""
if not resolved_path:
return ("__adc__",)
return None, ("__adc__",)
try:
_, key = _read_sa_file(resolved_path)
return key
return _read_sa_file(resolved_path)
except OSError:
return (resolved_path,)
return None, (resolved_path,)
def get_vertex_credentials(credentials_path: Optional[str] = None) -> Tuple[Optional[str], Optional[str]]:
@@ -163,17 +162,10 @@ def get_vertex_credentials(credentials_path: Optional[str] = None) -> Tuple[Opti
return None, None
resolved_path = _resolve_credentials_path(credentials_path)
sa_raw: Optional[bytes] = None
if resolved_path:
try:
# One read serves both the cache key and (on a miss) credential
# construction, so the credentials always match the bytes the
# key fingerprints — no stat/read or read/read TOCTOU.
sa_raw, cache_key = _read_sa_file(resolved_path)
except OSError:
cache_key = (resolved_path,)
else:
cache_key = ("__adc__",)
# One read serves both the cache key and (on a miss) credential
# construction, so the credentials always match the bytes the key
# fingerprints — no stat/read or read/read TOCTOU.
sa_raw, cache_key = _sa_snapshot(resolved_path)
try:
cached = _creds_cache.get(cache_key)
+8 -6
View File
@@ -207,18 +207,20 @@ def test_sa_file_rotation_invalidates_creds_cache(vertex_adapter, monkeypatch, t
assert vertex_adapter._creds_cache[key2][0] is not creds_obj_1
def test_creds_cache_stat_failure_falls_back_to_path_key(vertex_adapter, monkeypatch):
"""If the credentials file cannot be stat'ed the key degrades to the bare
path — same behavior as the pre-signature cache, never an exception."""
key = vertex_adapter._creds_cache_key("/nonexistent/sa.json")
def test_creds_cache_read_failure_falls_back_to_path_key(vertex_adapter, monkeypatch):
"""If the credentials file cannot be read the key degrades to the bare
path (no bytes) — same behavior as the pre-signature cache, never an
exception."""
raw, key = vertex_adapter._sa_snapshot("/nonexistent/sa.json")
assert raw is None
assert key == ("/nonexistent/sa.json",)
def test_adc_cache_key_is_stable_sentinel(vertex_adapter):
"""ADC has no file to fingerprint; both None and empty resolve to the
same sentinel so repeated ADC calls share one cache entry."""
assert vertex_adapter._creds_cache_key(None) == ("__adc__",)
assert vertex_adapter._creds_cache_key("") == ("__adc__",)
assert vertex_adapter._sa_snapshot(None) == (None, ("__adc__",))
assert vertex_adapter._sa_snapshot("") == (None, ("__adc__",))
def test_adc_failure_retries_with_late_added_sa_file(vertex_adapter, monkeypatch, tmp_path):