PYTHON-6036 Vendor the OpenTelemetry unified spec tests

.evergreen/resync-specs.sh

@@ -127,6 +127,9 @@ do
       cpjson command-logging-and-monitoring/tests/logging command_logging
       cpjson command-logging-and-monitoring/tests/monitoring command_monitoring
       ;;
+    open-telemetry|otel|open_telemetry)
+      cpjson open-telemetry/tests open_telemetry
+      ;;
     crud|CRUD)
       cpjson crud/tests/ crud
       ;;

test/asynchronous/test_open_telemetry_unified.py

@@ -0,0 +1,40 @@
+# Copyright 2026-present MongoDB, Inc.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+"""Run the OpenTelemetry unified format spec tests."""
+
+from __future__ import annotations
+
+import sys
+
+sys.path[0:0] = [""]
+
+import pytest
+
+from test import unittest
+from test.asynchronous.unified_format import generate_test_classes, get_test_path
+
+_IS_SYNC = False
+
+pytestmark = pytest.mark.otel
+
+globals().update(
+    generate_test_classes(
+        get_test_path("open_telemetry"),
+        module=__name__,
+    )
+)
+
+if __name__ == "__main__":
+    unittest.main()

test/asynchronous/unified_format.py

@@ -38,6 +38,7 @@
 import pytest
 
 import pymongo
+import pymongo._otel as _otel
 from bson import SON, json_util
 from bson.codec_options import DEFAULT_CODEC_OPTIONS
 from bson.objectid import ObjectId
@@ -93,6 +94,7 @@
     PLACEHOLDER_MAP,
     EventListenerUtil,
     MatchEvaluatorUtil,
+    _shared_test_provider,
     coerce_result,
     parse_bulk_write_error_result,
     parse_bulk_write_result,
@@ -113,6 +115,16 @@
 
 _IS_SYNC = False
 
+_HAS_OTEL_TEST_DEPS = False
+if _otel._HAS_OPENTELEMETRY:
+    try:
+        from opentelemetry.sdk.trace.export import SimpleSpanProcessor
+        from opentelemetry.sdk.trace.export.in_memory_span_exporter import InMemorySpanExporter
+
+        _HAS_OTEL_TEST_DEPS = True
+    except ImportError:
+        pass
+
 IS_INTERRUPTED = False
 
 
@@ -230,6 +242,11 @@ def __init__(self, test_class):
         self._entities: dict[str, Any] = {}
         self._listeners: dict[str, EventListenerUtil] = {}
         self._session_lsids: dict[str, Mapping[str, Any]] = {}
+        # The id of the (at most one, today) client entity created with
+        # observeTracingMessages. Spans carry no attribute identifying which
+        # client emitted them, so multi-client tracing correlation isn't
+        # supported; _create_entity fails loudly if a second one appears.
+        self._tracing_client_id: Optional[str] = None
         self.test: UnifiedSpecTestMixinV1 = test_class
 
     def __contains__(self, item):
@@ -311,6 +328,25 @@ async def _create_entity(self, entity_spec, uri=None):
             )
             self._listeners[spec["id"]] = listener
             kwargs["event_listeners"] = [listener]
+
+            observe_tracing = spec.get("observeTracingMessages")
+            if observe_tracing is not None:
+                if self._tracing_client_id is not None:
+                    self.test.fail(
+                        "Multiple clients with observeTracingMessages are not supported "
+                        f"by the unified test format runner (already tracking "
+                        f"{self._tracing_client_id!r}, got {spec['id']!r})"
+                    )
+                self._tracing_client_id = spec["id"]
+                enable_payload = observe_tracing.get("enableCommandPayload", False)
+                kwargs["tracing"] = {
+                    "enabled": True,
+                    # Tests asserting db.query.text match the full, untruncated
+                    # command, so an effectively-unlimited length avoids
+                    # truncating and failing that assertion.
+                    "query_text_max_length": 1_000_000 if enable_payload else None,
+                }
+
             if spec.get("useMultipleMongoses"):
                 if async_client_context.load_balancer:
                     kwargs["h"] = async_client_context.MULTI_MONGOS_LB_URI
@@ -482,6 +518,8 @@ class UnifiedSpecTestMixinV1(AsyncIntegrationTest):
     TEST_SPEC: Any
     TEST_PATH = ""  # This gets filled in by generate_test_classes
     mongos_clients: list[AsyncMongoClient] = []
+    # Set in setUpClass, only for test files that use observeTracingMessages.
+    _tracing_exporter: Optional[Any] = None
 
     @staticmethod
     async def should_run_on(run_on_spec):
@@ -526,6 +564,23 @@ async def insert_initial_data(self, initial_data):
 
     @classmethod
     def setUpClass(cls) -> None:
