100% coverage.

Author: mindflayer

mocket/socket.py

@@ -215,8 +215,9 @@ def recvmsg(
         Receive a message from the socket.
         This is a mock implementation that reads from the MocketSocketIO.
         """
-        data = self.recv(buffersize)
-        if not data:
+        try:
+            data = self.recv(buffersize)
+        except BlockingIOError:
             return b"", []
 
         # Mocking the ancillary data and flags as empty
@@ -236,8 +237,9 @@ def recvmsg_into(
         if not buffers:
             return 0
 
-        data = self.recv(len(buffers[0]))
-        if not data:
+        try:
+            data = self.recv(len(buffers[0]))
+        except BlockingIOError:
             return 0
 
         for i, buffer in enumerate(buffers):

tests/test_socket.py

@@ -90,3 +90,28 @@ def test_sendmsg_empty_buffers():
     sock = MocketSocket(socket.AF_INET, socket.SOCK_STREAM)
     result = sock.sendmsg([])
     assert result == 0
+
+
+def test_recvmsg_no_data():
+    sock = MocketSocket(socket.AF_INET, socket.SOCK_STREAM)
+    # Mock _io.read to return empty bytes
+    sock._io = type("MockIO", (), {"read": lambda self, n: b""})()
+    data, ancdata = sock.recvmsg(1024)
+    assert data == b""
+    assert ancdata == []
+
+
+def test_recvmsg_into_no_data():
+    sock = MocketSocket(socket.AF_INET, socket.SOCK_STREAM)
+    # Mock _io.read to return empty bytes
+    sock._io = type("MockIO", (), {"read": lambda self, n: b""})()
+    buf = bytearray(10)
+    nbytes = sock.recvmsg_into([buf])
+    assert nbytes == 0
+    assert buf == bytearray(10)
+
+
+def test_getsockopt():
+    # getsockopt is a static method, so we can call it directly
+    result = MocketSocket.getsockopt(0, 0)
+    assert result == socket.SOCK_STREAM