fix(cssc): 31646926 fix 'list' command filtering by '--run-status'

src/acrcssc/azext_acrcssc/helper/_taskoperations.py

@@ -350,10 +350,13 @@ def _retrieve_logs_for_image(cmd, registry, resource_group_name, schedule, workf
     progress_indicator = IndeterminateProgressBar(cmd.cli_ctx, message="Retrieving logs for images")
     progress_indicator.begin()
 
-    image_status = WorkflowTaskStatus.from_taskrun(cmd, acr_task_run_client, registry, scan_taskruns, patch_taskruns, progress_indicator=progress_indicator)
-    if workflow_status:
-        filtered_image_status = [image for image in image_status if image["patch_status"] == workflow_status or image["scan_status"] == workflow_status]
-        image_status = filtered_image_status
+    image_status = WorkflowTaskStatus.from_taskrun(cmd,
+                                                   acr_task_run_client,
+                                                   registry,
+                                                   scan_taskruns,
+                                                   patch_taskruns,
+                                                   progress_indicator=progress_indicator,
+                                                   workflow_status_filter=workflow_status)
 
     end_time = time.time()
     execution_time = end_time - start_time

src/acrcssc/azext_acrcssc/helper/_workflow_status.py

@@ -255,7 +255,13 @@ def process_taskrun(taskrun):
             concurrent.futures.wait(futures)
 
     @staticmethod
-    def from_taskrun(cmd, taskrun_client, registry, scan_taskruns, patch_taskruns, progress_indicator=None):
+    def from_taskrun(cmd,
+                     taskrun_client,
+                     registry,
+                     scan_taskruns,
+                     patch_taskruns,
+                     progress_indicator=None,
+                     workflow_status_filter=None):
         WorkflowTaskStatus._retrieve_all_tasklogs(cmd, taskrun_client, registry, scan_taskruns, progress_indicator)
         all_status = {}
 
@@ -313,7 +319,33 @@ def from_taskrun(cmd, taskrun_client, registry, scan_taskruns, patch_taskruns, p
             for workflow_status in failed_patch_tasklog_retrieval:
                 workflow_status.patch_logs = workflow_status.patch_task.task_log_result
 
-        return [status.get_status() for status in all_status.values()]
+        return [status.get_status()
+                for status in WorkflowTaskStatus._filter_taskruns(all_status, workflow_status_filter).values()]
+
+    @staticmethod
+    def _filter_taskruns(workflows, workflow_status_filter=None):
+        if not workflows:
+            return {}
+
+        if not workflow_status_filter:
+            return workflows
+
+        # SKIPPED is a special case, because it means that the patch task does not exist,
+        # but the scan task succeeded. Another special case that is not explicit here is SUCCEEDED,
+        # which will include both scan and patch tasks that succeeded, or the scan task succeeded
+        # and the patch task is skipped
+        if workflow_status_filter == WorkflowTaskState.SKIPPED.value:
+            filtered_workflow = {key: workflow
+                                 for key, workflow in workflows.items()
+                                 if workflow.scan_status() == WorkflowTaskState.SUCCEEDED.value and
+                                 workflow.patch_status() == WorkflowTaskState.SKIPPED.value}
+            return filtered_workflow
+
+        filtered_workflow = {
+            key: workflow
+            for key, workflow in workflows.items()
+            if workflow.status() == workflow_status_filter}
+        return filtered_workflow
 
     def get_status(self):
         scan_status = self.scan_status()

src/acrcssc/azext_acrcssc/tests/latest/test_helper_taskoperations.py

@@ -67,8 +67,11 @@ def test_create_continuous_patch_v1_create_run_immediately_triggers_task(self, m
     @mock.patch("azext_acrcssc.helper._taskoperations.create_oci_artifact_continuous_patch")
     @mock.patch("azext_acrcssc.helper._taskoperations.validate_and_deploy_template")
     @mock.patch("azext_acrcssc.helper._taskoperations._eval_trigger_run")
-    def test_update_continuous_patch_v1_schedule_update_should_not_update_config(self, mock_eval_trigger_run, mock_validate_and_deploy_template, mock_create_oci_artifact_continuous_patch, mock_convert_timespan_to_cron, mock_check_continuoustask_exists, mock_update_task_schedule):
+    @mock.patch('azext_acrcssc.helper._taskoperations.cf_acr_tasks')
+    def test_update_continuous_patch_v1_schedule_update_should_not_update_config(self, mock_cf_acr_tasks, mock_eval_trigger_run, mock_validate_and_deploy_template, mock_create_oci_artifact_continuous_patch, mock_convert_timespan_to_cron, mock_check_continuoustask_exists, mock_update_task_schedule):
         # Mock the necessary dependencies
+        mock_acr_tasks_client = mock.MagicMock()
+        mock_cf_acr_tasks.return_value = mock_acr_tasks_client
         with tempfile.NamedTemporaryFile(delete=False) as temp_file:
             temp_file_path = temp_file.name
         mock_check_continuoustask_exists.return_value = True, []
@@ -107,8 +110,14 @@ def test_update_continuous_patch_v1__update_without_tasks_workflow_should_fail(s
     @mock.patch("azext_acrcssc.helper._taskoperations.create_oci_artifact_continuous_patch")
     @mock.patch("azext_acrcssc.helper._taskoperations.validate_and_deploy_template")
     @mock.patch("azext_acrcssc.helper._taskoperations._eval_trigger_run")
-    def test_update_continuous_patch_v1_schedule_update_run_immediately_triggers_task(self, mock_eval_trigger_run, mock_validate_and_deploy_template, mock_create_oci_artifact_continuous_patch, mock_convert_timespan_to_cron, mock_check_continuoustask_exists, mock_update_task_schedule):
+    @mock.patch('azext_acrcssc.helper._taskoperations.cf_acr_tasks')
+    @mock.patch('azext_acrcssc.helper._taskoperations.cf_authorization')
+    def test_update_continuous_patch_v1_schedule_update_run_immediately_triggers_task(self, mock_cf_authorization, mock_cf_acr_tasks, mock_eval_trigger_run, mock_validate_and_deploy_template, mock_create_oci_artifact_continuous_patch, mock_convert_timespan_to_cron, mock_check_continuoustask_exists, mock_update_task_schedule):
         # Mock the necessary dependencies
+        mock_acr_tasks_client = mock.MagicMock()
+        mock_cf_acr_tasks.return_value = mock_acr_tasks_client
+        mock_role_client = mock.MagicMock()
+        mock_cf_authorization.return_value = mock_role_client
         with tempfile.NamedTemporaryFile(delete=False) as temp_file:
             temp_file_path = temp_file.name
         mock_check_continuoustask_exists.return_value = True, []
@@ -129,9 +138,10 @@ def test_update_continuous_patch_v1_schedule_update_run_immediately_triggers_tas
     @mock.patch("azext_acrcssc.helper._taskoperations.check_continuous_task_config_exists")
     @mock.patch('azext_acrcssc.helper._taskoperations.delete_oci_artifact_continuous_patch')
     @mock.patch("azext_acrcssc.helper._taskoperations.check_continuous_task_exists")
+    @mock.patch("azext_acrcssc.helper._taskoperations.cf_acr_runs")
     @mock.patch('azext_acrcssc.helper._taskoperations.cf_acr_tasks')
     @mock.patch('azext_acrcssc.helper._taskoperations.cf_authorization')
-    def test_delete_continuous_patch_v1(self, mock_cf_authorization, mock_cf_acr_tasks, mock_check_continuoustask_exists, mock_delete_oci_artifact_continuous_patch, mock_check_continuous_task_config_exists, mock_get_taskruns_with_filter, mock_cancel_task_runs):
+    def test_delete_continuous_patch_v1(self, mock_cf_authorization, mock_cf_acr_tasks, mock_cf_acr_runs, mock_check_continuoustask_exists, mock_delete_oci_artifact_continuous_patch, mock_check_continuous_task_config_exists, mock_get_taskruns_with_filter, mock_cancel_task_runs):
         # Mock the necessary dependencies
         mock_check_continuoustask_exists.return_value = True, []
         mock_check_continuous_task_config_exists.return_value = True
@@ -141,6 +151,8 @@ def test_delete_continuous_patch_v1(self, mock_cf_authorization, mock_cf_acr_tas
         mock_cf_acr_tasks.return_value = mock_acr_tasks_client
         mock_role_client = mock.MagicMock()
         mock_cf_authorization.return_value = mock_role_client
+        mock_acr_run_client = mock.MagicMock()
+        mock_cf_acr_runs.return_value = mock_acr_run_client
         mock_task = mock.MagicMock()
         mock_task.identity = mock.MagicMock()(principal_id='principal_id')
         mock_acr_tasks_client.get.return_value = mock_task
@@ -200,14 +212,17 @@ def test_list_continuous_patch_v1(self, mock_transform_task_list, mock_cf_acr_ta
     @mock.patch("azext_acrcssc.helper._taskoperations.prepare_source_location")
     @mock.patch("azext_acrcssc.helper._taskoperations.cf_acr_registries_tasks")
     @mock.patch("azext_acrcssc.helper._taskoperations.cf_acr_runs")
+    @mock.patch('azext_acrcssc.helper._taskoperations.cf_acr_tasks')
     @mock.patch("azext_acrcssc.helper._taskoperations.LongRunningOperation")
-    def test_acr_cssc_dry_run(self, mock_LongRunningOperation, mock_cf_acr_runs, mock_cf_acr_registries_tasks, mock_prepare_source_location, mock_delete_temporary_dry_run_file, mock_create_temporary_dry_run_file, mock_generate_logs):
+    def test_acr_cssc_dry_run(self, mock_LongRunningOperation, mock_cf_acr_tasks, mock_cf_acr_runs, mock_cf_acr_registries_tasks, mock_prepare_source_location, mock_delete_temporary_dry_run_file, mock_create_temporary_dry_run_file, mock_generate_logs):
         # Mock the necessary dependencies
         config_file_path = "test_config_file_path"
         mock_acr_registries_task_client = mock.MagicMock()
         mock_cf_acr_registries_tasks.return_value = mock_acr_registries_task_client
         mock_acr_run_client = mock.MagicMock()
         mock_cf_acr_runs.return_value = mock_acr_run_client
+        mock_acr_task_client = mock.MagicMock()
+        mock_cf_acr_tasks.return_value = mock_acr_task_client
         mock_LongRunningOperation.return_value.return_value.run_id = "test_run_id"
         mock_generate_logs.return_value = "mock_logs"
 

src/acrcssc/azext_acrcssc/tests/latest/test_workflow_Status.py → src/acrcssc/azext_acrcssc/tests/latest/test_workflow_status.py

@@ -28,7 +28,7 @@ def test_init(self):
     def test_image(self):
         self.assertEqual(self.workflow_task_status.image(), "repository:tag")
 
-    def test__task_status_to_workflow_status(self):
+    def test_task_status_to_workflow_status(self):
         from azext_acrcssc.helper._constants import TaskRunStatus
         task = mock.MagicMock()
         task.status = TaskRunStatus.Succeeded.value
@@ -159,11 +159,11 @@ def test_from_taskrun(self, mock_retrieve_all_tasklogs, mock_get_missing_taskrun
         self.assertTrue(result[0]["patch_status"] == WorkflowTaskState.SUCCEEDED.value)
 
         # Test with mixed scan and patch tasks status
-        patch_taskruns = [self._generate_test_taskrun(True, status=TaskRunStatus.Succeeded.value, tag="tag0"),
-                          self._generate_test_taskrun(True, status=TaskRunStatus.Canceled.value, tag="tag1"),
-                          self._generate_test_taskrun(True, status=TaskRunStatus.Queued.value, tag="tag2"),
-                          self._generate_test_taskrun(True, status=TaskRunStatus.Running.value, tag="tag3"),
-                          self._generate_test_taskrun(True, status=TaskRunStatus.Failed.value, tag="tag4")]
+        patch_taskruns = [self._generate_test_taskrun(False, status=TaskRunStatus.Succeeded.value, tag="tag0"),
+                          self._generate_test_taskrun(False, status=TaskRunStatus.Canceled.value, tag="tag1"),
+                          self._generate_test_taskrun(False, status=TaskRunStatus.Queued.value, tag="tag2"),
+                          self._generate_test_taskrun(False, status=TaskRunStatus.Running.value, tag="tag3"),
+                          self._generate_test_taskrun(False, status=TaskRunStatus.Failed.value, tag="tag4")]
 
         scan_taskruns = [self._generate_test_taskrun(True, patch_taskid_in_scan=patch_taskruns[0].run_id, tag="tag0"),
                          self._generate_test_taskrun(True, patch_taskid_in_scan=patch_taskruns[1].run_id, tag="tag1"),
@@ -193,6 +193,98 @@ def test_from_taskrun(self, mock_retrieve_all_tasklogs, mock_get_missing_taskrun
         # test that a successful patch has a patched image reference
         self.assertTrue(all(True if workflow["patch_status"] != TaskRunStatus.Succeeded.value or not workflow["last_patched_image"].startswith("---") else False for workflow in result))
 
+
+    @mock.patch('azext_acrcssc.helper._workflow_status.WorkflowTaskStatus._get_missing_taskrun')
+    @mock.patch('azext_acrcssc.helper._workflow_status.WorkflowTaskStatus._retrieve_all_tasklogs')
+    def test_from_taskrun_with_filter(self, mock_retrieve_all_tasklogs, mock_get_missing_taskrun):
+        cmd = mock.MagicMock()
+        cmd.cli_ctx = DummyCli()
+        taskrun_client = MagicMock()
+        registry = MagicMock()
+        scan_taskruns = []
+        patch_taskruns = []
+
+        mock_retrieve_all_tasklogs.return_value = scan_taskruns
+        mock_get_missing_taskrun.return_value = None
+
+         # Test with mixed scan and patch tasks status
+        patch_taskruns = [self._generate_test_taskrun(False, status=TaskRunStatus.Succeeded.value, tag="tag0"),
+                          self._generate_test_taskrun(False, status=TaskRunStatus.Canceled.value, tag="tag1"),
+                          self._generate_test_taskrun(False, status=TaskRunStatus.Queued.value, tag="tag2"),
+                          self._generate_test_taskrun(False, status=TaskRunStatus.Running.value, tag="tag3"),
+                          self._generate_test_taskrun(False, status=TaskRunStatus.Failed.value, tag="tag4")]
+
+        scan_taskruns = [self._generate_test_taskrun(True, patch_taskid_in_scan=patch_taskruns[0].run_id, tag="tag0"),
+                         self._generate_test_taskrun(True, patch_taskid_in_scan=patch_taskruns[1].run_id, tag="tag1"),
+                         self._generate_test_taskrun(True, patch_taskid_in_scan=patch_taskruns[2].run_id, tag="tag2"),
+                         self._generate_test_taskrun(True, patch_taskid_in_scan=patch_taskruns[3].run_id, tag="tag3"),
+                         self._generate_test_taskrun(True, patch_taskid_in_scan=patch_taskruns[4].run_id, tag="tag4"),
+                         self._generate_test_taskrun(True, status=TaskRunStatus.Failed.value, tag="tag5"),
+                         self._generate_test_taskrun(True, status=TaskRunStatus.Canceled.value, tag="tag6"),
+                         self._generate_test_taskrun(True, status=TaskRunStatus.Queued.value, tag="tag7"),
+                         self._generate_test_taskrun(True, status=TaskRunStatus.Running.value, tag="tag8"),
+                         self._generate_test_taskrun(True, status=TaskRunStatus.Succeeded.value, tag="tag9")]
+
+        result = WorkflowTaskStatus.from_taskrun(cmd,
+                                                 taskrun_client,
+                                                 registry,
+                                                 scan_taskruns,
+                                                 patch_taskruns,
+                                                 workflow_status_filter=WorkflowTaskState.SUCCEEDED.value)
+        self.assertTrue(len(result) > 0)
+        self.assertTrue(all(workflow["scan_status"] == WorkflowTaskState.SUCCEEDED.value 
+                            for workflow in result))
+        self.assertTrue(all(workflow["patch_status"] == WorkflowTaskState.SUCCEEDED.value or 
+                            workflow["patch_status"] == WorkflowTaskState.SKIPPED.value
+                            for workflow in result))
+
+        result = WorkflowTaskStatus.from_taskrun(cmd,
+                                                 taskrun_client,
+                                                 registry,
+                                                 scan_taskruns,
+                                                 patch_taskruns,
+                                                 workflow_status_filter=WorkflowTaskState.FAILED.value)
+        self.assertTrue(len(result) > 0)
+        self.assertTrue(all(workflow["scan_status"] == WorkflowTaskState.FAILED.value or 
+                            workflow["patch_status"] == WorkflowTaskState.FAILED.value 
+                            for workflow in result))
+
+        result = WorkflowTaskStatus.from_taskrun(cmd,
+                                                 taskrun_client,
+                                                 registry,
+                                                 scan_taskruns,
+                                                 patch_taskruns,
+                                                 workflow_status_filter=WorkflowTaskState.RUNNING.value)
+        self.assertTrue(len(result) > 0)
+        self.assertTrue(all(workflow["scan_status"] == WorkflowTaskState.RUNNING.value or
+                            workflow["patch_status"] == WorkflowTaskState.RUNNING.value or
+                            workflow["scan_status"] == WorkflowTaskState.QUEUED.value or
+                            workflow["patch_status"] == WorkflowTaskState.QUEUED.value
+                            for workflow in result))
+
+        result = WorkflowTaskStatus.from_taskrun(cmd,
+                                                 taskrun_client,
+                                                 registry,
+                                                 scan_taskruns,
+                                                 patch_taskruns,
+                                                 workflow_status_filter=WorkflowTaskState.CANCELED.value)
+        self.assertTrue(len(result) > 0)
+        self.assertTrue(all(workflow["scan_status"] == WorkflowTaskState.CANCELED.value or
+                            workflow["patch_status"] == WorkflowTaskState.CANCELED.value
+                            for workflow in result))
+
+        result = WorkflowTaskStatus.from_taskrun(cmd,
+                                                 taskrun_client,
+                                                 registry,
+                                                 scan_taskruns,
+                                                 patch_taskruns,
+                                                 workflow_status_filter=WorkflowTaskState.SKIPPED.value)
+        self.assertTrue(len(result) > 0)
+        self.assertTrue(all(workflow["scan_status"] == WorkflowTaskState.SUCCEEDED.value and
+                            workflow["patch_status"] == WorkflowTaskState.SKIPPED.value
+                            for workflow in result))
+
+
     # generate a random scan or patch taskrun with the desired properties
     def _generate_test_taskrun(self, scan_task=True, status=TaskRunStatus.Succeeded.value, repository="mock-repo", tag="mock-tag", patch_taskid_in_scan=""):
         import random