Merge pull request #577 from armosec/jfix

refactor cleanup methods in workflows to improve error handling and c…

tests_scripts/workflows/teams_workflows.py

@@ -133,8 +133,12 @@ def start(self):
 
 
     def cleanup(self, **kwargs):
+        super().cleanup_workflows()
         if self.webhook_name:
-            self.delete_channel_by_guid(self.get_channel_guid_by_name(self.webhook_name))
+            try:
+                self.delete_channel_by_guid(self.get_channel_guid_by_name(self.webhook_name))
+            except Exception as e:
+                Logger.logger.error(f"Failed to delete channel with name {self.webhook_name}, got exception {e}")
         if self.fw_name:
             self.wait_for_report(report_type=self.backend.delete_custom_framework, framework_name=self.fw_name)
         return super().cleanup(**kwargs)

tests_scripts/workflows/workflows.py

@@ -15,12 +15,7 @@ def __init__(self, test_obj=None, backend=None, test_driver=None, kubernetes_obj
 
 
     def cleanup(self, **kwargs):
-        for guid in self.test_workflows_guids:
-            try:
-                self.delete_and_assert_workflow(guid)
-            except Exception as e:
-                Logger.logger.error(f"Failed to delete workflow with guid {guid}, got exception {e}")
-        self.test_workflows_guids = []
+        self.cleanup_workflows()
         return super().cleanup(**kwargs)
 
     def active_workflow(self):
@@ -39,6 +34,14 @@ def install_kubescape(self, helm_kwargs: dict = None):
         self.add_and_upgrade_armo_to_repo()
         self.install_armo_helm_chart(helm_kwargs=helm_kwargs)
         self.verify_running_pods(namespace=statics.CA_NAMESPACE_FROM_HELM_NAME)
+
+    def cleanup_workflows(self):
+        for guid in self.test_workflows_guids:
+            try:
+                self.delete_and_assert_workflow(guid)
+            except Exception as e:
+                Logger.logger.error(f"Failed to delete workflow with guid {guid}, got exception {e}")
+        self.test_workflows_guids = []
 
     def create_and_assert_workflow(self, workflow_body, expected_response, update=False):
         if update: