chore: apply formatters and linters

Author: yiyuan-he

contract-tests/images/applications/botocore/botocore_server.py

@@ -251,7 +251,11 @@ def _handle_bedrock_request(self) -> None:
             set_main_status(200)
             bedrock_client.meta.events.register(
                 "before-call.bedrock.GetGuardrail",
-                lambda **kwargs: inject_200_success(guardrailId="bt4o77i015cu", guardrailArn="arn:aws:bedrock:us-east-1:000000000000:guardrail/bt4o77i015cu", **kwargs),
+                lambda **kwargs: inject_200_success(
+                    guardrailId="bt4o77i015cu",
+                    guardrailArn="arn:aws:bedrock:us-east-1:000000000000:guardrail/bt4o77i015cu",
+                    **kwargs,
+                ),
             )
             bedrock_client.get_guardrail(
                 guardrailIdentifier="arn:aws:bedrock:us-east-1:000000000000:guardrail/bt4o77i015cu"
@@ -320,7 +324,9 @@ def _handle_secretsmanager_request(self) -> None:
         elif self.in_path(_FAULT):
             set_main_status(500)
             try:
-                fault_client = boto3.client("secretsmanager", endpoint_url=_FAULT_ENDPOINT, region_name=_AWS_REGION, config=_NO_RETRY_CONFIG)
+                fault_client = boto3.client(
+                    "secretsmanager", endpoint_url=_FAULT_ENDPOINT, region_name=_AWS_REGION, config=_NO_RETRY_CONFIG
+                )
                 fault_client.get_secret_value(
                     SecretId="arn:aws:secretsmanager:us-west-2:000000000000:secret:nonexistent-secret"
                 )
@@ -338,7 +344,9 @@ def _handle_stepfunctions_request(self) -> None:
             set_main_status(400)
             try:
                 error_client = boto3.client("stepfunctions", endpoint_url=_ERROR_ENDPOINT, region_name=_AWS_REGION)
-                error_client.describe_state_machine(stateMachineArn="arn:aws:states:us-west-2:000000000000:stateMachine:unExistStateMachine")
+                error_client.describe_state_machine(
+                    stateMachineArn="arn:aws:states:us-west-2:000000000000:stateMachine:unExistStateMachine"
+                )
             except Exception as exception:
                 print("Expected exception occurred", exception)
         elif self.in_path(_FAULT):
@@ -356,7 +364,9 @@ def _handle_stepfunctions_request(self) -> None:
                 print("Expected exception occurred", exception)
         elif self.in_path("describestatemachine/my-state-machine"):
             set_main_status(200)
-            sfn_client.describe_state_machine(stateMachineArn="arn:aws:states:us-west-2:000000000000:stateMachine:testStateMachine")
+            sfn_client.describe_state_machine(
+                stateMachineArn="arn:aws:states:us-west-2:000000000000:stateMachine:testStateMachine"
+            )
         elif self.in_path("describeactivity/my-activity"):
             set_main_status(200)
             sfn_client.describe_activity(activityArn="arn:aws:states:us-west-2:000000000000:activity:testActivity")
@@ -372,6 +382,7 @@ def set_main_status(status: int) -> None:
     RequestHandler.main_status = status
 
 
+# pylint: disable=too-many-locals
 def prepare_aws_server() -> None:
     requests.Request(method="POST", url="http://localhost:4566/_localstack/state/reset")
     try:
@@ -409,7 +420,9 @@ def prepare_aws_server() -> None:
         kinesis_client.create_stream(StreamName="test_stream", ShardCount=1)
 
         # Set up Secrets Manager so tests can access a secret.
-        secretsmanager_client: BaseClient = boto3.client("secretsmanager", endpoint_url=_AWS_SDK_ENDPOINT, region_name=_AWS_REGION)
+        secretsmanager_client: BaseClient = boto3.client(
+            "secretsmanager", endpoint_url=_AWS_SDK_ENDPOINT, region_name=_AWS_REGION
+        )
         secretsmanager_response = secretsmanager_client.list_secrets()
         secret = next((s for s in secretsmanager_response["SecretList"] if s["Name"] == "testSecret"), None)
         if not secret:
@@ -431,41 +444,26 @@ def prepare_aws_server() -> None:
             trust_policy = {
                 "Version": "2012-10-17",
                 "Statement": [
-                    {
-                        "Effect": "Allow",
-                        "Principal": {
-                            "Service": "states.amazonaws.com"
-                        },
-                        "Action": "sts:AssumeRole"
-                    }
-                ]
+                    {"Effect": "Allow", "Principal": {"Service": "states.amazonaws.com"}, "Action": "sts:AssumeRole"}
+                ],
             }
             try:
                 iam_response = iam_client.create_role(
                     RoleName=iam_role_name, AssumeRolePolicyDocument=json.dumps(trust_policy)
                 )
                 iam_client.attach_role_policy(
-                    RoleName=iam_role_name,
-                    PolicyArn="arn:aws:iam::aws:policy/AWSStepFunctionsFullAccess"
+                    RoleName=iam_role_name, PolicyArn="arn:aws:iam::aws:policy/AWSStepFunctionsFullAccess"
                 )
                 print(f"IAM Role '{iam_role_name}' create successfully.")
                 iam_role_arn = iam_response["Role"]["Arn"]
                 sfn_defintion = {
                     "Comment": "A simple sequential workflow",
                     "StartAt": "FirstState",
-                    "States": {
-                        "FirstState": {
-                            "Type": "Pass",
-                            "Result": "Hello, World!",
-                            "End": True
-                        }
-                    }
+                    "States": {"FirstState": {"Type": "Pass", "Result": "Hello, World!", "End": True}},
                 }
                 definition_string = json.dumps(sfn_defintion)
                 sfn_client.create_state_machine(
-                    name=state_machine_name,
-                    definition=definition_string,
-                    roleArn=iam_role_arn
+                    name=state_machine_name, definition=definition_string, roleArn=iam_role_arn
                 )
                 sfn_client.create_activity(name=activity_name)
             except Exception as exception:

