(self)
| 3992 | self.assertEqual(s.recv_into(bytearray()), 0) |
| 3993 | |
| 3994 | def test_recv_into_buffer_protocol_len(self): |
| 3995 | server = ThreadedEchoServer(CERTFILE) |
| 3996 | self.enterContext(server) |
| 3997 | s = socket.create_connection((HOST, server.port)) |
| 3998 | self.addCleanup(s.close) |
| 3999 | s = test_wrap_socket(s, suppress_ragged_eofs=False) |
| 4000 | self.addCleanup(s.close) |
| 4001 | |
| 4002 | s.send(b"data") |
| 4003 | buf = array.array('I', [0, 0]) |
| 4004 | self.assertEqual(s.recv_into(buf), 4) |
| 4005 | self.assertEqual(bytes(buf)[:4], b"data") |
| 4006 | |
| 4007 | class B(bytearray): |
| 4008 | def __len__(self): |
| 4009 | 1/0 |
| 4010 | s.send(b"data") |
| 4011 | buf = B(6) |
| 4012 | self.assertEqual(s.recv_into(buf), 4) |
| 4013 | self.assertEqual(bytes(buf), b"data\0\0") |
| 4014 | |
| 4015 | def test_nonblocking_send(self): |
| 4016 | server = ThreadedEchoServer(CERTFILE, |
nothing calls this directly
no test coverage detected