MCPcopy Create free account
hub / github.com/apache/cassandra-python-driver / test_protocol_error

Method test_protocol_error

tests/unit/io/utils.py:323–338  ·  view source on GitHub ↗
(self)

Source from the content-addressed store, hash-verified

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()

Callers

nothing calls this directly

Calls 7

make_connectionMethod · 0.95
make_header_prefixMethod · 0.95
make_options_bodyMethod · 0.95
get_socketMethod · 0.95
make_msgMethod · 0.95
handle_writeMethod · 0.45
handle_readMethod · 0.45

Tested by

no test coverage detected