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

Author: blink1073

doc/changelog.rst

@@ -24,6 +24,7 @@ PyMongo 4.12 brings a number of changes including:
   :class:`~pymongo.read_preferences.SecondaryPreferred`,
   :class:`~pymongo.read_preferences.Nearest`. Support for ``hedge`` will be removed in PyMongo 5.0.
 - Removed PyOpenSSL support from the asynchronous API due to limitations of the CPython asyncio.Protocol SSL implementation.
+- Allow valid SRV hostnames with less than 3 parts.
 
 Issues Resolved
 ...............

pymongo/asynchronous/mongo_client.py

@@ -878,6 +878,7 @@ def __init__(
 
         self._opened = False
         self._closed = False
+        self._loop: Optional[asyncio.AbstractEventLoop] = None
         if not is_srv:
             self._init_background()
 
@@ -1709,6 +1710,13 @@ async def _get_topology(self) -> Topology:
         If this client was created with "connect=False", calling _get_topology
         launches the connection process in the background.
         """
+        if not _IS_SYNC:
+            if self._loop is None:
+                self._loop = asyncio.get_running_loop()
+            elif self._loop != asyncio.get_running_loop():
+                raise RuntimeError(
+                    "Cannot use AsyncMongoClient in different event loop. AsyncMongoClient uses low-level asyncio APIs that bind it to the event loop it was created on."
+                )
         if not self._opened:
             if self._resolve_srv_info["is_srv"]:
                 await self._resolve_srv()
@@ -2840,7 +2848,7 @@ async def _write(self) -> T:
                     _debug_log(
                         _COMMAND_LOGGER,
                         message=f"Retrying write attempt number {self._attempt_number}",
-                        clientId=self._client.client_id,
+                        clientId=self._client._topology_settings._topology_id,
                         commandName=self._operation,
                         operationId=self._operation_id,
                     )

pymongo/asynchronous/pool.py

@@ -931,13 +931,15 @@ async def remove_stale_sockets(self, reference_generation: int) -> None:
                 return
 
         if self.opts.max_idle_time_seconds is not None:
+            close_conns = []
             async with self.lock:
                 while (
                     self.conns
                     and self.conns[-1].idle_time_seconds() > self.opts.max_idle_time_seconds
                 ):
-                    conn = self.conns.pop()
-                    await conn.close_conn(ConnectionClosedReason.IDLE)
+                    close_conns.append(self.conns.pop())
+            for conn in close_conns:
+                await conn.close_conn(ConnectionClosedReason.IDLE)
 
         while True:
             async with self.size_cond:
@@ -957,14 +959,18 @@ async def remove_stale_sockets(self, reference_generation: int) -> None:
                     self._pending += 1
                     incremented = True
                 conn = await self.connect()
+                close_conn = False
                 async with self.lock:
                     # Close connection and return if the pool was reset during
                     # socket creation or while acquiring the pool lock.
                     if self.gen.get_overall() != reference_generation:
-                        await conn.close_conn(ConnectionClosedReason.STALE)
-                        return
-                    self.conns.appendleft(conn)
-                    self.active_contexts.discard(conn.cancel_context)
+                        close_conn = True
+                    if not close_conn:
+                        self.conns.appendleft(conn)
+                        self.active_contexts.discard(conn.cancel_context)
+                if close_conn:
+                    await conn.close_conn(ConnectionClosedReason.STALE)
+                    return
             finally:
                 if incremented:
                     # Notify after adding the socket to the pool.
@@ -1343,17 +1349,20 @@ async def checkin(self, conn: AsyncConnection) -> None:
                         error=ConnectionClosedReason.ERROR,
                     )
             else:
+                close_conn = False
                 async with self.lock:
                     # Hold the lock to ensure this section does not race with
                     # Pool.reset().
                     if self.stale_generation(conn.generation, conn.service_id):
-                        await conn.close_conn(ConnectionClosedReason.STALE)
+                        close_conn = True
                     else:
                         conn.update_last_checkin_time()
                         conn.update_is_writable(bool(self.is_writable))
                         self.conns.appendleft(conn)
                         # Notify any threads waiting to create a connection.
                         self._max_connecting_cond.notify()
+                if close_conn:
+                    await conn.close_conn(ConnectionClosedReason.STALE)
 
         async with self.size_cond:
             if txn:

pymongo/asynchronous/srv_resolver.py

@@ -90,14 +90,12 @@ def __init__(
             raise ConfigurationError(_INVALID_HOST_MSG % ("an IP address",))
         except ValueError:
             pass
-
         try:
-            self.__plist = self.__fqdn.split(".")[1:]
+            split_fqdn = self.__fqdn.split(".")
+            self.__plist = split_fqdn[1:] if len(split_fqdn) > 2 else split_fqdn
         except Exception:
             raise ConfigurationError(_INVALID_HOST_MSG % (fqdn,)) from None
         self.__slen = len(self.__plist)
-        if self.__slen < 2:
-            raise ConfigurationError(_INVALID_HOST_MSG % (fqdn,))
 
     async def get_options(self) -> Optional[str]:
         from dns import resolver
@@ -139,6 +137,10 @@ async def _get_srv_response_and_hosts(
 
         # Validate hosts
         for node in nodes:
+            if self.__fqdn == node[0].lower():
+                raise ConfigurationError(
+                    "Invalid SRV host: return address is identical to SRV hostname"
+                )
             try:
                 nlist = node[0].lower().split(".")[1:][-self.__slen :]
             except Exception:

pymongo/asynchronous/topology.py

@@ -244,8 +244,6 @@ async def open(self) -> None:
                 # Close servers and clear the pools.
                 for server in self._servers.values():
                     await server.close()
-                    if not _IS_SYNC:
-                        self._monitor_tasks.append(server._monitor)
                 # Reset the session pool to avoid duplicate sessions in
                 # the child process.
                 self._session_pool.reset()

pymongo/synchronous/mongo_client.py

@@ -876,6 +876,7 @@ def __init__(
 
         self._opened = False
         self._closed = False
+        self._loop: Optional[asyncio.AbstractEventLoop] = None
         if not is_srv:
             self._init_background()
 
@@ -1703,6 +1704,13 @@ def _get_topology(self) -> Topology:
         If this client was created with "connect=False", calling _get_topology
         launches the connection process in the background.
         """
+        if not _IS_SYNC:
+            if self._loop is None:
+                self._loop = asyncio.get_running_loop()
+            elif self._loop != asyncio.get_running_loop():
+                raise RuntimeError(
+                    "Cannot use MongoClient in different event loop. MongoClient uses low-level asyncio APIs that bind it to the event loop it was created on."
+                )
         if not self._opened:
             if self._resolve_srv_info["is_srv"]:
                 self._resolve_srv()
@@ -2826,7 +2834,7 @@ def _write(self) -> T:
                     _debug_log(
                         _COMMAND_LOGGER,
                         message=f"Retrying write attempt number {self._attempt_number}",
-                        clientId=self._client.client_id,
+                        clientId=self._client._topology_settings._topology_id,
                         commandName=self._operation,
                         operationId=self._operation_id,
                     )

pymongo/synchronous/pool.py

@@ -927,13 +927,15 @@ def remove_stale_sockets(self, reference_generation: int) -> None:
                 return
 
         if self.opts.max_idle_time_seconds is not None:
+            close_conns = []
             with self.lock:
                 while (
                     self.conns
                     and self.conns[-1].idle_time_seconds() > self.opts.max_idle_time_seconds
                 ):
-                    conn = self.conns.pop()
-                    conn.close_conn(ConnectionClosedReason.IDLE)
+                    close_conns.append(self.conns.pop())
+            for conn in close_conns:
+                conn.close_conn(ConnectionClosedReason.IDLE)
 
         while True:
             with self.size_cond:
@@ -953,14 +955,18 @@ def remove_stale_sockets(self, reference_generation: int) -> None:
                     self._pending += 1
                     incremented = True
                 conn = self.connect()
+                close_conn = False
                 with self.lock:
                     # Close connection and return if the pool was reset during
                     # socket creation or while acquiring the pool lock.
                     if self.gen.get_overall() != reference_generation:
-                        conn.close_conn(ConnectionClosedReason.STALE)
-                        return
-                    self.conns.appendleft(conn)
-                    self.active_contexts.discard(conn.cancel_context)
+                        close_conn = True
+                    if not close_conn:
+                        self.conns.appendleft(conn)
+                        self.active_contexts.discard(conn.cancel_context)
+                if close_conn:
+                    conn.close_conn(ConnectionClosedReason.STALE)
+                    return
             finally:
                 if incremented:
                     # Notify after adding the socket to the pool.
@@ -1339,17 +1345,20 @@ def checkin(self, conn: Connection) -> None:
                         error=ConnectionClosedReason.ERROR,
                     )
             else:
