black

Author: xiangyan99

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

@@ -115,9 +115,7 @@ def get_token(
           receive an access token.
         """
         if claims and claims.strip():
-            raise CredentialUnavailableError(
-                f"Fail to get token, please run az login --claims-challenge {claims}"
-            )
+            raise CredentialUnavailableError(f"Fail to get token, please run az login --claims-challenge {claims}")
 
         options: TokenRequestOptions = {}
         if tenant_id:
@@ -127,9 +125,7 @@ def get_token(
         return AccessToken(token_info.token, token_info.expires_on)
 
     @log_get_token
-    def get_token_info(
-        self, *scopes: str, options: Optional[TokenRequestOptions] = None
-    ) -> AccessTokenInfo:
+    def get_token_info(self, *scopes: str, options: Optional[TokenRequestOptions] = None) -> AccessTokenInfo:
         """Request an access token for `scopes`.
 
         This is an alternative to `get_token` to enable certain scenarios that require additional properties
@@ -230,9 +226,7 @@ def get_safe_working_dir() -> str:
     if sys.platform.startswith("win"):
         path = os.environ.get("SYSTEMROOT")
         if not path:
-            raise CredentialUnavailableError(
-                message="Environment variable 'SYSTEMROOT' has no value"
-            )
+            raise CredentialUnavailableError(message="Environment variable 'SYSTEMROOT' has no value")
         return path
 
     return "/bin"
@@ -252,9 +246,7 @@ def _run_command(command_args: List[str], timeout: int) -> str:
     # Ensure executable exists in PATH first. This avoids a subprocess call that would fail anyway.
     if sys.platform.startswith("win"):
         # On Windows, the expected executable is az.cmd, so we check for that first, falling back to 'az' in case.
-        az_path = shutil.which(EXECUTABLE_NAME + ".cmd") or shutil.which(
-            EXECUTABLE_NAME
-        )
+        az_path = shutil.which(EXECUTABLE_NAME + ".cmd") or shutil.which(EXECUTABLE_NAME)
     else:
         az_path = shutil.which(EXECUTABLE_NAME)
     if not az_path:
@@ -277,13 +269,10 @@ def _run_command(command_args: List[str], timeout: int) -> str:
     except subprocess.CalledProcessError as ex:
         # non-zero return from shell
         # Fallback check in case the executable is not found while executing subprocess.
-        if ex.returncode == 127 or (
-            ex.stderr is not None and ex.stderr.startswith("'az' is not recognized")
-        ):
+        if ex.returncode == 127 or (ex.stderr is not None and ex.stderr.startswith("'az' is not recognized")):
             raise CredentialUnavailableError(message=CLI_NOT_FOUND) from ex
         if ex.stderr is not None and (
-            ("az login" in ex.stderr or "az account set" in ex.stderr)
-            and "AADSTS" not in ex.stderr
+            ("az login" in ex.stderr or "az account set" in ex.stderr) and "AADSTS" not in ex.stderr
         ):
             raise CredentialUnavailableError(message=NOT_LOGGED_IN) from ex
 
@@ -297,9 +286,7 @@ def _run_command(command_args: List[str], timeout: int) -> str:
         raise ClientAuthenticationError(message=message) from ex
     except OSError as ex:
         # failed to execute 'cmd' or '/bin/sh'
-        error = CredentialUnavailableError(
-            message="Failed to execute '{}'".format(args[0])
-        )
+        error = CredentialUnavailableError(message="Failed to execute '{}'".format(args[0]))
         raise error from ex
     except Exception as ex:
         # could be a timeout, for example

sdk/identity/azure-identity/azure/identity/aio/_credentials/azure_cli.py

@@ -106,17 +106,11 @@ async def get_token(
           receive an access token.
         """
         if claims and claims.strip():
-            raise CredentialUnavailableError(
-                f"Fail to get token, please run az login --claims-challenge {claims}"
-            )
+            raise CredentialUnavailableError(f"Fail to get token, please run az login --claims-challenge {claims}")
 
         # only ProactorEventLoop supports subprocesses on Windows (and it isn't the default loop on Python < 3.8)
-        if sys.platform.startswith("win") and not isinstance(
-            asyncio.get_event_loop(), asyncio.ProactorEventLoop
-        ):
-            return _SyncAzureCliCredential().get_token(
-                *scopes, tenant_id=tenant_id, **kwargs
-            )
+        if sys.platform.startswith("win") and not isinstance(asyncio.get_event_loop(), asyncio.ProactorEventLoop):
+            return _SyncAzureCliCredential().get_token(*scopes, tenant_id=tenant_id, **kwargs)
 
         options: TokenRequestOptions = {}
         if tenant_id:
@@ -126,9 +120,7 @@ async def get_token(
         return AccessToken(token_info.token, token_info.expires_on)
 
     @log_get_token_async
-    async def get_token_info(
-        self, *scopes: str, options: Optional[TokenRequestOptions] = None
-    ) -> AccessTokenInfo:
+    async def get_token_info(self, *scopes: str, options: Optional[TokenRequestOptions] = None) -> AccessTokenInfo:
         """Request an access token for `scopes`.
 
         This is an alternative to `get_token` to enable certain scenarios that require additional properties
@@ -154,9 +146,7 @@ async def get_token_info(
                 f"Fail to get token, please run az login --claims-challenge {claims_value}"
             )
         # only ProactorEventLoop supports subprocesses on Windows (and it isn't the default loop on Python < 3.8)
-        if sys.platform.startswith("win") and not isinstance(
-            asyncio.get_event_loop(), asyncio.ProactorEventLoop
-        ):
+        if sys.platform.startswith("win") and not isinstance(asyncio.get_event_loop(), asyncio.ProactorEventLoop):
             return _SyncAzureCliCredential().get_token_info(*scopes, options=options)
         return await self._get_token_base(*scopes, options=options)
 
@@ -207,9 +197,7 @@ async def _run_command(command_args: List[str], timeout: int) -> str:
     # Ensure executable exists in PATH first. This avoids a subprocess call that would fail anyway.
     if sys.platform.startswith("win"):
         # On Windows, the expected executable is az.cmd, so we check for that first, falling back to 'az' in case.
-        az_path = shutil.which(EXECUTABLE_NAME + ".cmd") or shutil.which(
-            EXECUTABLE_NAME
-        )
+        az_path = shutil.which(EXECUTABLE_NAME + ".cmd") or shutil.which(EXECUTABLE_NAME)
     else:
         az_path = shutil.which(EXECUTABLE_NAME)
     if not az_path:
@@ -232,14 +220,10 @@ async def _run_command(command_args: List[str], timeout: int) -> str:
         stderr = stderr_b.decode()
     except asyncio.TimeoutError as ex:
         proc.kill()
-        raise CredentialUnavailableError(
-            message="Timed out waiting for Azure CLI"
-        ) from ex
+        raise CredentialUnavailableError(message="Timed out waiting for Azure CLI") from ex
     except OSError as ex:
         # failed to execute 'cmd' or '/bin/sh'
-        error = CredentialUnavailableError(
-            message="Failed to execute '{}'".format(args[0])
-        )
+        error = CredentialUnavailableError(message="Failed to execute '{}'".format(args[0]))
         raise error from ex
 
     if proc.returncode == 0:

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

@@ -400,10 +400,10 @@ def fake_check_output(command_line, **_):
 @pytest.mark.parametrize("get_token_method", GET_TOKEN_METHODS)
 def test_claims_challenge_raises_error(get_token_method):
     """The credential should raise CredentialUnavailableError when claims challenge is provided"""
-    
+
     claims = "test-claims-challenge"
     expected_message = f"Fail to get token, please run az login --claims-challenge {claims}"
-    
+
     if get_token_method == "get_token":
         with pytest.raises(CredentialUnavailableError, match=re.escape(expected_message)):
             AzureCliCredential().get_token("scope", claims=claims)
@@ -415,44 +415,46 @@ def test_claims_challenge_raises_error(get_token_method):
 @pytest.mark.parametrize("get_token_method", GET_TOKEN_METHODS)
 def test_empty_claims_does_not_raise_error(get_token_method):
     """The credential should not raise error when claims parameter is empty or None"""
-    
+
     # Mock the CLI to avoid actual invocation
     with mock.patch("shutil.which", return_value="az"):
-        with mock.patch(CHECK_OUTPUT, mock.Mock(return_value='{"accessToken": "token", "expiresOn": "2021-10-07 12:00:00.000000"}')):
-            
+        with mock.patch(
+            CHECK_OUTPUT, mock.Mock(return_value='{"accessToken": "token", "expiresOn": "2021-10-07 12:00:00.000000"}')
+        ):
+
             if get_token_method == "get_token":
                 # Test with None (default)
                 token = AzureCliCredential().get_token("scope")
                 assert token.token == "token"
-                
+
                 # Test with empty string
                 token = AzureCliCredential().get_token("scope", claims="")
                 assert token.token == "token"
-                
+
                 # Test with None explicitly
                 token = AzureCliCredential().get_token("scope", claims=None)
                 assert token.token == "token"
-                
+
                 # Test with whitespace-only string
                 token = AzureCliCredential().get_token("scope", claims="   ")
                 assert token.token == "token"
             else:  # get_token_info
                 # Test with None options
                 token = AzureCliCredential().get_token_info("scope")
                 assert token.token == "token"
-                
+
                 # Test with empty options
                 token = AzureCliCredential().get_token_info("scope", options={})
                 assert token.token == "token"
-                
+
                 # Test with None claims in options
                 token = AzureCliCredential().get_token_info("scope", options={"claims": None})
                 assert token.token == "token"
-                
-                # Test with empty string claims in options 
+
+                # Test with empty string claims in options
                 token = AzureCliCredential().get_token_info("scope", options={"claims": ""})
                 assert token.token == "token"
-                
+
                 # Test with whitespace-only claims in options
                 token = AzureCliCredential().get_token_info("scope", options={"claims": "   "})
                 assert token.token == "token"

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

@@ -394,10 +394,10 @@ async def fake_exec(*args, **_):
 @pytest.mark.parametrize("get_token_method", GET_TOKEN_METHODS)
 async def test_claims_challenge_raises_error(get_token_method):
     """The credential should raise CredentialUnavailableError when claims challenge is provided"""
-    
+
     claims = "test-claims-challenge"
     expected_message = f"Fail to get token, please run az login --claims-challenge {claims}"
-    
+
     if get_token_method == "get_token":
         with pytest.raises(CredentialUnavailableError, match=re.escape(expected_message)):
             await AzureCliCredential().get_token("scope", claims=claims)
@@ -409,52 +409,54 @@ async def test_claims_challenge_raises_error(get_token_method):
 @pytest.mark.parametrize("get_token_method", GET_TOKEN_METHODS)
 async def test_empty_claims_does_not_raise_error(get_token_method):
     """The credential should not raise error when claims parameter is empty or None"""
-    
-    successful_output = json.dumps({
-        "expiresOn": datetime.now().strftime("%Y-%m-%d %H:%M:%S.%f"),
-        "accessToken": "access-token",
-        "subscription": "subscription",
-        "tenant": "tenant",
-        "tokenType": "Bearer",
-    })
-    
+
+    successful_output = json.dumps(
+        {
+            "expiresOn": datetime.now().strftime("%Y-%m-%d %H:%M:%S.%f"),
+            "accessToken": "access-token",
+            "subscription": "subscription",
+            "tenant": "tenant",
+            "tokenType": "Bearer",
+        }
+    )
+
     # Mock the CLI to avoid actual invocation
     with mock.patch("shutil.which", return_value="az"):
         with mock.patch(SUBPROCESS_EXEC, mock_exec(successful_output)):
-            
+
             if get_token_method == "get_token":
                 # Test with None (default)
                 token = await AzureCliCredential().get_token("scope")
                 assert token.token == "access-token"
-                
+
                 # Test with empty string
                 token = await AzureCliCredential().get_token("scope", claims="")
                 assert token.token == "access-token"
-                
+
                 # Test with None explicitly
                 token = await AzureCliCredential().get_token("scope", claims=None)
                 assert token.token == "access-token"
-                
+
                 # Test with whitespace-only string
                 token = await AzureCliCredential().get_token("scope", claims="   ")
                 assert token.token == "access-token"
             else:  # get_token_info
                 # Test with None options
                 token = await AzureCliCredential().get_token_info("scope")
                 assert token.token == "access-token"
-                
+
                 # Test with empty options
                 token = await AzureCliCredential().get_token_info("scope", options={})
                 assert token.token == "access-token"
-                
+
                 # Test with None claims in options
                 token = await AzureCliCredential().get_token_info("scope", options={"claims": None})
                 assert token.token == "access-token"
-                
-                # Test with empty string claims in options 
+
+                # Test with empty string claims in options
                 token = await AzureCliCredential().get_token_info("scope", options={"claims": ""})
                 assert token.token == "access-token"
-                
+
                 # Test with whitespace-only claims in options
                 token = await AzureCliCredential().get_token_info("scope", options={"claims": "   "})
                 assert token.token == "access-token"