PYTHON-5781 Drop trivial tests, add sync-only receive_message/receive_data coverage

Remove the NetworkingInterfaceBase NotImplementedError tests and the
PyMongoProtocol timeout getter/setter tests (implementation, not behavior),
along with the now-empty synchro'd test_network_layer.py pair and its
synchro.py registration (including the unused AsyncMock->MagicMock rule).

Add a hand-maintained sync-only test/test_network_layer.py covering
receive_message and receive_data, mirroring the async-only protocol tests.
Neither file goes through synchro: the async (PyMongoProtocol) and sync
(receive_message) paths share no implementation to mirror.

Author: aclark4life

test/asynchronous/test_async_network_layer.py

@@ -49,19 +49,6 @@ async def _make_proto_with_header(self, header_bytes, max_size=MAX_MESSAGE_SIZE)
         protocol._header = memoryview(bytearray(header_bytes))
         return protocol
 
-    async def test_initial_timeout_from_constructor(self):
-        protocol = await _make_protocol(timeout=3.0)
-        self.assertEqual(protocol.gettimeout, 3.0)
-
-    async def test_settimeout_updates_value(self):
-        protocol = await _make_protocol()
-        protocol.settimeout(7.5)
-        self.assertEqual(protocol.gettimeout, 7.5)
-
-    async def test_default_timeout_is_none(self):
-        protocol = await _make_protocol()
-        self.assertIsNone(protocol.gettimeout)
-
     async def test_normal_op_msg(self):
         header = _make_header(length=32, request_id=1, response_to=99, op_code=2013)
         protocol = await self._make_proto_with_header(header)

test/asynchronous/test_network_layer.py

@@ -12,52 +12,137 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-"""Unit tests for code in network_layer.py shared between sync and async APIs.
+"""Sync-only unit tests for network_layer.py.
 
