Fix flake8 issues

Author: sfc-gh-pmansour

src/snowflake/connector/connection.py

@@ -1146,7 +1146,7 @@ def __open_connection(self):
             elif self._authenticator == PROGRAMMATIC_ACCESS_TOKEN:
                 self.auth_class = AuthByPAT(self._token)
             elif self._authenticator == WORKLOAD_IDENTITY_AUTHENTICATOR:
-                if not ENV_VAR_EXPERIMENTAL_AUTHENTICATION in os.environ:
+                if ENV_VAR_EXPERIMENTAL_AUTHENTICATION not in os.environ:
                     Error.errorhandler_wrapper(
                         self,
                         None,

src/snowflake/connector/wif_util.py

@@ -10,7 +10,6 @@
 from base64 import b64encode
 from dataclasses import dataclass
 from enum import Enum, unique
-from typing import Union
 
 import boto3
 import jwt
@@ -63,7 +62,7 @@ def try_metadata_service_call(
         )
         if not res.ok:
             return None
-    except:
+    except requests.RequestException:
         return None
     return res
 
@@ -82,7 +81,7 @@ def extract_iss_and_sub_without_signature_verification(jwt_str: str) -> tuple[st
     """
     try:
         claims = jwt.decode(jwt_str, options={"verify_signature": False})
-    except jwt.exceptions.InvalidTokenError as e:
+    except jwt.exceptions.InvalidTokenError:
         logger.warning("Token is not a valid JWT.", exc_info=True)
         return None, None
 
@@ -312,7 +311,7 @@ def create_attestation(
         attestation = create_gcp_attestation()
     elif provider == AttestationProvider.OIDC:
         attestation = create_oidc_attestation(token)
-    elif provider == None:
+    elif provider is None:
         attestation = create_autodetect_attestation(entra_resource, token)
 
     if not attestation:

test/csp_helpers.py

@@ -57,6 +57,7 @@ class FakeMetadataService(ABC):
     def __init__(self):
         self.reset_defaults()
 
+    @abstractmethod
     def reset_defaults(self):
         """Resets any default values for test parameters.
 
@@ -122,6 +123,9 @@ def __exit__(self, *args, **kwargs):
 class NoMetadataService(FakeMetadataService):
     """Emulates an environment without any metadata service."""
 
+    def reset_defaults(self):
+        pass
+
     @property
     def expected_hostname(self):
         return None  # Always raise a ConnectTimeout.
@@ -153,7 +157,7 @@ def handle_request(self, method, parsed_url, headers, timeout):
             and headers.get("Metadata") == "True"
             and query_string["resource"]
         ):
-            return HTTPError(status_code=400)
+            raise HTTPError()
 
         logger.debug("Received request for Azure VM metadata service")
 
@@ -191,7 +195,7 @@ def handle_request(self, method, parsed_url, headers, timeout):
             logger.warning(
                 f"Received malformed request: {method} {parsed_url.path} {str(headers)} {str(query_string)}"
             )
-            return HTTPError(status_code=400)
+            raise HTTPError()
 
         logger.debug("Received request for Azure Functions metadata service")
 
@@ -234,7 +238,7 @@ def handle_request(self, method, parsed_url, headers, timeout):
             and headers.get("Metadata-Flavor") == "Google"
             and query_string["audience"]
         ):
-            return HTTPError(status_code=400)
+            raise HTTPError()
 
         logger.debug("Received request for GCE metadata service")
 

test/unit/test_auth_workload_identity.py

@@ -132,7 +132,7 @@ def test_explicit_aws_uses_regional_hostname(fake_aws_environment: FakeAwsEnviro
     hostname_from_url = urlparse(decoded_token["url"]).hostname
     hostname_from_header = decoded_token["headers"]["Host"]
 
-    expected_hostname = f"sts.antarctica-northeast-3.amazonaws.com"
+    expected_hostname = "sts.antarctica-northeast-3.amazonaws.com"
     assert expected_hostname == hostname_from_url
     assert expected_hostname == hostname_from_header