PYTHON-5344 and PYTHON-5403 Allow Instantiated MongoClients to Send Client Metadata On-Demand

.evergreen/resync-specs.sh

@@ -131,6 +131,9 @@ do
     gridfs)
       cpjson gridfs/tests gridfs
       ;;
+    handshake)
+      cpjson mongodb-handshake/tests handshake
+      ;;
     index|index-management)
       cpjson index-management/tests index_management
       ;;

test/asynchronous/test_client_metadata.py

@@ -14,9 +14,12 @@
 from __future__ import annotations
 
 import asyncio
+import os
+import pathlib
 import time
 import unittest
 from test.asynchronous import AsyncIntegrationTest
+from test.asynchronous.unified_format import generate_test_classes
 from test.utils_shared import CMAPListener
 from typing import Any, Optional
 
@@ -37,6 +40,17 @@
 
 _IS_SYNC = False
 
+# Location of JSON test specifications.
+if _IS_SYNC:
+    _TEST_PATH = os.path.join(pathlib.Path(__file__).resolve().parent, "handshake", "unified")
+else:
+    _TEST_PATH = os.path.join(
+        pathlib.Path(__file__).resolve().parent.parent, "handshake", "unified"
+    )
+
+# Generate unified tests.
+globals().update(generate_test_classes(_TEST_PATH, module=__name__, RUN_ON_SERVERLESS=True))
+
 
 def _get_handshake_driver_info(request):
     assert "client" in request

test/asynchronous/unified_format.py

@@ -75,6 +75,7 @@
 from pymongo.asynchronous.database import AsyncDatabase
 from pymongo.asynchronous.encryption import AsyncClientEncryption
 from pymongo.asynchronous.helpers import anext
+from pymongo.driver_info import DriverInfo
 from pymongo.encryption_options import _HAVE_PYMONGOCRYPT
 from pymongo.errors import (
     AutoReconnect,
@@ -703,6 +704,7 @@ async def _collectionOperation_createChangeStream(self, target, *args, **kwargs)
     async def _databaseOperation_runCommand(self, target, **kwargs):
         self.__raise_if_unsupported("runCommand", target, AsyncDatabase)
         # Ensure the first key is the command name.
+        print(kwargs)
         ordered_command = SON([(kwargs.pop("command_name"), 1)])
         ordered_command.update(kwargs["command"])
         kwargs["command"] = ordered_command
@@ -840,6 +842,13 @@ async def _cursor_close(self, target, *args, **kwargs):
         self.__raise_if_unsupported("close", target, NonLazyCursor, AsyncCommandCursor)
         return await target.close()
 
+    async def _clientOperation_appendMetadata(self, target, *args, **kwargs):
+        print("IN MY FUNC")
+        print(kwargs)
+        info_opts = kwargs["driver_info_options"]
+        driver_info = DriverInfo(info_opts["name"], info_opts["version"], info_opts["platform"])
+        target.append_metadata(driver_info)
+
     async def _clientEncryptionOperation_createDataKey(self, target, *args, **kwargs):
         if "opts" in kwargs:
             kwargs.update(camel_to_snake_args(kwargs.pop("opts")))
@@ -925,11 +934,11 @@ async def run_entity_operation(self, spec):
             )
         else:
             arguments = {}
-
         if isinstance(target, AsyncMongoClient):
             method_name = f"_clientOperation_{opname}"
         elif isinstance(target, AsyncDatabase):
             method_name = f"_databaseOperation_{opname}"
+            print(f"{method_name=}")
         elif isinstance(target, AsyncCollection):
             method_name = f"_collectionOperation_{opname}"
             # contentType is always stored in metadata in pymongo.
@@ -976,6 +985,7 @@ async def run_entity_operation(self, spec):
                 with pymongo.timeout(timeout):
                     result = await cmd(**dict(arguments))
             else:
+                print(f"{cmd=} {dict=} {arguments=}")
                 result = await cmd(**dict(arguments))
         except Exception as exc:
             # Ignore all operation errors but to avoid masking bugs don't
@@ -1238,6 +1248,7 @@ async def run_special_operation(self, spec):
 
     async def run_operations(self, spec):
         for op in spec:
+            print(f"{op=}")
             if op["object"] == "testRunner":
                 await self.run_special_operation(op)
             else:
@@ -1440,6 +1451,7 @@ async def _run_scenario(self, spec, uri=None):
             await self.check_log_messages(spec["operations"], expect_log_messages)
         else:
             # process operations
+            print(f"{spec['operations']=}")
             await self.run_operations(spec["operations"])
 
         # process expectEvents

test/handshake/unified/metadata-not-propagated.json

