Revert "Revert "fix: use list for bundle analysis tasks (#565)" (#572)" (#576)

This reverts commit 507a065.

Author: thomasrockhu-codecov

apps/worker/tasks/bundle_analysis_notify.py

@@ -21,9 +21,9 @@ class BundleAnalysisNotifyTask(BaseCodecovTask, name=bundle_analysis_notify_task
     def run_impl(
         self,
         db_session,
-        # Celery `chain` injects this argument - it's the returned result
-        # from the prior task in the chain
-        previous_result: dict[str, Any],
+        # Celery `chain` injects this argument - it's the list of processing results
+        # from prior processor tasks in the chain
+        previous_result: list[dict[str, Any]],
         *,
         repoid: int,
         commitid: str,
@@ -72,7 +72,7 @@ def process_impl_within_lock(
         repoid: int,
         commitid: str,
         commit_yaml: UserYaml,
-        previous_result: dict[str, Any],
+        previous_result: list[dict[str, Any]],
         **kwargs,
     ):
         log.info(
@@ -90,9 +90,11 @@ def process_impl_within_lock(
         )
         assert commit, "commit not found"
 
-        # these are the task results from prior processor tasks in the chain
+        # previous_result is the list of processing results from prior processor tasks
         # (they get accumulated as we execute each task in succession)
-        processing_results = previous_result.get("results", [])
+        processing_results = (
+            previous_result if isinstance(previous_result, list) else []
+        )
 
         if all(result["error"] is not None for result in processing_results):
             # every processor errored, nothing to notify on

apps/worker/tasks/bundle_analysis_processor.py

@@ -31,9 +31,9 @@ class BundleAnalysisProcessorTask(
     def run_impl(
         self,
         db_session,
-        # Celery `chain` injects this argument - it's the returned result
-        # from the prior task in the chain
-        previous_result: dict[str, Any],
+        # Celery `chain` injects this argument - it's the list of processing results
+        # from prior tasks in the chain (accumulated as each task executes)
+        previous_result: list[dict[str, Any]],
         *args,
         repoid: int,
         commitid: str,
@@ -82,7 +82,7 @@ def process_impl_within_lock(
         commitid: str,
         commit_yaml: UserYaml,
         params: UploadArguments,
-        previous_result: dict[str, Any],
+        previous_result: list[dict[str, Any]],
     ):
         log.info(
             "Running bundle analysis processor",
@@ -100,9 +100,11 @@ def process_impl_within_lock(
 
         report_service = BundleAnalysisReportService(commit_yaml)
 
-        # these are the task results from prior processor tasks in the chain
+        # previous_result is the list of processing results from prior processor tasks
         # (they get accumulated as we execute each task in succession)
-        processing_results = previous_result.get("results", [])
+        processing_results = (
+            previous_result if isinstance(previous_result, list) else []
+        )
 
         # these are populated in the upload task
         # unless when this task is called on a non-BA upload then we have to create an empty upload
@@ -141,7 +143,7 @@ def process_impl_within_lock(
                             "commit": commit.commitid,
                         },
                     )
-                    return {"results": processing_results}
+                    return processing_results
             else:
                 # If the commit report does not exist, we will create a new one
                 commit_report = report_service.initialize_and_save_report(commit)
@@ -240,7 +242,7 @@ def process_impl_within_lock(
             },
         )
 
-        return {"results": processing_results}
+        return processing_results
 
 
 RegisteredBundleAnalysisProcessorTask = celery_app.register_task(

apps/worker/tasks/tests/unit/test_bundle_analysis_notify_task.py

@@ -30,7 +30,7 @@ def test_bundle_analysis_notify_task(
 
     result = BundleAnalysisNotifyTask().run_impl(
         dbsession,
-        {"results": [{"error": None}]},
+        [{"error": None}],
         repoid=commit.repoid,
         commitid=commit.commitid,
         commit_yaml={},
@@ -47,7 +47,7 @@ def test_bundle_analysis_notify_skips_if_all_processing_fail(dbsession):
     dbsession.flush()
     result = BundleAnalysisNotifyTask().run_impl(
         dbsession,
-        {"results": [{"error": True}]},
+        [{"error": True}],
         repoid=commit.repoid,
         commitid=commit.commitid,
         commit_yaml={},