Merge branch 'master' of github.com:mongodb/mongo-python-driver

Author: blink1073

doc/changelog.rst

@@ -42,13 +42,20 @@ PyMongo 4.9 brings a number of improvements including:
 - Fixed a bug where PyMongo would raise ``InvalidBSON: date value out of range``
   when using :attr:`~bson.codec_options.DatetimeConversion.DATETIME_CLAMP` or
   :attr:`~bson.codec_options.DatetimeConversion.DATETIME_AUTO` with a non-UTC timezone.
+- The default value for ``connect`` in ``MongoClient`` is changed to ``False`` when running on
+  unction-as-a-service (FaaS) like AWS Lambda, Google Cloud Functions, and Microsoft Azure Functions.
+  On some FaaS systems, there is a ``fork()`` operation at function
+  startup.  By delaying the connection to the first operation, we avoid a deadlock.  See
+  `Is PyMongo Fork-Safe`_ for more information.
+
 
 Issues Resolved
 ...............
 
 See the `PyMongo 4.9 release notes in JIRA`_ for the list of resolved issues
 in this release.
 
+.. _Is PyMongo Fork-Safe : https://www.mongodb.com/docs/languages/python/pymongo-driver/current/faq/#is-pymongo-fork-safe-
 .. _PyMongo 4.9 release notes in JIRA: https://jira.mongodb.org/secure/ReleaseNote.jspa?projectId=10004&version=39940
 
 

pymongo/_client_bulk_shared.py

@@ -74,4 +74,6 @@ def _throw_client_bulk_write_exception(
                 "to your connection string."
             )
             raise OperationFailure(errmsg, code, full_result)
+    if isinstance(full_result["error"], BaseException):
+        raise ClientBulkWriteException(full_result, verbose_results) from full_result["error"]
     raise ClientBulkWriteException(full_result, verbose_results)

pymongo/asynchronous/mongo_client.py

