Address review

Author: NoahStapp

bson/__init__.py

@@ -1109,9 +1109,7 @@ def _decode_all(data: _ReadableBuffer, opts: CodecOptions[_DocumentType]) -> lis
         while position < end:
             obj_size = _UNPACK_INT_FROM(data, position)[0]
             if data_len - position < obj_size:
-                raise InvalidBSON(
-                    f"invalid object size: expected {obj_size}, got {data_len - position}"
-                )
+                raise InvalidBSON("invalid object size")
             obj_end = position + obj_size - 1
             if data[obj_end] != 0:
                 raise InvalidBSON("bad eoo")

pymongo/message.py

@@ -1535,9 +1535,7 @@ def unpack(cls, msg: bytes) -> _OpMsg:
             raise ProtocolError(f"Unsupported OP_MSG payload type: 0x{first_payload_type:x}")
 
         if len(msg) != first_payload_size + 5:
-            raise ProtocolError(
-                f"Unsupported OP_MSG reply: >1 section, {len(msg)} vs {first_payload_size + 5}"
-            )
+            raise ProtocolError("Unsupported OP_MSG reply: >1 section")
 
         payload_document = msg[5:]
         return cls(flags, payload_document)

pymongo/network_layer.py

@@ -419,7 +419,7 @@ def settimeout(self, timeout: float | None) -> None:
         self.conn[1].settimeout(timeout)
 
     async def close(self) -> None:
-        self.conn[0].abort()
+        self.conn[1].close()
         await self.conn[1].wait_closed()
 
     def is_closing(self) -> bool:
@@ -656,7 +656,7 @@ def process_compression_header(self) -> tuple[int, int]:
         op_code, _, compressor_id = _UNPACK_COMPRESSION_HEADER(self._compression_header)
         return op_code, compressor_id
 
-    def _resolve_pending_messages(self, exc: Exception | None) -> None:
+    def _resolve_pending_messages(self, exc: Optional[Exception] = None) -> None:
         pending = list(self._pending_messages)
         for msg in pending:
             if not msg.done():
@@ -666,12 +666,12 @@ def _resolve_pending_messages(self, exc: Exception | None) -> None:
                     msg.set_exception(exc)
             self._done_messages.append(msg)
 
-    def close(self, exc: Exception | None) -> None:
-        self._connection_lost = True
+    def close(self, exc: Optional[Exception] = None) -> None:
+        self.transport.abort()
         self._resolve_pending_messages(exc)
-        self.transport.close()
+        self._connection_lost = True
 
-    def connection_lost(self, exc: Exception | None) -> None:
+    def connection_lost(self, exc: Optional[Exception] = None) -> None:
         self._resolve_pending_messages(exc)
         if not self._closed.done():
             self._closed.set_result(None)

pyproject.toml

@@ -116,12 +116,7 @@ filterwarnings = [
     "module:unclosed <socket.socket:ResourceWarning",
     "module:unclosed <ssl.SSLSocket:ResourceWarning",
     "module:unclosed <socket object:ResourceWarning",
-    # TODO: Remove the next five as part of PYTHON-5036.
     "module:unclosed transport:ResourceWarning",
-    "module:unclosed <_SelectorSocketTransport:ResourceWarning",
-    "module:Unclosed AsyncMongoClient:ResourceWarning",
-    "module:coroutine 'AsyncMongoClient.close' was never awaited:RuntimeWarning",
-    "module:coroutine 'UnifiedSpecTestMixinV1.kill_all_sessions' was never awaited:RuntimeWarning",
     # https://github.com/eventlet/eventlet/issues/818
     "module:please use dns.resolver.Resolver.resolve:DeprecationWarning",
     # https://github.com/dateutil/dateutil/issues/1314

test/asynchronous/test_client_bulk_write.py

@@ -676,8 +676,14 @@ async def test_timeout_in_multi_batch_bulk_write(self):
             listener = OvertCommandListener()
             client = await self.async_rs_or_single_client(
                 event_listeners=[listener],
+                readConcernLevel="majority",
+                readPreference="primary",
                 timeoutMS=2000,
+                w="majority",
             )
+            # Initialize the client with a larger timeout to help make test less flakey
+            with pymongo.timeout(10):
+                await client.admin.command("ping")
             with self.assertRaises(ClientBulkWriteException) as context:
                 await client.bulk_write(models=models)
             self.assertIsInstance(context.exception.error, NetworkTimeout)

test/test_client_bulk_write.py

@@ -672,8 +672,14 @@ def test_timeout_in_multi_batch_bulk_write(self):
             listener = OvertCommandListener()
             client = self.rs_or_single_client(
                 event_listeners=[listener],
+                readConcernLevel="majority",
+                readPreference="primary",
                 timeoutMS=2000,
+                w="majority",
             )
+            # Initialize the client with a larger timeout to help make test less flakey
+            with pymongo.timeout(10):
+                client.admin.command("ping")
             with self.assertRaises(ClientBulkWriteException) as context:
                 client.bulk_write(models=models)
             self.assertIsInstance(context.exception.error, NetworkTimeout)