style: apply black formatting to deploy_context and test_ecr_policy_helpers

Author: abhishektang

samcli/commands/deploy/deploy_context.py

@@ -167,9 +167,7 @@ def run(self):
         self.deployer = Deployer(cloudformation_client, client_sleep=self.poll_delay)
 
         if self.image_repositories or self.image_repository:
-            ecr_client = boto3.client(
-                "ecr", region_name=self.region if self.region else None, config=boto_config
-            )
+            ecr_client = boto3.client("ecr", region_name=self.region if self.region else None, config=boto_config)
             _ensure_ecr_lambda_pull_policy(
                 ecr_client,
                 self.image_repositories if isinstance(self.image_repositories, dict) else None,
@@ -361,7 +359,6 @@ def merge_parameters(template_dict: Dict, parameter_overrides: Dict) -> List[Dic
         return parameter_values
 
 
-
 def _extract_ecr_repo_name(ecr_uri: str) -> str:
     """
     Extract the ECR repository name from a full ECR URI.
@@ -410,7 +407,7 @@ def _upsert_ecr_lambda_policy(ecr_client, repo_name: str) -> None:
     Soft-fails on AccessDenied so users who have manually pre-configured
     policies or whose IAM principal lacks ecr:SetRepositoryPolicy are not blocked.
     """
-    # Step 1: Fetch current policy (if any)
+
     existing_statements = []
     try:
         response = ecr_client.get_repository_policy(repositoryName=repo_name)
@@ -430,16 +427,13 @@ def _upsert_ecr_lambda_policy(ecr_client, repo_name: str) -> None:
             return
         raise deploy_exceptions.ECRPolicySetError(repo_name=repo_name, msg=str(ex)) from ex
 
-    # Step 2: Remove any existing SAM-owned statement (idempotent upsert)
     filtered = [s for s in existing_statements if s.get("Sid") != _SAM_ECR_POLICY_SID]
 
-    # Step 3: Build merged policy
     merged_policy = {
         "Version": "2012-10-17",
         "Statement": filtered + [_LAMBDA_ECR_POLICY_STATEMENT],
     }
 
-    # Step 4: Write the merged policy back
     try:
         ecr_client.set_repository_policy(
             repositoryName=repo_name,

tests/unit/commands/deploy/test_ecr_policy_helpers.py

@@ -49,9 +49,7 @@ def _make_ecr_client(existing_policy_doc=None, get_side_effect=None, set_side_ef
     if get_side_effect is not None:
         ecr_client.get_repository_policy.side_effect = get_side_effect
     elif existing_policy_doc is not None:
-        ecr_client.get_repository_policy.return_value = {
-            "policyText": json.dumps(existing_policy_doc)
-        }
+        ecr_client.get_repository_policy.return_value = {"policyText": json.dumps(existing_policy_doc)}
     else:
         ecr_client.get_repository_policy.return_value = {"policyText": "{}"}
 
@@ -118,9 +116,7 @@ def test_empty_dict_and_none_returns_early(self):
     @patch("samcli.commands.deploy.deploy_context._upsert_ecr_lambda_policy")
     def test_deduplicates_same_repo(self, mock_upsert):
         uri = "123456789012.dkr.ecr.us-east-1.amazonaws.com/my-repo:latest"
-        _ensure_ecr_lambda_pull_policy(
-            MagicMock(), {"FnA": uri, "FnB": uri}, None
-        )
+        _ensure_ecr_lambda_pull_policy(MagicMock(), {"FnA": uri, "FnB": uri}, None)
         mock_upsert.assert_called_once()
 
     @patch("samcli.commands.deploy.deploy_context._upsert_ecr_lambda_policy")
@@ -137,9 +133,7 @@ def test_two_different_repos_calls_twice(self, mock_upsert):
 
     @patch("samcli.commands.deploy.deploy_context._upsert_ecr_lambda_policy")
     def test_singular_image_repository(self, mock_upsert):
-        _ensure_ecr_lambda_pull_policy(
-            MagicMock(), None, "123456789012.dkr.ecr.us-east-1.amazonaws.com/single:v2"
-        )
+        _ensure_ecr_lambda_pull_policy(MagicMock(), None, "123456789012.dkr.ecr.us-east-1.amazonaws.com/single:v2")
         mock_upsert.assert_called_once()
 
 
@@ -162,9 +156,10 @@ def test_no_existing_policy_sets_sam_statement(self):
         self.assertEqual(policy["Statement"][0]["Sid"], _SAM_ECR_POLICY_SID)
 
     def test_preserves_existing_statements_and_appends_sam(self):
-        existing = {"Version": "2012-10-17", "Statement": [
-            {"Sid": "CustomerPolicy", "Effect": "Allow", "Principal": "*", "Action": "ecr:*"}
-        ]}
+        existing = {
+            "Version": "2012-10-17",
+            "Statement": [{"Sid": "CustomerPolicy", "Effect": "Allow", "Principal": "*", "Action": "ecr:*"}],
+        }
         ecr_client = _make_ecr_client(existing_policy_doc=existing)
         _upsert_ecr_lambda_policy(ecr_client, "my-repo")
 
@@ -175,9 +170,17 @@ def test_preserves_existing_statements_and_appends_sam(self):
         self.assertEqual(len(policy["Statement"]), 2)
 
     def test_idempotent_replaces_existing_sam_statement(self):
-        stale = {"Version": "2012-10-17", "Statement": [
-            {"Sid": _SAM_ECR_POLICY_SID, "Effect": "Deny", "Principal": {"Service": "lambda.amazonaws.com"}, "Action": "ecr:*"}
-        ]}
+        stale = {
+            "Version": "2012-10-17",
+            "Statement": [
+                {
+                    "Sid": _SAM_ECR_POLICY_SID,
+                    "Effect": "Deny",
+                    "Principal": {"Service": "lambda.amazonaws.com"},
+                    "Action": "ecr:*",
+                }
+            ],
+        }
         ecr_client = _make_ecr_client(existing_policy_doc=stale)
         _upsert_ecr_lambda_policy(ecr_client, "my-repo")
 
@@ -271,9 +274,14 @@ def test_race_condition_without_fix(self):
         def cf_handler(function_name: str):
             current = json.loads(json.dumps(policy_store["doc"]))
             time.sleep(0.01)
-            current["Statement"] = [{"Sid": f"Grant_{function_name}", "Effect": "Allow",
-                                     "Principal": {"Service": "lambda.amazonaws.com"},
-                                     "Action": ["ecr:GetDownloadUrlForLayer"]}]
+            current["Statement"] = [
+                {
+                    "Sid": f"Grant_{function_name}",
+                    "Effect": "Allow",
+                    "Principal": {"Service": "lambda.amazonaws.com"},
+                    "Action": ["ecr:GetDownloadUrlForLayer"],
+                }
+            ]
             policy_store["doc"] = current
 
         threads = [threading.Thread(target=cf_handler, args=(n,)) for n in SEVEN_FUNCTIONS_SAME_REPO]