[Identity] Fix issue with cache option usage

sdk/identity/azure-identity/CHANGELOG.md

@@ -8,6 +8,8 @@
 
 ### Bugs Fixed
 
+- Fixed issue where user-supplied `TokenCachePersistenceOptions` weren't propagated when using `SharedTokenCacheCredential` ([#26982](https://github.com/Azure/azure-sdk-for-python/issues/26982))
+
 ### Other Changes
 
 ## 1.12.0b2 (2022-10-11)
@@ -133,7 +135,7 @@ Azure-identity is supported on Python 3.7 or later. For more details, please rea
 ### Bugs Fixed
 
 - Handle injected "tenant_id" and "claims" ([#23138](https://github.com/Azure/azure-sdk-for-python/issues/23138))
-  
+
   "tenant_id" argument in get_token() method is only supported by:
 
   - `AuthorizationCodeCredential`
@@ -163,7 +165,7 @@ Azure-identity is supported on Python 3.7 or later. For more details, please rea
 > Only code written against a beta version such as 1.7.0b1 may be affected.
 
 - The `allow_multitenant_authentication` argument has been removed and the default behavior is now as if it were true.
-  The multitenant authentication feature can be totally disabled by setting the environment variable 
+  The multitenant authentication feature can be totally disabled by setting the environment variable
   `AZURE_IDENTITY_DISABLE_MULTITENANTAUTH` to `True`.
 - `azure.identity.RegionalAuthority` is removed.
 - `regional_authority` argument is removed for `CertificateCredential` and `ClientSecretCredential`.

sdk/identity/azure-identity/azure/identity/_credentials/silent.py

@@ -36,6 +36,7 @@ def __init__(self, authentication_record, **kwargs):
         self._tenant_id = kwargs.pop("tenant_id", None) or self._auth_record.tenant_id
         validate_tenant_id(self._tenant_id)
         self._cache = kwargs.pop("_cache", None)
+        self._cache_persistence_options = kwargs.pop("cache_persistence_options", None)
         self._client_applications = {}  # type: Dict[str, PublicClientApplication]
         self._additionally_allowed_tenants = kwargs.pop("additionally_allowed_tenants", [])
         self._client = MsalClient(**kwargs)
@@ -64,10 +65,13 @@ def get_token(self, *scopes, **kwargs):  # pylint:disable=unused-argument
     def _initialize(self):
         if not self._cache and platform.system() in {"Darwin", "Linux", "Windows"}:
             try:
-                # This credential accepts the user's default cache regardless of whether it's encrypted. It doesn't
-                # create a new cache. If the default cache exists, the user must have created it earlier. If it's
-                # unencrypted, the user must have allowed that.
-                self._cache = _load_persistent_cache(TokenCachePersistenceOptions(allow_unencrypted_storage=True))
+                # If no cache options were provided, the default cache will be used. This credential accepts the
+                # user's default cache regardless of whether it's encrypted. It doesn't create a new cache. If the
+                # default cache exists, the user must have created it earlier. If it's unencrypted, the user must
+                # have allowed that.
+                options = self._cache_persistence_options or \
+                    TokenCachePersistenceOptions(allow_unencrypted_storage=True)
+                self._cache = _load_persistent_cache(options)
             except Exception:  # pylint:disable=broad-except
                 pass
 

sdk/identity/azure-identity/azure/identity/_internal/shared_token_cache.py

@@ -95,6 +95,7 @@ def __init__(self, username=None, **kwargs):  # pylint:disable=unused-argument
         self._username = username
         self._tenant_id = kwargs.pop("tenant_id", None)
         self._cache = kwargs.pop("_cache", None)
+        self._cache_persistence_options = kwargs.pop("cache_persistence_options", None)
         self._client = None  # type: Optional[AadClientBase]
         self._client_kwargs = kwargs
         self._client_kwargs["tenant_id"] = "organizations"
@@ -116,10 +117,13 @@ def _initialize(self):
     def _load_cache(self):
         if not self._cache and self.supported():
             try:
-                # This credential accepts the user's default cache regardless of whether it's encrypted. It doesn't
-                # create a new cache. If the default cache exists, the user must have created it earlier. If it's
-                # unencrypted, the user must have allowed that.
-                self._cache = _load_persistent_cache(TokenCachePersistenceOptions(allow_unencrypted_storage=True))
+                # If no cache options were provided, the default cache will be used. This credential accepts the
+                # user's default cache regardless of whether it's encrypted. It doesn't create a new cache. If the
+                # default cache exists, the user must have created it earlier. If it's unencrypted, the user must
+                # have allowed that.
+                options = self._cache_persistence_options or \
+                    TokenCachePersistenceOptions(allow_unencrypted_storage=True)
+                self._cache = _load_persistent_cache(options)
             except Exception:  # pylint:disable=broad-except
                 pass
 

sdk/identity/azure-identity/tests/test_shared_cache_credential.py

@@ -9,6 +9,7 @@
     AzureAuthorityHosts,
     CredentialUnavailableError,
     SharedTokenCacheCredential,
+    TokenCachePersistenceOptions,
 )
 from azure.identity._constants import DEVELOPER_SIGN_ON_CLIENT_ID, EnvironmentVariables
 from azure.identity._internal.shared_token_cache import (
@@ -764,6 +765,18 @@ def test_initialization():
             assert mock_cache_loader.call_count == 1
 
 
+def test_initialization_with_cache_options():
+    """the credential should use user-supplied persistence options"""
+
+    with patch("azure.identity._internal.shared_token_cache._load_persistent_cache") as mock_cache_loader:
+        options = TokenCachePersistenceOptions(name="foo.cache")
+        credential = SharedTokenCacheCredential(cache_persistence_options=options)
+
+        with pytest.raises(CredentialUnavailableError):
+            credential.get_token("scope")
+        mock_cache_loader.assert_called_once_with(options)
+
+
 def test_authentication_record_authenticating_tenant():
     """when given a record and 'tenant_id', the credential should authenticate in the latter"""
 

sdk/identity/azure-identity/tests/test_shared_cache_credential_async.py

@@ -7,7 +7,7 @@
 
 from azure.core.exceptions import ClientAuthenticationError
 from azure.core.pipeline.policies import SansIOHTTPPolicy
-from azure.identity import CredentialUnavailableError
+from azure.identity import CredentialUnavailableError, TokenCachePersistenceOptions
 from azure.identity.aio import SharedTokenCacheCredential
 from azure.identity._constants import EnvironmentVariables
 from azure.identity._internal.shared_token_cache import (
@@ -68,14 +68,14 @@ async def send(*_, **__):
         _cache=populated_cache(get_account_event("test@user", "uid", "utid")), transport=transport
     )
 
-    # async with before initialization: credential should call aexit but not aenter
+    # async with before initialization: credential should call __aexit__ but not __aenter__
     async with credential:
         await credential.get_token("scope")
 
     assert transport.__aenter__.call_count == 0
     assert transport.__aexit__.call_count == 1
 
-    # async with after initialization: credential should call aenter and aexit
+    # async with after initialization: credential should call __aenter__ and __aexit__
     async with credential:
         await credential.get_token("scope")
         assert transport.__aenter__.call_count == 1
@@ -621,6 +621,19 @@ async def test_initialization():
             assert mock_cache_loader.call_count == 1
 
 
+@pytest.mark.asyncio
+async def test_initialization_with_cache_options():
+    """the credential should use user-supplied persistence options"""
+
+    with patch("azure.identity._internal.shared_token_cache._load_persistent_cache") as mock_cache_loader:
+        options = TokenCachePersistenceOptions(name="foo.cache")
+        credential = SharedTokenCacheCredential(cache_persistence_options=options)
+
+        with pytest.raises(CredentialUnavailableError):
+            await credential.get_token("scope")
+        mock_cache_loader.assert_called_once_with(options)
+
+
 @pytest.mark.asyncio
 async def test_multitenant_authentication():
     first_token = "***"