+                close_conn = False
                 with self.lock:
                     # Hold the lock to ensure this section does not race with
                     # Pool.reset().
                     if self.stale_generation(conn.generation, conn.service_id):
-                        conn.close_conn(ConnectionClosedReason.STALE)
+                        close_conn = True
                     else:
                         conn.update_last_checkin_time()
                         conn.update_is_writable(bool(self.is_writable))
                         self.conns.appendleft(conn)
                         # Notify any threads waiting to create a connection.
                         self._max_connecting_cond.notify()
+                if close_conn:
+                    conn.close_conn(ConnectionClosedReason.STALE)
 
         with self.size_cond:
             if txn:

pymongo/synchronous/srv_resolver.py

@@ -90,14 +90,12 @@ def __init__(
             raise ConfigurationError(_INVALID_HOST_MSG % ("an IP address",))
         except ValueError:
             pass
-
         try:
-            self.__plist = self.__fqdn.split(".")[1:]
+            split_fqdn = self.__fqdn.split(".")
+            self.__plist = split_fqdn[1:] if len(split_fqdn) > 2 else split_fqdn
         except Exception:
             raise ConfigurationError(_INVALID_HOST_MSG % (fqdn,)) from None
         self.__slen = len(self.__plist)
-        if self.__slen < 2:
-            raise ConfigurationError(_INVALID_HOST_MSG % (fqdn,))
 
     def get_options(self) -> Optional[str]:
         from dns import resolver
@@ -139,6 +137,10 @@ def _get_srv_response_and_hosts(
 
         # Validate hosts
         for node in nodes:
+            if self.__fqdn == node[0].lower():
+                raise ConfigurationError(
+                    "Invalid SRV host: return address is identical to SRV hostname"
+                )
             try:
                 nlist = node[0].lower().split(".")[1:][-self.__slen :]
             except Exception:

pymongo/synchronous/topology.py

@@ -244,8 +244,6 @@ def open(self) -> None:
                 # Close servers and clear the pools.
                 for server in self._servers.values():
                     server.close()
-                    if not _IS_SYNC:
-                        self._monitor_tasks.append(server._monitor)
                 # Reset the session pool to avoid duplicate sessions in
                 # the child process.
                 self._session_pool.reset()

pyproject.toml

@@ -117,6 +117,8 @@ filterwarnings = [
     "module:unclosed <ssl.SSLSocket:ResourceWarning",
     "module:unclosed <socket object:ResourceWarning",
     "module:unclosed transport:ResourceWarning",
+    # pytest-asyncio known issue: https://github.com/pytest-dev/pytest-asyncio/issues/724
+    "module:unclosed event loop:ResourceWarning",
     # https://github.com/eventlet/eventlet/issues/818
     "module:please use dns.resolver.Resolver.resolve:DeprecationWarning",
     # https://github.com/dateutil/dateutil/issues/1314