test: add unit tests

Signed-off-by: Shivanshu Raj Shrivastava <[email protected]>

instrumentation/opentelemetry-instrumentation-celery/src/opentelemetry/instrumentation/celery/__init__.py

@@ -96,7 +96,7 @@ def add(x, y):
 _TASK_REVOKED_TERMINATED_SIGNAL_KEY = "celery.terminated.signal"
 _TASK_NAME_KEY = "celery.task_name"
 
-_QUEUE_NAME = "queue"
+_QUEUE_NAME = "celery"
 
 
 class CeleryGetter(Getter):

instrumentation/opentelemetry-instrumentation-celery/tests/test_tasks.py

@@ -92,6 +92,35 @@ def test_task(self):
         self.assertEqual(consumer.parent.span_id, producer.context.span_id)
         self.assertEqual(consumer.context.trace_id, producer.context.trace_id)
 
+    def test_queue_name(self):
+        CeleryInstrumentor().instrument()
+
+        result = task_add.delay(1, 2)
+
+        timeout = time.time() + 60 * 1  # 1 minutes from now
+        while not result.ready():
+            if time.time() > timeout:
+                break
+            time.sleep(0.05)
+
+        spans = self.sorted_spans(self.memory_exporter.get_finished_spans())
+        self.assertEqual(len(spans), 2)
+
+        consumer, producer = spans
+
+        self.assertEqual(consumer.name, "run/tests.celery_test_tasks.task_add")
+        self.assertEqual(consumer.kind, SpanKind.CONSUMER)
+        self.assertSpanHasAttributes(
+            consumer,
+            {
+                "celery.action": "run",
+                "celery.state": "SUCCESS",
+                SpanAttributes.MESSAGING_DESTINATION: "celery",
+                SpanAttributes.MESSAGING_SYSTEM: "celery",
+                "celery.task_name": "tests.celery_test_tasks.task_add",
+            },
+        )
+
     def test_task_raises(self):
         CeleryInstrumentor().instrument()