Fix black formatting issues with Azure SDK configuration

Co-authored-by: xiangyan99 <[email protected]>

Author: Copilot

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

@@ -156,9 +156,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
@@ -217,9 +215,7 @@ def run_command_line(command_line: List[str], timeout: int) -> str:
     try:
         proc = start_process(command_line)
         stdout, stderr = proc.communicate(**kwargs)
-        if sys.platform.startswith("win") and (
-            "' is not recognized" in stderr or proc.returncode == 9009
-        ):
+        if sys.platform.startswith("win") and ("' is not recognized" in stderr or proc.returncode == 9009):
             # pwsh.exe isn't on the path; try powershell.exe
             command_line[-1] = command_line[-1].replace("pwsh", "powershell", 1)
             proc = start_process(command_line)
@@ -261,12 +257,8 @@ def parse_token(output: str) -> AccessTokenInfo:
             return AccessTokenInfo(token, int(expires_on))
 
     if within_dac.get():
-        raise CredentialUnavailableError(
-            message='Unexpected output from Get-AzAccessToken: "{}"'.format(output)
-        )
-    raise ClientAuthenticationError(
-        message='Unexpected output from Get-AzAccessToken: "{}"'.format(output)
-    )
+        raise CredentialUnavailableError(message='Unexpected output from Get-AzAccessToken: "{}"'.format(output))
+    raise ClientAuthenticationError(message='Unexpected output from Get-AzAccessToken: "{}"'.format(output))
 
 
 def get_command_line(scopes: Tuple[str, ...], tenant_id: str) -> List[str]:

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

@@ -88,9 +88,7 @@ async def get_token(
             )
 
         # 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 _SyncCredential().get_token(*scopes, tenant_id=tenant_id, **kwargs)
 
         options: TokenRequestOptions = {}
@@ -101,9 +99,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
@@ -131,9 +127,7 @@ async def get_token_info(
                 message=f"Fail to get token, please run Connect-AzAccount --ClaimsChallenge {claims_value}"
             )
 
-        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 _SyncCredential().get_token_info(*scopes, options=options)
         return await self._get_token_base(*scopes, options=options)
 
@@ -173,9 +167,7 @@ async def run_command_line(command_line: List[str], timeout: int) -> str:
     try:
         proc = await start_process(command_line)
         stdout, stderr = await asyncio.wait_for(proc.communicate(), 10)
-        if sys.platform.startswith("win") and (
-            b"' is not recognized" in stderr or proc.returncode == 9009
-        ):
+        if sys.platform.startswith("win") and (b"' is not recognized" in stderr or proc.returncode == 9009):
             # pwsh.exe isn't on the path; try powershell.exe
             command_line[-1] = command_line[-1].replace("pwsh", "powershell", 1)
             proc = await start_process(command_line)
@@ -193,9 +185,7 @@ async def run_command_line(command_line: List[str], timeout: int) -> str:
         error = CredentialUnavailableError(
             message='Failed to execute "{}".\n'
             "To mitigate this issue, please refer to the troubleshooting guidelines here at "
-            "https://aka.ms/azsdk/python/identity/powershellcredential/troubleshoot.".format(
-                command_line[0]
-            )
+            "https://aka.ms/azsdk/python/identity/powershellcredential/troubleshoot.".format(command_line[0])
         )
         raise error from ex
 

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

@@ -38,9 +38,7 @@
 
 
 def raise_called_process_error(return_code, stdout, stderr="", cmd="..."):
-    error = subprocess.CalledProcessError(
-        return_code, cmd=cmd, output=stdout, stderr=stderr
-    )
+    error = subprocess.CalledProcessError(return_code, cmd=cmd, output=stdout, stderr=stderr)
     return Mock(side_effect=error)
 
 
@@ -62,9 +60,7 @@ def test_multiple_scopes(get_token_method):
     """The credential should raise ValueError when get_token is called with more than one scope"""
 
     with pytest.raises(ValueError):
-        getattr(AzurePowerShellCredential(), get_token_method)(
-            "one scope", "and another"
-        )
+        getattr(AzurePowerShellCredential(), get_token_method)("one scope", "and another")
 
 
 @pytest.mark.parametrize("get_token_method", GET_TOKEN_METHODS)
@@ -100,9 +96,7 @@ def test_invalid_scopes(get_token_method):
             getattr(AzurePowerShellCredential(), get_token_method)("scope" + c)
 
 
-@pytest.mark.parametrize(
-    "stderr,get_token_method", product(("", PREPARING_MODULES), GET_TOKEN_METHODS)
-)
+@pytest.mark.parametrize("stderr,get_token_method", product(("", PREPARING_MODULES), GET_TOKEN_METHODS))
 def test_get_token(stderr, get_token_method):
     """The credential should parse Azure PowerShell's output to an AccessToken"""
 
@@ -135,9 +129,7 @@ def test_get_token(stderr, get_token_method):
     args, kwargs = Popen().communicate.call_args
 
 
