diff --git a/src/tower/_storage.py b/src/tower/_storage.py index 9975a171..c0db1286 100644 --- a/src/tower/_storage.py +++ b/src/tower/_storage.py @@ -3,8 +3,10 @@ import hashlib import logging import time -from dataclasses import dataclass +from concurrent.futures import Future +from dataclasses import dataclass, field from datetime import datetime, timedelta, timezone +from threading import Lock from typing import TYPE_CHECKING import httpx @@ -15,6 +17,7 @@ from ._context import TowerContext from .exceptions import ( StorageConnectionError, + StorageError, StorageInvalidCredentialError, StorageMissingAuthenticationError, ) @@ -51,6 +54,12 @@ logger = logging.getLogger("tower.storage") +@dataclass(frozen=True, slots=True) +class _AccessCacheKey: + name: str + mode: str + + def _auth_from_context(context: TowerContext) -> tuple[str, str, str]: if context.jwt is not None: token = context.jwt @@ -69,7 +78,7 @@ def _auth_from_context(context: TowerContext) -> tuple[str, str, str]: if token.strip() == INVALID_CREDENTIAL_SENTINEL: raise StorageInvalidCredentialError( - f"{source} contains the {INVALID_CREDENTIAL_SENTINEL!r} placeholder, " + f"{source} contains the {INVALID_CREDENTIAL_SENTINEL} placeholder, " "not a usable Tower credential." ) if not token.strip(): @@ -104,6 +113,33 @@ def _auth_hash(token: str, auth_header_name: str, prefix: str) -> str: return hashlib.sha256(presented_auth.encode("utf-8")).hexdigest() +@dataclass(frozen=True) +class _ResolvedCatalogAccess: + target_environment: str + catalog_environment: str + catalog_name: str + catalog_uri: str + warehouse: str + mode: str + oauth_token: str = field(repr=False) + expires_at: datetime + + def is_inherited(self) -> bool: + return self.catalog_environment != self.target_environment + + def is_usable(self, now: datetime) -> bool: + return self.expires_at - now > CREDENTIAL_REFRESH_WINDOW + + def to_credentials(self) -> CatalogCredentials: + return CatalogCredentials( + catalog_uri=self.catalog_uri, + expires_at=self.expires_at, + mode=self.mode, + oauth_token=self.oauth_token, + warehouse=self.warehouse, + ) + + class _StorageResolver: """Private Tower configuration and authentication for catalog resolution.""" @@ -120,18 +156,20 @@ def __init__( if environment is not None and not environment.strip(): raise ValueError("environment must not be blank") - self._target_environment = ( + self._target_environment: str = ( environment or context.environment or DEFAULT_ENVIRONMENT_NAME ) - self._base_url = _api_base_url(context.tower_url) + self._base_url: str = _api_base_url(context.tower_url) + + self._token: str + self._auth_header_name: str + self._auth_prefix: str self._token, self._auth_header_name, self._auth_prefix = _auth_from_context( context ) - self._auth_hash = _auth_hash( - self._token, - self._auth_header_name, - self._auth_prefix, - ) + self._access_cache: dict[_AccessCacheKey, _ResolvedCatalogAccess] = {} + self._access_flights: dict[_AccessCacheKey, Future[_ResolvedCatalogAccess]] = {} + self._access_lock: Lock = Lock() def _new_client(self) -> AuthenticatedClient: return _new_tower_control_plane_client( @@ -142,6 +180,79 @@ def _new_client(self) -> AuthenticatedClient: timeout=DEFAULT_STORAGE_TIMEOUT_SECONDS, ) + def _resolve_catalog_access( + self, + name: str, + mode: str, + ) -> _ResolvedCatalogAccess: + mode = _normalize_mode(mode) + target_environment = self._target_environment + # Host, authentication, and target are fixed for this resolver. + cache_key = _AccessCacheKey(name=name, mode=mode) + + with self._access_lock: + cached = self._access_cache.get(cache_key) + if cached is not None and cached.is_usable(datetime.now(timezone.utc)): + return cached + _ = self._access_cache.pop(cache_key, None) + + flight = self._access_flights.get(cache_key) + if flight is None: + flight = Future() + self._access_flights[cache_key] = flight + should_vend = True + else: + should_vend = False + + if not should_vend: + return flight.result() + + try: + response = _vend_with_default_catalog_fallback( + self, + name, + mode, + ) + credentials = response.credentials + if credentials.mode != mode: + raise StorageError( + f"Tower returned {credentials.mode} credentials after " + f"{mode} access was requested." + ) + if response.environment not in ( + target_environment, + DEFAULT_ENVIRONMENT_NAME, + ): + raise StorageError( + f"Tower resolved catalog {name} from unexpected environment " + f"{response.environment}." + ) + + access = _ResolvedCatalogAccess( + target_environment=target_environment, + catalog_environment=response.environment, + catalog_name=name, + catalog_uri=credentials.catalog_uri, + warehouse=credentials.warehouse, + mode=credentials.mode, + oauth_token=credentials.oauth_token, + expires_at=_ensure_aware(credentials.expires_at), + ) + cacheable = access.is_usable(datetime.now(timezone.utc)) + except BaseException as error: + with self._access_lock: + flight.set_exception(error) + _ = self._access_flights.pop(cache_key, None) + raise + + with self._access_lock: + if cacheable: + self._access_cache[cache_key] = access + flight.set_result(access) + _ = self._access_flights.pop(cache_key, None) + + return access + def _request_catalog_credentials( self, name: str, @@ -173,13 +284,12 @@ def _api_base_url(tower_url: str) -> str: return str(url.copy_with(path="/v1", query=None, fragment=None)) -@dataclass -class _CachedCredentials: - credentials: CatalogCredentials - - def is_usable(self, now: datetime) -> bool: - expires_at = _ensure_aware(self.credentials.expires_at) - return now < expires_at - CREDENTIAL_REFRESH_WINDOW +@dataclass(frozen=True, slots=True) +class _CatalogTypeCacheKey: + base_url: str + auth_hash: str + name: str + environment: str @dataclass @@ -188,8 +298,7 @@ class _CachedCatalogType: retry_at: float | None = None -_credential_cache: dict[tuple[str, str, str, str, str], _CachedCredentials] = {} -_catalog_type_cache: dict[tuple[str, str, str, str], _CachedCatalogType] = {} +_catalog_type_cache: dict[_CatalogTypeCacheKey, _CachedCatalogType] = {} def get_tower_catalog( @@ -210,18 +319,8 @@ def get_tower_catalog_credentials( mode: str = "read", ) -> CatalogCredentials: storage_resolver = _StorageResolver(environment=environment) - mode = _normalize_mode(mode) - cache_key = _cache_key(storage_resolver, name, mode) - - now = datetime.now(timezone.utc) - _prune_credential_cache(now) - cached = _credential_cache.get(cache_key) - if cached is not None and cached.is_usable(now): - return cached.credentials - - credentials = _vend_with_default_catalog_fallback(storage_resolver, name, mode) - _credential_cache[cache_key] = _CachedCredentials(credentials) - return credentials + access = storage_resolver._resolve_catalog_access(name, mode) + return access.to_credentials() def load_vended_catalog(name: str, credentials: CatalogCredentials) -> Catalog: @@ -240,7 +339,7 @@ def _vend_with_default_catalog_fallback( storage_resolver: _StorageResolver, name: str, mode: str, -) -> CatalogCredentials: +) -> VendCatalogCredentialsResponse: environment = storage_resolver._target_environment result = storage_resolver._request_catalog_credentials(name, mode) if not _is_not_found(result): @@ -258,7 +357,7 @@ def _vend_with_default_catalog_fallback( return _unwrap_vend_result(result, name, environment) raise RuntimeError( - f"Tower catalog {name!r} does not exist in environment {environment!r}." + f"Tower catalog {name} does not exist in environment {environment}." ) @@ -271,7 +370,12 @@ def _describe_tower_catalog_type( token, auth_header_name, prefix = _auth_from_context(ctx) base_url = _api_base_url(ctx.tower_url) auth_hash = _auth_hash(token, auth_header_name, prefix) - cache_key = (base_url, auth_hash, name, environment) + cache_key = _CatalogTypeCacheKey( + base_url=base_url, + auth_hash=auth_hash, + name=name, + environment=environment, + ) cached = _catalog_type_cache.get(cache_key) if cached is not None: if cached.retry_at is None: @@ -346,45 +450,23 @@ def _unwrap_vend_result( result: ErrorModel | VendCatalogCredentialsResponse | None, name: str, environment: str, -) -> CatalogCredentials: +) -> VendCatalogCredentialsResponse: if isinstance(result, VendCatalogCredentialsResponse): - return result.credentials + return result if isinstance(result, ErrorModel): detail = _error_text(result) raise RuntimeError( - f"Failed to vend credentials for Tower catalog {name!r} " - f"in environment {environment!r}: {detail}" + f"Failed to vend credentials for Tower catalog {name} " + f"in environment {environment}: {detail}" ) raise RuntimeError( - f"Failed to vend credentials for Tower catalog {name!r} " - f"in environment {environment!r}." + f"Failed to vend credentials for Tower catalog {name} " + f"in environment {environment}." ) -def _cache_key( - storage_resolver: _StorageResolver, - name: str, - mode: str, -) -> tuple[str, str, str, str, str]: - return ( - storage_resolver._base_url, - storage_resolver._auth_hash, - name, - storage_resolver._target_environment, - mode, - ) - - -def _prune_credential_cache(now: datetime) -> None: - expired_keys = [ - key for key, cached in _credential_cache.items() if not cached.is_usable(now) - ] - for key in expired_keys: - _credential_cache.pop(key, None) - - def _normalize_mode(mode: str) -> str: if mode not in ("read", "read-write"): raise ValueError("mode must be 'read' or 'read-write'") @@ -416,6 +498,5 @@ def _ensure_aware(value: datetime) -> datetime: return value.astimezone(timezone.utc) -def _clear_credential_cache() -> None: - _credential_cache.clear() +def _clear_catalog_type_cache() -> None: _catalog_type_cache.clear() diff --git a/tests/tower/test_storage.py b/tests/tower/test_storage.py index 7f4307fd..2a0a136f 100644 --- a/tests/tower/test_storage.py +++ b/tests/tower/test_storage.py @@ -1,5 +1,7 @@ +from concurrent.futures import ThreadPoolExecutor from datetime import datetime, timedelta, timezone from http import HTTPStatus +from threading import Event import httpx import pytest @@ -8,6 +10,7 @@ from tower._context import TowerContext from tower.exceptions import ( StorageConnectionError, + StorageError, StorageInvalidCredentialError, StorageMissingAuthenticationError, ) @@ -20,6 +23,39 @@ ) +def make_vend_response( + environment, + token, + expires_at, + mode="read", +): + return VendCatalogCredentialsResponse( + credentials=CatalogCredentials( + catalog_uri="https://catalog.example.com", + expires_at=expires_at, + mode=mode, + oauth_token=token, + warehouse="warehouse-id", + ), + environment=environment, + ) + + +def script_vend(monkeypatch, results): + results = iter(results) + calls = [] + + def vend(client, name, mode): + calls.append((name, mode)) + result = next(results) + if isinstance(result, BaseException): + raise result + return result + + monkeypatch.setattr(_storage._StorageResolver, "_request_catalog_credentials", vend) + return calls + + @pytest.fixture(autouse=True) def isolate_tower_environment(monkeypatch, tmp_path): for name in ( @@ -115,13 +151,6 @@ def test_auth_hash_includes_how_the_credential_is_presented(): def test_missing_auth_fails_before_cache_or_vend(monkeypatch): - _storage._clear_credential_cache() - - monkeypatch.setattr( - _storage, - "_prune_credential_cache", - lambda now: pytest.fail("cache access must not run without authentication"), - ) monkeypatch.setattr( _storage.vend_catalog_credentials_api, "sync", @@ -152,9 +181,8 @@ def vend(**kwargs): monkeypatch.setattr(_storage.vend_catalog_credentials_api, "sync", vend) - result = _storage._StorageResolver()._request_catalog_credentials( - "analytics", "read" - ) + resolver = _storage._StorageResolver() + result = resolver._request_catalog_credentials("analytics", "read") assert result is rejected assert len(vend_calls) == 1 @@ -181,9 +209,8 @@ def vend(*, name, client, environment, body): monkeypatch.setattr(_storage.vend_catalog_credentials_api, "sync", vend) - result = _storage._StorageResolver( - environment="production" - )._request_catalog_credentials("analytics", "read") + resolver = _storage._StorageResolver(environment="production") + result = resolver._request_catalog_credentials("analytics", "read") assert result is response assert captured == { @@ -198,7 +225,7 @@ def vend(*, name, client, environment, body): def test_describe_and_vend_prefer_jwt_when_both_auth_vars_are_set(monkeypatch): - _storage._clear_credential_cache() + _storage._clear_catalog_type_cache() monkeypatch.setenv("TOWER_URL", "https://api.example.com") monkeypatch.setenv("TOWER_ENVIRONMENT", "production") monkeypatch.setenv("TOWER_API_KEY", "ambient-api-key") @@ -241,10 +268,8 @@ def vend(*, client, **kwargs): _storage._describe_tower_catalog_type(ctx, "analytics", "production") == _storage.TOWER_CATALOG_TYPE ) - assert ( - _storage._StorageResolver()._request_catalog_credentials("analytics", "read") - is vended - ) + resolver = _storage._StorageResolver() + assert resolver._request_catalog_credentials("analytics", "read") is vended assert captured_auth == [ ("describe", "Bearer ambient-jwt", None), ("vend", "Bearer ambient-jwt", None), @@ -261,7 +286,6 @@ def test_storage_resolver_allows_explicit_http_tower_url(monkeypatch): def test_get_tower_catalog_credentials_allows_http_and_reaches_vend(monkeypatch): - _storage._clear_credential_cache() monkeypatch.setenv("TOWER_URL", "http://localhost:9000") monkeypatch.setenv("TOWER_ENVIRONMENT", "production") monkeypatch.setenv("TOWER_API_KEY", "api-key") @@ -287,7 +311,7 @@ def vend(*, name, client, environment, body): result = _storage.get_tower_catalog_credentials("analytics") - assert result is credentials + assert result == credentials assert vend_calls == [ ( "analytics", @@ -316,11 +340,6 @@ def test_get_tower_catalog_credentials_rejects_invalid_mode_before_cache_or_vend monkeypatch, ): monkeypatch.setenv("TOWER_API_KEY", "api-key") - monkeypatch.setattr( - _storage, - "_prune_credential_cache", - lambda now: pytest.fail("cache access must not run for an invalid mode"), - ) monkeypatch.setattr( _storage.vend_catalog_credentials_api, "sync", @@ -350,7 +369,8 @@ def vend(*, client, **kwargs): monkeypatch.setattr(_storage.vend_catalog_credentials_api, "sync", vend) with pytest.raises(StorageConnectionError) as error: - _storage._StorageResolver()._request_catalog_credentials("analytics", "read") + resolver = _storage._StorageResolver() + resolver._request_catalog_credentials("analytics", "read") assert error.value.__cause__ is cause assert clients[0]._client is not None @@ -369,79 +389,203 @@ def test_storage_resolver_uses_runtime_environment_only_as_target_config(monkeyp ) -def test_get_tower_catalog_credentials_caches_vended_credentials(monkeypatch): - _storage._clear_credential_cache() +def test_access_cache_is_owned_by_resolver(monkeypatch): monkeypatch.setenv("TOWER_URL", "https://api.example.com") monkeypatch.setenv("TOWER_ENVIRONMENT", "production") monkeypatch.setenv("TOWER_API_KEY", "api-key") - expires_at = datetime.now(timezone.utc) + timedelta(hours=1) - credentials = CatalogCredentials( - catalog_uri="https://catalog.example.com", - expires_at=expires_at, - mode="read", - oauth_token="oauth-token", - warehouse="warehouse-id", + first_resolver = _storage._StorageResolver() + second_resolver = _storage._StorageResolver() + calls = script_vend( + monkeypatch, + [ + make_vend_response( + environment="production", + token=f"provider-token-{number}", + expires_at=datetime.now(timezone.utc) + timedelta(hours=1), + ) + for number in (1, 2) + ], ) - calls = [] - def vend(client, name, mode): - calls.append((name, client._target_environment, mode)) - return VendCatalogCredentialsResponse( - credentials=credentials, - environment=client._target_environment, - ) + first = first_resolver._resolve_catalog_access("analytics", "read") + first_again = first_resolver._resolve_catalog_access("analytics", "read") + second = second_resolver._resolve_catalog_access("analytics", "read") + second_again = second_resolver._resolve_catalog_access("analytics", "read") + + assert first == first_again + assert second == second_again + assert {first.oauth_token, second.oauth_token} == { + "provider-token-1", + "provider-token-2", + } + assert len(calls) == 2 - monkeypatch.setattr(_storage._StorageResolver, "_request_catalog_credentials", vend) - first = _storage.get_tower_catalog_credentials("default") - second = _storage.get_tower_catalog_credentials("default") +def test_one_shot_credential_loads_do_not_share_cache(monkeypatch): + monkeypatch.setenv("TOWER_API_KEY", "api-key") + calls = script_vend( + monkeypatch, + [ + make_vend_response( + environment="default", + token=f"provider-token-{number}", + expires_at=datetime.now(timezone.utc) + timedelta(hours=1), + ) + for number in (1, 2) + ], + ) - assert first is credentials - assert second is credentials - assert calls == [("default", "production", "read")] + first = _storage.get_tower_catalog_credentials("analytics") + second = _storage.get_tower_catalog_credentials("analytics") + assert first.oauth_token == "provider-token-1" + assert second.oauth_token == "provider-token-2" + assert len(calls) == 2 -def test_get_tower_catalog_credentials_prunes_expired_cache_entries(monkeypatch): - _storage._clear_credential_cache() - monkeypatch.setenv("TOWER_URL", "https://api.example.com") + +def test_near_expiry_access_retains_inherited_and_local_identity(monkeypatch): monkeypatch.setenv("TOWER_ENVIRONMENT", "production") monkeypatch.setenv("TOWER_API_KEY", "api-key") - expired_credentials = CatalogCredentials( - catalog_uri="https://old-catalog.example.com", - expires_at=datetime.now(timezone.utc) - timedelta(minutes=1), - mode="read", - oauth_token="old-oauth-token", - warehouse="old-warehouse-id", - ) - fresh_credentials = CatalogCredentials( - catalog_uri="https://catalog.example.com", - expires_at=datetime.now(timezone.utc) + timedelta(hours=1), - mode="read", - oauth_token="oauth-token", - warehouse="warehouse-id", - ) - storage_resolver = _storage._StorageResolver() - expired_key = _storage._cache_key(storage_resolver, "stale", "read") - _storage._credential_cache[expired_key] = _storage._CachedCredentials( - expired_credentials - ) + responses = [ + make_vend_response( + environment="default", + token="inherited-token", + expires_at=datetime.now(timezone.utc) + + _storage.CREDENTIAL_REFRESH_WINDOW + - timedelta(seconds=1), + ), + make_vend_response( + environment="production", + token="local-token", + expires_at=datetime.now(timezone.utc) + timedelta(hours=1), + ), + ] + calls = script_vend(monkeypatch, responses) + + resolver = _storage._StorageResolver() + inherited = resolver._resolve_catalog_access("analytics", "read") + local = resolver._resolve_catalog_access("analytics", "read") + resolver._resolve_catalog_access("analytics", "read") + + assert inherited.target_environment == "production" + assert inherited.catalog_environment == "default" + assert inherited.is_inherited() is True + assert inherited.oauth_token == "inherited-token" + assert local.catalog_environment == "production" + assert local.is_inherited() is False + assert local.oauth_token == "local-token" + assert len(calls) == 2 + + +def test_concurrent_access_shares_one_vend_request(monkeypatch): + monkeypatch.setenv("TOWER_API_KEY", "api-key") + timeout = 10 + vend_started = Event() + waiter_joined = Event() + release_vend = Event() + calls = [] + + class ObservableFuture(_storage.Future): + def result(self, timeout=None): + waiter_joined.set() + return super().result(timeout) def vend(client, name, mode): - return VendCatalogCredentialsResponse( - credentials=fresh_credentials, - environment=client._target_environment, + calls.append((name, mode)) + vend_started.set() + assert release_vend.wait(timeout=timeout) + return make_vend_response( + environment="default", + token="shared-token", + expires_at=datetime.now(timezone.utc) + timedelta(hours=1), ) + def resolve(resolver): + return resolver._resolve_catalog_access("analytics", "read") + + monkeypatch.setattr(_storage, "Future", ObservableFuture) monkeypatch.setattr(_storage._StorageResolver, "_request_catalog_credentials", vend) - result = _storage.get_tower_catalog_credentials("default") + resolver = _storage._StorageResolver() + with ThreadPoolExecutor(max_workers=2) as executor: + leader = executor.submit(resolve, resolver) + started = vend_started.wait(timeout=timeout) + waiter = executor.submit(resolve, resolver) + joined = waiter_joined.wait(timeout=timeout) + release_vend.set() + accesses = [ + leader.result(timeout=timeout), + waiter.result(timeout=timeout), + ] + + assert started + assert joined + assert len(calls) == 1 + assert all(access == accesses[0] for access in accesses) + + +def test_failed_vend_is_not_cached(monkeypatch): + monkeypatch.setenv("TOWER_API_KEY", "api-key") + calls = script_vend( + monkeypatch, + [ + RuntimeError("vend failed"), + make_vend_response( + environment="default", + token="retry-token", + expires_at=datetime.now(timezone.utc) + timedelta(hours=1), + ), + ], + ) + + resolver = _storage._StorageResolver() + with pytest.raises(RuntimeError, match="vend failed"): + resolver._resolve_catalog_access("analytics", "read") + access = resolver._resolve_catalog_access("analytics", "read") + + assert access.oauth_token == "retry-token" + assert len(calls) == 2 + + +@pytest.mark.parametrize( + ("environment", "mode", "message"), + [ + ("staging", "read", "unexpected environment"), + ("default", "read-write", "after read access was requested"), + ], +) +def test_inconsistent_vend_response_is_not_cached( + monkeypatch, + environment, + mode, + message, +): + monkeypatch.setenv("TOWER_API_KEY", "api-key") + responses = [ + make_vend_response( + environment=environment, + token="invalid-token", + expires_at=datetime.now(timezone.utc) + timedelta(hours=1), + mode=mode, + ), + make_vend_response( + environment="default", + token="valid-token", + expires_at=datetime.now(timezone.utc) + timedelta(hours=1), + ), + ] + calls = script_vend(monkeypatch, responses) + + resolver = _storage._StorageResolver() + with pytest.raises(StorageError, match=message): + resolver._resolve_catalog_access("analytics", "read") + access = resolver._resolve_catalog_access("analytics", "read") - assert result is fresh_credentials - assert expired_key not in _storage._credential_cache + assert access.oauth_token == "valid-token" + assert len(calls) == 2 def test_default_catalog_retries_reuse_client_auth_snapshot(monkeypatch): - _storage._clear_credential_cache() monkeypatch.setenv("TOWER_API_KEY", "operation-token") credentials = CatalogCredentials( catalog_uri="https://catalog.example.com", @@ -483,7 +627,7 @@ def legacy_default(*, client): result = _storage.get_tower_catalog_credentials("default") - assert result is credentials + assert result == credentials assert vend_tokens == ["operation-token"] * 3 assert legacy_tokens == ["operation-token"] * 2 assert len({id(client) for client in clients}) == 5 @@ -494,7 +638,7 @@ def legacy_default(*, client): def test_describe_tower_catalog_type_uses_timeout_and_recovers_after_cooldown( monkeypatch, ): - _storage._clear_credential_cache() + _storage._clear_catalog_type_cache() ctx = TowerContext( tower_url="https://api.example.com", environment="production", diff --git a/tests/tower/test_tables.py b/tests/tower/test_tables.py index 4bec5997..acd428f7 100644 --- a/tests/tower/test_tables.py +++ b/tests/tower/test_tables.py @@ -155,7 +155,7 @@ def sql_catalog(): def test_string_catalog_precedence( monkeypatch, tower_credentials, catalog_type, has_pyiceberg_config, expected_source ): - _storage._clear_credential_cache() + _storage._clear_catalog_type_cache() patch_tower_context(monkeypatch) vend_catalog = FakeCatalog("vend") configured_catalog = FakeCatalog("configured") @@ -271,7 +271,7 @@ def unexpected_call(*args, **kwargs): def test_no_tower_auth_preserves_ambient_pyiceberg_catalog(monkeypatch): - _storage._clear_credential_cache() + _storage._clear_catalog_type_cache() patch_tower_context(monkeypatch, api_key=None) monkeypatch.setenv( "PYICEBERG_CATALOG__S3_TABLES__URI", "https://s3tables.example.com" @@ -298,7 +298,7 @@ def load_catalog(name): def test_managed_catalog_vend_failure_does_not_fall_back_to_pyiceberg(monkeypatch): - _storage._clear_credential_cache() + _storage._clear_catalog_type_cache() patch_tower_context(monkeypatch) calls = [] @@ -334,7 +334,7 @@ def load_catalog(name): def test_external_catalog_write_mode_keeps_ambient_pyiceberg_catalog( monkeypatch, catalog_type ): - _storage._clear_credential_cache() + _storage._clear_catalog_type_cache() patch_tower_context(monkeypatch) catalog = FakeCatalog("configured") @@ -357,7 +357,7 @@ def unexpected_vend(*args, **kwargs): def test_string_catalog_type_describe_is_cached(monkeypatch): - _storage._clear_credential_cache() + _storage._clear_catalog_type_cache() patch_tower_context(monkeypatch) calls = [] vend_catalogs = []