feat: propagate MessageGroupId entry on bulk upload

Author: joybytes

app/job/rest.py

@@ -176,7 +176,12 @@ def create_job(service_id):
     sender_id = data.get("sender_id")
 
     if job.job_status == JOB_STATUS_PENDING:
-        process_job.apply_async([str(job.id)], {"sender_id": sender_id}, queue=QueueNames.JOBS)
+        process_job.apply_async(
+            [str(job.id)],
+            {"sender_id": sender_id},
+            queue=QueueNames.JOBS,
+            MessageGroupId=str(job.service_id),
+        )
 
     job_json = job_schema.dump(job)
     job_json["statistics"] = []

tests/app/job/test_rest.py

@@ -152,7 +152,10 @@ def test_create_unscheduled_job(client, sample_template, mocker, mock_celery_tas
     assert response.status_code == 201
 
     app.celery.tasks.process_job.apply_async.assert_called_once_with(
-        ([str(fake_uuid)]), {"sender_id": None}, queue="job-tasks"
+        [str(fake_uuid)],
+        {"sender_id": None},
+        queue="job-tasks",
+        MessageGroupId=str(sample_template.service_id),
     )
 
     resp_json = json.loads(response.get_data(as_text=True))
@@ -193,7 +196,10 @@ def test_create_unscheduled_job_with_sender_id_in_metadata(
     assert response.status_code == 201
 
     app.celery.tasks.process_job.apply_async.assert_called_once_with(
-        ([str(fake_uuid)]), {"sender_id": fake_uuid}, queue="job-tasks"
+        [str(fake_uuid)],
+        {"sender_id": fake_uuid},
+        queue="job-tasks",
+        MessageGroupId=str(sample_template.service_id),
     )