-@pytest.mark.parametrize(
-    "stderr,get_token_method", product(("", PREPARING_MODULES), GET_TOKEN_METHODS)
-)
+@pytest.mark.parametrize("stderr,get_token_method", product(("", PREPARING_MODULES), GET_TOKEN_METHODS))
 def test_get_token_tenant_id(stderr, get_token_method):
     expected_access_token = "access"
     expected_expires_on = 1617923581
@@ -160,9 +152,7 @@ def test_ignores_extraneous_stdout_content(get_token_method):
     expected_access_token = "access"
     expected_expires_on = 1617923581
     motd = "MOTD: Customize your experience: save your profile to $HOME/.config/PowerShell\n"
-    Popen = get_mock_Popen(
-        stdout=motd + "azsdk%{}%{}".format(expected_access_token, expected_expires_on)
-    )
+    Popen = get_mock_Popen(stdout=motd + "azsdk%{}%{}".format(expected_access_token, expected_expires_on))
 
     with patch(POPEN, Popen):
         token = getattr(AzurePowerShellCredential(), get_token_method)("scope")
@@ -233,9 +223,7 @@ def test_blocked_by_execution_policy(get_token_method):
 <Objs Version="1.1.0.1" xmlns="http://schemas.microsoft.com/powershell/2004/04"><S S="Error">Import-Module : Errors occurred while loading the format data file: _x000D__x000A_</S><S S="Error">C:\Users\foo\Documents\WindowsPowerShell\Modules\Az.Accounts\2.2.7\Accounts.format.ps1xml, , C:\Users\foo\Documents\WindowsPowerShell\Modules\Az.Accounts\2.2.7\Accounts.format.ps1xml: The file was skipped because of the _x000D__x000A_</S><S S="Error">following validation exception: AuthorizationManager check failed.._x000D__x000A_</S><S S="Error">C:\Users\foo\Documents\WindowsPowerShell\Modules\Az.Accounts\2.2.7\Accounts.generated.format.ps1xml, , C:\Users\foo\Documents\WindowsPowerShell\Modules\Az.Accounts\2.2.7\Accounts.generated.format.ps1xml: The file was skipped _x000D__x000A_</S><S S="Error">because of the following validation exception: AuthorizationManager check failed.._x000D__x000A_</S><S S="Error">At line:4 char:6_x000D__x000A_</S><S S="Error">+ $m = Import-Module Az.Accounts -MinimumVersion $minimumVersion -PassT ..._x000D__x000A_</S><S S="Error">+      ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~_x000D__x000A_</S><S S="Error">    + CategoryInfo          : InvalidOperation: (:) [Import-Module], RuntimeException_x000D__x000A_</S><S S="Error">    + FullyQualifiedErrorId : FormatXmlUpdateException,Microsoft.PowerShell.Commands.ImportModuleCommand_x000D__x000A_</S><S S="Error"> _x000D__x000A_</S></Objs>"""
     Popen = get_mock_Popen(return_code=1, stderr=stderr)
     with patch(POPEN, Popen):
-        with pytest.raises(
-            CredentialUnavailableError, match=BLOCKED_BY_EXECUTION_POLICY
-        ):
+        with pytest.raises(CredentialUnavailableError, match=BLOCKED_BY_EXECUTION_POLICY):
             getattr(AzurePowerShellCredential(), get_token_method)("scope")
 
 
@@ -248,9 +236,7 @@ def test_timeout(get_token_method):
     proc = Mock(communicate=Mock(side_effect=TimeoutExpired("", 42)), returncode=None)
     with patch(POPEN, Mock(return_value=proc)):
         with pytest.raises(CredentialUnavailableError):
-            getattr(AzurePowerShellCredential(process_timeout=42), get_token_method)(
-                "scope"
-            )
+            getattr(AzurePowerShellCredential(process_timeout=42), get_token_method)("scope")
 
     assert proc.communicate.call_count == 1
     # Ensure custom timeout is passed to subprocess
@@ -314,23 +300,17 @@ def Popen(args, **kwargs):
             stderr = error_message
             return_code = 9009
         else:
-            assert args[-1].startswith(
-                "powershell"
-            ), 'credential should fall back to "powershell"'
+            assert args[-1].startswith("powershell"), 'credential should fall back to "powershell"'
             stdout = NO_AZ_ACCOUNT_MODULE
             stderr = ""
             return_code = 0
 
-        return Mock(
-            communicate=Mock(return_value=(stdout, stderr)), returncode=return_code
-        )
+        return Mock(communicate=Mock(return_value=(stdout, stderr)), returncode=return_code)
 
     with patch(AzurePowerShellCredential.__module__ + ".sys.platform", "win32"):
         with patch.dict("os.environ", {"SYSTEMROOT": "foo"}):
             with patch(POPEN, Popen):
-                with pytest.raises(
-                    CredentialUnavailableError, match=AZ_ACCOUNT_NOT_INSTALLED
-                ):
+                with pytest.raises(CredentialUnavailableError, match=AZ_ACCOUNT_NOT_INSTALLED):
                     getattr(AzurePowerShellCredential(), get_token_method)("scope")
 
     assert Fake.calls == 2
@@ -343,9 +323,7 @@ def test_multitenant_authentication(get_token_method):
     second_token = first_token * 2
 
     def fake_Popen(command, **_):
-        assert command[-1].startswith(
-            "pwsh -NoProfile -NonInteractive -EncodedCommand "
-        )
+        assert command[-1].startswith("pwsh -NoProfile -NonInteractive -EncodedCommand ")
         match = re.search(r"-EncodedCommand\s+(\S+)", command[-1])
         assert match, "couldn't find encoded script in command line"
         encoded_script = match.groups()[0]
@@ -354,9 +332,7 @@ def fake_Popen(command, **_):
         assert match
         tenant = match.group(1)
 
-        assert not tenant or tenant == second_tenant, 'unexpected tenant "{}"'.format(
-            tenant
-        )
+        assert not tenant or tenant == second_tenant, 'unexpected tenant "{}"'.format(tenant)
         token = first_token if not tenant else second_token
         stdout = "azsdk%{}%{}".format(token, int(time.time()) + 3600)
 
@@ -384,9 +360,7 @@ def test_multitenant_authentication_not_allowed(get_token_method):
     expected_token = "***"
 
     def fake_Popen(command, **_):
-        assert command[-1].startswith(
-            "pwsh -NoProfile -NonInteractive -EncodedCommand "
-        )
+        assert command[-1].startswith("pwsh -NoProfile -NonInteractive -EncodedCommand ")
         match = re.search(r"-EncodedCommand\s+(\S+)", command[-1])
         assert match, "couldn't find encoded script in command line"
         encoded_script = match.groups()[0]
@@ -424,23 +398,13 @@ def test_claims_challenge_error(get_token_method):
     if get_token_method == "get_token":
         # Test claims parameter in get_token method
         with pytest.raises(CredentialUnavailableError) as exc_info:
-            getattr(AzurePowerShellCredential(), get_token_method)(
-                "scope", claims="some-claims"
-            )
-        assert (
-            "Fail to get token, please run Connect-AzAccount --ClaimsChallenge some-claims"
-            in str(exc_info.value)
-        )
+            getattr(AzurePowerShellCredential(), get_token_method)("scope", claims="some-claims")
+        assert "Fail to get token, please run Connect-AzAccount --ClaimsChallenge some-claims" in str(exc_info.value)
     else:
         # Test claims in options for get_token_info method
         with pytest.raises(CredentialUnavailableError) as exc_info:
-            getattr(AzurePowerShellCredential(), get_token_method)(
-                "scope", options={"claims": "some-claims"}
-            )
-        assert (
-            "Fail to get token, please run Connect-AzAccount --ClaimsChallenge some-claims"
-            in str(exc_info.value)
-        )
+            getattr(AzurePowerShellCredential(), get_token_method)("scope", options={"claims": "some-claims"})
+        assert "Fail to get token, please run Connect-AzAccount --ClaimsChallenge some-claims" in str(exc_info.value)
 
 
 @pytest.mark.parametrize("get_token_method", GET_TOKEN_METHODS)
@@ -456,31 +420,21 @@ def test_empty_claims_no_error(get_token_method):
     with patch(POPEN, Popen):
         if get_token_method == "get_token":
             # Test None claims parameter in get_token method
-            token = getattr(AzurePowerShellCredential(), get_token_method)(
-                "scope", claims=None
-            )
+            token = getattr(AzurePowerShellCredential(), get_token_method)("scope", claims=None)
             assert token.token == expected_access_token
 
             # Test empty string claims
-            token = getattr(AzurePowerShellCredential(), get_token_method)(
-                "scope", claims=""
-            )
+            token = getattr(AzurePowerShellCredential(), get_token_method)("scope", claims="")
             assert token.token == expected_access_token
         else:
             # Test None claims in options for get_token_info method
-            token = getattr(AzurePowerShellCredential(), get_token_method)(
-                "scope", options={"claims": None}
-            )
+            token = getattr(AzurePowerShellCredential(), get_token_method)("scope", options={"claims": None})
             assert token.token == expected_access_token
 
             # Test empty string claims in options
-            token = getattr(AzurePowerShellCredential(), get_token_method)(
-                "scope", options={"claims": ""}
-            )
+            token = getattr(AzurePowerShellCredential(), get_token_method)("scope", options={"claims": ""})
             assert token.token == expected_access_token
 
             # Test missing claims key in options
-            token = getattr(AzurePowerShellCredential(), get_token_method)(
-                "scope", options={}
-            )
+            token = getattr(AzurePowerShellCredential(), get_token_method)("scope", options={})
             assert token.token == expected_access_token