-Async-only tests live in ``test_async_network_layer.py``.
+These cover ``receive_message`` and ``receive_data``, which only exist on the
+synchronous receive path (the async path uses ``PyMongoProtocol`` instead).
+The async-only tests live in ``test/asynchronous/test_async_network_layer.py``.
 """
 
 from __future__ import annotations
 
+import struct
 import sys
-from unittest.mock import MagicMock
+from unittest.mock import MagicMock, patch
 
 sys.path[0:0] = [""]
 
 from test import UnitTest, unittest
 
-from pymongo.network_layer import NetworkingInterfaceBase
-
-_IS_SYNC = True
-
-
-class TestNetworkingInterfaceBase(UnitTest):
-    def setUp(self):
-        self.base = NetworkingInterfaceBase(MagicMock())
-
-    def test_gettimeout_raises(self):
-        with self.assertRaises(NotImplementedError):
-            _ = self.base.gettimeout
-
-    def test_settimeout_raises(self):
-        with self.assertRaises(NotImplementedError):
-            self.base.settimeout(1.0)
-
-    def test_close_raises(self):
-        with self.assertRaises(NotImplementedError):
-            self.base.close()
-
-    def test_is_closing_raises(self):
-        with self.assertRaises(NotImplementedError):
-            self.base.is_closing()
-
-    def test_get_conn_raises(self):
-        with self.assertRaises(NotImplementedError):
-            _ = self.base.get_conn
-
-    def test_sock_raises(self):
-        with self.assertRaises(NotImplementedError):
-            _ = self.base.sock
+from pymongo import network_layer
+from pymongo.common import MAX_MESSAGE_SIZE
+from pymongo.errors import ProtocolError
+
+
+def _make_header(length, request_id, response_to, op_code):
+    return struct.pack("<iiii", length, request_id, response_to, op_code)
+
+
+def _make_compression_header(op_code, uncompressed_size, compressor_id):
+    return struct.pack("<iiB", op_code, uncompressed_size, compressor_id)
+
+
+def _make_conn():
+    conn = MagicMock()
+    conn.conn.gettimeout.return_value = None
+    return conn
+
+
+class TestReceiveMessage(UnitTest):
+    def _patch_receive_data(self, *chunks):
+        """Make receive_data return the given byte strings on successive calls."""
+        mock = patch.object(network_layer, "receive_data", side_effect=list(chunks))
+        self.addCleanup(mock.stop)
+        return mock.start()
+
+    def test_request_id_mismatch_raises(self):
+        self._patch_receive_data(
+            _make_header(length=32, request_id=0, response_to=99, op_code=2013)
+        )
+        with self.assertRaises(ProtocolError):
+            network_layer.receive_message(_make_conn(), request_id=1)
+
+    def test_length_too_small_raises(self):
+        self._patch_receive_data(_make_header(length=16, request_id=0, response_to=0, op_code=2013))
+        with self.assertRaisesRegex(ProtocolError, "not longer than standard message header"):
+            network_layer.receive_message(_make_conn(), request_id=None)
+
+    def test_length_exceeds_max_raises(self):
+        self._patch_receive_data(
+            _make_header(length=MAX_MESSAGE_SIZE + 1, request_id=0, response_to=0, op_code=2013)
+        )
+        with self.assertRaisesRegex(ProtocolError, "larger than server max"):
+            network_layer.receive_message(_make_conn(), request_id=None)
+
+    def test_normal_op_msg_unpacks(self):
+        body = b"x" * 16
+        self._patch_receive_data(
+            _make_header(length=32, request_id=0, response_to=0, op_code=2013), body
+        )
+        unpack = MagicMock(return_value="REPLY")
+        with patch.object(network_layer, "_UNPACK_REPLY", {2013: unpack}):
+            result = network_layer.receive_message(_make_conn(), request_id=None)
+        unpack.assert_called_once_with(body)
+        self.assertEqual(result, "REPLY")
+
+    def test_op_compressed_decompresses(self):
+        # length=35 -> body length = 35 - 25 = 10 (header 16 + compression sub-header 9).
+        compressed_body = b"y" * 10
+        self._patch_receive_data(
+            _make_header(length=35, request_id=0, response_to=0, op_code=2012),
+            _make_compression_header(op_code=2013, uncompressed_size=0, compressor_id=1),
+            compressed_body,
+        )
+        unpack = MagicMock(return_value="REPLY")
+        with (
+            patch.object(network_layer, "decompress", return_value=b"decompressed") as decompress,
+            patch.object(network_layer, "_UNPACK_REPLY", {2013: unpack}),
+        ):
+            result = network_layer.receive_message(_make_conn(), request_id=None)
+        decompress.assert_called_once_with(compressed_body, 1)
+        unpack.assert_called_once_with(b"decompressed")
+        self.assertEqual(result, "REPLY")
+
+    def test_unknown_opcode_raises(self):
+        self._patch_receive_data(
+            _make_header(length=20, request_id=0, response_to=0, op_code=9999), b"data"
+        )
+        with patch.object(network_layer, "_UNPACK_REPLY", {2013: MagicMock()}):
+            with self.assertRaises(ProtocolError):
+                network_layer.receive_message(_make_conn(), request_id=None)
+
+
+class TestReceiveData(UnitTest):
+    def test_reads_data_in_multiple_chunks(self):
+        # Covers the loop in receive_data that accumulates short reads until the
+        # requested length has been received.
+        data = b"abcdefgh"
+        chunk1, chunk2 = data[:4], data[4:]
+        conn = _make_conn()
+        calls = 0
+
+        def fake_recv_into(buf):
+            nonlocal calls
+            if calls == 0:
+                buf[: len(chunk1)] = chunk1
+                calls += 1
+                return len(chunk1)
+            buf[: len(chunk2)] = chunk2
+            calls += 1
+            return len(chunk2)
+
+        conn.conn.recv_into.side_effect = fake_recv_into
+        result = network_layer.receive_data(conn, len(data), deadline=None)
+        self.assertEqual(bytes(result), data)
+        self.assertEqual(calls, 2)
+
+    def test_raises_on_connection_closed(self):
+        # Covers the explicit `raise OSError("connection closed")` branch when
+        # recv_into returns 0.
+        conn = _make_conn()
+        conn.conn.recv_into.return_value = 0
+        with self.assertRaisesRegex(OSError, "connection closed"):
+            network_layer.receive_data(conn, 10, deadline=None)
 
 
 if __name__ == "__main__":

test/test_network_layer.py

@@ -128,7 +128,6 @@
     "SpecRunnerTask": "SpecRunnerThread",
     "AsyncMockConnection": "MockConnection",
     "AsyncMockPool": "MockPool",
-    "AsyncMock": "MagicMock",
     "StopAsyncIteration": "StopIteration",
     "create_async_event": "create_event",
     "async_create_barrier": "create_barrier",
@@ -255,7 +254,6 @@ def async_only_test(f: str) -> bool:
     "test_monitor.py",
     "test_monitoring.py",
     "test_mongos_load_balancing.py",
-    "test_network_layer.py",
     "test_on_demand_csfle.py",
     "test_periodic_executor.py",
     "test_pooling.py",