Correctly pause and resume reading

Author: NoahStapp

pymongo/network_layer.py

@@ -193,9 +193,11 @@ async def write(self, message: bytes):
             raise OSError("Connection is closed")
         self.transport.write(message)
         await self._drain_helper()
+        self.transport.resume_reading()
 
     async def read(self, request_id: Optional[int], max_message_size: int, debug: bool = False):
         """Read a single MongoDB Wire Protocol message from this connection."""
+        self.transport.resume_reading()
         if self._done_messages:
             message = await self._done_messages.popleft()
         else:
@@ -297,6 +299,7 @@ def buffer_updated(self, nbytes: int):
                     self._length -= extra
                     self._expecting_header = True
                     self.buffer_updated(extra)
+                self.transport.pause_reading()
 
     def process_header(self):
         """Unpack a MongoDB Wire Protocol header."""

test/asynchronous/test_collection.py

@@ -1792,8 +1792,6 @@ async def test_cursor_timeout(self):
         await self.db.test.find(no_cursor_timeout=True).to_list()
         await self.db.test.find(no_cursor_timeout=False).to_list()
 
-    # TODO: fix exhaust cursor + batch_size
-    @async_client_context.require_sync
     async def test_exhaust(self):
         if await async_is_mongos(self.db.client):
             with self.assertRaises(InvalidOperation):

test/asynchronous/test_monitoring.py

@@ -421,8 +421,6 @@ async def test_not_primary_error(self):
         self.assertTrue(isinstance(failed.duration_micros, int))
         self.assertEqual(error, failed.failure)
 
-    # TODO: fix exhaust cursor + batch_size
-    @async_client_context.require_sync
     @async_client_context.require_no_mongos
     async def test_exhaust(self):
         await self.client.pymongo_test.test.drop()

test/test_collection.py

@@ -1774,8 +1774,6 @@ def test_cursor_timeout(self):
         self.db.test.find(no_cursor_timeout=True).to_list()
         self.db.test.find(no_cursor_timeout=False).to_list()
 
-    # TODO: fix exhaust cursor + batch_size
-    @client_context.require_sync
     def test_exhaust(self):
         if is_mongos(self.db.client):
             with self.assertRaises(InvalidOperation):

test/test_monitoring.py

@@ -419,8 +419,6 @@ def test_not_primary_error(self):
         self.assertTrue(isinstance(failed.duration_micros, int))
         self.assertEqual(error, failed.failure)
 
-    # TODO: fix exhaust cursor + batch_size
-    @client_context.require_sync
     @client_context.require_no_mongos
     def test_exhaust(self):
         self.client.pymongo_test.test.drop()