@@ -0,0 +1,100 @@
+{
+  "description": "client metadata is not propagated to the server",
+  "schemaVersion": "1.9",
+  "runOnRequirements": [
+    {
+      "minServerVersion": "6.0"
+    }
+  ],
+  "createEntities": [
+    {
+      "client": {
+        "id": "client",
+        "observeEvents": [
+          "commandSucceededEvent",
+          "commandFailedEvent",
+          "connectionClosedEvent",
+          "connectionCreatedEvent"
+        ]
+      }
+    },
+    {
+      "database": {
+        "id": "database",
+        "client": "client",
+        "databaseName": "test"
+      }
+    }
+  ],
+  "tests": [
+    {
+      "description": "metadata append does not create new connections or close existing ones and no hello command is sent",
+      "operations": [
+        {
+          "name": "runCommand",
+          "object": "database",
+          "arguments": {
+            "commandName": "ping",
+            "command": {
+              "ping": 1
+            }
+          },
+          "expectResult": {
+            "ok": 1
+          }
+        },
+        {
+          "name": "appendMetadata",
+          "object": "client",
+          "arguments": {
+            "driverInfoOptions": {
+              "name": "framework",
+              "version": "2.0",
+              "platform": "Framework Platform"
+            }
+          }
+        },
+        {
+          "name": "runCommand",
+          "object": "database",
+          "arguments": {
+            "commandName": "ping",
+            "command": {
+              "ping": 1
+            }
+          },
+          "expectResult": {
+            "ok": 1
+          }
+        }
+      ],
+      "expectEvents": [
+        {
+          "client": "client",
+          "eventType": "cmap",
+          "events": [
+            {
+              "connectionCreatedEvent": {}
+            }
+          ]
+        },
+        {
+          "client": "client",
+          "eventType": "command",
+          "events": [
+            {
+              "commandSucceededEvent": {
+                "commandName": "ping"
+              }
+            },
+            {
+              "commandSucceededEvent": {
+                "commandName": "ping"
+              }
+            }
+          ]
+        }
+      ]
+    }
+  ]
+}

test/test_client_metadata.py

@@ -14,9 +14,12 @@
 from __future__ import annotations
 
 import asyncio
+import os
+import pathlib
 import time
 import unittest
 from test import IntegrationTest
+from test.unified_format import generate_test_classes
 from test.utils_shared import CMAPListener
 from typing import Any, Optional
 
@@ -37,6 +40,17 @@
 
 _IS_SYNC = True
 
+# Location of JSON test specifications.
+if _IS_SYNC:
+    _TEST_PATH = os.path.join(pathlib.Path(__file__).resolve().parent, "handshake", "unified")
+else:
+    _TEST_PATH = os.path.join(
+        pathlib.Path(__file__).resolve().parent.parent, "handshake", "unified"
+    )
+
+# Generate unified tests.
+globals().update(generate_test_classes(_TEST_PATH, module=__name__, RUN_ON_SERVERLESS=True))
+
 
 def _get_handshake_driver_info(request):
     assert "client" in request

test/unified_format.py

@@ -67,6 +67,7 @@
 from bson.objectid import ObjectId
 from gridfs import GridFSBucket, GridOut, NoFile
 from pymongo import ASCENDING, CursorType, MongoClient, _csot
+from pymongo.driver_info import DriverInfo
 from pymongo.encryption_options import _HAVE_PYMONGOCRYPT
 from pymongo.errors import (
     AutoReconnect,
@@ -700,6 +701,7 @@ def _collectionOperation_createChangeStream(self, target, *args, **kwargs):
     def _databaseOperation_runCommand(self, target, **kwargs):
         self.__raise_if_unsupported("runCommand", target, Database)
         # Ensure the first key is the command name.
+        print(kwargs)
         ordered_command = SON([(kwargs.pop("command_name"), 1)])
         ordered_command.update(kwargs["command"])
         kwargs["command"] = ordered_command
@@ -837,6 +839,13 @@ def _cursor_close(self, target, *args, **kwargs):
         self.__raise_if_unsupported("close", target, NonLazyCursor, CommandCursor)
         return target.close()
 
+    def _clientOperation_appendMetadata(self, target, *args, **kwargs):
+        print("IN MY FUNC")
+        print(kwargs)
+        info_opts = kwargs["driver_info_options"]
+        driver_info = DriverInfo(info_opts["name"], info_opts["version"], info_opts["platform"])
+        target.append_metadata(driver_info)
+
     def _clientEncryptionOperation_createDataKey(self, target, *args, **kwargs):
         if "opts" in kwargs:
             kwargs.update(camel_to_snake_args(kwargs.pop("opts")))
@@ -916,11 +925,11 @@ def run_entity_operation(self, spec):
             )
         else:
             arguments = {}
-
         if isinstance(target, MongoClient):
             method_name = f"_clientOperation_{opname}"
         elif isinstance(target, Database):
             method_name = f"_databaseOperation_{opname}"
+            print(f"{method_name=}")
         elif isinstance(target, Collection):
             method_name = f"_collectionOperation_{opname}"
             # contentType is always stored in metadata in pymongo.
@@ -967,6 +976,7 @@ def run_entity_operation(self, spec):
                 with pymongo.timeout(timeout):
                     result = cmd(**dict(arguments))
             else:
+                print(f"{cmd=} {dict=} {arguments=}")
                 result = cmd(**dict(arguments))
         except Exception as exc:
             # Ignore all operation errors but to avoid masking bugs don't
@@ -1225,6 +1235,7 @@ def run_special_operation(self, spec):
 
     def run_operations(self, spec):
         for op in spec:
+            print(f"{op=}")
             if op["object"] == "testRunner":
                 self.run_special_operation(op)
             else:
@@ -1425,6 +1436,7 @@ def _run_scenario(self, spec, uri=None):
             self.check_log_messages(spec["operations"], expect_log_messages)
         else:
             # process operations
+            print(f"{spec['operations']=}")
             self.run_operations(spec["operations"])
 
         # process expectEvents