(self)
| 391 | self.assertTrue(self.get_socket(c).send.call_args is not None) |
| 392 | |
| 393 | def test_partial_send(self): |
| 394 | c = self.make_connection() |
| 395 | |
| 396 | # only write the first four bytes of the OptionsMessage |
| 397 | write_size = 4 |
| 398 | self.get_socket(c).send.side_effect = None |
| 399 | self.get_socket(c).send.return_value = write_size |
| 400 | c.handle_write(*self.null_handle_function_args) |
| 401 | |
| 402 | msg_size = 9 # v3+ frame header |
| 403 | expected_writes = int(math.ceil(float(msg_size) / write_size)) |
| 404 | size_mod = msg_size % write_size |
| 405 | last_write_size = size_mod if size_mod else write_size |
| 406 | self.assertFalse(c.is_defunct) |
| 407 | self.assertEqual(expected_writes, self.get_socket(c).send.call_count) |
| 408 | self.assertEqual(last_write_size, |
| 409 | len(self.get_socket(c).send.call_args[0][0])) |
| 410 | |
| 411 | def test_socket_error_on_read(self): |
| 412 | c = self.make_connection() |
nothing calls this directly
no test coverage detected