PYTHON-5055 - Convert test_client.py to unittest

.evergreen/run-tests.sh

@@ -31,7 +31,7 @@ set -o xtrace
 AUTH=${AUTH:-noauth}
 SSL=${SSL:-nossl}
 TEST_SUITES=${TEST_SUITES:-}
-TEST_ARGS="${*:1}"
+TEST_ARGS=("${*:1}")
 
 export PIP_QUIET=1  # Quiet by default
 export PIP_PREFER_BINARY=1 # Prefer binary dists by default
@@ -206,6 +206,7 @@ if [ -n "$TEST_INDEX_MANAGEMENT" ]; then
     TEST_SUITES="index_management"
 fi
 
+# shellcheck disable=SC2128
 if [ -n "$TEST_DATA_LAKE" ] && [ -z "$TEST_ARGS" ]; then
     TEST_SUITES="data_lake"
 fi
@@ -235,7 +236,7 @@ if [ -n "$PERF_TEST" ]; then
     TEST_SUITES="perf"
     # PYTHON-4769 Run perf_test.py directly otherwise pytest's test collection negatively
     # affects the benchmark results.
-    TEST_ARGS="test/performance/perf_test.py $TEST_ARGS"
+    TEST_ARGS+=("test/performance/perf_test.py")
 fi
 
 echo "Running $AUTH tests over $SSL with python $(uv python find)"
@@ -251,7 +252,7 @@ if [ -n "$COVERAGE" ] && [ "$PYTHON_IMPL" = "CPython" ]; then
     # Keep in sync with combine-coverage.sh.
     # coverage >=5 is needed for relative_files=true.
     UV_ARGS+=("--group coverage")
-    TEST_ARGS="$TEST_ARGS --cov"
+    TEST_ARGS+=("--cov")
 fi
 
 if [ -n "$GREEN_FRAMEWORK" ]; then
@@ -265,15 +266,35 @@ PIP_QUIET=0 uv run ${UV_ARGS[*]} --with pip pip list
 if [ -z "$GREEN_FRAMEWORK" ]; then
     # Use --capture=tee-sys so pytest prints test output inline:
     # https://docs.pytest.org/en/stable/how-to/capture-stdout-stderr.html
-    PYTEST_ARGS="-v --capture=tee-sys --durations=5 $TEST_ARGS"
+    PYTEST_ARGS=("-v" "--capture=tee-sys" "--durations=5" "${TEST_ARGS[@]}")
     if [ -n "$TEST_SUITES" ]; then
-      PYTEST_ARGS="-m $TEST_SUITES $PYTEST_ARGS"
+      # Workaround until unittest -> pytest conversion is complete
+      if [[ "$TEST_SUITES" == *"default_async"* ]]; then
+        ASYNC_PYTEST_ARGS=("-m asyncio" "--junitxml=xunit-results/TEST-asyncresults.xml" "${PYTEST_ARGS[@]}")
+      else
+        ASYNC_PYTEST_ARGS=("-m asyncio and $TEST_SUITES" "--junitxml=xunit-results/TEST-asyncresults.xml" "${PYTEST_ARGS[@]}")
+      fi
+      PYTEST_ARGS=("-m $TEST_SUITES and not asyncio" "${PYTEST_ARGS[@]}")
     fi
     # shellcheck disable=SC2048
-    uv run ${UV_ARGS[*]} pytest $PYTEST_ARGS
+    uv run ${UV_ARGS[*]} pytest "${PYTEST_ARGS[@]}"
+
+    # Workaround until unittest -> pytest conversion is complete
+    if [ -n "$TEST_SUITES" ]; then
+      set +o errexit
+      # shellcheck disable=SC2048
+      uv run ${UV_ARGS[*]} pytest "${ASYNC_PYTEST_ARGS[@]}" "--collect-only"
+      collected=$?
+      set -o errexit
+      # If we collected at least one async test, run all collected tests
+      if [ $collected -ne 5 ]; then
+        # shellcheck disable=SC2048
+        uv run ${UV_ARGS[*]} pytest "${ASYNC_PYTEST_ARGS[@]}"
+      fi
+    fi
 else
     # shellcheck disable=SC2048
-    uv run ${UV_ARGS[*]} green_framework_test.py $GREEN_FRAMEWORK -v $TEST_ARGS
+    uv run ${UV_ARGS[*]} green_framework_test.py $GREEN_FRAMEWORK -v "${TEST_ARGS[@]}"
 fi
 
 # Handle perf test post actions.

.github/workflows/test-python.yml

@@ -94,8 +94,10 @@ jobs:
         run: |
           if [[ "${{ matrix.python-version }}" == "3.13t" ]]; then
             pytest -v --durations=5 --maxfail=10
+            pytest -v --durations=5 --maxfail=10 -m asyncio
           else
             just test
+            just test-async
           fi
 
   doctest:

justfile

@@ -62,6 +62,10 @@ lint-manual:
 test *args="-v --durations=5 --maxfail=10":
     {{uv_run}} --extra test pytest {{args}}
 
