feat: add v2 API config for bulk mute finding to the existing mute config Python client library or samples (GoogleCloudPlatform#12609)

vijaykanthm · gcf-owl-bot[bot] · web-flow · commit 612323322d7e · 2024-10-08T17:18:33.000-04:00
* Add bulk mute finding config * 🦉 Updates from OwlBot post-processor See https://github.com/googleapis/repo-automation-bots/blob/main/packages/owl-bot/README.md * fix the build error * remove unused import * 🦉 Updates from OwlBot post-processor See https://github.com/googleapis/repo-automation-bots/blob/main/packages/owl-bot/README.md * Address suggestions * 🦉 Updates from OwlBot post-processor See https://github.com/googleapis/repo-automation-bots/blob/main/packages/owl-bot/README.md * Add type annotation for location id * Address suggestions * 🦉 Updates from OwlBot post-processor See https://github.com/googleapis/repo-automation-bots/blob/main/packages/owl-bot/README.md * refactor the filter value --------- Co-authored-by: Owl Bot <gcf-owl-bot[bot]@users.noreply.github.com>
diff --git a/securitycenter/snippets_v2/mute_findings_v2.py b/securitycenter/snippets_v2/mute_findings_v2.py
@@ -76,30 +76,36 @@ def set_unmute_finding(finding_path: str) -> None:
 
 
 # [START securitycenter_bulk_mute_v2]
-# def bulk_mute_findings(parent_path: str, location_id, mute_rule: str) -> None:
-#     """
-#       Kicks off a long-running operation (LRO) to bulk mute findings for a parent based on a filter.
-#       The parent can be either an organization, folder, or project. The findings
-#       matched by the filter will be muted after the LRO is done.
-#     Args:
-#         parent_path: use any one of the following options:
-#                      - organizations/{organization}
-#                      - folders/{folder}
-#                      - projects/{project}
-#         mute_rule: Expression that identifies findings that should be updated.
-#     """
-#     from google.cloud import securitycenter_v2
-
-#     client = securitycenter_v2.SecurityCenterClient()
-
-#     request = securitycenter_v2.BulkMuteFindingsRequest()
-#     request.parent = parent_path + "/locations/" + location_id
-#     # To create mute rules, see:
-#     # https://cloud.google.com/security-command-center/docs/how-to-mute-findings#create_mute_rules
-#     request.filter = mute_rule
-
-#     response = client.bulk_mute_findings(request)
-#     print(f"Bulk mute findings completed successfully! : {response}")
-#     return response
+def bulk_mute_findings(parent_path: str, location_id: str, mute_rule: str) -> None:
+    """
+      Kicks off a long-running operation (LRO) to bulk mute findings for a parent based on a filter.
+
+      The parent can be either an organization, folder, or project. The findings
+      matched by the filter will be muted after the LRO is done.
+
+    Args:
+        parent_path: Used to specify the resource under which the mute rule operates.
+          Use any one of the following options:
+            - organizations/{organization}
+            - folders/{folder}
+            - projects/{project}
+        location_id: The Google Cloud location ID, for example: 'global'.
+        mute_rule: Expression that identifies findings that should be updated.
+    """
+    from google.cloud import securitycenter_v2
+
+    client = securitycenter_v2.SecurityCenterClient()
+
+    request = securitycenter_v2.BulkMuteFindingsRequest(
+        parent=f"{parent_path}/locations/{location_id}",
+        # To create mute rules, see:
+        # https://cloud.google.com/security-command-center/docs/how-to-mute-findings#create_mute_rules
+        filter=mute_rule,
+    )
+
+    response = client.bulk_mute_findings(request)
+    print(f"Bulk mute findings completed successfully: {response}")
+    return response
+
 
 # [END securitycenter_bulk_mute_v2]
diff --git a/securitycenter/snippets_v2/mute_findings_v2_test.py b/securitycenter/snippets_v2/mute_findings_v2_test.py
@@ -67,14 +67,14 @@ def test_set_unmute_finding(finding):
     assert response.mute.name == "UNMUTED"
 
 
-# @backoff.on_exception(
-#     backoff.expo, (InternalServerError, ServiceUnavailable, NotFound), max_tries=3
-# )
-# def test_bulk_mute_findings(finding):
-#     # Mute findings that belong to this project.
-#     response = mute_findings_v2.bulk_mute_findings(
-#         f"organizations/{ORGANIZATION_ID}",
-#         "global",
-#         f'resource.project_display_name="{ORGANIZATION_ID}"',
-#     )
-#     assert response.done
+@backoff.on_exception(
+    backoff.expo, (InternalServerError, ServiceUnavailable, NotFound), max_tries=3
+)
+def test_bulk_mute_findings(finding):
+    # Mute findings that belong to this project.
+    response = mute_findings_v2.bulk_mute_findings(
+        f"organizations/{ORGANIZATION_ID}",
+        "global",
+        f'resource.project_display_name="{ORGANIZATION_ID}"',
+    )
+    assert response.done
