chore: Enable pyupgrade check (#3019)

Author: aahung

bin/_file_formatter.py

@@ -45,7 +45,7 @@ def config_additional_args(cls) -> None:  # noqa: empty-method-without-abstract-
         """Optionally configure additional args to arg parser."""
 
     def process_file(self, file_path: str) -> None:
-        with open(file_path, "r", encoding="utf-8") as f:
+        with open(file_path, encoding="utf-8") as f:
             file_str = f.read()
             try:
                 formatted_file_str = self.format_str(file_str)

bin/add_transform_test.py

@@ -75,7 +75,7 @@ def replace_aws_partition(partition: str, file_path: str) -> None:
 def generate_transform_test_output_files(input_file_path: str, file_basename: str) -> None:
     output_file_option = file_basename + ".json"
 
-    with open(os.path.join(input_file_path), "r") as f:
+    with open(os.path.join(input_file_path)) as f:
         manifest = yaml_parse(f)  # type: ignore[no-untyped-call]
 
     transform_test_output_paths = {

bin/sam-translate.py

@@ -102,7 +102,7 @@ def package(input_file_path, output_file_path):  # type: ignore[no-untyped-def]
 
 
 def transform_template(input_file_path, output_file_path):  # type: ignore[no-untyped-def]
-    with open(input_file_path, "r") as f:
+    with open(input_file_path) as f:
         sam_template = yaml_parse(f)  # type: ignore[no-untyped-call]
 
     try:

integration/combination/test_api_with_authorizer_apikey.py

@@ -85,7 +85,7 @@ def verify_authorized_request(
         status = response.status_code
         if status != expected_status_code:
             raise StatusCodeError(
-                "Request to {} failed with status: {}, expected status: {}".format(url, status, expected_status_code)
+                f"Request to {url} failed with status: {status}, expected status: {expected_status_code}"
             )
 
         if not header_key or not header_value:

integration/combination/test_api_with_authorizers.py

@@ -444,7 +444,7 @@ def verify_authorized_request(
 
         if status != expected_status_code:
             raise StatusCodeError(
-                "Request to {} failed with status: {}, expected status: {}".format(url, status, expected_status_code)
+                f"Request to {url} failed with status: {status}, expected status: {expected_status_code}"
             )
 
         if not header_key or not header_value:

integration/combination/test_function_with_api.py

@@ -28,9 +28,9 @@ def test_function_with_api(self):
 
         self.assertTrue(
             get_api_policy_expectation in policy,
-            "{} should be present in policy {}".format(get_api_policy_expectation, policy),
+            f"{get_api_policy_expectation} should be present in policy {policy}",
         )
         self.assertTrue(
             post_api_policy_expectation in policy,
-            "{} should be present in policy {}".format(post_api_policy_expectation, policy),
+            f"{post_api_policy_expectation} should be present in policy {policy}",
         )

integration/combination/test_intrinsic_function_support.py

@@ -53,11 +53,11 @@ def test_severless_api_properties_support(self):
 
         self.assertTrue(
             get_api_policy_expectation in policy,
-            "{} should be present in policy {}".format(get_api_policy_expectation, policy),
+            f"{get_api_policy_expectation} should be present in policy {policy}",
         )
         self.assertTrue(
             post_api_policy_expectation in policy,
-            "{} should be present in policy {}".format(post_api_policy_expectation, policy),
+            f"{post_api_policy_expectation} should be present in policy {policy}",
         )
 
         # Test for tags

integration/conftest.py

@@ -171,16 +171,16 @@ def upload_resources(get_s3):
 
 def get_s3_uri(file_name, uri_type, bucket, region):
     if uri_type == "s3":
-        return "s3://{}/{}".format(bucket, file_name)
+        return f"s3://{bucket}/{file_name}"
 
     if region == "us-east-1":
-        return "https://s3.amazonaws.com/{}/{}".format(bucket, file_name)
+        return f"https://s3.amazonaws.com/{bucket}/{file_name}"
     if region == "us-iso-east-1":
-        return "https://s3.us-iso-east-1.c2s.ic.gov/{}/{}".format(bucket, file_name)
+        return f"https://s3.us-iso-east-1.c2s.ic.gov/{bucket}/{file_name}"
     if region == "us-isob-east-1":
-        return "https://s3.us-isob-east-1.sc2s.sgov.gov/{}/{}".format(bucket, file_name)
+        return f"https://s3.us-isob-east-1.sc2s.sgov.gov/{bucket}/{file_name}"
 
-    return "https://s3-{}.amazonaws.com/{}/{}".format(region, bucket, file_name)
+    return f"https://s3-{region}.amazonaws.com/{bucket}/{file_name}"
 
 
 @pytest.fixture()

integration/helpers/base_test.py

@@ -416,7 +416,7 @@ def _fill_template(self, folder, file_name):
             data = f.read()
         for key, _ in self.code_key_to_file.items():
             # We must double the {} to escape them so they will survive a round of unescape
-            data = data.replace("${{{}}}".format(key), self.get_code_key_s3_uri(key))
+            data = data.replace(f"${{{key}}}", self.get_code_key_s3_uri(key))
         yaml_doc = yaml_parse(data)
 
         dump_yaml(updated_template_path, yaml_doc)

integration/helpers/deployer/deployer.py

@@ -140,7 +140,7 @@ def create_changeset(
             "ChangeSetType": changeset_type,
             "Parameters": parameter_values,
             "Capabilities": capabilities,
-            "Description": "Created by SAM CLI at {0} UTC".format(datetime.utcnow().isoformat()),
+            "Description": f"Created by SAM CLI at {datetime.utcnow().isoformat()} UTC",
             "Tags": tags,
         }
 
@@ -172,7 +172,7 @@ def _create_change_set(self, stack_name, changeset_type, **kwargs):
         except botocore.exceptions.ClientError as ex:
             if "The bucket you are attempting to access must be addressed using the specified endpoint" in str(ex):
                 raise deploy_exceptions.DeployBucketInDifferentRegionError(
-                    "Failed to create/update stack {}".format(stack_name)
+                    f"Failed to create/update stack {stack_name}"
                 )
             raise deploy_exceptions.ChangeSetError(stack_name=stack_name, msg=str(ex))
 
@@ -278,7 +278,7 @@ def wait_for_changeset(self, changeset_id, stack_name):
                 raise deploy_exceptions.ChangeEmptyError(stack_name=stack_name)
 
             raise deploy_exceptions.ChangeSetError(
-                stack_name=stack_name, msg="ex: {0} Status: {1}. Reason: {2}".format(ex, status, reason)
+                stack_name=stack_name, msg=f"ex: {ex} Status: {status}. Reason: {reason}"
             )
 
     def execute_changeset(self, changeset_id, stack_name):
@@ -323,7 +323,7 @@ def wait_for_execute(self, stack_name, changeset_type):
         elif changeset_type == "UPDATE":
             waiter = self._client.get_waiter("stack_update_complete")
         else:
-            raise RuntimeError("Invalid changeset type {0}".format(changeset_type))
+            raise RuntimeError(f"Invalid changeset type {changeset_type}")
 
         # Poll every 30 seconds. Polling too frequently risks hitting rate limits
         # on CloudFormation's DescribeStacks API
@@ -408,7 +408,7 @@ def get_stack_outputs(self, stack_name, echo=True):
             try:
                 outputs = stacks_description["Stacks"][0]["Outputs"]
                 if echo:
-                    sys.stdout.write("\nStack {stack_name} outputs:\n".format(stack_name=stack_name))
+                    sys.stdout.write(f"\nStack {stack_name} outputs:\n")
                     sys.stdout.flush()
                     self._display_stack_outputs(stack_outputs=outputs)
                 return outputs