+[group('test')]
+test-async *args="-v --durations=5 --maxfail=10 -m asyncio":
+    {{uv_run}} --extra test pytest {{args}}
+
 [group('test')]
 test-mockupdb *args:
     {{uv_run}} -v --extra test --group mockupdb pytest -m mockupdb {{args}}

pymongo/asynchronous/mongo_client.py

@@ -1418,6 +1418,9 @@ def __next__(self) -> NoReturn:
         raise TypeError("'AsyncMongoClient' object is not iterable")
 
     next = __next__
+    if not _IS_SYNC:
+        anext = next
+        __anext__ = next
 
     async def _server_property(self, attr_name: str) -> Any:
         """An attribute of the current server's description.

pymongo/synchronous/mongo_client.py

@@ -1414,6 +1414,9 @@ def __next__(self) -> NoReturn:
         raise TypeError("'MongoClient' object is not iterable")
 
     next = __next__
+    if not _IS_SYNC:
+        next = next
+        __next__ = next
 
     def _server_property(self, attr_name: str) -> Any:
         """An attribute of the current server's description.

pyproject.toml

@@ -135,6 +135,8 @@ markers = [
     "mockupdb: tests that rely on mockupdb",
     "default: default test suite",
     "default_async: default async test suite",
+    "unit: tests that don't require a connection to MongoDB",
+    "integration: tests that require a connection to MongoDB",
 ]
 
 [tool.mypy]

test/__init__.py

@@ -15,18 +15,14 @@
 """Synchronous test suite for pymongo, bson, and gridfs."""
 from __future__ import annotations
 
-import asyncio
 import gc
-import logging
 import multiprocessing
 import os
 import signal
 import socket
 import subprocess
 import sys
-import threading
 import time
-import traceback
 import unittest
 import warnings
 from asyncio import iscoroutinefunction
@@ -53,6 +49,7 @@
     sanitize_reply,
 )
 
+from pymongo.lock import _create_lock
 from pymongo.uri_parser import parse_uri
 
 try:
@@ -116,7 +113,7 @@ def __init__(self):
         self.default_client_options: Dict = {}
         self.sessions_enabled = False
         self.client = None  # type: ignore
-        self.conn_lock = threading.Lock()
+        self.conn_lock = _create_lock()
         self.is_data_lake = False
         self.load_balancer = TEST_LOADBALANCER
         self.serverless = TEST_SERVERLESS
@@ -518,6 +515,12 @@ def require_data_lake(self, func):
             func=func,
         )
 
+    @property
+    def is_not_mmap(self):
+        if self.is_mongos:
+            return True
+        return self.storage_engine != "mmapv1"
+
     def require_no_mmap(self, func):
         """Run a test only if the server is not using the MMAPv1 storage
         engine. Only works for standalone and replica sets; tests are
@@ -571,6 +574,10 @@ def require_replica_set(self, func):
         """Run a test only if the client is connected to a replica set."""
         return self._require(lambda: self.is_rs, "Not connected to a replica set", func=func)
 
+    @property
+    def secondaries_count(self):
+        return 0 if not self.client else len(self.client.secondaries)
+
     def require_secondaries_count(self, count):
         """Run a test only if the client is connected to a replica set that has
         `count` secondaries.
@@ -589,7 +596,7 @@ def supports_secondary_read_pref(self):
         if self.has_secondaries:
             return True
         if self.is_mongos:
-            shard = self.client.config.shards.find_one()["host"]  # type:ignore[index]
+            shard = (self.client.config.shards.find_one())["host"]  # type:ignore[index]
             num_members = shard.count(",") + 1
             return num_members > 1
         return False
@@ -690,7 +697,7 @@ def is_topology_type(self, topologies):
         if "sharded" in topologies and self.is_mongos:
             return True
         if "sharded-replicaset" in topologies and self.is_mongos:
-            shards = client_context.client.config.shards.find().to_list()
+            shards = self.client.config.shards.find().to_list()
             for shard in shards:
                 # For a 3-member RS-backed sharded cluster, shard['host']
                 # will be 'replicaName/ip1:port1,ip2:port2,ip3:port3'
@@ -864,14 +871,16 @@ def max_message_size_bytes(self):
 client_context = ClientContext()
 
 
-def reset_client_context():
-    if _IS_SYNC:
-        # sync tests don't need to reset a client context
-        return
-    elif client_context.client is not None:
-        client_context.client.close()
-        client_context.client = None
-    client_context._init_client()
+class PyMongoTestCasePyTest:
+    @contextmanager
+    def fail_point(self, client, command_args):
+        cmd_on = SON([("configureFailPoint", "failCommand")])
+        cmd_on.update(command_args)
+        client.admin.command(cmd_on)
+        try:
+            yield
+        finally:
+            client.admin.command("configureFailPoint", cmd_on["configureFailPoint"], mode="off")
 
 
 class PyMongoTestCase(unittest.TestCase):
