test(arq): Update streaming tests for enqueue guard

The retry re-enqueue happens without an active span, so no producer
(queue.submit.arq) span is created for it. Adjust the positional span
assertions in test_job_retry and test_span_origin_consumer to account
for the suppressed producer span.

Author: ericapisani

tests/integrations/arq/test_arq.py

@@ -245,18 +245,22 @@ async def retry_job(ctx):
         sentry_sdk.flush()
         spans = [item.payload for item in items]
 
-        assert spans[4]["attributes"]["sentry.op"] == "queue.task.arq"
-        assert spans[4]["status"] == "ok"
-        assert spans[4]["name"] == "retry_job"
+        # The retry re-enqueue happens without an active span, so no producer
+        # (queue.submit.arq) span is created for it; only the consumer segment
+        # is emitted. The consumer segment is preceded by the redis spans for
+        # the re-enqueue, so it lands at index 3.
+        assert spans[3]["attributes"]["sentry.op"] == "queue.task.arq"
+        assert spans[3]["status"] == "ok"
+        assert spans[3]["name"] == "retry_job"
 
         await worker.run_job(job.job_id, timestamp_ms())
 
         sentry_sdk.flush()
         spans = [item.payload for item in items]
 
-        assert spans[7]["attributes"]["sentry.op"] == "queue.task.arq"
-        assert spans[7]["status"] == "ok"
-        assert spans[7]["name"] == "retry_job"
+        assert spans[6]["attributes"]["sentry.op"] == "queue.task.arq"
+        assert spans[6]["status"] == "ok"
+        assert spans[6]["name"] == "retry_job"
     else:
         events = capture_events()
 
@@ -594,10 +598,13 @@ async def job(ctx):
         sentry_sdk.flush()
         spans = [item.payload for item in items]
 
-        assert spans[4]["attributes"]["sentry.op"] == "queue.task.arq"
-        assert spans[4]["attributes"]["sentry.origin"] == "auto.queue.arq"
-        assert spans[3]["attributes"]["sentry.origin"] == "auto.db.redis"
+        # No producer (queue.submit.arq) span is created for the re-enqueue
+        # triggered by the retry, since it happens without an active span, so
+        # the consumer segment lands at index 3.
+        assert spans[3]["attributes"]["sentry.op"] == "queue.task.arq"
+        assert spans[3]["attributes"]["sentry.origin"] == "auto.queue.arq"
         assert spans[2]["attributes"]["sentry.origin"] == "auto.db.redis"
+        assert spans[1]["attributes"]["sentry.origin"] == "auto.db.redis"
     else:
         job = await pool.enqueue_job("retry_job")