(self)
| 321 | c.process_io_buffer.assert_called_once_with() |
| 322 | |
| 323 | def test_protocol_error(self): |
| 324 | c = self.make_connection() |
| 325 | |
| 326 | # let it write the OptionsMessage |
| 327 | c.handle_write(*self.null_handle_function_args) |
| 328 | |
| 329 | # read in a SupportedMessage response |
| 330 | header = self.make_header_prefix(SupportedMessage, version=0xa4) |
| 331 | options = self.make_options_body() |
| 332 | self.get_socket(c).recv.return_value = self.make_msg(header, options) |
| 333 | c.handle_read(*self.null_handle_function_args) |
| 334 | |
| 335 | # make sure it errored correctly |
| 336 | self.assertTrue(c.is_defunct) |
| 337 | self.assertTrue(c.connected_event.is_set()) |
| 338 | self.assertIsInstance(c.last_error, ProtocolError) |
| 339 | |
| 340 | def test_error_message_on_startup(self): |
| 341 | c = self.make_connection() |
nothing calls this directly
no test coverage detected