contract-tests/tests/test/amazon/botocore/botocore_test.py

@@ -265,7 +265,9 @@ def test_sqs_create_queue(self):
             remote_operation="CreateQueue",
             remote_resource_type="AWS::SQS::Queue",
             remote_resource_identifier="test_queue",
-            cloudformation_primary_identifier="http://sqs.us-west-2.localhost.localstack.cloud:4566/000000000000/test_queue",
+            cloudformation_primary_identifier=(
+                "http://sqs.us-west-2.localhost.localstack.cloud:4566/000000000000/test_queue"
+            ),
             request_specific_attributes={
                 _AWS_SQS_QUEUE_NAME: "test_queue",
             },
@@ -547,7 +549,9 @@ def test_secretsmanager_describe_secret(self):
             remote_operation="DescribeSecret",
             remote_resource_type="AWS::SecretsManager::Secret",
             remote_resource_identifier=r"testSecret-[a-zA-Z0-9]{6}$",
-            cloudformation_primary_identifier=r"arn:aws:secretsmanager:us-west-2:000000000000:secret:testSecret-[a-zA-Z0-9]{6}$",
+            cloudformation_primary_identifier=(
+                r"arn:aws:secretsmanager:us-west-2:000000000000:secret:testSecret-[a-zA-Z0-9]{6}$"
+            ),
             response_specific_attributes={
                 _AWS_SECRET_ARN: r"arn:aws:secretsmanager:us-west-2:000000000000:secret:testSecret-[a-zA-Z0-9]{6}$",
             },
@@ -661,7 +665,9 @@ def test_stepfunctions_fault(self):
             remote_operation="ListStateMachineVersions",
             remote_resource_type="AWS::StepFunctions::StateMachine",
             remote_resource_identifier="invalid-state-machine",
-            cloudformation_primary_identifier="arn:aws:states:us-west-2:000000000000:stateMachine:invalid-state-machine",
+            cloudformation_primary_identifier=(
+                "arn:aws:states:us-west-2:000000000000:stateMachine:invalid-state-machine"
+            ),
             request_specific_attributes={
                 _AWS_STATE_MACHINE_ARN: "arn:aws:states:us-west-2:000000000000:stateMachine:invalid-state-machine",
             },
@@ -708,14 +714,20 @@ def _assert_aws_attributes(
             self._assert_str_attribute(attributes_dict, AWS_REMOTE_RESOURCE_TYPE, remote_resource_type)
         if remote_resource_identifier != "None":
             if self._is_valid_regex(remote_resource_identifier):
-                self._assert_match_attribute(attributes_dict, AWS_REMOTE_RESOURCE_IDENTIFIER, remote_resource_identifier)
+                self._assert_match_attribute(
+                    attributes_dict, AWS_REMOTE_RESOURCE_IDENTIFIER, remote_resource_identifier
+                )
             else:
                 self._assert_str_attribute(attributes_dict, AWS_REMOTE_RESOURCE_IDENTIFIER, remote_resource_identifier)
         if cloudformation_primary_identifier != "None":
             if self._is_valid_regex(remote_resource_identifier):
-                self._assert_match_attribute(attributes_dict, AWS_CLOUDFORMATION_PRIMARY_IDENTIFIER, cloudformation_primary_identifier)
+                self._assert_match_attribute(
+                    attributes_dict, AWS_CLOUDFORMATION_PRIMARY_IDENTIFIER, cloudformation_primary_identifier
+                )
             else:
-                self._assert_str_attribute(attributes_dict, AWS_CLOUDFORMATION_PRIMARY_IDENTIFIER, cloudformation_primary_identifier)
+                self._assert_str_attribute(
+                    attributes_dict, AWS_CLOUDFORMATION_PRIMARY_IDENTIFIER, cloudformation_primary_identifier
+                )
         # See comment above AWS_LOCAL_OPERATION
         self._assert_str_attribute(attributes_dict, AWS_SPAN_KIND, span_kind)