Make pytest import lazy in test/__init__.py

The integration tests import from test/__init__.py but don't have
pytest installed. Wrap the pytest import and skip_if_rust_bson
marker definition in a try/except to gracefully handle when pytest
is not available.

Author: aclark4life

test/__init__.py

@@ -32,19 +32,17 @@
 import warnings
 from inspect import iscoroutinefunction
 
-import pytest
-
+from pymongo.synchronous.uri_parser import parse_uri
 from pymongo.encryption_options import _HAVE_PYMONGOCRYPT
 from pymongo.errors import AutoReconnect
-from pymongo.synchronous.uri_parser import parse_uri
 
 try:
     import ipaddress
 
     HAVE_IPADDRESS = True
 except ImportError:
     HAVE_IPADDRESS = False
-from contextlib import contextmanager
+from contextlib import contextmanager, contextmanager
 from functools import partial, wraps
 from typing import Any, Callable, Dict, Generator, overload
 from unittest import SkipTest
@@ -53,12 +51,12 @@
 import pymongo
 import pymongo.errors
 from bson.son import SON
+from pymongo.synchronous.database import Database
+from pymongo.synchronous.mongo_client import MongoClient
 from pymongo.common import partition_node
 from pymongo.hello import HelloCompat
 from pymongo.server_api import ServerApi
 from pymongo.ssl_support import HAVE_SSL, _ssl  # type:ignore[attr-defined]
-from pymongo.synchronous.database import Database
-from pymongo.synchronous.mongo_client import MongoClient
 
 sys.path[0:0] = [""]
 
@@ -86,13 +84,20 @@
 
 _IS_SYNC = True
 
-import bson
-
 # Skip tests when using Rust BSON extension for features not yet implemented
-skip_if_rust_bson = pytest.mark.skipif(
-    bson.get_bson_implementation() == "rust",
-    reason="Feature not yet implemented in Rust BSON extension",
-)
+# Import pytest lazily to avoid requiring it for integration tests
+try:
+    import bson
+    import pytest
+
+    skip_if_rust_bson = pytest.mark.skipif(
+        bson.get_bson_implementation() == "rust",
+        reason="Feature not yet implemented in Rust BSON extension",
+    )
+except ImportError:
+    # pytest not available, define a no-op decorator
+    def skip_if_rust_bson(func):
+        return func
 
 
 def _connection_string(h):
@@ -1016,7 +1021,9 @@ def _unmanaged_async_mongo_client(
             client._connect()
         return client
 
-    def _async_mongo_client(self, host, port, authenticate=True, directConnection=None, **kwargs):
+    def _async_mongo_client(
+        self, host, port, authenticate=True, directConnection=None, **kwargs
+    ):
         """Create a new client over SSL/TLS if necessary."""
         host = host or client_context.host
         port = port or client_context.port
@@ -1063,7 +1070,9 @@ def unmanaged_single_client(
         return cls._unmanaged_async_mongo_client(h, p, directConnection=True, **kwargs)
 
     @classmethod
-    def unmanaged_rs_client(cls, h: Any = None, p: Any = None, **kwargs: Any) -> MongoClient[dict]:
+    def unmanaged_rs_client(
+        cls, h: Any = None, p: Any = None, **kwargs: Any
+    ) -> MongoClient[dict]:
         """Connect to the replica set and authenticate if necessary."""
         return cls._unmanaged_async_mongo_client(h, p, **kwargs)
 
@@ -1092,17 +1101,25 @@ def single_client_noauth(
         self, h: Any = None, p: Any = None, **kwargs: Any
     ) -> MongoClient[dict]:
         """Make a direct connection. Don't authenticate."""
-        return self._async_mongo_client(h, p, authenticate=False, directConnection=True, **kwargs)
+        return self._async_mongo_client(
+            h, p, authenticate=False, directConnection=True, **kwargs
+        )
 
-    def single_client(self, h: Any = None, p: Any = None, **kwargs: Any) -> MongoClient[dict]:
+    def single_client(
+        self, h: Any = None, p: Any = None, **kwargs: Any
+    ) -> MongoClient[dict]:
         """Make a direct connection, and authenticate if necessary."""
         return self._async_mongo_client(h, p, directConnection=True, **kwargs)
 
-    def rs_client_noauth(self, h: Any = None, p: Any = None, **kwargs: Any) -> MongoClient[dict]:
+    def rs_client_noauth(
+        self, h: Any = None, p: Any = None, **kwargs: Any
+    ) -> MongoClient[dict]:
         """Connect to the replica set. Don't authenticate."""
         return self._async_mongo_client(h, p, authenticate=False, **kwargs)
 
-    def rs_client(self, h: Any = None, p: Any = None, **kwargs: Any) -> MongoClient[dict]:
+    def rs_client(
+        self, h: Any = None, p: Any = None, **kwargs: Any
+    ) -> MongoClient[dict]:
         """Connect to the replica set and authenticate if necessary."""
         return self._async_mongo_client(h, p, **kwargs)
 
@@ -1115,7 +1132,9 @@ def rs_or_single_client_noauth(
         """
         return self._async_mongo_client(h, p, authenticate=False, **kwargs)
 
-    def rs_or_single_client(self, h: Any = None, p: Any = None, **kwargs: Any) -> MongoClient[Any]:
+    def rs_or_single_client(
+        self, h: Any = None, p: Any = None, **kwargs: Any
+    ) -> MongoClient[Any]:
         """Connect to the replica set if there is one, otherwise the standalone.
 
         Authenticates if necessary.
@@ -1131,7 +1150,9 @@ def simple_client(self, h: Any = None, p: Any = None, **kwargs: Any) -> MongoCli
         return client
 
     @classmethod
-    def unmanaged_simple_client(cls, h: Any = None, p: Any = None, **kwargs: Any) -> MongoClient:
+    def unmanaged_simple_client(
+        cls, h: Any = None, p: Any = None, **kwargs: Any
+    ) -> MongoClient:
         if not h and not p:
             client = MongoClient(**kwargs)
         else:

test/asynchronous/__init__.py

@@ -32,8 +32,6 @@
 import warnings
 from inspect import iscoroutinefunction
 
-import pytest
-
 from pymongo.asynchronous.uri_parser import parse_uri
 from pymongo.encryption_options import _HAVE_PYMONGOCRYPT
 from pymongo.errors import AutoReconnect
@@ -86,13 +84,20 @@
 
 _IS_SYNC = False
 
-import bson
-
 # Skip tests when using Rust BSON extension for features not yet implemented
-skip_if_rust_bson = pytest.mark.skipif(
-    bson.get_bson_implementation() == "rust",
-    reason="Feature not yet implemented in Rust BSON extension",
-)
+# Import pytest lazily to avoid requiring it for integration tests
+try:
+    import bson
+    import pytest
+
+    skip_if_rust_bson = pytest.mark.skipif(
+        bson.get_bson_implementation() == "rust",
+        reason="Feature not yet implemented in Rust BSON extension",
+    )
+except ImportError:
+    # pytest not available, define a no-op decorator
+    def skip_if_rust_bson(func):
+        return func
 
 
 def _connection_string(h):