instrumentation/asyncio: catch CancelledError exception in tests (#2690)

* instrumentation/asyncio: catch CancelledError exception in tests

After a29242f49386c097defce500b138dc00f06ce300 we are re-raising the
CancelledError so we need to catch it on the caller side.

Fix #2688

* instrument/asyncio: don't test anext on python < 3.10

Since it's not available there.

* instrumentation/asyncio: use unittest.skipIf instead of reimplementing it

---------

Co-authored-by: Leighton Chen <lechen@microsoft.com>

instrumentation/opentelemetry-instrumentation-asyncio/tests/test_asyncio_anext.py

@@ -12,6 +12,8 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 import asyncio
+import sys
+from unittest import skipIf
 from unittest.mock import patch
 
 # pylint: disable=no-name-in-module
@@ -41,6 +43,9 @@ class TestAsyncioAnext(TestBase):
 
     # Asyncio anext() does not have __name__ attribute, which is used to determine if the coroutine should be traced.
     # This test is to ensure that the instrumentation does not break when the coroutine does not have __name__ attribute.
+    @skipIf(
+        sys.version_info < (3, 10), "anext is only available in Python 3.10+"
+    )
     def test_asyncio_anext(self):
         async def main():
             async def async_gen():

instrumentation/opentelemetry-instrumentation-asyncio/tests/test_asyncio_cancellation.py

@@ -45,7 +45,10 @@ class TestAsyncioCancel(TestBase):
 
     def test_cancel(self):
         with self._tracer.start_as_current_span("root", kind=SpanKind.SERVER):
-            asyncio.run(cancellation_create_task())
+            try:
+                asyncio.run(cancellation_create_task())
+            except asyncio.CancelledError:
+                pass
         spans = self.memory_exporter.get_finished_spans()
         self.assertEqual(len(spans), 3)
         self.assertEqual(spans[0].context.trace_id, spans[1].context.trace_id)

instrumentation/opentelemetry-instrumentation-asyncio/tests/test_asyncio_taskgroup.py

@@ -13,6 +13,7 @@
 # limitations under the License.
 import asyncio
 import sys
+from unittest import skipIf
 from unittest.mock import patch
 
 # pylint: disable=no-name-in-module
@@ -25,10 +26,6 @@ from opentelemetry.trace import get_tracer
 
 from .common_test_func import async_func
 
-py11 = False
-if sys.version_info >= (3, 11):
-    py11 = True
-
 
 class TestAsyncioTaskgroup(TestBase):
     @patch.dict(
@@ -46,11 +43,11 @@ class TestAsyncioTaskgroup(TestBase):
         super().tearDown()
         AsyncioInstrumentor().uninstrument()
 
+    @skipIf(
+        sys.version_info < (3, 11),
+        "TaskGroup is only available in Python 3.11+",
+    )
     def test_task_group_create_task(self):
-        # TaskGroup is only available in Python 3.11+
-        if not py11:
-            return
-
         async def main():
             async with asyncio.TaskGroup() as tg:  # pylint: disable=no-member
                 for _ in range(10):

instrumentation/opentelemetry-instrumentation-asyncio/tests/test_asyncio_to_thread.py

@@ -13,6 +13,7 @@
 # limitations under the License.
 import asyncio
 import sys
+from unittest import skipIf
 from unittest.mock import patch
 
 # pylint: disable=no-name-in-module
@@ -40,34 +41,34 @@ class TestAsyncioToThread(TestBase):
         super().tearDown()
         AsyncioInstrumentor().uninstrument()
 
+    @skipIf(
+        sys.version_info < (3, 9), "to_thread is only available in Python 3.9+"
+    )
     def test_to_thread(self):
-        # to_thread is only available in Python 3.9+
+        def multiply(x, y):
-        if sys.version_info >= (3, 9):
+            return x * y
 
-            def multiply(x, y):
+        async def to_thread():
-                return x * y
+            result = await asyncio.to_thread(multiply, 2, 3)
+            assert result == 6
 
-            async def to_thread():
+        with self._tracer.start_as_current_span("root"):
-                result = await asyncio.to_thread(multiply, 2, 3)
+            asyncio.run(to_thread())
-                assert result == 6
+        spans = self.memory_exporter.get_finished_spans()
 
-            with self._tracer.start_as_current_span("root"):
+        self.assertEqual(len(spans), 2)
-                asyncio.run(to_thread())
+        assert spans[0].name == "asyncio to_thread-multiply"
-            spans = self.memory_exporter.get_finished_spans()
+        for metric in (
-
+            self.memory_metrics_reader.get_metrics_data()
-            self.assertEqual(len(spans), 2)
+            .resource_metrics[0]
-            assert spans[0].name == "asyncio to_thread-multiply"
+            .scope_metrics[0]
-            for metric in (
+            .metrics
-                self.memory_metrics_reader.get_metrics_data()
+        ):
-                .resource_metrics[0]
+            if metric.name == "asyncio.process.duration":
-                .scope_metrics[0]
+                for point in metric.data.data_points:
-                .metrics
+                    self.assertEqual(point.attributes["type"], "to_thread")
-            ):
+                    self.assertEqual(point.attributes["name"], "multiply")
-                if metric.name == "asyncio.process.duration":
+            if metric.name == "asyncio.process.created":
-                    for point in metric.data.data_points:
+                for point in metric.data.data_points:
-                        self.assertEqual(point.attributes["type"], "to_thread")
+                    self.assertEqual(point.attributes["type"], "to_thread")
-                        self.assertEqual(point.attributes["name"], "multiply")
+                    self.assertEqual(point.attributes["name"], "multiply")
-                if metric.name == "asyncio.process.created":
-                    for point in metric.data.data_points:
-                        self.assertEqual(point.attributes["type"], "to_thread")
-                        self.assertEqual(point.attributes["name"], "multiply")