openstack/deb-python-cassandra-driver
Code Issues Proposed changes

fix unit tests following heartbeat update

Browse Source
This commit is contained in:
Adam Holmberg
2016-04-12 16:51:25 -05:00
parent 5b7627a848
commit c030d3e2e7
1 changed files with 9 additions and 6 deletions
Download Patch File Download Diff File Expand all files Collapse all files

15
tests/unit/test_connection.py
View File

@@ -22,9 +22,11 @@ from six import BytesIO
import time import time
from threading import Lock from threading import Lock
from cassandra.cluster import Cluster, Session from cassandra import OperationTimedOut
from cassandra.cluster import Cluster
from cassandra.connection import (Connection, HEADER_DIRECTION_TO_CLIENT, ProtocolError, from cassandra.connection import (Connection, HEADER_DIRECTION_TO_CLIENT, ProtocolError,
locally_supported_compressions, ConnectionHeartbeat, _Frame, Timer, TimerManager) locally_supported_compressions, ConnectionHeartbeat, _Frame, Timer, TimerManager,
ConnectionException)
from cassandra.marshal import uint8_pack, uint32_pack, int32_pack from cassandra.marshal import uint8_pack, uint32_pack, int32_pack
from cassandra.protocol import (write_stringmultimap, write_int, write_string, from cassandra.protocol import (write_stringmultimap, write_int, write_string,
SupportedMessage, ProtocolHandler) SupportedMessage, ProtocolHandler)
@@ -382,8 +384,8 @@ class ConnectionHeartbeatTest(unittest.TestCase):
connection.send_msg.assert_has_calls([call(ANY, request_id, ANY)] * get_holders.call_count) connection.send_msg.assert_has_calls([call(ANY, request_id, ANY)] * get_holders.call_count)
connection.defunct.assert_has_calls([call(ANY)] * get_holders.call_count) connection.defunct.assert_has_calls([call(ANY)] * get_holders.call_count)
exc = connection.defunct.call_args_list[0][0][0] exc = connection.defunct.call_args_list[0][0][0]
self.assertIsInstance(exc, Exception) self.assertIsInstance(exc, ConnectionException)
self.assertEqual(exc.args, Exception('Connection heartbeat failure').args) self.assertRegex(exc.args[0], r'^Received unexpected response to OptionsMessage.*')
holder.return_connection.assert_has_calls([call(connection)] * get_holders.call_count) holder.return_connection.assert_has_calls([call(connection)] * get_holders.call_count)
def test_timeout(self, *args): def test_timeout(self, *args):
@@ -410,8 +412,9 @@ class ConnectionHeartbeatTest(unittest.TestCase):
connection.send_msg.assert_has_calls([call(ANY, request_id, ANY)] * get_holders.call_count) connection.send_msg.assert_has_calls([call(ANY, request_id, ANY)] * get_holders.call_count)
connection.defunct.assert_has_calls([call(ANY)] * get_holders.call_count) connection.defunct.assert_has_calls([call(ANY)] * get_holders.call_count)
exc = connection.defunct.call_args_list[0][0][0] exc = connection.defunct.call_args_list[0][0][0]
self.assertIsInstance(exc, Exception) self.assertIsInstance(exc, OperationTimedOut)
self.assertEqual(exc.args, Exception('Connection heartbeat failure').args) self.assertEqual(exc.errors, 'Connection heartbeat timeout after 0.05 seconds')
self.assertEqual(exc.last_host, 'localhost')
holder.return_connection.assert_has_calls([call(connection)] * get_holders.call_count) holder.return_connection.assert_has_calls([call(connection)] * get_holders.call_count)