Revert all cleanup changes

Author: NoahStapp

pymongo/asynchronous/mongo_client.py

@@ -1559,8 +1559,6 @@ async def close(self) -> None:
         # Stop the periodic task thread and then send pending killCursor
         # requests before closing the topology.
         self._kill_cursors_executor.close()
-        if not _IS_SYNC:
-            await self._kill_cursors_executor.join()
         await self._process_kill_cursors()
         await self._topology.close()
         if self._encrypter:

pymongo/asynchronous/monitor.py

@@ -191,8 +191,6 @@ def gc_safe_close(self) -> None:
 
     async def close(self) -> None:
         self.gc_safe_close()
-        if not _IS_SYNC:
-            await self._executor.join()
         await self._rtt_monitor.close()
         # Increment the generation and maybe close the socket. If the executor
         # thread has the socket checked out, it will be closed when checked in.
@@ -460,8 +458,6 @@ def __init__(self, topology: Topology, topology_settings: TopologySettings, pool
 
     async def close(self) -> None:
         self.gc_safe_close()
-        if not _IS_SYNC:
-            await self._executor.join()
         # Increment the generation and maybe close the socket. If the executor
         # thread has the socket checked out, it will be closed when checked in.
         await self._pool.reset()

pymongo/synchronous/mongo_client.py

@@ -1553,8 +1553,6 @@ def close(self) -> None:
         # Stop the periodic task thread and then send pending killCursor
         # requests before closing the topology.
         self._kill_cursors_executor.close()
-        if not _IS_SYNC:
-            self._kill_cursors_executor.join()
         self._process_kill_cursors()
         self._topology.close()
         if self._encrypter:

pymongo/synchronous/monitor.py

@@ -191,8 +191,6 @@ def gc_safe_close(self) -> None:
 
     def close(self) -> None:
         self.gc_safe_close()
-        if not _IS_SYNC:
-            self._executor.join()
         self._rtt_monitor.close()
         # Increment the generation and maybe close the socket. If the executor
         # thread has the socket checked out, it will be closed when checked in.
@@ -460,8 +458,6 @@ def __init__(self, topology: Topology, topology_settings: TopologySettings, pool
 
     def close(self) -> None:
         self.gc_safe_close()
-        if not _IS_SYNC:
-            self._executor.join()
         # Increment the generation and maybe close the socket. If the executor
         # thread has the socket checked out, it will be closed when checked in.
         self._pool.reset()

test/__init__.py

@@ -864,6 +864,16 @@ def max_message_size_bytes(self):
 client_context = ClientContext()
 
 
+def reset_client_context():
+    if _IS_SYNC:
+        # sync tests don't need to reset a client context
+        return
+    elif client_context.client is not None:
+        client_context.client.close()
+        client_context.client = None
+    client_context._init_client()
+
+
 class PyMongoTestCase(unittest.TestCase):
     def assertEqualCommand(self, expected, actual, msg=None):
         self.assertEqual(sanitize_cmd(expected), sanitize_cmd(actual), msg)
@@ -1126,6 +1136,8 @@ class IntegrationTest(PyMongoTestCase):
 
     @client_context.require_connection
     def setUp(self) -> None:
+        if not _IS_SYNC:
+            reset_client_context()
         if client_context.load_balancer and not getattr(self, "RUN_ON_LOAD_BALANCER", False):
             raise SkipTest("this test does not support load balancers")
         if client_context.serverless and not getattr(self, "RUN_ON_SERVERLESS", False):

test/asynchronous/__init__.py

@@ -866,6 +866,16 @@ async def max_message_size_bytes(self):
 async_client_context = AsyncClientContext()
 
 
+async def reset_client_context():
+    if _IS_SYNC:
+        # sync tests don't need to reset a client context
+        return
+    elif async_client_context.client is not None:
+        await async_client_context.client.close()
+        async_client_context.client = None
+    await async_client_context._init_client()
+
+
 class AsyncPyMongoTestCase(unittest.IsolatedAsyncioTestCase):
     def assertEqualCommand(self, expected, actual, msg=None):
         self.assertEqual(sanitize_cmd(expected), sanitize_cmd(actual), msg)
@@ -1144,6 +1154,8 @@ class AsyncIntegrationTest(AsyncPyMongoTestCase):
 
     @async_client_context.require_connection
     async def asyncSetUp(self) -> None:
+        if not _IS_SYNC:
+            await reset_client_context()
         if async_client_context.load_balancer and not getattr(self, "RUN_ON_LOAD_BALANCER", False):
             raise SkipTest("this test does not support load balancers")
         if async_client_context.serverless and not getattr(self, "RUN_ON_SERVERLESS", False):

test/asynchronous/utils_spec_runner.py

@@ -647,11 +647,10 @@ async def setup_scenario(self, scenario_def):
     async def run_scenario(self, scenario_def, test):
         self.maybe_skip_scenario(test)
 
-        # Kill all sessions before (sync only) and after each test to prevent an open
+        # Kill all sessions before and after each test to prevent an open
         # transaction (from a test failure) from blocking collection/database
         # operations during test set up and tear down.
-        if _IS_SYNC:
-            await self.kill_all_sessions()
+        await self.kill_all_sessions()
         self.addAsyncCleanup(self.kill_all_sessions)
         await self.setup_scenario(scenario_def)
         database_name = self.get_scenario_db_name(scenario_def)

test/utils_spec_runner.py

@@ -647,11 +647,10 @@ def setup_scenario(self, scenario_def):
     def run_scenario(self, scenario_def, test):
         self.maybe_skip_scenario(test)
 
-        # Kill all sessions before (sync only) and after each test to prevent an open
+        # Kill all sessions before and after each test to prevent an open
         # transaction (from a test failure) from blocking collection/database
         # operations during test set up and tear down.
-        if _IS_SYNC:
-            self.kill_all_sessions()
+        self.kill_all_sessions()
         self.addCleanup(self.kill_all_sessions)
         self.setup_scenario(scenario_def)
         database_name = self.get_scenario_db_name(scenario_def)