+        # Only register a span exporter (and the shared SDK TracerProvider it
+        # depends on) for test files that actually use observeTracingMessages,
+        # to avoid needlessly accumulating span processors on the process-wide
+        # provider for the (vast majority of) unified-format suites that don't.
+        cls._tracing_exporter = None
+        uses_tracing = any(
+            "observeTracingMessages" in entity.get("client", {})
+            for entity in cls.TEST_SPEC.get("createEntities", [])
+        )
+        if uses_tracing:
+            if not _HAS_OTEL_TEST_DEPS:
+                raise unittest.SkipTest(
+                    "observeTracingMessages requires opentelemetry-sdk to be installed"
+                )
+            cls._tracing_exporter = InMemorySpanExporter()
+            _shared_test_provider().add_span_processor(SimpleSpanProcessor(cls._tracing_exporter))
+
         # Speed up the tests by decreasing the heartbeat frequency.
         cls.knobs = client_knobs(
             heartbeat_frequency=0.1,
@@ -538,6 +593,14 @@ def setUpClass(cls) -> None:
     @classmethod
     def tearDownClass(cls) -> None:
         cls.knobs.disable()
+        # The exporter's span processor can never be removed from the shared process-wide
+        # TracerProvider (see _shared_test_provider), so without this, every span emitted by any
+        # client anywhere in the process for the rest of the test run keeps getting appended to this
+        # (otherwise dead) class's exporter: an unbounded memory leak across a full test run, and
+        # needless per-span export overhead for every other tracing-enabled test class that runs
+        # afterwards. shutdown() makes further export() calls into this exporter no-ops.
+        if cls._tracing_exporter is not None:
+            cls._tracing_exporter.shutdown()
 
     async def asyncSetUp(self):
         # super call creates internal client cls.client
@@ -576,6 +639,14 @@ def maybe_skip_test(self, spec):
             self.skipTest("PyMongo does not support the symbol type")
         if "timeoutms applied to entire download" in description:
             self.skipTest("PyMongo's open_download_stream does not cap the stream's lifetime")
+        # Removed API: PyMongo no longer exposes map_reduce/inline_map_reduce at
+        # all (mapReduce is deprecated server-side), so there's no code path left
+        # that could send this command; this operation can never be exercised.
+        if class_name == "testoperationmapreduce" and description == "mapreduce":
+            self.skipTest(
+                "PyMongo removed the map_reduce/inline_map_reduce Collection methods "
+                "(mapReduce is deprecated server-side); this operation cannot be exercised"
+            )
         if any(
             x in description
             for x in [
@@ -1463,6 +1534,77 @@ def format_logs(log_list):
                     self.match_evaluator.match_result(expected_data, actual_data)
                     self.match_evaluator.match_result(expected_msg, actual_msg)
 
+    async def check_tracing_messages(self, operations, spec):
+        # Like expectLogMessages/expectEvents, expectTracingMessages is a list of
+        # per-client blocks (even though only one client with
+        # observeTracingMessages is currently supported, see entity.py above).
+        exporter = self._tracing_exporter
+        if exporter is None:
+            self.fail(
+                "expectTracingMessages requires a client entity created with observeTracingMessages"
+            )
+
+        exporter.clear()
+        await self.run_operations(operations)
+        finished_spans = exporter.get_finished_spans()
+
+        # Reconstruct the parent/child span tree from the flat, finish-ordered
+        # list the in-memory exporter records, keyed by each span's parent id.
+        children_by_parent_id = defaultdict(list)
+        for span in finished_spans:
+            parent_id = span.parent.span_id if span.parent is not None else None
+            children_by_parent_id[parent_id].append(span)
+
+        def check_span_list(expected_list, actual_list, ignore_extra_spans):
+            if ignore_extra_spans:
+                # Per the unified-test-format spec, "additional unexpected spans
+                # are allowed". Unlike ignoreExtraEvents (which only tolerates
+                # a trailing tail), spans from concurrent/out-of-band activity
+                # (e.g. a testRunner-issued configureFailPoint command) can
+                # finish interleaved anywhere among the expected ones, not just
+                # at the end. Filter down to just the spans that line up (by
+                # name, in order) with the expected list, dropping anything
+                # else, instead of naively truncating the tail.
+                filtered = []
+                expected_iter = iter(expected_list)
+                current_expected = next(expected_iter, None)
+                for actual in actual_list:
+                    if current_expected is not None and actual.name == current_expected["name"]:
+                        filtered.append(actual)
+                        current_expected = next(expected_iter, None)
+                actual_list = filtered
+            self.assertEqual(
+                len(expected_list),
+                len(actual_list),
+                f"expected spans {[e['name'] for e in expected_list]} but got "
+                f"{[a.name for a in actual_list]}",
+            )
+            for expected, actual in zip(expected_list, actual_list):
+                self.assertEqual(expected["name"], actual.name)
+                self.match_evaluator.match_span_attributes(
+                    expected["attributes"], actual.attributes
+                )
+                expected_nested = expected.get("nested")
+                if expected_nested is not None:
+                    actual_children = children_by_parent_id[actual.context.span_id]
+                    check_span_list(expected_nested, actual_children, ignore_extra_spans)
+
+        for client_spec in spec:
+            expected_client_id = client_spec["client"]
+            tracing_client_id = self.entity_map._tracing_client_id
+            self.assertEqual(
+                expected_client_id,
+                tracing_client_id,
+                f"expectTracingMessages.client {expected_client_id!r} does not match the "
+                f"client with observeTracingMessages enabled ({tracing_client_id!r})",
+            )
+
+            ignore_extra_spans = client_spec.get("ignoreExtraSpans", False)
+            expected_spans = client_spec["spans"]
+            self.assertTrue(expected_spans, "expectTracingMessages spans must be non-empty")
+
+            check_span_list(expected_spans, children_by_parent_id[None], ignore_extra_spans)
+
     async def verify_outcome(self, spec):
         for collection_data in spec:
             coll_name = collection_data["collectionName"]
@@ -1551,6 +1693,10 @@ async def _run_scenario(self, spec, uri=None):
             expect_log_messages = spec["expectLogMessages"]
             self.assertTrue(expect_log_messages, "expectEvents must be non-empty")
             await self.check_log_messages(spec["operations"], expect_log_messages)
+        elif "expectTracingMessages" in spec:
+            expect_tracing_messages = spec["expectTracingMessages"]
+            self.assertTrue(expect_tracing_messages, "expectTracingMessages must be non-empty")
+            await self.check_tracing_messages(spec["operations"], expect_tracing_messages)
         else:
             # process operations
             await self.run_operations(spec["operations"])