PYTHON-5366 - test_pool_reset waits until Pool.reset() increments gen… (#2797)

Co-authored-by: Qi Deng <qdeng@aurascape.ai>

.evergreen/generated_configs/variants.yml

@@ -615,6 +615,7 @@ buildvariants:
   - name: test-win64
     tasks:
       - name: .test-standard !.pypy
+      - name: .test-no-orchestration !.pypy
     display_name: "* Test Win64"
     run_on:
       - windows-2022-latest-small

.evergreen/scripts/generate_config.py

@@ -97,6 +97,8 @@ def create_standard_nonlinux_variants() -> list[BuildVariant]:
             tasks = [
                 f".test-standard !.pypy .server-{version}" for version in get_versions_from("6.0")
             ]
+        if host_name == "win64":
+            tasks.append(".test-no-orchestration !.pypy")
         host = HOSTS[host_name]
         tags = ["standard-non-linux"]
         expansions = dict()

.evergreen/spec-patch/PYTHON-2673.patch

@@ -1,64 +0,0 @@
-diff --git a/test/load_balancer/cursors.json b/test/load_balancer/cursors.json
-index 43e4fbb4f..4e2a55fd4 100644
---- a/test/load_balancer/cursors.json
-+++ b/test/load_balancer/cursors.json
-@@ -376,7 +376,7 @@
-       ]
-     },
-     {
-+      "description": "pinned connections are not returned after an network error during getMore",
--      "description": "pinned connections are returned after an network error during getMore",
-       "operations": [
-         {
-           "name": "failPoint",
-@@ -440,7 +440,7 @@
-           "object": "testRunner",
-           "arguments": {
-             "client": "client0",
-+            "connections": 1
--            "connections": 0
-           }
-         },
-         {
-@@ -659,7 +659,7 @@
-       ]
-     },
-     {
-+      "description": "pinned connections are not returned to the pool after a non-network error on getMore",
--      "description": "pinned connections are returned to the pool after a non-network error on getMore",
-       "operations": [
-         {
-           "name": "failPoint",
-@@ -715,7 +715,7 @@
-           "object": "testRunner",
-           "arguments": {
-             "client": "client0",
-+            "connections": 1
--            "connections": 0
-           }
-         },
-         {
-diff --git a/test/load_balancer/sdam-error-handling.json b/test/load_balancer/sdam-error-handling.json
-index 63aabc04d..462fa0aac 100644
---- a/test/load_balancer/sdam-error-handling.json
-+++ b/test/load_balancer/sdam-error-handling.json
-@@ -366,6 +366,9 @@
-             {
-               "connectionCreatedEvent": {}
-             },
-+            {
-+              "poolClearedEvent": {}
-+            },
-             {
-               "connectionClosedEvent": {
-                 "reason": "error"
-@@ -378,9 +375,6 @@
-               "connectionCheckOutFailedEvent": {
-                 "reason": "connectionError"
-               }
--            },
--            {
--              "poolClearedEvent": {}
-             }
-           ]
-         }

.evergreen/spec-patch/PYTHON-3712.patch

@@ -1,14 +0,0 @@
-diff --git a/test/discovery_and_monitoring/unified/serverMonitoringMode.json b/test/discovery_and_monitoring/unified/serverMonitoringMode.json
-index e44fad1b..4b492f7d 100644
---- a/test/discovery_and_monitoring/unified/serverMonitoringMode.json
-+++ b/test/discovery_and_monitoring/unified/serverMonitoringMode.json
-@@ -5,7 +5,8 @@
-     {
-       "topologies": [
-         "single",
--        "sharded"
-+        "sharded",
-+        "sharded-replicaset"
-       ],
-       "serverless": "forbid"
-     }

.evergreen/spec-patch/PYTHON-4261.patch

@@ -1,61 +0,0 @@
-diff --git a/test/server_selection_logging/replica-set.json b/test/server_selection_logging/replica-set.json
-index 830b1ea51..5eba784bf 100644
---- a/test/server_selection_logging/replica-set.json
-+++ b/test/server_selection_logging/replica-set.json
-@@ -184,7 +184,7 @@
-               }
-             },
-             {
--              "level": "debug",
-+              "level": "info",
-               "component": "serverSelection",
-               "data": {
-                 "message": "Waiting for suitable server to become available",
-diff --git a/test/server_selection_logging/standalone.json b/test/server_selection_logging/standalone.json
-index 830b1ea51..5eba784bf 100644
---- a/test/server_selection_logging/standalone.json
-+++ b/test/server_selection_logging/standalone.json
-@@ -191,7 +191,7 @@
-               }
-             },
-             {
--              "level": "debug",
-+              "level": "info",
-               "component": "serverSelection",
-               "data": {
-                 "message": "Waiting for suitable server to become available",
-diff --git a/test/server_selection_logging/sharded.json b/test/server_selection_logging/sharded.json
-index 830b1ea51..5eba784bf 100644
---- a/test/server_selection_logging/sharded.json
-+++ b/test/server_selection_logging/sharded.json
-@@ -193,7 +193,7 @@
-               }
-             },
-             {
--              "level": "debug",
-+              "level": "info",
-               "component": "serverSelection",
-               "data": {
-                 "message": "Waiting for suitable server to become available",
-diff --git a/test/server_selection_logging/sharded.json b/test/server_selection_logging/operation-id.json
-index 830b1ea51..5eba784bf 100644
---- a/test/server_selection_logging/operation-id.json
-+++ b/test/server_selection_logging/operation-id.json
-@@ -197,7 +197,7 @@
-               }
-             },
-             {
--              "level": "debug",
-+              "level": "info",
-               "component": "serverSelection",
-               "data": {
-                 "message": "Waiting for suitable server to become available",
-@@ -383,7 +383,7 @@
-               }
-             },
-             {
--              "level": "debug",
-+              "level": "info",
-               "component": "serverSelection",
-               "data": {
-                 "message": "Waiting for suitable server to become available",

.evergreen/spec-patch/PYTHON-5517.patch

@@ -1,31 +0,0 @@
-diff --git a/test/discovery_and_monitoring/errors/error_handling_handshake.json b/test/discovery_and_monitoring/errors/error_handling_handshake.json
-index 56ca7d113..bf83f46f6 100644
---- a/test/discovery_and_monitoring/errors/error_handling_handshake.json
-+++ b/test/discovery_and_monitoring/errors/error_handling_handshake.json
-@@ -97,14 +97,22 @@
-       "outcome": {
-         "servers": {
-           "a:27017": {
--            "type": "Unknown",
--            "topologyVersion": null,
-+            "type": "RSPrimary",
-+            "setName": "rs",
-+            "topologyVersion": {
-+              "processId": {
-+                "$oid": "000000000000000000000001"
-+              },
-+              "counter": {
-+                "$numberLong": "1"
-+              }
-+            },
-             "pool": {
--              "generation": 1
-+              "generation": 0
-             }
-           }
-         },
--        "topologyType": "ReplicaSetNoPrimary",
-+        "topologyType": "ReplicaSetWithPrimary",
-         "logicalSessionTimeoutMinutes": null,
-         "setName": "rs"
-       }

.github/workflows/release-python.yml

@@ -16,7 +16,7 @@ env:
   # Changes per repo
   PRODUCT_NAME: PyMongo
   # Changes per branch
-  EVERGREEN_PROJECT: mongo-python-driver-release
+  EVERGREEN_PROJECT: mongo-python-driver
   # Constant
   # inputs will be empty on a scheduled run.  so, we only set dry_run
   # to 'false' when the input is set to 'false'.

CONTRIBUTING.md

@@ -505,13 +505,20 @@ python3 ./.evergreen/scripts/resync-all-specs.py
 
 Follow the [Python Driver Release Process Wiki](https://wiki.corp.mongodb.com/display/DRIVERS/Python+Driver+Release+Process).
 
-## Asyncio considerations
+## Project Structure and Asyncio Considerations
 
-PyMongo adds asyncio capability by modifying the source files in `*/asynchronous` to `*/synchronous` using
-[unasync](https://github.com/python-trio/unasync/) and some custom transforms.
+This section describes the layout of the `pymongo/` package.
 
-Where possible, edit the code in `*/asynchronous/*.py` and not the synchronous files.
-You can run `pre-commit run --all-files synchro` before running tests if you are testing synchronous code.
+Within `pymongo/`, the code is further divided into the `pymongo/asynchronous` and `pymongo/synchronous` subdirectories.
+Files in `pymongo/synchronous` are generated from `pymongo/asynchronous` using the `synchro` pre-commit hook, which uses [unasync](https://github.com/python-trio/unasync/) and some custom transforms.
+
+As a result, **all modifications** within `pymongo` must be made in either the top-level `pymongo` directory when they have to exhibit differing behavior between sync and async contexts or the `pymongo/asynchronous` directory, not `pymongo/synchronous`.
+Any changes made directly to files in the `pymongo/synchronous` directory will be overwritten by the `synchro` hook when it is run, which happens automatically on commit.
+
+Some top-level files (e.g. `pymongo/collection.py`) are re-export files for existing import compatibility and should not be modified directly.
+The other top-level files (e.g. `pymongo/network_layer.py`, `pymongo/pool_shared.py`) contain either shared code used in both the asynchronous and synchronous APIs, or code that is very different between the two APIs and therefore cannot be generated from the async version using `synchro`.
+
+Run `pre-commit run --all-files synchro` before running tests to generate the latest version of the synchronous code.
 
 To prevent the `synchro` hook from accidentally overwriting code, it first checks to see whether a sync version
 of a file is changing and not its async counterpart, and will fail.

bson/_cbsonmodule.c

@@ -109,6 +109,7 @@ struct module_state {
 #define DATETIME_CLAMP 2
 #define DATETIME_MS 3
 #define DATETIME_AUTO 4
+#define PYTHON_3_12 0x030C0000
 
 /* Converts integer to its string representation in decimal notation. */
 extern int cbson_long_long_to_str(long long num, char* str, size_t size) {
@@ -249,6 +250,67 @@ static int _write_element_to_buffer(PyObject* self, buffer_t buffer,
  */
 static int write_raw_doc(buffer_t buffer, PyObject* raw, PyObject* _raw);
 
+#if PY_VERSION_HEX >= PYTHON_3_12
+/* Transfer traceback from old_exc to new_exc.
+ * Steals reference to old_exc. */
+static PyObject* _transfer_traceback(PyObject *old_exc, PyObject *new_exc) {
+    PyObject *tb = PyException_GetTraceback(old_exc);
+    if (tb) {
+        PyException_SetTraceback(new_exc, tb);
+        Py_DECREF(tb);
+    }
+    Py_DECREF(old_exc);
+    return new_exc;
+}
+#endif
+
+/* Rewrap the current exception as InvalidBSON(str(e)) if it is not already an InvalidBSON error. */
+static void _rewrap_as_invalid_bson(void) {
+#if PY_VERSION_HEX >= PYTHON_3_12
+    PyObject *exc = PyErr_GetRaisedException();
+    if (exc && PyErr_GivenExceptionMatches(exc, PyExc_Exception)) {
+        PyObject *InvalidBSON = _error("InvalidBSON");
+        if (InvalidBSON) {
+            if (!PyErr_GivenExceptionMatches(exc, InvalidBSON)) {
+                PyObject *err_msg = PyObject_Str(exc);
+                if (err_msg) {
+                    PyObject *new_exc = PyObject_CallOneArg(InvalidBSON, err_msg);
+                    if (new_exc) {
+                        exc = _transfer_traceback(exc, new_exc);
+                    }
+                }
+                Py_XDECREF(err_msg);
+            }
+            Py_DECREF(InvalidBSON);
+        }
+    }
+    /* Steals reference to exc. */
+    PyErr_SetRaisedException(exc);
+#else
+    PyObject *etype = NULL, *evalue = NULL, *etrace = NULL;
+    PyObject *InvalidBSON = NULL;
+    PyErr_Fetch(&etype, &evalue, &etrace);
+    if (PyErr_GivenExceptionMatches(etype, PyExc_Exception)) {
+        InvalidBSON = _error("InvalidBSON");
+        if (InvalidBSON) {
+            if (!PyErr_GivenExceptionMatches(etype, InvalidBSON)) {
+                Py_DECREF(etype);
+                etype = InvalidBSON;
+                if (evalue) {
+                    PyObject *msg = PyObject_Str(evalue);
+                    Py_DECREF(evalue);
+                    evalue = msg;
+                }
+                PyErr_NormalizeException(&etype, &evalue, &etrace);
+            } else {
+                Py_DECREF(InvalidBSON);
+            }
+        }
+    }
+    PyErr_Restore(etype, evalue, etrace);
+#endif
+}
+
 /* Date stuff */
 static PyObject* datetime_from_millis(long long millis) {
     /* To encode a datetime instance like datetime(9999, 12, 31, 23, 59, 59, 999999)
@@ -294,34 +356,57 @@ static PyObject* datetime_from_millis(long long millis) {
                                           timeinfo.tm_sec,
                                           microseconds);
     if(!datetime) {
-        PyObject *etype = NULL, *evalue = NULL, *etrace = NULL;
+        #if PY_VERSION_HEX >= PYTHON_3_12
+            PyObject *exc = PyErr_GetRaisedException();
 
-        /*
-        * Calling _error clears the error state, so fetch it first.
-        */
-        PyErr_Fetch(&etype, &evalue, &etrace);
-
-        /* Only add addition error message on ValueError exceptions. */
-        if (PyErr_GivenExceptionMatches(etype, PyExc_ValueError)) {
-            if (evalue) {
-                PyObject* err_msg = PyObject_Str(evalue);
+            /* Only add additional error message on ValueError exceptions. */
+            if (exc && PyErr_GivenExceptionMatches(exc, PyExc_ValueError)) {
+                PyObject* err_msg = PyObject_Str(exc);
                 if (err_msg) {
                     PyObject* appendage = PyUnicode_FromString(" (Consider Using CodecOptions(datetime_conversion=DATETIME_AUTO) or MongoClient(datetime_conversion='DATETIME_AUTO')). See: https://www.mongodb.com/docs/languages/python/pymongo-driver/current/data-formats/dates-and-times/#handling-out-of-range-datetimes");
                     if (appendage) {
                         PyObject* msg = PyUnicode_Concat(err_msg, appendage);
                         if (msg) {
-                            Py_DECREF(evalue);
-                            evalue = msg;
+                            PyObject* new_exc = PyObject_CallOneArg(PyExc_ValueError, msg);
+                            if (new_exc) {
+                                exc = _transfer_traceback(exc, new_exc);
+                            }
+                            Py_DECREF(msg);
                         }
                     }
                     Py_XDECREF(appendage);
                 }
                 Py_XDECREF(err_msg);
             }
-            PyErr_NormalizeException(&etype, &evalue, &etrace);
-        }
-        /* Steals references to args. */
-        PyErr_Restore(etype, evalue, etrace);
+            /* Steals reference to exc. */
+            PyErr_SetRaisedException(exc);
+        #else
+            /* Calling _error clears the error state, so fetch it first.*/
+            PyObject *etype = NULL, *evalue = NULL, *etrace = NULL;
+            PyErr_Fetch(&etype, &evalue, &etrace);
+
+            /* Only add additional error message on ValueError exceptions. */
+            if (PyErr_GivenExceptionMatches(etype, PyExc_ValueError)) {
+                if (evalue) {
+                    PyObject* err_msg = PyObject_Str(evalue);
+                    if (err_msg) {
+                        PyObject* appendage = PyUnicode_FromString(" (Consider Using CodecOptions(datetime_conversion=DATETIME_AUTO) or MongoClient(datetime_conversion='DATETIME_AUTO')). See: https://www.mongodb.com/docs/languages/python/pymongo-driver/current/data-formats/dates-and-times/#handling-out-of-range-datetimes");
+                        if (appendage) {
+                            PyObject* msg = PyUnicode_Concat(err_msg, appendage);
+                            if (msg) {
+                                Py_DECREF(evalue);
+                                evalue = msg;
+                            }
+                        }
+                        Py_XDECREF(appendage);
+                    }
+                    Py_XDECREF(err_msg);
+                }
+                PyErr_NormalizeException(&etype, &evalue, &etrace);
+            }
+            /* Steals references to args. */
+            PyErr_Restore(etype, evalue, etrace);
+        #endif
     }
     return datetime;
 }
@@ -1681,6 +1766,46 @@ fail:
 /* Update Invalid Document error to include doc as a property.
  */
 void handle_invalid_doc_error(PyObject* dict) {
+#if PY_VERSION_HEX >= PYTHON_3_12
+    PyObject *exc = PyErr_GetRaisedException();
+    PyObject *msg = NULL, *new_msg = NULL;
+    PyObject *InvalidDocument = NULL;
+
+    if (exc == NULL) {
+        return;
+    }
+
+    InvalidDocument = _error("InvalidDocument");
+    if (InvalidDocument == NULL) {
+        goto cleanup;
+    }
+
+    if (PyErr_GivenExceptionMatches(exc, InvalidDocument)) {
+        msg = PyObject_Str(exc);
+        if (msg) {
+            const char *msg_utf8 = PyUnicode_AsUTF8(msg);
+            if (msg_utf8 == NULL) {
+                goto cleanup;
+            }
+            new_msg = PyUnicode_FromFormat("Invalid document: %s", msg_utf8);
+            if (new_msg == NULL) {
+                goto cleanup;
+            }
+            /* Add doc to the error instance as a property. */
+            PyObject* exc_args[2] = {new_msg, dict};
+            PyObject* new_exc = PyObject_Vectorcall(InvalidDocument, exc_args, 2, NULL);
+            if (new_exc) {
+                exc = _transfer_traceback(exc, new_exc);
+            }
+        }
+    }
+cleanup:
+    /* Steals reference to exc. */
+    PyErr_SetRaisedException(exc);
+    Py_XDECREF(msg);
+    Py_XDECREF(InvalidDocument);
+    Py_XDECREF(new_msg);
+#else
     PyObject *etype = NULL, *evalue = NULL, *etrace = NULL;
     PyObject *msg = NULL, *new_msg = NULL, *new_evalue = NULL;
     PyErr_Fetch(&etype, &evalue, &etrace);
@@ -1723,6 +1848,7 @@ cleanup:
     Py_XDECREF(InvalidDocument);
     Py_XDECREF(new_evalue);
     Py_XDECREF(new_msg);
+#endif
 }
 
 
@@ -2155,7 +2281,7 @@ static PyObject* get_value(PyObject* self, PyObject* name, const char* buffer,
             }
             memcpy(&length, buffer + *position, 4);
             length = BSON_UINT32_FROM_LE(length);
-            if (max < length) {
+            if (max - 5 < length) { // Account for 5-byte header. max >= 5 guaranteed above
                 goto invalid;
             }
 
@@ -2654,42 +2780,7 @@ static PyObject* get_value(PyObject* self, PyObject* name, const char* buffer,
      * Wrap any non-InvalidBSON errors in InvalidBSON.
      */
     if (PyErr_Occurred()) {
-        PyObject *etype = NULL, *evalue = NULL, *etrace = NULL;
-        PyObject *InvalidBSON = NULL;
-
-        /*
-         * Calling _error clears the error state, so fetch it first.
-         */
-        PyErr_Fetch(&etype, &evalue, &etrace);
-
-        /* Dont reraise anything but PyExc_Exceptions as InvalidBSON. */
-        if (PyErr_GivenExceptionMatches(etype, PyExc_Exception)) {
-            InvalidBSON = _error("InvalidBSON");
-            if (InvalidBSON) {
-                if (!PyErr_GivenExceptionMatches(etype, InvalidBSON)) {
-                    /*
-                     * Raise InvalidBSON(str(e)).
-                     */
-                    Py_DECREF(etype);
-                    etype = InvalidBSON;
-
-                    if (evalue) {
-                        PyObject *msg = PyObject_Str(evalue);
-                        Py_DECREF(evalue);
-                        evalue = msg;
-                    }
-                    PyErr_NormalizeException(&etype, &evalue, &etrace);
-                } else {
-                    /*
-                     * The current exception matches InvalidBSON, so we don't
-                     * need this reference after all.
-                     */
-                    Py_DECREF(InvalidBSON);
-                }
-            }
-        }
-        /* Steals references to args. */
-        PyErr_Restore(etype, evalue, etrace);
+        _rewrap_as_invalid_bson();
     } else {
         PyObject *InvalidBSON = _error("InvalidBSON");
         if (InvalidBSON) {
@@ -2727,25 +2818,7 @@ static int _element_to_dict(PyObject* self, const char* string,
     if (!*name) {
         /* If NULL is returned then wrap the UnicodeDecodeError
            in an InvalidBSON error */
-        PyObject *etype = NULL, *evalue = NULL, *etrace = NULL;
-        PyObject *InvalidBSON = NULL;
-
-        PyErr_Fetch(&etype, &evalue, &etrace);
-        if (PyErr_GivenExceptionMatches(etype, PyExc_Exception)) {
-            InvalidBSON = _error("InvalidBSON");
-            if (InvalidBSON) {
-                Py_DECREF(etype);
-                etype = InvalidBSON;
-
-                if (evalue) {
-                    PyObject *msg = PyObject_Str(evalue);
-                    Py_DECREF(evalue);
-                    evalue = msg;
-                }
-                PyErr_NormalizeException(&etype, &evalue, &etrace);
-            }
-        }
-        PyErr_Restore(etype, evalue, etrace);
+        _rewrap_as_invalid_bson();
         return -1;
     }
     position += (unsigned)name_length + 1;

justfile

@@ -16,64 +16,78 @@ default:
 resync:
  @uv sync --quiet
 
+# Set up the development environment
 install:
    bash .evergreen/scripts/setup-dev-env.sh
 
+# Build the HTML documentation
 [group('docs')]
 docs: && resync
     {{docs_run}} sphinx-build -W -b html doc {{doc_build}}/html
 
+# Serve the docs locally with live-reload
 [group('docs')]
 docs-serve: && resync
     {{docs_run}} sphinx-autobuild -W -b html doc --watch ./pymongo --watch ./bson --watch ./gridfs {{doc_build}}/serve
 
+# Check documentation hyperlinks for broken URLs
 [group('docs')]
 docs-linkcheck: && resync
     {{docs_run}} sphinx-build -E -b linkcheck doc {{doc_build}}/linkcheck
 
+# Run mypy and pyright
 [group('typing')]
 typing: && resync
     just typing-mypy
     just typing-pyright
 
+# Run mypy against the library source and test suite
 [group('typing')]
 typing-mypy: && resync
     {{typing_run}} python -m mypy {{mypy_args}} bson gridfs tools pymongo
     {{typing_run}} python -m mypy {{mypy_args}} --config-file mypy_test.ini test
     {{typing_run}} python -m mypy {{mypy_args}} test/test_typing.py test/test_typing_strict.py
 
+# Run pyright against the typing test files
 [group('typing')]
 typing-pyright: && resync
     {{typing_run}} python -m pyright test/test_typing.py test/test_typing_strict.py
     {{typing_run}} python -m pyright -p strict_pyrightconfig.json test/test_typing_strict.py
 
+# Run all pre-commit hooks across the repository
 [group('lint')]
 lint *args="": && resync
     uvx pre-commit run --all-files {{args}}
 
+# Run shellcheck, doc8, and slotscheck
 [group('lint')]
 lint-manual *args="": && resync
     uvx pre-commit run --all-files --hook-stage manual {{args}}
 
+# Run pytest (e.g. just test test/test_uri_parser.py)
 [group('test')]
 test *args="-v --durations=5 --maxfail=10": && resync
     #!/usr/bin/env bash
     set -euo pipefail
     uv run ${USE_ACTIVE_VENV:+--active} --extra test python -m pytest {{args}}
 
+# Run the BSON test suite with numpy
 [group('test')]
 test-numpy *args="": && resync
     just setup-tests numpy {{args}}
     just run-tests test/test_bson.py
 
+# Run tests via the Evergreen test runner script
 [group('test')]
 run-tests *args: && resync
     bash ./.evergreen/run-tests.sh {{args}}
 
+# Set up the test environment (auth, TLS, etc.)
 [group('test')]
 setup-tests *args="":
     bash .evergreen/scripts/setup-tests.sh {{args}}
 
+# Tear down resources created by setup-tests
 [group('test')]
 teardown-tests:
     bash .evergreen/scripts/teardown-tests.sh
@@ -82,25 +96,30 @@ teardown-tests:
 integration-tests:
     bash integration_tests/run.sh
 
+# Run the full test suite with coverage
 [group('test')]
 test-coverage *args="":
     just setup-tests --cov
     just run-tests {{args}}
 
+# Print the coverage summary to the terminal
 [group('coverage')]
 coverage-report:
     uv tool run --with "coverage[toml]" coverage report
 
+# Generate an HTML coverage report in htmlcov/
 [group('coverage')]
 coverage-html:
     uv tool run --with "coverage[toml]" coverage html
     @echo "Coverage report generated in htmlcov/index.html"
 
+# Generate an XML coverage report at coverage.xml
 [group('coverage')]
 coverage-xml:
     uv tool run --with "coverage[toml]" coverage xml
     @echo "Coverage report generated in coverage.xml"
 
+# Start a MongoDB server via drivers-evergreen-tools
 [group('server')]
 run-server *args="":
     bash .evergreen/scripts/run-server.sh {{args}}

pymongo/_azure_helpers.py

@@ -17,6 +17,7 @@ from __future__ import annotations
 
 import json
 from typing import Any, Optional
+from urllib.parse import quote
 
 
 def _get_azure_response(
@@ -29,7 +30,7 @@ def _get_azure_response(
     url += "?api-version=2018-02-01"
     url += f"&resource={resource}"
     if client_id:
-        url += f"&client_id={client_id}"
+        url += f"&client_id={quote(client_id)}"
     headers = {"Metadata": "true", "Accept": "application/json"}
     request = Request(url, headers=headers)  # noqa: S310
     try:

pymongo/_version.py

@@ -18,7 +18,7 @@ from __future__ import annotations
 import re
 from typing import List, Tuple, Union
 
-__version__ = "4.17.0"
+__version__ = "4.18.0.dev0"
 
 
 def get_version_tuple(version: str) -> Tuple[Union[int, str], ...]:

pymongo/asynchronous/client_session.py

@@ -1101,7 +1101,11 @@ class AsyncClientSession:
         read_preference: _ServerMode,
         conn: AsyncConnection,
     ) -> None:
-        if not conn.supports_sessions:
+        # getMores must be sent with a session if the cursor was opened with one
+        operation = next(iter(command))
+        if not conn.supports_sessions and (
+            isinstance(self._server_session, _EmptyServerSession) or operation != "getMore"
+        ):
             if not self._implicit:
                 raise ConfigurationError("Sessions are not supported by this MongoDB deployment")
             return

pymongo/asynchronous/pool.py

@@ -760,11 +760,7 @@ class Pool:
         self._pending = 0
         self._max_connecting = self.opts.max_connecting
         self._client_id = client_id
-        if self.enabled_for_cmap:
-            assert self.opts._event_listeners is not None
-            self.opts._event_listeners.publish_pool_created(
-                self.address, self.opts.non_default_options
-            )
+        # Log before publishing event to prevent potential listener preemption in tests
         if self.enabled_for_logging and _CONNECTION_LOGGER.isEnabledFor(logging.DEBUG):
             _debug_log(
                 _CONNECTION_LOGGER,
@@ -774,6 +770,11 @@ class Pool:
                 serverPort=self.address[1],
                 **self.opts.non_default_options,
             )
+        if self.enabled_for_cmap:
+            assert self.opts._event_listeners is not None
+            self.opts._event_listeners.publish_pool_created(
+                self.address, self.opts.non_default_options
+            )
         # Similar to active_sockets but includes threads in the wait queue.
         self.operation_count: int = 0
         # Retain references to pinned connections to prevent the CPython GC
@@ -788,9 +789,6 @@ class Pool:
         async with self.lock:
             if self.state != PoolState.READY:
                 self.state = PoolState.READY
-                if self.enabled_for_cmap:
-                    assert self.opts._event_listeners is not None
-                    self.opts._event_listeners.publish_pool_ready(self.address)
                 if self.enabled_for_logging and _CONNECTION_LOGGER.isEnabledFor(logging.DEBUG):
                     _debug_log(
                         _CONNECTION_LOGGER,
@@ -799,6 +797,9 @@ class Pool:
                         serverHost=self.address[0],
                         serverPort=self.address[1],
                     )
+                if self.enabled_for_cmap:
+                    assert self.opts._event_listeners is not None
+                    self.opts._event_listeners.publish_pool_ready(self.address)
 
     @property
     def closed(self) -> bool:
@@ -859,9 +860,6 @@ class Pool:
             else:
                 for conn in sockets:
                     await conn.close_conn(ConnectionClosedReason.POOL_CLOSED)
-            if self.enabled_for_cmap:
-                assert listeners is not None
-                listeners.publish_pool_closed(self.address)
             if self.enabled_for_logging and _CONNECTION_LOGGER.isEnabledFor(logging.DEBUG):
                 _debug_log(
                     _CONNECTION_LOGGER,
@@ -870,15 +868,11 @@ class Pool:
                     serverHost=self.address[0],
                     serverPort=self.address[1],
                 )
+            if self.enabled_for_cmap:
+                assert listeners is not None
+                listeners.publish_pool_closed(self.address)
         else:
             if old_state != PoolState.PAUSED:
-                if self.enabled_for_cmap:
-                    assert listeners is not None
-                    listeners.publish_pool_cleared(
-                        self.address,
-                        service_id=service_id,
-                        interrupt_connections=interrupt_connections,
-                    )
                 if self.enabled_for_logging and _CONNECTION_LOGGER.isEnabledFor(logging.DEBUG):
                     _debug_log(
                         _CONNECTION_LOGGER,
@@ -888,6 +882,13 @@ class Pool:
                         serverPort=self.address[1],
                         serviceId=service_id,
                     )
+                if self.enabled_for_cmap:
+                    assert listeners is not None
+                    listeners.publish_pool_cleared(
+                        self.address,
+                        service_id=service_id,
+                        interrupt_connections=interrupt_connections,
+                    )
             if not _IS_SYNC:
                 await asyncio.gather(
                     *[conn.close_conn(ConnectionClosedReason.STALE) for conn in sockets],  # type: ignore[func-returns-value]

pymongo/synchronous/client_session.py

@@ -1097,7 +1097,11 @@ class ClientSession:
         read_preference: _ServerMode,
         conn: Connection,
     ) -> None:
-        if not conn.supports_sessions:
+        # getMores must be sent with a session if the cursor was opened with one
+        operation = next(iter(command))
+        if not conn.supports_sessions and (
+            isinstance(self._server_session, _EmptyServerSession) or operation != "getMore"
+        ):
             if not self._implicit:
                 raise ConfigurationError("Sessions are not supported by this MongoDB deployment")
             return

pymongo/synchronous/pool.py

@@ -758,11 +758,7 @@ class Pool:
         self._pending = 0
         self._max_connecting = self.opts.max_connecting
         self._client_id = client_id
-        if self.enabled_for_cmap:
-            assert self.opts._event_listeners is not None
-            self.opts._event_listeners.publish_pool_created(
-                self.address, self.opts.non_default_options
-            )
+        # Log before publishing event to prevent potential listener preemption in tests
         if self.enabled_for_logging and _CONNECTION_LOGGER.isEnabledFor(logging.DEBUG):
             _debug_log(
                 _CONNECTION_LOGGER,
@@ -772,6 +768,11 @@ class Pool:
                 serverPort=self.address[1],
                 **self.opts.non_default_options,
             )
+        if self.enabled_for_cmap:
+            assert self.opts._event_listeners is not None
+            self.opts._event_listeners.publish_pool_created(
+                self.address, self.opts.non_default_options
+            )
         # Similar to active_sockets but includes threads in the wait queue.
         self.operation_count: int = 0
         # Retain references to pinned connections to prevent the CPython GC
@@ -786,9 +787,6 @@ class Pool:
         with self.lock:
             if self.state != PoolState.READY:
                 self.state = PoolState.READY
-                if self.enabled_for_cmap:
-                    assert self.opts._event_listeners is not None
-                    self.opts._event_listeners.publish_pool_ready(self.address)
                 if self.enabled_for_logging and _CONNECTION_LOGGER.isEnabledFor(logging.DEBUG):
                     _debug_log(
                         _CONNECTION_LOGGER,
@@ -797,6 +795,9 @@ class Pool:
                         serverHost=self.address[0],
                         serverPort=self.address[1],
                     )
+                if self.enabled_for_cmap:
+                    assert self.opts._event_listeners is not None
+                    self.opts._event_listeners.publish_pool_ready(self.address)
 
     @property
     def closed(self) -> bool:
@@ -857,9 +858,6 @@ class Pool:
             else:
                 for conn in sockets:
                     conn.close_conn(ConnectionClosedReason.POOL_CLOSED)
-            if self.enabled_for_cmap:
-                assert listeners is not None
-                listeners.publish_pool_closed(self.address)
             if self.enabled_for_logging and _CONNECTION_LOGGER.isEnabledFor(logging.DEBUG):
                 _debug_log(
                     _CONNECTION_LOGGER,
@@ -868,15 +866,11 @@ class Pool:
                     serverHost=self.address[0],
                     serverPort=self.address[1],
                 )
+            if self.enabled_for_cmap:
+                assert listeners is not None
+                listeners.publish_pool_closed(self.address)
         else:
             if old_state != PoolState.PAUSED:
-                if self.enabled_for_cmap:
-                    assert listeners is not None
-                    listeners.publish_pool_cleared(
-                        self.address,
-                        service_id=service_id,
-                        interrupt_connections=interrupt_connections,
-                    )
                 if self.enabled_for_logging and _CONNECTION_LOGGER.isEnabledFor(logging.DEBUG):
                     _debug_log(
                         _CONNECTION_LOGGER,
@@ -886,6 +880,13 @@ class Pool:
                         serverPort=self.address[1],
                         serviceId=service_id,
                     )
+                if self.enabled_for_cmap:
+                    assert listeners is not None
+                    listeners.publish_pool_cleared(
+                        self.address,
+                        service_id=service_id,
+                        interrupt_connections=interrupt_connections,
+                    )
             if not _IS_SYNC:
                 asyncio.gather(
                     *[conn.close_conn(ConnectionClosedReason.STALE) for conn in sockets],  # type: ignore[func-returns-value]

test/asynchronous/test_client.py

@@ -2711,11 +2711,11 @@ class TestClientPool(AsyncMockClientTest):
 
         await async_wait_until(lambda: len(c.nodes) == 1, "connect")
         self.assertEqual(await c.address, ("c", 3))
-        # Assert that we create 1 pooled connection.
+        # Wait for the pooled connection to be registered
         await listener.async_wait_for_event(monitoring.ConnectionReadyEvent, 1)
         self.assertEqual(listener.event_count(monitoring.ConnectionCreatedEvent), 1)
         arbiter = c._topology.get_server_by_address(("c", 3))
-        self.assertEqual(len(arbiter.pool.conns), 1)
+        await async_wait_until(lambda: len(arbiter.pool.conns) == 1, "create 1 pooled connection")
         # Arbiter pool is marked ready.
         self.assertEqual(listener.event_count(monitoring.PoolReadyEvent), 1)
 

test/asynchronous/test_encryption.py

@@ -876,8 +876,6 @@ class TestViews(AsyncEncryptionIntegrationTest):
 
 
 class TestCorpus(AsyncEncryptionIntegrationTest):
-    # PYTHON-5708: Encryption tests sending large payloads fail on some mongocryptd versions.
-    @async_client_context.require_version_max(6, 99)
     @unittest.skipUnless(any(AWS_CREDS.values()), "AWS environment credentials are not set")
     async def asyncSetUp(self):
         await super().asyncSetUp()
@@ -1054,8 +1052,6 @@ class TestBsonSizeBatches(AsyncEncryptionIntegrationTest):
     client_encrypted: AsyncMongoClient
     listener: OvertCommandListener
 
-    # PYTHON-5708: Encryption tests sending large payloads fail on some mongocryptd versions.
-    @async_client_context.require_version_max(6, 99)
     async def asyncSetUp(self):
         await super().asyncSetUp()
         db = async_client_context.client.db
@@ -3326,6 +3322,7 @@ class TestAutomaticDecryptionKeys(AsyncEncryptionIntegrationTest):
 class TestExplicitTextEncryptionProse(AsyncEncryptionIntegrationTest):
     @async_client_context.require_no_standalone
     @async_client_context.require_version_min(8, 2, -1)
+    @async_client_context.require_version_max(8, 99, 99)
     @async_client_context.require_libmongocrypt_min(1, 15, 1)
     @async_client_context.require_pymongocrypt_min(1, 16, 0)
     async def asyncSetUp(self):

test/asynchronous/test_session.py

@@ -15,7 +15,6 @@
 """Test the client_session module."""
 from __future__ import annotations
 
-import asyncio
 import copy
 import sys
 import time
@@ -24,8 +23,6 @@ from io import BytesIO
 from test.asynchronous.helpers import ExceptionCatchingTask
 from typing import Any, Callable, List, Set, Tuple
 
-from pymongo.synchronous.mongo_client import MongoClient
-
 sys.path[0:0] = [""]
 
 from test.asynchronous import (
@@ -45,7 +42,7 @@ from test.utils_shared import (
 
 from bson import DBRef
 from gridfs.asynchronous.grid_file import AsyncGridFS, AsyncGridFSBucket
-from pymongo import ASCENDING, AsyncMongoClient, _csot, monitoring
+from pymongo import ASCENDING, AsyncMongoClient, monitoring
 from pymongo.asynchronous.command_cursor import AsyncCommandCursor
 from pymongo.asynchronous.cursor import AsyncCursor
 from pymongo.asynchronous.helpers import anext
@@ -938,6 +935,39 @@ class TestSession(AsyncIntegrationTest):
 
         await s2.end_session()
 
+    async def test_getmore_preserves_lsid_after_session_support_lost(self):
+        listener = OvertCommandListener()
+        client = await self.async_rs_or_single_client(event_listeners=[listener], maxPoolSize=1)
+        coll = client.pymongo_test.test
+        await coll.drop()
+        await coll.insert_many([{"x": i} for i in range(10)])
+        self.addAsyncCleanup(coll.drop)
+
+        async with client.start_session() as s:
+            cursor = coll.find({}, batch_size=2, session=s)
+            await anext(cursor)
+
+            find_event = next(e for e in listener.started_events if e.command_name == "find")
+            lsid = find_event.command["lsid"]
+
+            # Simulate a node stepping down: mark idle connections as not supporting sessions.
+            for server in client._topology._servers.values():
+                for conn in server.pool.conns:
+                    conn.supports_sessions = False
+
+            listener.reset()
+            await cursor.to_list()
+
+        getmore_events = [e for e in listener.started_events if e.command_name == "getMore"]
+        self.assertGreater(len(getmore_events), 0, "expected at least one getMore command")
+        for event in getmore_events:
+            self.assertIn(
+                "lsid", event.command, "getMore must include lsid when session is materialized"
+            )
+            self.assertEqual(
+                lsid, event.command["lsid"], "getMore lsid must match the session lsid from find"
+            )
+
 
 class TestCausalConsistency(AsyncUnitTest):
     listener: SessionTestListener

test/change_streams/unified/change-streams-disambiguatedPaths.json

@@ -42,6 +42,91 @@
     }
   ],
   "tests": [
+    {
+      "description": "disambiguatedPaths is not present when showExpandedEvents is false/unset",
+      "runOnRequirements": [
+        {
+          "minServerVersion": "6.1.0",
+          "maxServerVersion": "8.1.99",
+          "topologies": [
+            "replicaset",
+            "load-balanced",
+            "sharded"
+          ],
+          "serverless": "forbid"
+        },
+        {
+          "minServerVersion": "8.2.1",
+          "topologies": [
+            "replicaset",
+            "load-balanced",
+            "sharded"
+          ],
+          "serverless": "forbid"
+        }
+      ],
+      "operations": [
+        {
+          "name": "insertOne",
+          "object": "collection0",
+          "arguments": {
+            "document": {
+              "_id": 1,
+              "a": {
+                "1": 1
+              }
+            }
+          }
+        },
+        {
+          "name": "createChangeStream",
+          "object": "collection0",
+          "arguments": {
+            "pipeline": []
+          },
+          "saveResultAsEntity": "changeStream0"
+        },
+        {
+          "name": "updateOne",
+          "object": "collection0",
+          "arguments": {
+            "filter": {
+              "_id": 1
+            },
+            "update": {
+              "$set": {
+                "a.1": 2
+              }
+            }
+          }
+        },
+        {
+          "name": "iterateUntilDocumentOrError",
+          "object": "changeStream0",
+          "expectResult": {
+            "operationType": "update",
+            "ns": {
+              "db": "database0",
+              "coll": "collection0"
+            },
+            "updateDescription": {
+              "updatedFields": {
+                "$$exists": true
+              },
+              "removedFields": {
+                "$$exists": true
+              },
+              "truncatedArrays": {
+                "$$exists": true
+              },
+              "disambiguatedPaths": {
+                "$$exists": false
+              }
+            }
+          }
+        }
+      ]
+    },
     {
       "description": "disambiguatedPaths is present on updateDescription when an ambiguous path is present",
       "operations": [

test/change_streams/unified/change-streams-nsType.json

@@ -63,47 +63,6 @@
         }
       ]
     },
-    {
-      "description": "nsType is present when creating timeseries",
-      "operations": [
-        {
-          "name": "dropCollection",
-          "object": "database0",
-          "arguments": {
-            "collection": "foo"
-          }
-        },
-        {
-          "name": "createChangeStream",
-          "object": "database0",
-          "arguments": {
-            "pipeline": [],
-            "showExpandedEvents": true
-          },
-          "saveResultAsEntity": "changeStream0"
-        },
-        {
-          "name": "createCollection",
-          "object": "database0",
-          "arguments": {
-            "collection": "foo",
-            "timeseries": {
-              "timeField": "time",
-              "metaField": "meta",
-              "granularity": "minutes"
-            }
-          }
-        },
-        {
-          "name": "iterateUntilDocumentOrError",
-          "object": "changeStream0",
-          "expectResult": {
-            "operationType": "create",
-            "nsType": "timeseries"
-          }
-        }
-      ]
-    },
     {
       "description": "nsType is present when creating views",
       "operations": [

test/client-side-encryption/spec/unified/QE-Text-cleanupStructuredEncryptionData.json

@@ -4,6 +4,7 @@
   "runOnRequirements": [
     {
       "minServerVersion": "8.2.0",
+      "maxServerVersion": "8.99.99",
       "topologies": [
         "replicaset",
         "sharded",

test/client-side-encryption/spec/unified/QE-Text-compactStructuredEncryptionData.json

@@ -4,6 +4,7 @@
   "runOnRequirements": [
     {
       "minServerVersion": "8.2.0",
+      "maxServerVersion": "8.99.99",
       "topologies": [
         "replicaset",
         "sharded",

test/client-side-encryption/spec/unified/QE-Text-prefixPreview.json

@@ -4,6 +4,7 @@
   "runOnRequirements": [
     {
       "minServerVersion": "8.2.0",
+      "maxServerVersion": "8.99.99",
       "topologies": [
         "replicaset",
         "sharded",

test/client-side-encryption/spec/unified/QE-Text-substringPreview.json

@@ -126,7 +126,7 @@
   ],
   "tests": [
     {
-      "description": "Insert QE suffixPreview",
+      "description": "Insert QE substringPreview",
       "operations": [
         {
           "name": "insertOne",

test/client-side-encryption/spec/unified/QE-Text-suffixPreview.json

@@ -4,6 +4,7 @@
   "runOnRequirements": [
     {
       "minServerVersion": "8.2.0",
+      "maxServerVersion": "8.99.99",
       "topologies": [
         "replicaset",
         "sharded",

test/client-side-encryption/spec/unified/fle2v2-InsertFind-keyAltName.json

@@ -0,0 +1,485 @@
+{
+  "description": "fle2v2-InsertFind-keyAltName",
+  "schemaVersion": "1.25",
+  "runOnRequirements": [
+    {
+      "minServerVersion": "7.0.0",
+      "topologies": [
+        "replicaset",
+        "sharded",
+        "load-balanced"
+      ],
+      "csfle": {
+        "minLibmongocryptVersion": "1.18.0"
+      }
+    }
+  ],
+  "createEntities": [
+    {
+      "client": {
+        "id": "client0",
+        "autoEncryptOpts": {
+          "keyVaultNamespace": "keyvault.datakeys",
+          "kmsProviders": {
+            "local": {
+              "key": "Mng0NCt4ZHVUYUJCa1kxNkVyNUR1QURhZ2h2UzR2d2RrZzh0cFBwM3R6NmdWMDFBMUN3YkQ5aXRRMkhGRGdQV09wOGVNYUMxT2k3NjZKelhaQmRCZGJkTXVyZG9uSjFk"
+            }
+          },
+          "encryptedFieldsMap": {
+            "default.default": {
+              "fields": [
+                {
+                  "path": "encryptedIndexed",
+                  "bsonType": "string",
+                  "queries": {
+                    "queryType": "equality",
+                    "contention": {
+                      "$numberLong": "0"
+                    }
+                  },
+                  "keyAltName": "altname"
+                }
+              ]
+            }
+          }
+        },
+        "observeEvents": [
+          "commandStartedEvent"
+        ]
+      }
+    },
+    {
+      "database": {
+        "id": "db",
+        "client": "client0",
+        "databaseName": "default"
+      }
+    },
+    {
+      "collection": {
+        "id": "coll",
+        "database": "db",
+        "collectionName": "default"
+      }
+    },
+    {
+      "client": {
+        "id": "client_unencrypted",
+        "observeEvents": [
+          "commandStartedEvent"
+        ]
+      }
+    },
+    {
+      "database": {
+        "id": "db_unencrypted",
+        "client": "client_unencrypted",
+        "databaseName": "default"
+      }
+    },
+    {
+      "collection": {
+        "id": "coll_unencrypted",
+        "database": "db_unencrypted",
+        "collectionName": "default"
+      }
+    }
+  ],
+  "initialData": [
+    {
+      "databaseName": "default",
+      "collectionName": "default",
+      "documents": [],
+      "createOptions": {
+        "encryptedFields": {
+          "fields": [
+            {
+              "keyId": {
+                "$binary": {
+                  "base64": "EjRWeBI0mHYSNBI0VniQEg==",
+                  "subType": "04"
+                }
+              },
+              "path": "encryptedIndexed",
+              "bsonType": "string",
+              "queries": {
+                "queryType": "equality",
+                "contention": {
+                  "$numberLong": "0"
+                }
+              }
+            }
+          ]
+        }
+      }
+    },
+    {
+      "databaseName": "keyvault",
+      "collectionName": "datakeys",
+      "documents": [
+        {
+          "_id": {
+            "$binary": {
+              "base64": "EjRWeBI0mHYSNBI0VniQEg==",
+              "subType": "04"
+            }
+          },
+          "keyMaterial": {
+            "$binary": {
+              "base64": "sHe0kz57YW7v8g9VP9sf/+K1ex4JqKc5rf/URX3n3p8XdZ6+15uXPaSayC6adWbNxkFskuMCOifDoTT+rkqMtFkDclOy884RuGGtUysq3X7zkAWYTKi8QAfKkajvVbZl2y23UqgVasdQu3OVBQCrH/xY00nNAs/52e958nVjBuzQkSb1T8pKJAyjZsHJ60+FtnfafDZSTAIBJYn7UWBCwQ==",
+              "subType": "00"
+            }
+          },
+          "creationDate": {
+            "$date": {
+              "$numberLong": "1648914851981"
+            }
+          },
+          "updateDate": {
+            "$date": {
+              "$numberLong": "1648914851981"
+            }
+          },
+          "status": {
+            "$numberInt": "0"
+          },
+          "masterKey": {
+            "provider": "local"
+          },
+          "keyAltNames": [
+            "altname"
+          ]
+        }
+      ]
+    }
+  ],
+  "tests": [
+    {
+      "description": "Insert and find FLE2 indexed field",
+      "operations": [
+        {
+          "name": "insertOne",
+          "arguments": {
+            "document": {
+              "_id": 1,
+              "encryptedIndexed": "123"
+            }
+          },
+          "object": "coll"
+        },
+        {
+          "name": "find",
+          "arguments": {
+            "filter": {
+              "encryptedIndexed": "123"
+            }
+          },
+          "object": "coll",
+          "expectResult": [
+            {
+              "_id": 1,
+              "encryptedIndexed": "123"
+            }
+          ]
+        },
+        {
+          "name": "find",
+          "object": "coll_unencrypted",
+          "arguments": {
+            "filter": {}
+          },
+          "expectResult": [
+            {
+              "_id": 1,
+              "encryptedIndexed": {
+                "$$type": "binData"
+              },
+              "__safeContent__": [
+                {
+                  "$binary": {
+                    "base64": "31eCYlbQoVboc5zwC8IoyJVSkag9PxREka8dkmbXJeY=",
+                    "subType": "00"
+                  }
+                }
+              ]
+            }
+          ]
+        }
+      ],
+      "expectEvents": [
+        {
+          "client": "client0",
+          "events": [
+            {
+              "commandStartedEvent": {
+                "command": {
+                  "find": "datakeys",
+                  "filter": {
+                    "$or": [
+                      {
+                        "_id": {
+                          "$in": []
+                        }
+                      },
+                      {
+                        "keyAltNames": {
+                          "$in": [
+                            "altname"
+                          ]
+                        }
+                      }
+                    ]
+                  },
+                  "$db": "keyvault",
+                  "readConcern": {
+                    "level": "majority"
+                  }
+                },
+                "commandName": "find"
+              }
+            },
+            {
+              "commandStartedEvent": {
+                "command": {
+                  "insert": "default",
+                  "documents": [
+                    {
+                      "_id": 1,
+                      "encryptedIndexed": {
+                        "$$type": "binData"
+                      }
+                    }
+                  ],
+                  "ordered": true,
+                  "encryptionInformation": {
+                    "type": 1,
+                    "schema": {
+                      "default.default": {
+                        "escCollection": "enxcol_.default.esc",
+                        "ecocCollection": "enxcol_.default.ecoc",
+                        "fields": [
+                          {
+                            "keyId": {
+                              "$binary": {
+                                "base64": "EjRWeBI0mHYSNBI0VniQEg==",
+                                "subType": "04"
+                              }
+                            },
+                            "path": "encryptedIndexed",
+                            "bsonType": "string",
+                            "queries": {
+                              "queryType": "equality",
+                              "contention": {
+                                "$numberLong": "0"
+                              }
+                            }
+                          }
+                        ]
+                      }
+                    }
+                  }
+                },
+                "commandName": "insert"
+              }
+            },
+            {
+              "commandStartedEvent": {
+                "command": {
+                  "find": "default",
+                  "filter": {
+                    "encryptedIndexed": {
+                      "$eq": {
+                        "$binary": {
+                          "base64": "DIkAAAAFZAAgAAAAAPGmZcUzdE/FPILvRSyAScGvZparGI2y9rJ/vSBxgCujBXMAIAAAAACi1RjmndKqgnXy7xb22RzUbnZl1sOZRXPOC0KcJkAxmQVsACAAAAAApJtKPW4+o9B7gAynNLL26jtlB4+hq5TXResijcYet8USY20AAAAAAAAAAAAA",
+                          "subType": "06"
+                        }
+                      }
+                    }
+                  },
+                  "encryptionInformation": {
+                    "type": 1,
+                    "schema": {
+                      "default.default": {
+                        "escCollection": "enxcol_.default.esc",
+                        "ecocCollection": "enxcol_.default.ecoc",
+                        "fields": [
+                          {
+                            "keyId": {
+                              "$binary": {
+                                "base64": "EjRWeBI0mHYSNBI0VniQEg==",
+                                "subType": "04"
+                              }
+                            },
+                            "path": "encryptedIndexed",
+                            "bsonType": "string",
+                            "queries": {
+                              "queryType": "equality",
+                              "contention": {
+                                "$numberLong": "0"
+                              }
+                            }
+                          }
+                        ]
+                      }
+                    }
+                  }
+                },
+                "commandName": "find"
+              }
+            }
+          ]
+        }
+      ]
+    },
+    {
+      "description": "Create translates keyAltName",
+      "operations": [
+        {
+          "name": "dropCollection",
+          "object": "db",
+          "arguments": {
+            "collection": "default"
+          }
+        },
+        {
+          "name": "createCollection",
+          "object": "db",
+          "arguments": {
+            "collection": "default"
+          }
+        }
+      ],
+      "expectEvents": [
+        {
+          "client": "client0",
+          "events": [
+            {
+              "commandStartedEvent": {
+                "command": {
+                  "drop": "enxcol_.default.esc"
+                },
+                "commandName": "drop"
+              }
+            },
+            {
+              "commandStartedEvent": {
+                "command": {
+                  "drop": "enxcol_.default.ecoc"
+                },
+                "commandName": "drop"
+              }
+            },
+            {
+              "commandStartedEvent": {
+                "command": {
+                  "drop": "default"
+                },
+                "commandName": "drop"
+              }
+            },
+            {
+              "commandStartedEvent": {
+                "command": {
+                  "create": "enxcol_.default.esc",
+                  "clusteredIndex": {
+                    "key": {
+                      "_id": 1
+                    },
+                    "unique": true
+                  }
+                },
+                "commandName": "create"
+              }
+            },
+            {
+              "commandStartedEvent": {
+                "command": {
+                  "create": "enxcol_.default.ecoc",
+                  "clusteredIndex": {
+                    "key": {
+                      "_id": 1
+                    },
+                    "unique": true
+                  }
+                },
+                "commandName": "create"
+              }
+            },
+            {
+              "commandStartedEvent": {
+                "command": {
+                  "find": "datakeys",
+                  "filter": {
+                    "$or": [
+                      {
+                        "_id": {
+                          "$in": []
+                        }
+                      },
+                      {
+                        "keyAltNames": {
+                          "$in": [
+                            "altname"
+                          ]
+                        }
+                      }
+                    ]
+                  },
+                  "$db": "keyvault",
+                  "readConcern": {
+                    "level": "majority"
+                  }
+                },
+                "commandName": "find"
+              }
+            },
+            {
+              "commandStartedEvent": {
+                "command": {
+                  "create": "default",
+                  "encryptedFields": {
+                    "fields": [
+                      {
+                        "path": "encryptedIndexed",
+                        "bsonType": "string",
+                        "queries": {
+                          "queryType": "equality",
+                          "contention": {
+                            "$numberLong": "0"
+                          }
+                        },
+                        "keyId": {
+                          "$binary": {
+                            "base64": "EjRWeBI0mHYSNBI0VniQEg==",
+                            "subType": "04"
+                          }
+                        }
+                      }
+                    ]
+                  }
+                },
+                "commandName": "create"
+              }
+            },
+            {
+              "commandStartedEvent": {
+                "command": {
+                  "createIndexes": "default",
+                  "indexes": [
+                    {
+                      "name": "__safeContent___1",
+                      "key": {
+                        "__safeContent__": 1
+                      }
+                    }
+                  ]
+                },
+                "commandName": "createIndexes"
+              }
+            }
+          ]
+        }
+      ]
+    }
+  ]
+}

test/connection_monitoring/pool-create-min-size-error.json

@@ -9,7 +9,9 @@
   ],
   "failPoint": {
     "configureFailPoint": "failCommand",
-    "mode": "alwaysOn",
+    "mode": {
+      "times": 50
+    },
     "data": {
       "failCommands": [
         "isMaster",

test/discovery_and_monitoring/unified/backpressure-network-error-fail-replicaset.json

@@ -1,142 +0,0 @@
-{
-  "description": "backpressure-network-error-fail-replicaset",
-  "schemaVersion": "1.17",
-  "runOnRequirements": [
-    {
-      "minServerVersion": "4.4",
-      "serverless": "forbid",
-      "topologies": [
-        "replicaset"
-      ]
-    }
-  ],
-  "createEntities": [
-    {
-      "client": {
-        "id": "setupClient",
-        "useMultipleMongoses": false
-      }
-    }
-  ],
-  "initialData": [
-    {
-      "collectionName": "backpressure-network-error-fail",
-      "databaseName": "sdam-tests",
-      "documents": [
-        {
-          "_id": 1
-        },
-        {
-          "_id": 2
-        }
-      ]
-    }
-  ],
-  "tests": [
-    {
-      "description": "apply backpressure on network connection errors during connection establishment",
-      "operations": [
-        {
-          "name": "createEntities",
-          "object": "testRunner",
-          "arguments": {
-            "entities": [
-              {
-                "client": {
-                  "id": "client",
-                  "useMultipleMongoses": false,
-                  "observeEvents": [
-                    "serverDescriptionChangedEvent",
-                    "poolClearedEvent"
-                  ],
-                  "uriOptions": {
-                    "retryWrites": false,
-                    "heartbeatFrequencyMS": 1000000,
-                    "serverMonitoringMode": "poll",
-                    "appname": "backpressureNetworkErrorFailTest"
-                  }
-                }
-              },
-              {
-                "database": {
-                  "id": "database",
-                  "client": "client",
-                  "databaseName": "sdam-tests"
-                }
-              },
-              {
-                "collection": {
-                  "id": "collection",
-                  "database": "database",
-                  "collectionName": "backpressure-network-error-fail"
-                }
-              }
-            ]
-          }
-        },
-        {
-          "name": "waitForEvent",
-          "object": "testRunner",
-          "arguments": {
-            "client": "client",
-            "event": {
-              "serverDescriptionChangedEvent": {
-                "newDescription": {
-                  "type": "RSPrimary"
-                }
-              }
-            },
-            "count": 1
-          }
-        },
-        {
-          "name": "failPoint",
-          "object": "testRunner",
-          "arguments": {
-            "client": "setupClient",
-            "failPoint": {
-              "configureFailPoint": "failCommand",
-              "mode": "alwaysOn",
-              "data": {
-                "failCommands": [
-                  "isMaster",
-                  "hello"
-                ],
-                "appName": "backpressureNetworkErrorFailTest",
-                "closeConnection": true
-              }
-            }
-          }
-        },
-        {
-          "name": "insertMany",
-          "object": "collection",
-          "arguments": {
-            "documents": [
-              {
-                "_id": 3
-              },
-              {
-                "_id": 4
-              }
-            ]
-          },
-          "expectError": {
-            "isError": true,
-            "errorLabelsContain": [
-              "SystemOverloadedError",
-              "RetryableError"
-            ]
-          }
-        }
-      ],
-      "expectEvents": [
-        {
-          "client": "client",
-          "eventType": "cmap",
-          "events": []
-        }
-      ]
-    }
-  ]
-}

test/discovery_and_monitoring/unified/backpressure-network-error-fail-single.json

@@ -1,142 +0,0 @@
-{
-  "description": "backpressure-network-error-fail-single",
-  "schemaVersion": "1.17",
-  "runOnRequirements": [
-    {
-      "minServerVersion": "4.4",
-      "serverless": "forbid",
-      "topologies": [
-        "single"
-      ]
-    }
-  ],
-  "createEntities": [
-    {
-      "client": {
-        "id": "setupClient",
-        "useMultipleMongoses": false
-      }
-    }
-  ],
-  "initialData": [
-    {
-      "collectionName": "backpressure-network-error-fail",
-      "databaseName": "sdam-tests",
-      "documents": [
-        {
-          "_id": 1
-        },
-        {
-          "_id": 2
-        }
-      ]
-    }
-  ],
-  "tests": [
-    {
-      "description": "apply backpressure on network connection errors during connection establishment",
-      "operations": [
-        {
-          "name": "createEntities",
-          "object": "testRunner",
-          "arguments": {
-            "entities": [
-              {
-                "client": {
-                  "id": "client",
-                  "useMultipleMongoses": false,
-                  "observeEvents": [
-                    "serverDescriptionChangedEvent",
-                    "poolClearedEvent"
-                  ],
-                  "uriOptions": {
-                    "retryWrites": false,
-                    "heartbeatFrequencyMS": 1000000,
-                    "serverMonitoringMode": "poll",
-                    "appname": "backpressureNetworkErrorFailTest"
-                  }
-                }
-              },
-              {
-                "database": {
-                  "id": "database",
-                  "client": "client",
-                  "databaseName": "sdam-tests"
-                }
-              },
-              {
-                "collection": {
-                  "id": "collection",
-                  "database": "database",
-                  "collectionName": "backpressure-network-error-fail"
-                }
-              }
-            ]
-          }
-        },
-        {
-          "name": "waitForEvent",
-          "object": "testRunner",
-          "arguments": {
-            "client": "client",
-            "event": {
-              "serverDescriptionChangedEvent": {
-                "newDescription": {
-                  "type": "Standalone"
-                }
-              }
-            },
-            "count": 1
-          }
-        },
-        {
-          "name": "failPoint",
-          "object": "testRunner",
-          "arguments": {
-            "client": "setupClient",
-            "failPoint": {
-              "configureFailPoint": "failCommand",
-              "mode": "alwaysOn",
-              "data": {
-                "failCommands": [
-                  "isMaster",
-                  "hello"
-                ],
-                "appName": "backpressureNetworkErrorFailTest",
-                "closeConnection": true
-              }
-            }
-          }
-        },
-        {
-          "name": "insertMany",
-          "object": "collection",
-          "arguments": {
-            "documents": [
-              {
-                "_id": 3
-              },
-              {
-                "_id": 4
-              }
-            ]
-          },
-          "expectError": {
-            "isError": true,
-            "errorLabelsContain": [
-              "SystemOverloadedError",
-              "RetryableError"
-            ]
-          }
-        }
-      ],
-      "expectEvents": [
-        {
-          "client": "client",
-          "eventType": "cmap",
-          "events": []
-        }
-      ]
-    }
-  ]
-}

test/discovery_and_monitoring/unified/backpressure-network-timeout-fail-replicaset.json

@@ -1,145 +0,0 @@
-{
-  "description": "backpressure-network-timeout-error-replicaset",
-  "schemaVersion": "1.17",
-  "runOnRequirements": [
-    {
-      "minServerVersion": "4.4",
-      "serverless": "forbid",
-      "topologies": [
-        "replicaset"
-      ]
-    }
-  ],
-  "createEntities": [
-    {
-      "client": {
-        "id": "setupClient",
-        "useMultipleMongoses": false
-      }
-    }
-  ],
-  "initialData": [
-    {
-      "collectionName": "backpressure-network-timeout-error",
-      "databaseName": "sdam-tests",
-      "documents": [
-        {
-          "_id": 1
-        },
-        {
-          "_id": 2
-        }
-      ]
-    }
-  ],
-  "tests": [
-    {
-      "description": "apply backpressure on network timeout error during connection establishment",
-      "operations": [
-        {
-          "name": "createEntities",
-          "object": "testRunner",
-          "arguments": {
-            "entities": [
-              {
-                "client": {
-                  "id": "client",
-                  "useMultipleMongoses": false,
-                  "observeEvents": [
-                    "serverDescriptionChangedEvent",
-                    "poolClearedEvent"
-                  ],
-                  "uriOptions": {
-                    "retryWrites": false,
-                    "heartbeatFrequencyMS": 1000000,
-                    "appname": "backpressureNetworkTimeoutErrorTest",
-                    "serverMonitoringMode": "poll",
-                    "connectTimeoutMS": 250,
-                    "socketTimeoutMS": 250
-                  }
-                }
-              },
-              {
-                "database": {
-                  "id": "database",
-                  "client": "client",
-                  "databaseName": "sdam-tests"
-                }
-              },
-              {
-                "collection": {
-                  "id": "collection",
-                  "database": "database",
-                  "collectionName": "backpressure-network-timeout-error"
-                }
-              }
-            ]
-          }
-        },
-        {
-          "name": "waitForEvent",
-          "object": "testRunner",
-          "arguments": {
-            "client": "client",
-            "event": {
-              "serverDescriptionChangedEvent": {
-                "newDescription": {
-                  "type": "RSPrimary"
-                }
-              }
-            },
-            "count": 1
-          }
-        },
-        {
-          "name": "failPoint",
-          "object": "testRunner",
-          "arguments": {
-            "client": "setupClient",
-            "failPoint": {
-              "configureFailPoint": "failCommand",
-              "mode": "alwaysOn",
-              "data": {
-                "failCommands": [
-                  "isMaster",
-                  "hello"
-                ],
-                "blockConnection": true,
-                "blockTimeMS": 500,
-                "appName": "backpressureNetworkTimeoutErrorTest"
-              }
-            }
-          }
-        },
-        {
-          "name": "insertMany",
-          "object": "collection",
-          "arguments": {
-            "documents": [
-              {
-                "_id": 3
-              },
-              {
-                "_id": 4
-              }
-            ]
-          },
-          "expectError": {
-            "isError": true,
-            "errorLabelsContain": [
-              "SystemOverloadedError",
-              "RetryableError"
-            ]
-          }
-        }
-      ],
-      "expectEvents": [
-        {
-          "client": "client",
-          "eventType": "cmap",
-          "events": []
-        }
-      ]
-    }
-  ]
-}

test/discovery_and_monitoring/unified/backpressure-network-timeout-fail-single.json

@@ -1,145 +0,0 @@
-{
-  "description": "backpressure-network-timeout-error-single",
-  "schemaVersion": "1.17",
-  "runOnRequirements": [
-    {
-      "minServerVersion": "4.4",
-      "serverless": "forbid",
-      "topologies": [
-        "single"
-      ]
-    }
-  ],
-  "createEntities": [
-    {
-      "client": {
-        "id": "setupClient",
-        "useMultipleMongoses": false
-      }
-    }
-  ],
-  "initialData": [
-    {
-      "collectionName": "backpressure-network-timeout-error",
-      "databaseName": "sdam-tests",
-      "documents": [
-        {
-          "_id": 1
-        },
-        {
-          "_id": 2
-        }
-      ]
-    }
-  ],
-  "tests": [
-    {
-      "description": "apply backpressure on network timeout error during connection establishment",
-      "operations": [
-        {
-          "name": "createEntities",
-          "object": "testRunner",
-          "arguments": {
-            "entities": [
-              {
-                "client": {
-                  "id": "client",
-                  "useMultipleMongoses": false,
-                  "observeEvents": [
-                    "serverDescriptionChangedEvent",
-                    "poolClearedEvent"
-                  ],
-                  "uriOptions": {
-                    "retryWrites": false,
-                    "heartbeatFrequencyMS": 1000000,
-                    "appname": "backpressureNetworkTimeoutErrorTest",
-                    "serverMonitoringMode": "poll",
-                    "connectTimeoutMS": 250,
-                    "socketTimeoutMS": 250
-                  }
-                }
-              },
-              {
-                "database": {
-                  "id": "database",
-                  "client": "client",
-                  "databaseName": "sdam-tests"
-                }
-              },
-              {
-                "collection": {
-                  "id": "collection",
-                  "database": "database",
-                  "collectionName": "backpressure-network-timeout-error"
-                }
-              }
-            ]
-          }
-        },
-        {
-          "name": "waitForEvent",
-          "object": "testRunner",
-          "arguments": {
-            "client": "client",
-            "event": {
-              "serverDescriptionChangedEvent": {
-                "newDescription": {
-                  "type": "Standalone"
-                }
-              }
-            },
-            "count": 1
-          }
-        },
-        {
-          "name": "failPoint",
-          "object": "testRunner",
-          "arguments": {
-            "client": "setupClient",
-            "failPoint": {
-              "configureFailPoint": "failCommand",
-              "mode": "alwaysOn",
-              "data": {
-                "failCommands": [
-                  "isMaster",
-                  "hello"
-                ],
-                "blockConnection": true,
-                "blockTimeMS": 500,
-                "appName": "backpressureNetworkTimeoutErrorTest"
-              }
-            }
-          }
-        },
-        {
-          "name": "insertMany",
-          "object": "collection",
-          "arguments": {
-            "documents": [
-              {
-                "_id": 3
-              },
-              {
-                "_id": 4
-              }
-            ]
-          },
-          "expectError": {
-            "isError": true,
-            "errorLabelsContain": [
-              "SystemOverloadedError",
-              "RetryableError"
-            ]
-          }
-        }
-      ],
-      "expectEvents": [
-        {
-          "client": "client",
-          "eventType": "cmap",
-          "events": []
-        }
-      ]
-    }
-  ]
-}

test/discovery_and_monitoring/unified/backpressure-server-description-unchanged-on-min-pool-size-population-error.json

@@ -1,106 +0,0 @@
-{
-  "description": "backpressure-server-description-unchanged-on-min-pool-size-population-error",
-  "schemaVersion": "1.17",
-  "runOnRequirements": [
-    {
-      "minServerVersion": "4.4",
-      "serverless": "forbid",
-      "topologies": [
-        "single"
-      ]
-    }
-  ],
-  "createEntities": [
-    {
-      "client": {
-        "id": "setupClient",
-        "useMultipleMongoses": false
-      }
-    }
-  ],
-  "tests": [
-    {
-      "description": "the server description is not changed on handshake error during minPoolSize population",
-      "operations": [
-        {
-          "name": "failPoint",
-          "object": "testRunner",
-          "arguments": {
-            "client": "setupClient",
-            "failPoint": {
-              "configureFailPoint": "failCommand",
-              "mode": {
-                "skip": 1
-              },
-              "data": {
-                "failCommands": [
-                  "hello",
-                  "isMaster"
-                ],
-                "appName": "authErrorTest",
-                "closeConnection": true
-              }
-            }
-          }
-        },
-        {
-          "name": "createEntities",
-          "object": "testRunner",
-          "arguments": {
-            "entities": [
-              {
-                "client": {
-                  "id": "client",
-                  "observeEvents": [
-                    "serverDescriptionChangedEvent",
-                    "connectionClosedEvent"
-                  ],
-                  "uriOptions": {
-                    "appname": "authErrorTest",
-                    "minPoolSize": 5,
-                    "maxConnecting": 1,
-                    "serverMonitoringMode": "poll",
-                    "heartbeatFrequencyMS": 1000000
-                  }
-                }
-              }
-            ]
-          }
-        },
-        {
-          "name": "waitForEvent",
-          "object": "testRunner",
-          "arguments": {
-            "client": "client",
-            "event": {
-              "serverDescriptionChangedEvent": {}
-            },
-            "count": 1
-          }
-        },
-        {
-          "name": "waitForEvent",
-          "object": "testRunner",
-          "arguments": {
-            "client": "client",
-            "event": {
-              "connectionClosedEvent": {}
-            },
-            "count": 1
-          }
-        }
-      ],
-      "expectEvents": [
-        {
-          "client": "client",
-          "eventType": "sdam",
-          "events": [
-            {
-              "serverDescriptionChangedEvent": {}
-            }
-          ]
-        }
-      ]
-    }
-  ]
-}

test/test_azure_helpers.py

@@ -150,6 +150,20 @@ class TestGetAzureResponse(unittest.TestCase):
         _, kwargs = mock_open.call_args
         self.assertEqual(kwargs["timeout"], 42)
 
+    def test_client_id_is_url_encoded(self):
+        """Ensure special characters in client_id are percent-encoded."""
+        body = json.dumps({"access_token": "tok", "expires_in": "3600"})
+        with _mock_urlopen(200, body) as mock_open:
+            self._call(client_id="id with spaces&special=chars")
+
+        url = mock_open.call_args[0][0].full_url
+        # '&' and '=' must be percent-encoded so they don't inject extra query params
+        self.assertIn("client_id=id%20with%20spaces%26special%3Dchars", url)
+        # The encoded client_id should not introduce a raw '&'
+        # Count params: api-version, resource, client_id — exactly 3
+        query_string = url.split("?", 1)[1]
+        self.assertEqual(query_string.count("&"), 2)
+
 
 if __name__ == "__main__":
     unittest.main()

test/test_bson.py

@@ -1269,6 +1269,22 @@ class TestBSON(unittest.TestCase):
             encode(doc)
         self.assertEqual(cm.exception.document, doc)
 
+    def test_binary_length_accounts_for_header(self):
+        size = 20
+        binary_length = 12  # 5 more than the actual 7 bytes
+
+        payload = b""
+        payload += struct.pack("<i", size)  # document size
+        payload += b"\x05"  # type = Binary
+        payload += b"a\x00"  # key "a"
+        payload += struct.pack("<I", binary_length)  # Binary length (inflated)
+        payload += b"\x00"  # subtype 0
+        payload += b"\x41" * 7  # value
+        payload += b"\x00"  # EOO
+
+        with self.assertRaises(InvalidBSON):
+            decode(payload)
+
 
 class TestCodecOptions(unittest.TestCase):
     def test_document_class(self):

test/test_client.py

@@ -2666,11 +2666,11 @@ class TestClientPool(MockClientTest):
 
         wait_until(lambda: len(c.nodes) == 1, "connect")
         self.assertEqual(c.address, ("c", 3))
-        # Assert that we create 1 pooled connection.
+        # Wait for the pooled connection to be registered
         listener.wait_for_event(monitoring.ConnectionReadyEvent, 1)
         self.assertEqual(listener.event_count(monitoring.ConnectionCreatedEvent), 1)
         arbiter = c._topology.get_server_by_address(("c", 3))
-        self.assertEqual(len(arbiter.pool.conns), 1)
+        wait_until(lambda: len(arbiter.pool.conns) == 1, "create 1 pooled connection")
         # Arbiter pool is marked ready.
         self.assertEqual(listener.event_count(monitoring.PoolReadyEvent), 1)
 

test/test_daemon.py

@@ -0,0 +1,183 @@
+# 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.
+
+"""Test the pymongo daemon module."""
+from __future__ import annotations
+
+import subprocess
+import sys
+import warnings
+from unittest.mock import MagicMock, patch
+
+sys.path[0:0] = [""]
+
+from test import unittest
+
+import pymongo.daemon as daemon_module
+from pymongo.daemon import _popen_wait, _silence_resource_warning, _spawn_daemon
+
+
+class TestPopenWait(unittest.TestCase):
+    def test_returns_returncode_on_success(self):
+        mock_popen = MagicMock()
+        mock_popen.wait.return_value = 0
+        self.assertEqual(0, _popen_wait(mock_popen, timeout=5))
+        mock_popen.wait.assert_called_once_with(timeout=5)
+
+    def test_returns_none_on_timeout_expired(self):
+        mock_popen = MagicMock()
+        mock_popen.wait.side_effect = subprocess.TimeoutExpired(cmd="foo", timeout=5)
+        self.assertIsNone(_popen_wait(mock_popen, timeout=5))
+
+    def test_none_timeout_passes_through(self):
+        mock_popen = MagicMock()
+        mock_popen.wait.return_value = 1
+        self.assertEqual(1, _popen_wait(mock_popen, timeout=None))
+        mock_popen.wait.assert_called_once_with(timeout=None)
+
+
+class TestSilenceResourceWarning(unittest.TestCase):
+    def test_sets_returncode_to_zero(self):
+        mock_popen = MagicMock()
+        mock_popen.returncode = None
+        _silence_resource_warning(mock_popen)
+        self.assertEqual(0, mock_popen.returncode)
+
+    def test_no_op_for_none(self):
+        # Should not raise when popen is None (mongocryptd spawn failed).
+        _silence_resource_warning(None)
+
+
+@unittest.skipIf(sys.platform == "win32", "Unix only")
+class TestSpawnUnix(unittest.TestCase):
+    def setUp(self):
+        from pymongo.daemon import _spawn
+
+        self._spawn = _spawn
+
+    def test_returns_popen_on_success(self):
+        mock_popen = MagicMock()
+        with patch("subprocess.Popen", return_value=mock_popen):
+            result = self._spawn(["somecommand"])
+        self.assertIs(mock_popen, result)
+
+    def test_filenotfound_warns_and_returns_none(self):
+        with patch("subprocess.Popen", side_effect=FileNotFoundError("not found")):
+            with warnings.catch_warnings(record=True) as w:
+                warnings.simplefilter("always")
+                result = self._spawn(["nonexistent_command"])
+        self.assertIsNone(result)
+        self.assertEqual(1, len(w))
+        self.assertIs(RuntimeWarning, w[0].category)
+        self.assertIn("nonexistent_command", str(w[0].message))
+
+
+@unittest.skipIf(sys.platform == "win32", "Unix only")
+class TestSpawnDaemonDoublePopen(unittest.TestCase):
+    def setUp(self):
+        from pymongo.daemon import _spawn_daemon_double_popen
+
+        self._spawn_daemon_double_popen = _spawn_daemon_double_popen
+
+    def test_spawns_this_file_as_intermediate(self):
+        mock_popen = MagicMock()
+        mock_popen.wait.return_value = 0
+        with patch("subprocess.Popen", return_value=mock_popen) as mock_cls:
+            self._spawn_daemon_double_popen(["somecommand", "--arg"])
+        spawner_args = mock_cls.call_args[0][0]
+        self.assertEqual(sys.executable, spawner_args[0])
+        self.assertIn("daemon.py", spawner_args[1])
+        self.assertIn("somecommand", spawner_args)
+
+    def test_waits_for_intermediate_process(self):
+        mock_popen = MagicMock()
+        with patch("subprocess.Popen", return_value=mock_popen):
+            self._spawn_daemon_double_popen(["somecommand"])
+        mock_popen.wait.assert_called_once_with(timeout=daemon_module._WAIT_TIMEOUT)
+
+    def test_continues_on_timeout(self):
+        # _popen_wait swallows TimeoutExpired — double Popen must not raise.
+        mock_popen = MagicMock()
+        mock_popen.wait.side_effect = subprocess.TimeoutExpired(cmd="foo", timeout=10)
+        with patch("subprocess.Popen", return_value=mock_popen):
+            self._spawn_daemon_double_popen(["somecommand"])  # must not raise
+
+
+@unittest.skipIf(sys.platform == "win32", "Unix only")
+class TestSpawnDaemonUnix(unittest.TestCase):
+    def test_uses_double_popen_when_executable_set(self):
+        with patch("pymongo.daemon._spawn_daemon_double_popen") as mock_double:
+            _spawn_daemon(["somecommand"])
+        mock_double.assert_called_once_with(["somecommand"])
+
+    def test_fallback_to_spawn_when_no_executable(self):
+        with patch("pymongo.daemon._spawn") as mock_spawn:
+            with patch.object(sys, "executable", ""):
+                _spawn_daemon(["somecommand"])
+        mock_spawn.assert_called_once_with(["somecommand"])
+
+
+@unittest.skipUnless(sys.platform == "win32", "Windows only")
+class TestSpawnDaemonWindows(unittest.TestCase):
+    def test_silences_resource_warning_on_success(self):
+        mock_popen = MagicMock()
+        with patch("subprocess.Popen", return_value=mock_popen):
+            _spawn_daemon(["somecommand"])
+        self.assertEqual(0, mock_popen.returncode)
+
+    def test_filenotfound_warns(self):
+        with patch("subprocess.Popen", side_effect=FileNotFoundError("not found")):
+            with warnings.catch_warnings(record=True) as w:
+                warnings.simplefilter("always")
+                _spawn_daemon(["nonexistent_command"])
+        self.assertEqual(1, len(w))
+        self.assertIs(RuntimeWarning, w[0].category)
+        self.assertIn("nonexistent_command", str(w[0].message))
+
+    def test_uses_detached_process_flag(self):
+        # DETACHED_PROCESS must be passed so the child survives parent exit.
+        mock_popen = MagicMock()
+        with patch("subprocess.Popen", return_value=mock_popen) as mock_cls:
+            _spawn_daemon(["somecommand"])
+        kwargs = mock_cls.call_args[1]
+        self.assertEqual(daemon_module._DETACHED_PROCESS, kwargs["creationflags"])
+
+    def test_uses_devnull_for_stdio(self):
+        # stdin/stdout/stderr must be redirected to devnull to fully detach.
+        mock_popen = MagicMock()
+        with patch("subprocess.Popen", return_value=mock_popen) as mock_cls:
+            _spawn_daemon(["somecommand"])
+        kwargs = mock_cls.call_args[1]
+        self.assertIsNotNone(kwargs.get("stdin"))
+        self.assertIsNotNone(kwargs.get("stdout"))
+        self.assertIsNotNone(kwargs.get("stderr"))
+
+    def test_detached_process_constant_value(self):
+        # Value must match the Windows DETACHED_PROCESS process creation flag.
+        self.assertEqual(0x00000008, daemon_module._DETACHED_PROCESS)
+
+
+@unittest.skipIf(sys.platform == "win32", "Unix only")
+class TestMainBlock(unittest.TestCase):
+    def test_exits_with_zero(self):
+        # Run daemon.py as a script with a no-op subprocess; verify it exits cleanly.
+        result = subprocess.run(
+            [sys.executable, "-m", "pymongo.daemon", sys.executable, "-c", "pass"],
+            timeout=15,
+        )
+        self.assertEqual(0, result.returncode)
+
+
+if __name__ == "__main__":
+    unittest.main()

test/test_encryption.py

@@ -872,8 +872,6 @@ class TestViews(EncryptionIntegrationTest):
 
 
 class TestCorpus(EncryptionIntegrationTest):
-    # PYTHON-5708: Encryption tests sending large payloads fail on some mongocryptd versions.
-    @client_context.require_version_max(6, 99)
     @unittest.skipUnless(any(AWS_CREDS.values()), "AWS environment credentials are not set")
     def setUp(self):
         super().setUp()
@@ -1050,8 +1048,6 @@ class TestBsonSizeBatches(EncryptionIntegrationTest):
     client_encrypted: MongoClient
     listener: OvertCommandListener
 
-    # PYTHON-5708: Encryption tests sending large payloads fail on some mongocryptd versions.
-    @client_context.require_version_max(6, 99)
     def setUp(self):
         super().setUp()
         db = client_context.client.db
@@ -3308,6 +3304,7 @@ class TestAutomaticDecryptionKeys(EncryptionIntegrationTest):
 class TestExplicitTextEncryptionProse(EncryptionIntegrationTest):
     @client_context.require_no_standalone
     @client_context.require_version_min(8, 2, -1)
+    @client_context.require_version_max(8, 99, 99)
     @client_context.require_libmongocrypt_min(1, 15, 1)
     @client_context.require_pymongocrypt_min(1, 16, 0)
     def setUp(self):

test/test_event_loggers.py

@@ -0,0 +1,374 @@
+# 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.
+
+"""Tests for pymongo.event_loggers."""
+from __future__ import annotations
+
+import sys
+from unittest.mock import MagicMock, patch
+
+sys.path[0:0] = [""]
+
+from test import unittest
+
+from pymongo.event_loggers import (
+    CommandLogger,
+    ConnectionPoolLogger,
+    HeartbeatLogger,
+    ServerLogger,
+    TopologyLogger,
+)
+
+
+class TestCommandLogger(unittest.TestCase):
+    def setUp(self):
+        self.logger = CommandLogger()
+
+    def test_started_logs_info(self):
+        event = MagicMock()
+        event.command_name = "find"
+        event.request_id = 42
+        event.connection_id = ("localhost", 27017)
+        with self.assertLogs(level="INFO") as logs:
+            self.logger.started(event)
+        log = logs.records[0].getMessage()
+        self.assertIn("find", log)
+        self.assertIn("42", log)
+        self.assertIn("started", log)
+
+    def test_succeeded_logs_info(self):
+        event = MagicMock()
+        event.command_name = "insert"
+        event.request_id = 7
+        event.connection_id = ("localhost", 27017)
+        event.duration_micros = 500
+        with self.assertLogs(level="INFO") as logs:
+            self.logger.succeeded(event)
+        log = logs.records[0].getMessage()
+        self.assertIn("insert", log)
+        self.assertIn("7", log)
+        self.assertIn("500", log)
+        self.assertIn("microseconds", log)
+        self.assertIn("succeeded", log)
+
+    def test_failed_logs_info(self):
+        event = MagicMock()
+        event.command_name = "delete"
+        event.request_id = 3
+        event.connection_id = ("localhost", 27017)
+        event.duration_micros = 300
+        with self.assertLogs(level="INFO") as logs:
+            self.logger.failed(event)
+        log = logs.records[0].getMessage()
+        self.assertIn("delete", log)
+        self.assertIn("3", log)
+        self.assertIn("300", log)
+        self.assertIn("microseconds", log)
+        self.assertIn("failed", log)
+
+
+class TestServerLogger(unittest.TestCase):
+    def setUp(self):
+        self.logger = ServerLogger()
+
+    def test_opened_logs_info(self):
+        event = MagicMock()
+        event.server_address = ("host1", 27017)
+        event.topology_id = "topology-abc"
+        with self.assertLogs(level="INFO") as logs:
+            self.logger.opened(event)
+        log = logs.records[0].getMessage()
+        self.assertIn("host1", log)
+        self.assertIn("topology-abc", log)
+
+    def test_closed_logs_warning(self):
+        event = MagicMock()
+        event.server_address = ("host1", 27017)
+        event.topology_id = "topology-abc"
+        with self.assertLogs(level="WARNING") as logs:
+            self.logger.closed(event)
+        log = logs.records[0].getMessage()
+        self.assertIn("host1", log)
+        self.assertIn("topology-abc", log)
+
+    def test_description_changed_logs_when_type_changes(self):
+        event = MagicMock()
+        event.server_address = ("host1", 27017)
+        event.previous_description.server_type = 1
+        event.previous_description.server_type_name = "Unknown"
+        event.new_description.server_type = 2
+        event.new_description.server_type_name = "Standalone"
+        with self.assertLogs(level="INFO") as logs:
+            self.logger.description_changed(event)
+        log = logs.records[0].getMessage()
+        self.assertIn("Unknown", log)
+        self.assertIn("Standalone", log)
+
+    def test_description_changed_no_log_when_type_same(self):
+        event = MagicMock()
+        event.previous_description.server_type = 2
+        event.new_description.server_type = 2
+        with patch("logging.info") as mock_info:
+            self.logger.description_changed(event)
+        mock_info.assert_not_called()
+
+
+class TestHeartbeatLogger(unittest.TestCase):
+    def setUp(self):
+        self.logger = HeartbeatLogger()
+
+    def test_started_logs_info(self):
+        event = MagicMock()
+        event.connection_id = ("mongo.host", 27017)
+        with self.assertLogs(level="INFO") as logs:
+            self.logger.started(event)
+        log = logs.records[0].getMessage()
+        self.assertIn("mongo.host", log)
+
+    def test_succeeded_logs_info(self):
+        event = MagicMock()
+        event.connection_id = ("mongo.host", 27017)
+        event.reply.document = {"ok": 1, "maxWireVersion": 17}
+        with self.assertLogs(level="INFO") as logs:
+            self.logger.succeeded(event)
+        log = logs.records[0].getMessage()
+        self.assertIn("mongo.host", log)
+        self.assertIn("succeeded", log)
+        self.assertIn("maxWireVersion", log)
+
+    def test_failed_logs_warning(self):
+        event = MagicMock()
+        event.connection_id = ("mongo.host", 27017)
+        event.reply = TimeoutError("timed out")
+        with self.assertLogs(level="WARNING") as logs:
+            self.logger.failed(event)
+        log = logs.records[0].getMessage()
+        self.assertIn("mongo.host", log)
+        self.assertIn("failed", log)
+        self.assertIn("timed out", log)
+
+
+class TestTopologyLogger(unittest.TestCase):
+    def setUp(self):
+        self.logger = TopologyLogger()
+
+    def test_opened_logs_info(self):
+        event = MagicMock()
+        event.topology_id = "topo-1"
+        with self.assertLogs(level="INFO") as logs:
+            self.logger.opened(event)
+        log = logs.records[0].getMessage()
+        self.assertIn("topo-1", log)
+        self.assertIn("opened", log)
+
+    def test_closed_logs_info(self):
+        event = MagicMock()
+        event.topology_id = "topo-1"
+        with self.assertLogs(level="INFO") as logs:
+            self.logger.closed(event)
+        log = logs.records[0].getMessage()
+        self.assertIn("topo-1", log)
+        self.assertIn("closed", log)
+
+    def test_description_changed_always_logs_update(self):
+        event = MagicMock()
+        event.topology_id = "topo-1"
+        event.previous_description.topology_type = 1
+        event.new_description.topology_type = 1
+        event.new_description.has_writable_server.return_value = True
+        event.new_description.has_readable_server.return_value = True
+        with self.assertLogs(level="INFO") as logs:
+            self.logger.description_changed(event)
+        messages = [r.getMessage() for r in logs.records]
+        self.assertTrue(any("updated" in m for m in messages))
+        self.assertTrue(any("topo-1" in m for m in messages))
+
+    def test_description_changed_logs_type_change(self):
+        event = MagicMock()
+        event.topology_id = "topo-2"
+        event.previous_description.topology_type = 0
+        event.previous_description.topology_type_name = "Unknown"
+        event.new_description.topology_type = 1
+        event.new_description.topology_type_name = "Single"
+        event.new_description.has_writable_server.return_value = True
+        event.new_description.has_readable_server.return_value = True
+        with self.assertLogs(level="INFO") as logs:
+            self.logger.description_changed(event)
+        messages = [r.getMessage() for r in logs.records]
+        self.assertTrue(any("Unknown" in m and "Single" in m for m in messages))
+
+    def test_description_changed_no_type_change_log_when_same(self):
+        event = MagicMock()
+        event.topology_id = "topo-1"
+        event.previous_description.topology_type = 1
+        event.new_description.topology_type = 1
+        event.new_description.has_writable_server.return_value = True
+        event.new_description.has_readable_server.return_value = True
+        with self.assertLogs(level="INFO") as logs:
+            self.logger.description_changed(event)
+        messages = [r.getMessage() for r in logs.records]
+        self.assertFalse(any("changed type" in m for m in messages))
+
+    def test_description_changed_warns_no_writable_server(self):
+        event = MagicMock()
+        event.previous_description.topology_type = 1
+        event.new_description.topology_type = 1
+        event.new_description.has_writable_server.return_value = False
+        event.new_description.has_readable_server.return_value = True
+        with self.assertLogs(level="WARNING") as logs:
+            self.logger.description_changed(event)
+        messages = [r.getMessage() for r in logs.records]
+        self.assertTrue(any("writable" in m for m in messages))
+
+    def test_description_changed_warns_no_readable_server(self):
+        event = MagicMock()
+        event.previous_description.topology_type = 1
+        event.new_description.topology_type = 1
+        event.new_description.has_writable_server.return_value = True
+        event.new_description.has_readable_server.return_value = False
+        with self.assertLogs(level="WARNING") as logs:
+            self.logger.description_changed(event)
+        messages = [r.getMessage() for r in logs.records]
+        self.assertTrue(any("readable" in m for m in messages))
+
+    def test_description_changed_warns_both_unavailable(self):
+        event = MagicMock()
+        event.previous_description.topology_type = 1
+        event.new_description.topology_type = 1
+        event.new_description.has_writable_server.return_value = False
+        event.new_description.has_readable_server.return_value = False
+        with self.assertLogs(level="WARNING") as logs:
+            self.logger.description_changed(event)
+        warning_messages = [r.getMessage() for r in logs.records if r.levelname == "WARNING"]
+        self.assertEqual(len(warning_messages), 2)
+
+
+class TestConnectionPoolLogger(unittest.TestCase):
+    def setUp(self):
+        self.logger = ConnectionPoolLogger()
+
+    def test_pool_created(self):
+        event = MagicMock()
+        event.address = ("localhost", 27017)
+        with self.assertLogs(level="INFO") as logs:
+            self.logger.pool_created(event)
+        log = logs.records[0].getMessage()
+        self.assertIn("pool created", log)
+        self.assertIn("localhost", log)
+
+    def test_pool_ready(self):
+        event = MagicMock()
+        event.address = ("localhost", 27017)
+        with self.assertLogs(level="INFO") as logs:
+            self.logger.pool_ready(event)
+        log = logs.records[0].getMessage()
+        self.assertIn("pool ready", log)
+        self.assertIn("localhost", log)
+
+    def test_pool_cleared(self):
+        event = MagicMock()
+        event.address = ("localhost", 27017)
+        with self.assertLogs(level="INFO") as logs:
+            self.logger.pool_cleared(event)
+        log = logs.records[0].getMessage()
+        self.assertIn("pool cleared", log)
+        self.assertIn("localhost", log)
+
+    def test_pool_closed(self):
+        event = MagicMock()
+        event.address = ("localhost", 27017)
+        with self.assertLogs(level="INFO") as logs:
+            self.logger.pool_closed(event)
+        log = logs.records[0].getMessage()
+        self.assertIn("pool closed", log)
+        self.assertIn("localhost", log)
+
+    def test_connection_created(self):
+        event = MagicMock()
+        event.address = ("localhost", 27017)
+        event.connection_id = 5
+        with self.assertLogs(level="INFO") as logs:
+            self.logger.connection_created(event)
+        log = logs.records[0].getMessage()
+        self.assertIn("connection created", log)
+        self.assertIn("5", log)
+        self.assertIn("localhost", log)
+
+    def test_connection_ready(self):
+        event = MagicMock()
+        event.address = ("localhost", 27017)
+        event.connection_id = 5
+        with self.assertLogs(level="INFO") as logs:
+            self.logger.connection_ready(event)
+        log = logs.records[0].getMessage()
+        self.assertIn("connection setup succeeded", log)
+        self.assertIn("5", log)
+
+    def test_connection_closed(self):
+        event = MagicMock()
+        event.address = ("localhost", 27017)
+        event.connection_id = 5
+        event.reason = "stale"
+        with self.assertLogs(level="INFO") as logs:
+            self.logger.connection_closed(event)
+        log = logs.records[0].getMessage()
+        self.assertIn("connection closed", log)
+        self.assertIn("5", log)
+        self.assertIn("stale", log)
+
+    def test_connection_check_out_started(self):
+        event = MagicMock()
+        event.address = ("localhost", 27017)
+        with self.assertLogs(level="INFO") as logs:
+            self.logger.connection_check_out_started(event)
+        log = logs.records[0].getMessage()
+        self.assertIn("check out started", log)
+        self.assertIn("localhost", log)
+
+    def test_connection_check_out_failed(self):
+        event = MagicMock()
+        event.address = ("localhost", 27017)
+        event.reason = "timeout"
+        with self.assertLogs(level="INFO") as logs:
+            self.logger.connection_check_out_failed(event)
+        log = logs.records[0].getMessage()
+        self.assertIn("check out failed", log)
+        self.assertIn("timeout", log)
+        self.assertIn("localhost", log)
+
+    def test_connection_checked_out(self):
+        event = MagicMock()
+        event.address = ("localhost", 27017)
+        event.connection_id = 3
+        with self.assertLogs(level="INFO") as logs:
+            self.logger.connection_checked_out(event)
+        log = logs.records[0].getMessage()
+        self.assertIn("checked out", log)
+        self.assertIn("3", log)
+        self.assertIn("localhost", log)
+
+    def test_connection_checked_in(self):
+        event = MagicMock()
+        event.address = ("localhost", 27017)
+        event.connection_id = 3
+        with self.assertLogs(level="INFO") as logs:
+            self.logger.connection_checked_in(event)
+        log = logs.records[0].getMessage()
+        self.assertIn("checked into", log)
+        self.assertIn("3", log)
+        self.assertIn("localhost", log)
+
+
+if __name__ == "__main__":
+    unittest.main()

test/test_pyopenssl_context.py

@@ -0,0 +1,271 @@
+# 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.
+
+"""Unit tests for pyopenssl_context.py.
+
+These tests require PyOpenSSL (install via: pip install pymongo[ocsp]).
+Tests are automatically skipped when PyOpenSSL is not available.
+"""
+
+from __future__ import annotations
+
+import ssl
+import sys
+from unittest.mock import patch
+
+sys.path[0:0] = [""]
+
+from test import unittest
+
+try:
+    from pymongo import pyopenssl_context as _ctx_module
+    from pymongo.pyopenssl_context import (
+        PROTOCOL_SSLv23,
+        SSLContext,
+        _is_ip_address,
+        _ragged_eof,
+    )
+
+    _HAVE_PYOPENSSL = True
+except ImportError:
+    _HAVE_PYOPENSSL = False
+
+
+# ---------------------------------------------------------------------------
+# Pure functions (no SSL context required)
+# ---------------------------------------------------------------------------
+
+
+class TestIsIpAddress(unittest.TestCase):
+    @unittest.skipUnless(_HAVE_PYOPENSSL, "PyOpenSSL is not available.")
+    def test_ipv4(self):
+        self.assertTrue(_is_ip_address("192.168.1.1"))
+
+    @unittest.skipUnless(_HAVE_PYOPENSSL, "PyOpenSSL is not available.")
+    def test_ipv6(self):
+        self.assertTrue(_is_ip_address("::1"))
+        self.assertTrue(_is_ip_address("2001:db8::1"))
+
+    @unittest.skipUnless(_HAVE_PYOPENSSL, "PyOpenSSL is not available.")
+    def test_hostname_is_not_ip(self):
+        self.assertFalse(_is_ip_address("example.com"))
+        self.assertFalse(_is_ip_address("localhost"))
+
+    @unittest.skipUnless(_HAVE_PYOPENSSL, "PyOpenSSL is not available.")
+    def test_invalid_string_returns_false(self):
+        self.assertFalse(_is_ip_address("not-an-ip"))
+
+    @unittest.skipUnless(_HAVE_PYOPENSSL, "PyOpenSSL is not available.")
+    def test_unicode_error_returns_false(self):
+        # UnicodeError path: some inputs that can't be decoded.
+        # ip_address raises UnicodeError for byte strings with non-ASCII.
+        self.assertFalse(_is_ip_address(b"\xff\xfe"))
+
+
+class TestRaggedEof(unittest.TestCase):
+    @unittest.skipUnless(_HAVE_PYOPENSSL, "PyOpenSSL is not available.")
+    def test_matching_args_returns_true(self):
+        from OpenSSL.SSL import SysCallError
+
+        exc = SysCallError(-1, "Unexpected EOF")
+        self.assertTrue(_ragged_eof(exc))
+
+    @unittest.skipUnless(_HAVE_PYOPENSSL, "PyOpenSSL is not available.")
+    def test_non_matching_args_returns_false(self):
+        from OpenSSL.SSL import SysCallError
+
+        exc = SysCallError(0, "something else")
+        self.assertFalse(_ragged_eof(exc))
+
+    @unittest.skipUnless(_HAVE_PYOPENSSL, "PyOpenSSL is not available.")
+    def test_wrong_code_returns_false(self):
+        from OpenSSL.SSL import SysCallError
+
+        exc = SysCallError(5, "Unexpected EOF")
+        self.assertFalse(_ragged_eof(exc))
+
+
+# ---------------------------------------------------------------------------
+# SSLContext — construction and properties
+# ---------------------------------------------------------------------------
+
+
+class TestSSLContextConstruction(unittest.TestCase):
+    def _make(self):
+        return SSLContext(PROTOCOL_SSLv23)
+
+    @unittest.skipUnless(_HAVE_PYOPENSSL, "PyOpenSSL is not available.")
+    def test_protocol_property(self):
+        ctx = self._make()
+        self.assertEqual(ctx.protocol, PROTOCOL_SSLv23)
+
+    @unittest.skipUnless(_HAVE_PYOPENSSL, "PyOpenSSL is not available.")
+    def test_default_check_hostname(self):
+        ctx = self._make()
+        self.assertTrue(ctx.check_hostname)
+
+    @unittest.skipUnless(_HAVE_PYOPENSSL, "PyOpenSSL is not available.")
+    def test_set_check_hostname_false(self):
+        ctx = self._make()
+        ctx.check_hostname = False
+        self.assertFalse(ctx.check_hostname)
+
+    @unittest.skipUnless(_HAVE_PYOPENSSL, "PyOpenSSL is not available.")
+    def test_set_check_hostname_invalid_raises(self):
+        ctx = self._make()
+        with self.assertRaises(TypeError):
+            ctx.check_hostname = "yes"
+
+    @unittest.skipUnless(_HAVE_PYOPENSSL, "PyOpenSSL is not available.")
+    def test_default_check_ocsp_endpoint(self):
+        ctx = self._make()
+        self.assertTrue(ctx.check_ocsp_endpoint)
+
+    @unittest.skipUnless(_HAVE_PYOPENSSL, "PyOpenSSL is not available.")
+    def test_set_check_ocsp_endpoint_false(self):
+        ctx = self._make()
+        ctx.check_ocsp_endpoint = False
+        self.assertFalse(ctx.check_ocsp_endpoint)
+
+    @unittest.skipUnless(_HAVE_PYOPENSSL, "PyOpenSSL is not available.")
+    def test_verify_mode_roundtrip(self):
+        ctx = self._make()
+        ctx.verify_mode = ssl.CERT_REQUIRED
+        self.assertEqual(ctx.verify_mode, ssl.CERT_REQUIRED)
+
+    @unittest.skipUnless(_HAVE_PYOPENSSL, "PyOpenSSL is not available.")
+    def test_verify_mode_cert_none(self):
+        ctx = self._make()
+        ctx.verify_mode = ssl.CERT_NONE
+        self.assertEqual(ctx.verify_mode, ssl.CERT_NONE)
+
+    @unittest.skipUnless(_HAVE_PYOPENSSL, "PyOpenSSL is not available.")
+    def test_options_setter_and_getter(self):
+        ctx = self._make()
+        from pymongo.pyopenssl_context import OP_NO_SSLv3
+
+        ctx.options = OP_NO_SSLv3
+        self.assertTrue(ctx.options & OP_NO_SSLv3)
+
+
+# ---------------------------------------------------------------------------
+# SSLContext._load_certifi
+# ---------------------------------------------------------------------------
+
+
+class TestLoadCertifi(unittest.TestCase):
+    @unittest.skipUnless(_HAVE_PYOPENSSL, "PyOpenSSL is not available.")
+    def test_raises_when_certifi_unavailable(self):
+        from pymongo.errors import ConfigurationError
+
+        ctx = SSLContext(PROTOCOL_SSLv23)
+        with patch.object(_ctx_module, "_HAVE_CERTIFI", False):
+            with self.assertRaises(ConfigurationError) as exc_ctx:
+                ctx._load_certifi()
+        self.assertIn("certifi", str(exc_ctx.exception))
+
+    @unittest.skipUnless(_HAVE_PYOPENSSL, "PyOpenSSL is not available.")
+    def test_loads_when_certifi_available(self):
+        if not _ctx_module._HAVE_CERTIFI:
+            self.skipTest("certifi not installed")
+        ctx = SSLContext(PROTOCOL_SSLv23)
+        ctx.verify_mode = ssl.CERT_NONE
+        # Should not raise.
+        ctx._load_certifi()
+
+
+# ---------------------------------------------------------------------------
+# SSLContext.load_default_certs — platform branching
+# ---------------------------------------------------------------------------
+
+
+class TestLoadDefaultCerts(unittest.TestCase):
+    @unittest.skipUnless(_HAVE_PYOPENSSL, "PyOpenSSL is not available.")
+    def test_darwin_calls_load_certifi(self):
+        with patch.object(_ctx_module._sys, "platform", "darwin"):
+            with patch.object(SSLContext, "_load_certifi") as mock_certifi:
+                with patch("OpenSSL.SSL.Context.set_default_verify_paths"):
+                    ctx = SSLContext(PROTOCOL_SSLv23)
+                    ctx.load_default_certs()
+        mock_certifi.assert_called()
+
+    @unittest.skipUnless(_HAVE_PYOPENSSL, "PyOpenSSL is not available.")
+    def test_win32_calls_load_wincerts(self):
+        with patch.object(_ctx_module._sys, "platform", "win32"):
+            with patch.object(SSLContext, "_load_wincerts") as mock_wincerts:
+                with patch("OpenSSL.SSL.Context.set_default_verify_paths"):
+                    ctx = SSLContext(PROTOCOL_SSLv23)
+                    ctx.load_default_certs()
+        calls = [call.args[0] for call in mock_wincerts.call_args_list]
+        self.assertIn("CA", calls)
+        self.assertIn("ROOT", calls)
+
+    @unittest.skipUnless(_HAVE_PYOPENSSL, "PyOpenSSL is not available.")
+    def test_win32_falls_back_to_certifi_on_exception(self):
+        with patch.object(_ctx_module._sys, "platform", "win32"):
+            with patch.object(SSLContext, "_load_wincerts", side_effect=Exception("no certs")):
+                with patch.object(SSLContext, "_load_certifi") as mock_certifi:
+                    with patch("OpenSSL.SSL.Context.set_default_verify_paths"):
+                        ctx = SSLContext(PROTOCOL_SSLv23)
+                        ctx.load_default_certs()
+        mock_certifi.assert_called()
+
+    @unittest.skipUnless(_HAVE_PYOPENSSL, "PyOpenSSL is not available.")
+    def test_linux_no_certifi_call(self):
+        with patch.object(_ctx_module._sys, "platform", "linux"):
+            with patch.object(SSLContext, "_load_certifi") as mock_certifi:
+                with patch("OpenSSL.SSL.Context.set_default_verify_paths"):
+                    ctx = SSLContext(PROTOCOL_SSLv23)
+                    ctx.load_default_certs()
+        mock_certifi.assert_not_called()
+
+    @unittest.skipUnless(_HAVE_PYOPENSSL, "PyOpenSSL is not available.")
+    def test_calls_set_default_verify_paths(self):
+        with patch.object(_ctx_module._sys, "platform", "linux"):
+            ctx = SSLContext(PROTOCOL_SSLv23)
+            with patch.object(ctx._ctx, "set_default_verify_paths") as mock_sdvp:
+                ctx.load_default_certs()
+        mock_sdvp.assert_called_once()
+
+
+# ---------------------------------------------------------------------------
+# SSLContext.set_default_verify_paths
+# ---------------------------------------------------------------------------
+
+
+class TestSetDefaultVerifyPaths(unittest.TestCase):
+    @unittest.skipUnless(_HAVE_PYOPENSSL, "PyOpenSSL is not available.")
+    def test_delegates_to_ctx(self):
+        ctx = SSLContext(PROTOCOL_SSLv23)
+        with patch.object(ctx._ctx, "set_default_verify_paths") as mock_sdvp:
+            ctx.set_default_verify_paths()
+        mock_sdvp.assert_called_once()
+
+
+# ---------------------------------------------------------------------------
+# SSLContext.load_verify_locations
+# ---------------------------------------------------------------------------
+
+
+class TestLoadVerifyLocations(unittest.TestCase):
+    @unittest.skipUnless(_HAVE_PYOPENSSL, "PyOpenSSL is not available.")
+    def test_delegates_to_ctx(self):
+        ctx = SSLContext(PROTOCOL_SSLv23)
+        with patch.object(ctx._ctx, "load_verify_locations") as mock_lvl:
+            ctx.load_verify_locations(cafile="/tmp/ca.pem")
+        mock_lvl.assert_called_once_with("/tmp/ca.pem", None)
+
+
+if __name__ == "__main__":
+    unittest.main()

test/test_session.py

@@ -15,7 +15,6 @@
 """Test the client_session module."""
 from __future__ import annotations
 
-import asyncio
 import copy
 import sys
 import time
@@ -24,8 +23,6 @@ from io import BytesIO
 from test.helpers import ExceptionCatchingTask
 from typing import Any, Callable, List, Set, Tuple
 
-from pymongo.synchronous.mongo_client import MongoClient
-
 sys.path[0:0] = [""]
 
 from test import (
@@ -45,7 +42,7 @@ from test.utils_shared import (
 
 from bson import DBRef
 from gridfs.synchronous.grid_file import GridFS, GridFSBucket
-from pymongo import ASCENDING, MongoClient, _csot, monitoring
+from pymongo import ASCENDING, MongoClient, monitoring
 from pymongo.common import _MAX_END_SESSIONS
 from pymongo.errors import ConfigurationError, InvalidOperation, OperationFailure
 from pymongo.operations import IndexModel, InsertOne, UpdateOne
@@ -938,6 +935,39 @@ class TestSession(IntegrationTest):
 
         s2.end_session()
 
+    def test_getmore_preserves_lsid_after_session_support_lost(self):
+        listener = OvertCommandListener()
+        client = self.rs_or_single_client(event_listeners=[listener], maxPoolSize=1)
+        coll = client.pymongo_test.test
+        coll.drop()
+        coll.insert_many([{"x": i} for i in range(10)])
+        self.addCleanup(coll.drop)
+
+        with client.start_session() as s:
+            cursor = coll.find({}, batch_size=2, session=s)
+            next(cursor)
+
+            find_event = next(e for e in listener.started_events if e.command_name == "find")
+            lsid = find_event.command["lsid"]
+
+            # Simulate a node stepping down: mark idle connections as not supporting sessions.
+            for server in client._topology._servers.values():
+                for conn in server.pool.conns:
+                    conn.supports_sessions = False
+
+            listener.reset()
+            cursor.to_list()
+
+        getmore_events = [e for e in listener.started_events if e.command_name == "getMore"]
+        self.assertGreater(len(getmore_events), 0, "expected at least one getMore command")
+        for event in getmore_events:
+            self.assertIn(
+                "lsid", event.command, "getMore must include lsid when session is materialized"
+            )
+            self.assertEqual(
+                lsid, event.command["lsid"], "getMore lsid must match the session lsid from find"
+            )
+
 
 class TestCausalConsistency(UnitTest):
     listener: SessionTestListener

test/test_topology.py

@@ -23,7 +23,7 @@ sys.path[0:0] = [""]
 
 from test import client_knobs, unittest
 from test.pymongo_mocks import DummyMonitor
-from test.utils import MockPool, flaky
+from test.utils import MockPool
 from test.utils_shared import wait_until
 
 from bson.objectid import ObjectId
@@ -755,7 +755,6 @@ def wait_for_primary(topology):
 class TestTopologyErrors(TopologyTest):
     # Errors when calling hello.
 
-    @flaky(reason="PYTHON-5366")
     def test_pool_reset(self):
         # hello succeeds at first, then always raises socket error.
         hello_count = [0]
@@ -776,7 +775,11 @@ class TestTopologyErrors(TopologyTest):
 
         # Pool is reset by hello failure.
         t.request_check_all()
-        self.assertNotEqual(generation, server.pool.gen.get_overall())
+        # Wait for the monitor's hello failure to trigger Pool.reset() and bump the generation.
+        wait_until(
+            lambda: server.pool.gen.get_overall() != generation,
+            "pool reset after failed monitor check",
+        )
 
     def test_hello_retry(self):
         # hello succeeds at first, then raises socket error, then succeeds.