@@ -720,6 +720,10 @@ def __init__(
 
         .. versionchanged:: 4.7
             Deprecated parameter ``wTimeoutMS``, use :meth:`~pymongo.timeout`.
+
+        .. versionchanged:: 4.9
+            The default value of ``connect`` is changed to ``False`` when running in a
+            Function-as-a-service environment.
         """
         doc_class = document_class or dict
         self._init_kwargs: dict[str, Any] = {
@@ -803,7 +807,10 @@ def __init__(
         if tz_aware is None:
             tz_aware = opts.get("tz_aware", False)
         if connect is None:
-            connect = opts.get("connect", True)
+            # Default to connect=True unless on a FaaS system, which might use fork.
+            from pymongo.pool_options import _is_faas
+
+            connect = opts.get("connect", not _is_faas())
         keyword_opts["tz_aware"] = tz_aware
         keyword_opts["connect"] = connect
 

pymongo/synchronous/mongo_client.py

@@ -263,7 +263,8 @@ def __init__(
             aware (otherwise they will be naive)
         :param connect: If ``True`` (the default), immediately
             begin connecting to MongoDB in the background. Otherwise connect
-            on the first operation.
+            on the first operation.  The default value is ``False`` when
+            running in a Function-as-a-service environment.
         :param type_registry: instance of
             :class:`~bson.codec_options.TypeRegistry` to enable encoding
             and decoding of custom types.
@@ -719,6 +720,10 @@ def __init__(
 
         .. versionchanged:: 4.7
             Deprecated parameter ``wTimeoutMS``, use :meth:`~pymongo.timeout`.
+
+        .. versionchanged:: 4.9
+            The default value of ``connect`` is changed to ``False`` when running in a
+            Function-as-a-service environment.
         """
         doc_class = document_class or dict
         self._init_kwargs: dict[str, Any] = {
@@ -802,7 +807,10 @@ def __init__(
         if tz_aware is None:
             tz_aware = opts.get("tz_aware", False)
         if connect is None:
-            connect = opts.get("connect", True)
+            # Default to connect=True unless on a FaaS system, which might use fork.
+            from pymongo.pool_options import _is_faas
+
+            connect = opts.get("connect", not _is_faas())
         keyword_opts["tz_aware"] = tz_aware
         keyword_opts["connect"] = connect
 

pyproject.toml

@@ -74,6 +74,7 @@ addopts = ["-ra", "--strict-config", "--strict-markers", "--junitxml=xunit-resul
 testpaths = ["test"]
 log_cli_level = "INFO"
 faulthandler_timeout = 1500
+asyncio_default_fixture_loop_scope = "session"
 xfail_strict = true
 filterwarnings = [
     "error",

requirements/test.txt

@@ -1,2 +1,2 @@
-pytest>=7
-pytest-asyncio
+pytest>=8.2
+pytest-asyncio>=0.24.0

test/asynchronous/conftest.py

@@ -17,6 +17,8 @@ def event_loop_policy():
     # has issues with sharing sockets across loops (https://github.com/python/cpython/issues/122240)
     # We explicitly use a different loop implementation here to prevent that issue
     if sys.platform == "win32":
+        # Needed for Python 3.8.
+        asyncio.set_event_loop_policy(asyncio.WindowsSelectorEventLoopPolicy())
         return asyncio.WindowsSelectorEventLoopPolicy()  # type: ignore[attr-defined]
 
     return asyncio.get_event_loop_policy()

test/asynchronous/test_auth_spec.py

@@ -0,0 +1,108 @@
+# Copyright 2018-present MongoDB, Inc.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+"""Run the auth spec tests."""
+from __future__ import annotations
+
+import glob
+import json
+import os
+import sys
+import warnings
+
+sys.path[0:0] = [""]
+
+from test import unittest
+from test.unified_format import generate_test_classes
+
+from pymongo import AsyncMongoClient
+from pymongo.asynchronous.auth_oidc import OIDCCallback
+
+_IS_SYNC = False
+
+_TEST_PATH = os.path.join(os.path.dirname(os.path.realpath(__file__)), "auth")
+
+
+class TestAuthSpec(unittest.IsolatedAsyncioTestCase):
+    pass
+
+
+class SampleHumanCallback(OIDCCallback):
+    def fetch(self, context):
+        pass
+
+
+def create_test(test_case):
+    def run_test(self):
+        uri = test_case["uri"]
+        valid = test_case["valid"]
+        credential = test_case.get("credential")
+
+        if not valid:
+            with warnings.catch_warnings():
+                warnings.simplefilter("default")
+                self.assertRaises(Exception, AsyncMongoClient, uri, connect=False)
+        else:
+            client = AsyncMongoClient(uri, connect=False)
+            credentials = client.options.pool_options._credentials
+            if credential is None:
+                self.assertIsNone(credentials)
+            else:
+                self.assertIsNotNone(credentials)
+                self.assertEqual(credentials.username, credential["username"])
+                self.assertEqual(credentials.password, credential["password"])
+                self.assertEqual(credentials.source, credential["source"])
+                if credential["mechanism"] is not None:
+                    self.assertEqual(credentials.mechanism, credential["mechanism"])
+                else:
+                    self.assertEqual(credentials.mechanism, "DEFAULT")
+                expected = credential["mechanism_properties"]
+                if expected is not None:
+                    actual = credentials.mechanism_properties
+                    for key, value in expected.items():
+                        self.assertEqual(getattr(actual, key.lower()), value)
+                else:
+                    if credential["mechanism"] == "MONGODB-AWS":
+                        self.assertIsNone(credentials.mechanism_properties.aws_session_token)
+                    else:
+                        self.assertIsNone(credentials.mechanism_properties)
+
+    return run_test
+
+
+def create_tests():
+    for filename in glob.glob(os.path.join(_TEST_PATH, "legacy", "*.json")):
+        test_suffix, _ = os.path.splitext(os.path.basename(filename))
+        with open(filename) as auth_tests:
+            test_cases = json.load(auth_tests)["tests"]
+            for test_case in test_cases:
+                if test_case.get("optional", False):
+                    continue
+                test_method = create_test(test_case)
+                name = str(test_case["description"].lower().replace(" ", "_"))
+                setattr(TestAuthSpec, f"test_{test_suffix}_{name}", test_method)
+
+
+create_tests()
+
+
+globals().update(
+    generate_test_classes(
+        os.path.join(_TEST_PATH, "unified"),
+        module=__name__,
+    )
+)
+
+if __name__ == "__main__":
+    unittest.main()

test/asynchronous/test_bulk.py

@@ -976,6 +976,7 @@ async def cause_wtimeout(self, requests, ordered):
         finally:
             await self.secondary.admin.command("configureFailPoint", "rsSyncApplyStop", mode="off")
 
+    @async_client_context.require_version_max(7, 1)  # PYTHON-4560
     @async_client_context.require_replica_set
     @async_client_context.require_secondaries_count(1)
     async def test_write_concern_failure_ordered(self):
@@ -1055,6 +1056,7 @@ async def test_write_concern_failure_ordered(self):
         failed = details["writeErrors"][0]
         self.assertTrue("duplicate" in failed["errmsg"])
 
+    @async_client_context.require_version_max(7, 1)  # PYTHON-4560
     @async_client_context.require_replica_set
     @async_client_context.require_secondaries_count(1)
     async def test_write_concern_failure_unordered(self):

test/asynchronous/test_transactions.py

@@ -15,16 +15,14 @@
 """Execute Transactions Spec tests."""
 from __future__ import annotations
 
-import os
 import sys
 from io import BytesIO
 
 from gridfs.asynchronous.grid_file import AsyncGridFS, AsyncGridFSBucket
 
 sys.path[0:0] = [""]
 
-from test.asynchronous import async_client_context, unittest
-from test.asynchronous.utils_spec_runner import AsyncSpecRunner
+from test.asynchronous import AsyncIntegrationTest, async_client_context, unittest
 from test.utils import (
     OvertCommandListener,
     async_rs_client,
@@ -54,40 +52,14 @@
 
 _IS_SYNC = False
 
-_TXN_TESTS_DEBUG = os.environ.get("TRANSACTION_TESTS_DEBUG")
-
 # Max number of operations to perform after a transaction to prove unpinning
 # occurs. Chosen so that there's a low false positive rate. With 2 mongoses,
 # 50 attempts yields a one in a quadrillion chance of a false positive
 # (1/(0.5^50)).
 UNPIN_TEST_MAX_ATTEMPTS = 50
 
 
-class AsyncTransactionsBase(AsyncSpecRunner):
-    @classmethod
-    async def _setup_class(cls):
-        await super()._setup_class()
-        if async_client_context.supports_transactions():
-            for address in async_client_context.mongoses:
-                cls.mongos_clients.append(await async_single_client("{}:{}".format(*address)))
-
-    @classmethod
-    async def _tearDown_class(cls):
-        for client in cls.mongos_clients:
-            await client.close()
-        await super()._tearDown_class()
-
-    def maybe_skip_scenario(self, test):
-        super().maybe_skip_scenario(test)
-        if (
-            "secondary" in self.id()
-            and not async_client_context.is_mongos
-            and not async_client_context.has_secondaries
-        ):
-            raise unittest.SkipTest("No secondaries")
-
-
-class TestTransactions(AsyncTransactionsBase):
+class TestTransactions(AsyncIntegrationTest):
     RUN_ON_SERVERLESS = True
 
     @async_client_context.require_transactions
@@ -421,7 +393,31 @@ def __exit__(self, exc_type, exc_val, exc_tb):
         client_session._WITH_TRANSACTION_RETRY_TIME_LIMIT = self.real_timeout
 
 
-class TestTransactionsConvenientAPI(AsyncTransactionsBase):
+class TestTransactionsConvenientAPI(AsyncIntegrationTest):
+    @classmethod
+    async def _setup_class(cls):
+        await super()._setup_class()
+        cls.mongos_clients = []
+        if async_client_context.supports_transactions():
+            for address in async_client_context.mongoses:
+                cls.mongos_clients.append(await async_single_client("{}:{}".format(*address)))
+
+    @classmethod
+    async def _tearDown_class(cls):
+        for client in cls.mongos_clients:
+            await client.close()
+        await super()._tearDown_class()
+
+    async def _set_fail_point(self, client, command_args):
+        cmd = {"configureFailPoint": "failCommand"}
+        cmd.update(command_args)
+        await client.admin.command(cmd)
+
+    async def set_fail_point(self, command_args):
+        clients = self.mongos_clients if self.mongos_clients else [self.client]
+        for client in clients:
+            await self._set_fail_point(client, command_args)
+
     @async_client_context.require_transactions
     async def test_callback_raises_custom_error(self):
         class _MyException(Exception):