@@ -1136,8 +1145,6 @@ class IntegrationTest(PyMongoTestCase):
 
     @client_context.require_connection
     def setUp(self) -> None:
-        if not _IS_SYNC:
-            reset_client_context()
         if client_context.load_balancer and not getattr(self, "RUN_ON_LOAD_BALANCER", False):
             raise SkipTest("this test does not support load balancers")
         if client_context.serverless and not getattr(self, "RUN_ON_SERVERLESS", False):

test/asynchronous/__init__.py

@@ -15,18 +15,14 @@
 """Asynchronous test suite for pymongo, bson, and gridfs."""
 from __future__ import annotations
 
-import asyncio
 import gc
-import logging
 import multiprocessing
 import os
 import signal
 import socket
 import subprocess
 import sys
-import threading
 import time
-import traceback
 import unittest
 import warnings
 from asyncio import iscoroutinefunction
@@ -53,6 +49,7 @@
     sanitize_reply,
 )
 
+from pymongo.lock import _async_create_lock
 from pymongo.uri_parser import parse_uri
 
 try:
@@ -116,7 +113,7 @@ def __init__(self):
         self.default_client_options: Dict = {}
         self.sessions_enabled = False
         self.client = None  # type: ignore
-        self.conn_lock = threading.Lock()
+        self.conn_lock = _async_create_lock()
         self.is_data_lake = False
         self.load_balancer = TEST_LOADBALANCER
         self.serverless = TEST_SERVERLESS
@@ -337,7 +334,7 @@ async def _init_client(self):
                         await mongos_client.close()
 
     async def init(self):
-        with self.conn_lock:
+        async with self.conn_lock:
             if not self.client and not self.connection_attempts:
                 await self._init_client()
 
@@ -520,6 +517,12 @@ def require_data_lake(self, func):
             func=func,
         )
 
+    @property
+    def is_not_mmap(self):
+        if self.is_mongos:
+            return True
+        return self.storage_engine != "mmapv1"
+
     def require_no_mmap(self, func):
         """Run a test only if the server is not using the MMAPv1 storage
         engine. Only works for standalone and replica sets; tests are
@@ -573,6 +576,10 @@ def require_replica_set(self, func):
         """Run a test only if the client is connected to a replica set."""
         return self._require(lambda: self.is_rs, "Not connected to a replica set", func=func)
 
+    @property
+    async def secondaries_count(self):
+        return 0 if not self.client else len(await self.client.secondaries)
+
     def require_secondaries_count(self, count):
         """Run a test only if the client is connected to a replica set that has
         `count` secondaries.
@@ -588,10 +595,10 @@ async def check():
 
     @property
     async def supports_secondary_read_pref(self):
-        if self.has_secondaries:
+        if await self.has_secondaries:
             return True
         if self.is_mongos:
-            shard = await self.client.config.shards.find_one()["host"]  # type:ignore[index]
+            shard = (await self.client.config.shards.find_one())["host"]  # type:ignore[index]
             num_members = shard.count(",") + 1
             return num_members > 1
         return False
@@ -692,7 +699,7 @@ async def is_topology_type(self, topologies):
         if "sharded" in topologies and self.is_mongos:
             return True
         if "sharded-replicaset" in topologies and self.is_mongos:
-            shards = await async_client_context.client.config.shards.find().to_list()
+            shards = await self.client.config.shards.find().to_list()
             for shard in shards:
                 # For a 3-member RS-backed sharded cluster, shard['host']
                 # will be 'replicaName/ip1:port1,ip2:port2,ip3:port3'
@@ -866,14 +873,18 @@ async def max_message_size_bytes(self):
 async_client_context = AsyncClientContext()
 
 
-async def reset_client_context():
-    if _IS_SYNC:
-        # sync tests don't need to reset a client context
-        return
-    elif async_client_context.client is not None:
-        await async_client_context.client.close()
-        async_client_context.client = None
-    await async_client_context._init_client()
+class AsyncPyMongoTestCasePyTest:
+    @asynccontextmanager
+    async def fail_point(self, client, command_args):
+        cmd_on = SON([("configureFailPoint", "failCommand")])
+        cmd_on.update(command_args)
+        await client.admin.command(cmd_on)
+        try:
+            yield
+        finally:
+            await client.admin.command(
+                "configureFailPoint", cmd_on["configureFailPoint"], mode="off"
+            )
 
 
 class AsyncPyMongoTestCase(unittest.IsolatedAsyncioTestCase):
@@ -1154,8 +1165,6 @@ class AsyncIntegrationTest(AsyncPyMongoTestCase):
 
     @async_client_context.require_connection
     async def asyncSetUp(self) -> None:
-        if not _IS_SYNC:
-            await reset_client_context()
         if async_client_context.load_balancer and not getattr(self, "RUN_ON_LOAD_BALANCER", False):
             raise SkipTest("this test does not support load balancers")
         if async_client_context.serverless and not getattr(self, "RUN_ON_SERVERLESS", False):