Revert "fix: use list for bundle analysis tasks"

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 list of processing results
-        # from prior processor tasks in the chain
-        previous_result: list[dict[str, Any]],
+        # Celery `chain` injects this argument - it's the returned result
+        # from the prior task in the chain
+        previous_result: 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: list[dict[str, Any]],
+        previous_result: dict[str, Any],
         **kwargs,
     ):
         log.info(
@@ -90,11 +90,9 @@ def process_impl_within_lock(
         )
         assert commit, "commit not found"
 
-        # previous_result is the list of processing results from prior processor tasks
+        # these are the task results from prior processor tasks in the chain
         # (they get accumulated as we execute each task in succession)
-        processing_results = (
-            previous_result if isinstance(previous_result, list) else []
-        )
+        processing_results = previous_result.get("results", [])
 
         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 list of processing results
-        # from prior tasks in the chain (accumulated as each task executes)
-        previous_result: list[dict[str, Any]],
+        # Celery `chain` injects this argument - it's the returned result
+        # from the prior task in the chain
+        previous_result: 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: list[dict[str, Any]],
+        previous_result: dict[str, Any],
     ):
         log.info(
             "Running bundle analysis processor",
@@ -100,11 +100,9 @@ def process_impl_within_lock(
 
         report_service = BundleAnalysisReportService(commit_yaml)
 
-        # previous_result is the list of processing results from prior processor tasks
+        # these are the task results from prior processor tasks in the chain
         # (they get accumulated as we execute each task in succession)
-        processing_results = (
-            previous_result if isinstance(previous_result, list) else []
-        )
+        processing_results = previous_result.get("results", [])
 
         # 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
@@ -143,7 +141,7 @@ def process_impl_within_lock(
                             "commit": commit.commitid,
                         },
                     )
-                    return processing_results
+                    return {"results": 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)
@@ -242,7 +240,7 @@ def process_impl_within_lock(
             },
         )
 
-        return processing_results
+        return {"results": 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,
-        [{"error": None}],
+        {"results": [{"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,
-        [{"error": True}],
+        {"results": [{"error": True}]},
         repoid=commit.repoid,
         commitid=commit.commitid,
         commit_yaml={},