Move Celery error tests to the functional tests

The celery tests failed on Python 3.11. This is most likely due to this
issue in billiard, a celery dependency, about it not working on Python
3.11 because of the error reported in the CI:
celery/billiard#377

It's been fixed in billiard 4.1.0, but celery is locked on billiard
version lower than 4, so it cannot use this version with the fix.

This issue does not arise on the Docker tests, because they use Python
3.9.16.

I've moved the error test span event assertions to the error test that
is available in the functional tests, and removed the unit test. That
way, the build will run successfully.

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

@@ -83,64 +83,6 @@ 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_task_with_error(self):
-        CeleryInstrumentor().instrument()
-
-        result = task_error.delay()
-
-        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_error"
-        )
-        self.assertEqual(consumer.kind, SpanKind.CONSUMER)
-        self.assertSpanHasAttributes(
-            consumer,
-            {
-                "celery.action": "run",
-                "celery.state": "FAILURE",
-                SpanAttributes.MESSAGING_DESTINATION: "celery",
-                "celery.task_name": "tests.celery_test_tasks.task_error",
-            },
-        )
-        self.assertEqual(consumer.status.status_code, StatusCode.ERROR)
-        self.assertEqual(len(consumer.events), 1)
-        event = consumer.events[0]
-        self.assertEqual(event.name, "exception")
-        self.assertEqual(
-            event.attributes[SpanAttributes.EXCEPTION_TYPE], "ExceptionInfo"
-        )
-        self.assertIn(SpanAttributes.EXCEPTION_MESSAGE, event.attributes)
-
-        self.assertEqual(
-            producer.name, "apply_async/tests.celery_test_tasks.task_error"
-        )
-        self.assertEqual(producer.kind, SpanKind.PRODUCER)
-        self.assertSpanHasAttributes(
-            producer,
-            {
-                "celery.action": "apply_async",
-                "celery.task_name": "tests.celery_test_tasks.task_error",
-                SpanAttributes.MESSAGING_DESTINATION_KIND: "queue",
-                SpanAttributes.MESSAGING_DESTINATION: "celery",
-            },
-        )
-        self.assertEqual(producer.status.status_code, StatusCode.UNSET)
-        self.assertEqual(len(producer.events), 0)
-
-        self.assertNotEqual(consumer.parent, producer.context)
-        self.assertEqual(consumer.parent.span_id, producer.context.span_id)
-        self.assertEqual(consumer.context.trace_id, producer.context.trace_id)
-
     def test_uninstrument(self):
         CeleryInstrumentor().instrument()
         CeleryInstrumentor().uninstrument()

tests/opentelemetry-docker-tests/tests/celery/test_celery_functional.py

@@ -249,6 +249,7 @@ def fn_task_parameters(user, force_logout=False):
         run_span.attributes.get("celery.task_name")
         == "test_celery_functional.fn_task_parameters"
     )
+    assert len(run_span.events) == 0
 
 
 def test_fn_exception(celery_app, memory_exporter):
@@ -275,6 +276,11 @@ def fn_exception():
         == "test_celery_functional.fn_exception"
     )
     assert span.status.status_code == StatusCode.ERROR
+    assert len(span.events) == 1
+    event = span.events[0]
+    assert event.name == "exception"
+    assert event.attributes[SpanAttributes.EXCEPTION_TYPE] == "ExceptionInfo"
+    assert SpanAttributes.EXCEPTION_MESSAGE in event.attributes
     assert (
         span.attributes.get(SpanAttributes.MESSAGING_MESSAGE_ID)
         == result.task_id