PYTHON-5672 Add coverage for _PoolCheckout leak guard; remove dead server.checkout()

server.checkout() was a thin wrapper around pool.checkout() that nothing
called after _ClientCheckout was changed to call server.pool.checkout()
directly.

Add a test that mocks publish_connection_checked_out to raise and verifies
the connection is returned to the pool rather than leaked.

Author: blink1073

pymongo/asynchronous/server.py

@@ -43,7 +43,7 @@
     from bson.objectid import ObjectId
     from pymongo.asynchronous.mongo_client import AsyncMongoClient
     from pymongo.asynchronous.monitor import Monitor
-    from pymongo.asynchronous.pool import AsyncConnection, Pool, _PoolCheckout
+    from pymongo.asynchronous.pool import AsyncConnection, Pool
     from pymongo.monitoring import _EventListeners
     from pymongo.read_preferences import _ServerMode
     from pymongo.server_description import ServerDescription
@@ -226,9 +226,6 @@ async def run_operation(
 
         return response
 
-    def checkout(self) -> _PoolCheckout:
-        return self.pool.checkout()
-
     @property
     def description(self) -> ServerDescription:
         return self._description

pymongo/synchronous/server.py

@@ -46,7 +46,7 @@
     from pymongo.server_description import ServerDescription
     from pymongo.synchronous.mongo_client import MongoClient
     from pymongo.synchronous.monitor import Monitor
-    from pymongo.synchronous.pool import Connection, Pool, _PoolCheckout
+    from pymongo.synchronous.pool import Connection, Pool
     from pymongo.typings import _DocumentOut
 
 _IS_SYNC = True
@@ -226,9 +226,6 @@ def run_operation(
 
         return response
 
-    def checkout(self) -> _PoolCheckout:
-        return self.pool.checkout()
-
     @property
     def description(self) -> ServerDescription:
         return self._description

test/asynchronous/test_pooling.py

@@ -215,6 +215,34 @@ async def test_get_socket_and_exception(self):
 
         self.assertEqual(1, len(cx_pool.conns))
 
+    async def test_checkout_event_listener_failure_no_leak(self):
+        # Connection is returned to the pool when publish_connection_checked_out raises.
+        from unittest.mock import patch
+
+        from pymongo.monitoring import _EventListeners
+        from test.utils_shared import CMAPListener
+
+        cx_pool = await self.create_pool(
+            max_pool_size=1, event_listeners=_EventListeners([CMAPListener()])
+        )
+
+        with patch.object(
+            cx_pool.opts._event_listeners,
+            "publish_connection_checked_out",
+            side_effect=RuntimeError("simulated failure"),
+        ):
+            with self.assertRaises(RuntimeError):
+                async with cx_pool.checkout():
+                    pass
+
+        # Connection was returned to the pool — not leaked.
+        self.assertEqual(1, len(cx_pool.conns))
+        self.assertEqual(0, cx_pool.active_sockets)
+
+        # Pool is still functional.
+        async with cx_pool.checkout():
+            pass
+
     async def test_pool_removes_closed_socket(self):
         # Test that Pool removes explicitly closed socket.
         cx_pool = await self.create_pool()

test/test_pooling.py

@@ -215,6 +215,34 @@ def test_get_socket_and_exception(self):
 
         self.assertEqual(1, len(cx_pool.conns))
 
+    def test_checkout_event_listener_failure_no_leak(self):
+        # Connection is returned to the pool when publish_connection_checked_out raises.
+        from unittest.mock import patch
+
+        from pymongo.monitoring import _EventListeners
+        from test.utils_shared import CMAPListener
+
+        cx_pool = self.create_pool(
+            max_pool_size=1, event_listeners=_EventListeners([CMAPListener()])
+        )
+
+        with patch.object(
+            cx_pool.opts._event_listeners,
+            "publish_connection_checked_out",
+            side_effect=RuntimeError("simulated failure"),
+        ):
+            with self.assertRaises(RuntimeError):
+                with cx_pool.checkout():
+                    pass
+
+        # Connection was returned to the pool — not leaked.
+        self.assertEqual(1, len(cx_pool.conns))
+        self.assertEqual(0, cx_pool.active_sockets)
+
+        # Pool is still functional.
+        with cx_pool.checkout():
+            pass
+
     def test_pool_removes_closed_socket(self):
         # Test that Pool removes explicitly closed socket.
         cx_pool = self.create_pool()