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):
@@ -257,7 +259,7 @@ def recvfrom_into(
         Receive data into a buffer and return the number of bytes received.
         This is a mock implementation that reads from the MocketSocketIO.
         """
-        return self.recv_into(buffer, buffersize, flags)
+        return self.recv_into(buffer, buffersize, flags), self._address
 
     def recv_into(
         self,

tests/test_socket.py

@@ -90,3 +90,39 @@ 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
+
+
+def test_recvfrom_into():
+    sock = MocketSocket(socket.AF_INET, socket.SOCK_STREAM)
+    test_data = b"abc123"
+    sock._io = type("MockIO", (), {"read": lambda self, n: test_data})()
+    buf = bytearray(10)
+    nbytes, addr = sock.recvfrom_into(buf)
+    assert nbytes == len(test_data)
+    assert buf[:nbytes] == test_data
+    assert addr == sock._address