# Copyright DataStax, Inc. # # Licensed under the Apache License, Version 2.0 (the "License"); # you may not use this file except in compliance with the License. # You may obtain a copy of the License at # # http://www.apache.org/licenses/LICENSE-2.0 # # Unless required by applicable law or agreed to in writing, software # distributed under the License is distributed on an "AS IS" BASIS, # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. # See the License for the specific language governing permissions and # limitations under the License. import unittest from collections import deque from copy import copy from unittest.mock import Mock, call, patch, ANY import time from uuid import uuid4 import logging import warnings from packaging.version import Version import cassandra from cassandra.cluster import NoHostAvailable, ExecutionProfile, EXEC_PROFILE_DEFAULT, ControlConnection, Cluster from cassandra.concurrent import execute_concurrent from cassandra.policies import (RoundRobinPolicy, ExponentialReconnectionPolicy, RetryPolicy, SimpleConvictionPolicy, HostDistance, AddressTranslator, TokenAwarePolicy, HostFilterPolicy) from cassandra import ConsistencyLevel from cassandra.query import SimpleStatement, TraceUnavailable, tuple_factory from cassandra.auth import PlainTextAuthProvider, SaslAuthProvider from cassandra import connection from cassandra.connection import DefaultEndPoint from tests import notwindows from tests.integration import use_singledc, get_server_versions, CASSANDRA_VERSION, \ execute_until_pass, execute_with_long_wait_retry, get_node, MockLoggingHandler, get_unsupported_lower_protocol, \ get_unsupported_upper_protocol, protocolv6, local, CASSANDRA_IP, greaterthanorequalcass30, lessthanorequalcass40, \ DSE_VERSION, TestCluster, PROTOCOL_VERSION from tests.integration.util import assert_quiescent_pool_state import sys log = logging.getLogger(__name__) def setup_module(): use_singledc() warnings.simplefilter("always") class IgnoredHostPolicy(RoundRobinPolicy): def __init__(self, ignored_hosts): self.ignored_hosts = ignored_hosts RoundRobinPolicy.__init__(self) def distance(self, host): if(host.address in self.ignored_hosts): return HostDistance.IGNORED else: return HostDistance.LOCAL class ClusterTests(unittest.TestCase): @local def test_ignored_host_up(self): """ Test to ensure that is_up is not set by default on ignored hosts @since 3.6 @jira_ticket PYTHON-551 @expected_result ignored hosts should have None set for is_up @test_category connection """ ignored_host_policy = IgnoredHostPolicy(["127.0.0.2", "127.0.0.3"]) cluster = TestCluster( execution_profiles={EXEC_PROFILE_DEFAULT: ExecutionProfile(load_balancing_policy=ignored_host_policy)} ) cluster.connect() for host in cluster.metadata.all_hosts(): if str(host) == "127.0.0.1:9042": self.assertTrue(host.is_up) else: self.assertIsNone(host.is_up) cluster.shutdown() @local def test_host_resolution(self): """ Test to insure A records are resolved appropriately. @since 3.3 @jira_ticket PYTHON-415 @expected_result hostname will be transformed into IP @test_category connection """ cluster = TestCluster(contact_points=["localhost"], connect_timeout=1) self.assertTrue(DefaultEndPoint('127.0.0.1') in cluster.endpoints_resolved) @local def test_host_duplication(self): """ Ensure that duplicate hosts in the contact points are surfaced in the cluster metadata @since 3.3 @jira_ticket PYTHON-103 @expected_result duplicate hosts aren't surfaced in cluster.metadata @test_category connection """ cluster = TestCluster( contact_points=["localhost", "127.0.0.1", "localhost", "localhost", "localhost"], connect_timeout=1 ) cluster.connect(wait_for_all_pools=True) self.assertEqual(len(cluster.metadata.all_hosts()), 3) cluster.shutdown() cluster = TestCluster(contact_points=["127.0.0.1", "localhost"], connect_timeout=1) cluster.connect(wait_for_all_pools=True) self.assertEqual(len(cluster.metadata.all_hosts()), 3) cluster.shutdown() @local def test_raise_error_on_control_connection_timeout(self): """ Test for initial control connection timeout test_raise_error_on_control_connection_timeout tests that the driver times out after the set initial connection timeout. It first pauses node1, essentially making it unreachable. It then attempts to create a Cluster object via connecting to node1 with a timeout of 1 second, and ensures that a NoHostAvailable is raised, along with an OperationTimedOut for 1 second. @expected_errors NoHostAvailable When node1 is paused, and a connection attempt is made. @since 2.6.0 @jira_ticket PYTHON-206 @expected_result NoHostAvailable exception should be raised after 1 second. @test_category connection """ get_node(1).pause() cluster = TestCluster(contact_points=['127.0.0.1'], connect_timeout=1) with self.assertRaisesRegex(NoHostAvailable, "OperationTimedOut\('errors=Timed out creating connection \(1 seconds\)"): cluster.connect() cluster.shutdown() get_node(1).resume() def test_basic(self): """ Test basic connection and usage """ cluster = TestCluster() session = cluster.connect() result = execute_until_pass(session, """ CREATE KEYSPACE clustertests WITH replication = {'class': 'SimpleStrategy', 'replication_factor': '1'} """) self.assertFalse(result) result = execute_with_long_wait_retry(session, """ CREATE TABLE clustertests.cf0 ( a text, b text, c text, PRIMARY KEY (a, b) ) """) self.assertFalse(result) result = session.execute( """ INSERT INTO clustertests.cf0 (a, b, c) VALUES ('a', 'b', 'c') """) self.assertFalse(result) result = session.execute("SELECT * FROM clustertests.cf0") self.assertEqual([('a', 'b', 'c')], result) execute_with_long_wait_retry(session, "DROP KEYSPACE clustertests") cluster.shutdown() def test_session_host_parameter(self): """ Test for protocol negotiation Very that NoHostAvailable is risen in Session.__init__ when there are no valid connections and that no error is arisen otherwise, despite maybe being some invalid hosts @since 3.9 @jira_ticket PYTHON-665 @expected_result NoHostAvailable when the driver is unable to connect to a valid host, no exception otherwise @test_category connection """ def cleanup(): """ When this test fails, the inline .shutdown() calls don't get called, so we register this as a cleanup. """ self.cluster_to_shutdown.shutdown() self.addCleanup(cleanup) # Test with empty list self.cluster_to_shutdown = TestCluster(contact_points=[]) with self.assertRaises(NoHostAvailable): self.cluster_to_shutdown.connect() self.cluster_to_shutdown.shutdown() # Test with only invalid self.cluster_to_shutdown = TestCluster(contact_points=('1.2.3.4',)) with self.assertRaises(NoHostAvailable): self.cluster_to_shutdown.connect() self.cluster_to_shutdown.shutdown() # Test with valid and invalid hosts self.cluster_to_shutdown = TestCluster(contact_points=("127.0.0.1", "127.0.0.2", "1.2.3.4")) self.cluster_to_shutdown.connect() self.cluster_to_shutdown.shutdown() def test_protocol_negotiation(self): """ Test for protocol negotiation test_protocol_negotiation tests that the driver will select the correct protocol version to match the correct cassandra version. Please note that 2.1.5 has a bug https://issues.apache.org/jira/browse/CASSANDRA-9451 that will cause this test to fail that will cause this to not pass. It was rectified in 2.1.6 @since 2.6.0 @jira_ticket PYTHON-240 @expected_result the correct protocol version should be selected @test_category connection """ cluster = Cluster() self.assertLessEqual(cluster.protocol_version, cassandra.ProtocolVersion.MAX_SUPPORTED) session = cluster.connect() updated_protocol_version = session._protocol_version updated_cluster_version = cluster.protocol_version # Make sure the correct protocol was selected by default if DSE_VERSION and DSE_VERSION >= Version("6.0"): self.assertEqual(updated_protocol_version, cassandra.ProtocolVersion.DSE_V2) self.assertEqual(updated_cluster_version, cassandra.ProtocolVersion.DSE_V2) elif DSE_VERSION and DSE_VERSION >= Version("5.1"): self.assertEqual(updated_protocol_version, cassandra.ProtocolVersion.DSE_V1) self.assertEqual(updated_cluster_version, cassandra.ProtocolVersion.DSE_V1) elif CASSANDRA_VERSION >= Version('4.0-beta5'): self.assertEqual(updated_protocol_version, cassandra.ProtocolVersion.V5) self.assertEqual(updated_cluster_version, cassandra.ProtocolVersion.V5) elif CASSANDRA_VERSION >= Version('4.0-a'): self.assertEqual(updated_protocol_version, cassandra.ProtocolVersion.V4) self.assertEqual(updated_cluster_version, cassandra.ProtocolVersion.V4) elif CASSANDRA_VERSION >= Version('3.11'): self.assertEqual(updated_protocol_version, cassandra.ProtocolVersion.V4) self.assertEqual(updated_cluster_version, cassandra.ProtocolVersion.V4) elif CASSANDRA_VERSION >= Version('3.0'): self.assertEqual(updated_protocol_version, cassandra.ProtocolVersion.V4) self.assertEqual(updated_cluster_version, cassandra.ProtocolVersion.V4) elif CASSANDRA_VERSION >= Version('2.2'): self.assertEqual(updated_protocol_version, 4) self.assertEqual(updated_cluster_version, 4) elif CASSANDRA_VERSION >= Version('2.1'): self.assertEqual(updated_protocol_version, 3) self.assertEqual(updated_cluster_version, 3) elif CASSANDRA_VERSION >= Version('2.0'): self.assertEqual(updated_protocol_version, 2) self.assertEqual(updated_cluster_version, 2) else: self.assertEqual(updated_protocol_version, 1) self.assertEqual(updated_cluster_version, 1) cluster.shutdown() def test_invalid_protocol_negotation(self): """ Test for protocol negotiation when explicit versions are set If an explicit protocol version that is not compatible with the server version is set an exception should be thrown. It should not attempt to negotiate for reference supported protocol version to server versions is as follows/ 1.2 -> 1 2.0 -> 2, 1 2.1 -> 3, 2, 1 2.2 -> 4, 3, 2, 1 3.X -> 4, 3 @since 3.6.0 @jira_ticket PYTHON-537 @expected_result downgrading should not be allowed when explicit protocol versions are set. @test_category connection """ upper_bound = get_unsupported_upper_protocol() log.debug('got upper_bound of {}'.format(upper_bound)) if upper_bound is not None: cluster = TestCluster(protocol_version=upper_bound) with self.assertRaises(NoHostAvailable): cluster.connect() cluster.shutdown() lower_bound = get_unsupported_lower_protocol() log.debug('got lower_bound of {}'.format(lower_bound)) if lower_bound is not None: cluster = TestCluster(protocol_version=lower_bound) with self.assertRaises(NoHostAvailable): cluster.connect() cluster.shutdown() def test_connect_on_keyspace(self): """ Ensure clusters that connect on a keyspace, do """ cluster = TestCluster() session = cluster.connect() result = session.execute( """ INSERT INTO test1rf.test (k, v) VALUES (8889, 8889) """) self.assertFalse(result) result = session.execute("SELECT * FROM test1rf.test") self.assertEqual([(8889, 8889)], result, "Rows in ResultSet are {0}".format(result.current_rows)) # test_connect_on_keyspace session2 = cluster.connect('test1rf') result2 = session2.execute("SELECT * FROM test") self.assertEqual(result, result2) cluster.shutdown() def test_set_keyspace_twice(self): cluster = TestCluster() session = cluster.connect() session.execute("USE system") session.execute("USE system") cluster.shutdown() def test_default_connections(self): """ Ensure errors are not thrown when using non-default policies """ TestCluster( reconnection_policy=ExponentialReconnectionPolicy(1.0, 600.0), conviction_policy_factory=SimpleConvictionPolicy, protocol_version=PROTOCOL_VERSION ) def test_connect_to_already_shutdown_cluster(self): """ Ensure you cannot connect to a cluster that's been shutdown """ cluster = TestCluster() cluster.shutdown() self.assertRaises(Exception, cluster.connect) def test_auth_provider_is_callable(self): """ Ensure that auth_providers are always callable """ self.assertRaises(TypeError, Cluster, auth_provider=1, protocol_version=1) c = TestCluster(protocol_version=1) self.assertRaises(TypeError, setattr, c, 'auth_provider', 1) def test_v2_auth_provider(self): """ Check for v2 auth_provider compliance """ bad_auth_provider = lambda x: {'username': 'foo', 'password': 'bar'} self.assertRaises(TypeError, Cluster, auth_provider=bad_auth_provider, protocol_version=2) c = TestCluster(protocol_version=2) self.assertRaises(TypeError, setattr, c, 'auth_provider', bad_auth_provider) def test_conviction_policy_factory_is_callable(self): """ Ensure that conviction_policy_factory are always callable """ self.assertRaises(ValueError, Cluster, conviction_policy_factory=1) def test_connect_to_bad_hosts(self): """ Ensure that a NoHostAvailable Exception is thrown when a cluster cannot connect to given hosts """ cluster = TestCluster(contact_points=['127.1.2.9', '127.1.2.10'], protocol_version=PROTOCOL_VERSION) self.assertRaises(NoHostAvailable, cluster.connect) def test_cluster_settings(self): """ Test connection setting getters and setters """ if PROTOCOL_VERSION >= 3: raise unittest.SkipTest("min/max requests and core/max conns aren't used with v3 protocol") cluster = TestCluster() min_requests_per_connection = cluster.get_min_requests_per_connection(HostDistance.LOCAL) self.assertEqual(cassandra.cluster.DEFAULT_MIN_REQUESTS, min_requests_per_connection) cluster.set_min_requests_per_connection(HostDistance.LOCAL, min_requests_per_connection + 1) self.assertEqual(cluster.get_min_requests_per_connection(HostDistance.LOCAL), min_requests_per_connection + 1) max_requests_per_connection = cluster.get_max_requests_per_connection(HostDistance.LOCAL) self.assertEqual(cassandra.cluster.DEFAULT_MAX_REQUESTS, max_requests_per_connection) cluster.set_max_requests_per_connection(HostDistance.LOCAL, max_requests_per_connection + 1) self.assertEqual(cluster.get_max_requests_per_connection(HostDistance.LOCAL), max_requests_per_connection + 1) core_connections_per_host = cluster.get_core_connections_per_host(HostDistance.LOCAL) self.assertEqual(cassandra.cluster.DEFAULT_MIN_CONNECTIONS_PER_LOCAL_HOST, core_connections_per_host) cluster.set_core_connections_per_host(HostDistance.LOCAL, core_connections_per_host + 1) self.assertEqual(cluster.get_core_connections_per_host(HostDistance.LOCAL), core_connections_per_host + 1) max_connections_per_host = cluster.get_max_connections_per_host(HostDistance.LOCAL) self.assertEqual(cassandra.cluster.DEFAULT_MAX_CONNECTIONS_PER_LOCAL_HOST, max_connections_per_host) cluster.set_max_connections_per_host(HostDistance.LOCAL, max_connections_per_host + 1) self.assertEqual(cluster.get_max_connections_per_host(HostDistance.LOCAL), max_connections_per_host + 1) def test_refresh_schema(self): cluster = TestCluster() session = cluster.connect() original_meta = cluster.metadata.keyspaces # full schema refresh, with wait cluster.refresh_schema_metadata() self.assertIsNot(original_meta, cluster.metadata.keyspaces) self.assertEqual(original_meta, cluster.metadata.keyspaces) cluster.shutdown() def test_refresh_schema_keyspace(self): cluster = TestCluster() session = cluster.connect() original_meta = cluster.metadata.keyspaces original_system_meta = original_meta['system'] # only refresh one keyspace cluster.refresh_keyspace_metadata('system') current_meta = cluster.metadata.keyspaces self.assertIs(original_meta, current_meta) current_system_meta = current_meta['system'] self.assertIsNot(original_system_meta, current_system_meta) self.assertEqual(original_system_meta.as_cql_query(), current_system_meta.as_cql_query()) cluster.shutdown() def test_refresh_schema_table(self): cluster = TestCluster() session = cluster.connect() original_meta = cluster.metadata.keyspaces original_system_meta = original_meta['system'] original_system_schema_meta = original_system_meta.tables['local'] # only refresh one table cluster.refresh_table_metadata('system', 'local') current_meta = cluster.metadata.keyspaces current_system_meta = current_meta['system'] current_system_schema_meta = current_system_meta.tables['local'] self.assertIs(original_meta, current_meta) self.assertIs(original_system_meta, current_system_meta) self.assertIsNot(original_system_schema_meta, current_system_schema_meta) self.assertEqual(original_system_schema_meta.as_cql_query(), current_system_schema_meta.as_cql_query()) cluster.shutdown() def test_refresh_schema_type(self): if get_server_versions()[0] < (2, 1, 0): raise unittest.SkipTest('UDTs were introduced in Cassandra 2.1') if PROTOCOL_VERSION < 3: raise unittest.SkipTest('UDTs are not specified in change events for protocol v2') # We may want to refresh types on keyspace change events in that case(?) cluster = TestCluster() session = cluster.connect() keyspace_name = 'test1rf' type_name = self._testMethodName execute_until_pass(session, 'CREATE TYPE IF NOT EXISTS %s.%s (one int, two text)' % (keyspace_name, type_name)) original_meta = cluster.metadata.keyspaces original_test1rf_meta = original_meta[keyspace_name] original_type_meta = original_test1rf_meta.user_types[type_name] # only refresh one type cluster.refresh_user_type_metadata('test1rf', type_name) current_meta = cluster.metadata.keyspaces current_test1rf_meta = current_meta[keyspace_name] current_type_meta = current_test1rf_meta.user_types[type_name] self.assertIs(original_meta, current_meta) self.assertEqual(original_test1rf_meta.export_as_string(), current_test1rf_meta.export_as_string()) self.assertIsNot(original_type_meta, current_type_meta) self.assertEqual(original_type_meta.as_cql_query(), current_type_meta.as_cql_query()) cluster.shutdown() @local @notwindows def test_refresh_schema_no_wait(self): original_wait_for_responses = connection.Connection.wait_for_responses def patched_wait_for_responses(*args, **kwargs): # When selecting schema version, replace the real schema UUID with an unexpected UUID response = original_wait_for_responses(*args, **kwargs) if len(args) > 2 and hasattr(args[2], "query") and args[2].query == "SELECT schema_version FROM system.local WHERE key='local'": new_uuid = uuid4() response[1].parsed_rows[0] = (new_uuid,) return response with patch.object(connection.Connection, "wait_for_responses", patched_wait_for_responses): agreement_timeout = 1 # cluster agreement wait exceeded c = TestCluster(max_schema_agreement_wait=agreement_timeout) c.connect() self.assertTrue(c.metadata.keyspaces) # cluster agreement wait used for refresh original_meta = c.metadata.keyspaces start_time = time.time() self.assertRaisesRegex(Exception, r"Schema metadata was not refreshed.*", c.refresh_schema_metadata) end_time = time.time() self.assertGreaterEqual(end_time - start_time, agreement_timeout) self.assertIs(original_meta, c.metadata.keyspaces) # refresh wait overrides cluster value original_meta = c.metadata.keyspaces start_time = time.time() c.refresh_schema_metadata(max_schema_agreement_wait=0) end_time = time.time() self.assertLess(end_time - start_time, agreement_timeout) self.assertIsNot(original_meta, c.metadata.keyspaces) self.assertEqual(original_meta, c.metadata.keyspaces) c.shutdown() refresh_threshold = 0.5 # cluster agreement bypass c = TestCluster(max_schema_agreement_wait=0) start_time = time.time() s = c.connect() end_time = time.time() self.assertLess(end_time - start_time, refresh_threshold) self.assertTrue(c.metadata.keyspaces) # cluster agreement wait used for refresh original_meta = c.metadata.keyspaces start_time = time.time() c.refresh_schema_metadata() end_time = time.time() self.assertLess(end_time - start_time, refresh_threshold) self.assertIsNot(original_meta, c.metadata.keyspaces) self.assertEqual(original_meta, c.metadata.keyspaces) # refresh wait overrides cluster value original_meta = c.metadata.keyspaces start_time = time.time() self.assertRaisesRegex(Exception, r"Schema metadata was not refreshed.*", c.refresh_schema_metadata, max_schema_agreement_wait=agreement_timeout) end_time = time.time() self.assertGreaterEqual(end_time - start_time, agreement_timeout) self.assertIs(original_meta, c.metadata.keyspaces) c.shutdown() def test_trace(self): """ Ensure trace can be requested for async and non-async queries """ cluster = TestCluster() session = cluster.connect() result = session.execute( "SELECT * FROM system.local", trace=True) self._check_trace(result.get_query_trace()) query = "SELECT * FROM system.local" statement = SimpleStatement(query) result = session.execute(statement, trace=True) self._check_trace(result.get_query_trace()) query = "SELECT * FROM system.local" statement = SimpleStatement(query) result = session.execute(statement) self.assertIsNone(result.get_query_trace()) statement2 = SimpleStatement(query) future = session.execute_async(statement2, trace=True) future.result() self._check_trace(future.get_query_trace()) statement2 = SimpleStatement(query) future = session.execute_async(statement2) future.result() self.assertIsNone(future.get_query_trace()) prepared = session.prepare("SELECT * FROM system.local") future = session.execute_async(prepared, parameters=(), trace=True) future.result() self._check_trace(future.get_query_trace()) cluster.shutdown() def test_trace_unavailable(self): """ First checks that TraceUnavailable is arisen if the max_wait parameter is negative Then checks that TraceUnavailable is arisen if the result hasn't been set yet @since 3.10 @jira_ticket PYTHON-196 @expected_result TraceUnavailable is arisen in both cases @test_category query """ cluster = TestCluster() self.addCleanup(cluster.shutdown) session = cluster.connect() query = "SELECT * FROM system.local" statement = SimpleStatement(query) max_retry_count = 10 for i in range(max_retry_count): future = session.execute_async(statement, trace=True) future.result() try: result = future.get_query_trace(-1.0) # In case the result has time to come back before this timeout due to a race condition self._check_trace(result) except TraceUnavailable: break else: raise Exception("get_query_trace didn't raise TraceUnavailable after {} tries".format(max_retry_count)) for i in range(max_retry_count): future = session.execute_async(statement, trace=True) try: result = future.get_query_trace(max_wait=120) # In case the result has been set check the trace self._check_trace(result) except TraceUnavailable: break else: raise Exception("get_query_trace didn't raise TraceUnavailable after {} tries".format(max_retry_count)) def test_one_returns_none(self): """ Test ResulSet.one returns None if no rows where found @since 3.14 @jira_ticket PYTHON-947 @expected_result ResulSet.one is None @test_category query """ with TestCluster() as cluster: session = cluster.connect() self.assertIsNone(session.execute("SELECT * from system.local WHERE key='madeup_key'").one()) def test_string_coverage(self): """ Ensure str(future) returns without error """ cluster = TestCluster() session = cluster.connect() query = "SELECT * FROM system.local" statement = SimpleStatement(query) future = session.execute_async(statement) self.assertIn(query, str(future)) future.result() self.assertIn(query, str(future)) self.assertIn('result', str(future)) cluster.shutdown() def test_can_connect_with_plainauth(self): """ Verify that we can connect setting PlainTextAuthProvider against a C* server without authentication set. We also verify a warning is issued per connection. This test is here instead of in test_authentication.py because the C* server running in that module has auth set. @since 3.14 @jira_ticket PYTHON-940 @expected_result we can connect, query C* and warning are issued @test_category auth """ auth_provider = PlainTextAuthProvider( username="made_up_username", password="made_up_password" ) self._warning_are_issued_when_auth(auth_provider) def test_can_connect_with_sslauth(self): """ Verify that we can connect setting SaslAuthProvider against a C* server without authentication set. We also verify a warning is issued per connection. This test is here instead of in test_authentication.py because the C* server running in that module has auth set. @since 3.14 @jira_ticket PYTHON-940 @expected_result we can connect, query C* and warning are issued @test_category auth """ sasl_kwargs = {'service': 'cassandra', 'mechanism': 'PLAIN', 'qops': ['auth'], 'username': "made_up_username", 'password': "made_up_password"} auth_provider = SaslAuthProvider(**sasl_kwargs) self._warning_are_issued_when_auth(auth_provider) def _warning_are_issued_when_auth(self, auth_provider): with MockLoggingHandler().set_module_name(connection.__name__) as mock_handler: with TestCluster(auth_provider=auth_provider) as cluster: session = cluster.connect() self.assertIsNotNone(session.execute("SELECT * from system.local")) # Three conenctions to nodes plus the control connection auth_warning = mock_handler.get_message_count('warning', "An authentication challenge was not sent") self.assertGreaterEqual(auth_warning, 4) self.assertEqual( auth_warning, mock_handler.get_message_count("debug", "Got ReadyMessage on new connection") ) def test_idle_heartbeat(self): interval = 2 cluster = TestCluster(idle_heartbeat_interval=interval, monitor_reporting_enabled=False) if PROTOCOL_VERSION < 3: cluster.set_core_connections_per_host(HostDistance.LOCAL, 1) session = cluster.connect(wait_for_all_pools=True) # This test relies on impl details of connection req id management to see if heartbeats # are being sent. May need update if impl is changed connection_request_ids = {} for h in cluster.get_connection_holders(): for c in h.get_connections(): # make sure none are idle (should have startup messages self.assertFalse(c.is_idle) with c.lock: connection_request_ids[id(c)] = deque(c.request_ids) # copy of request ids # let two heatbeat intervals pass (first one had startup messages in it) time.sleep(2 * interval + interval/2) connections = [c for holders in cluster.get_connection_holders() for c in holders.get_connections()] # make sure requests were sent on all connections for c in connections: expected_ids = connection_request_ids[id(c)] expected_ids.rotate(-1) with c.lock: self.assertListEqual(list(c.request_ids), list(expected_ids)) # assert idle status self.assertTrue(all(c.is_idle for c in connections)) # send messages on all connections statements_and_params = [("SELECT release_version FROM system.local", ())] * len(cluster.metadata.all_hosts()) results = execute_concurrent(session, statements_and_params) for success, result in results: self.assertTrue(success) # assert not idle status self.assertFalse(any(c.is_idle if not c.is_control_connection else False for c in connections)) # holders include session pools and cc holders = cluster.get_connection_holders() self.assertIn(cluster.control_connection, holders) self.assertEqual(len(holders), len(cluster.metadata.all_hosts()) + 1) # hosts pools, 1 for cc # include additional sessions session2 = cluster.connect(wait_for_all_pools=True) holders = cluster.get_connection_holders() self.assertIn(cluster.control_connection, holders) self.assertEqual(len(holders), 2 * len(cluster.metadata.all_hosts()) + 1) # 2 sessions' hosts pools, 1 for cc cluster._idle_heartbeat.stop() cluster._idle_heartbeat.join() assert_quiescent_pool_state(self, cluster) cluster.shutdown() @patch('cassandra.cluster.Cluster.idle_heartbeat_interval', new=0.1) def test_idle_heartbeat_disabled(self): self.assertTrue(Cluster.idle_heartbeat_interval) # heartbeat disabled with '0' cluster = TestCluster(idle_heartbeat_interval=0) self.assertEqual(cluster.idle_heartbeat_interval, 0) session = cluster.connect() # let two heatbeat intervals pass (first one had startup messages in it) time.sleep(2 * Cluster.idle_heartbeat_interval) connections = [c for holders in cluster.get_connection_holders() for c in holders.get_connections()] # assert not idle status (should never get reset because there is not heartbeat) self.assertFalse(any(c.is_idle for c in connections)) cluster.shutdown() def test_pool_management(self): # Ensure that in_flight and request_ids quiesce after cluster operations cluster = TestCluster(idle_heartbeat_interval=0) # no idle heartbeat here, pool management is tested in test_idle_heartbeat session = cluster.connect() session2 = cluster.connect() # prepare p = session.prepare("SELECT * FROM system.local WHERE key=?") self.assertTrue(session.execute(p, ('local',))) # simple self.assertTrue(session.execute("SELECT * FROM system.local WHERE key='local'")) # set keyspace session.set_keyspace('system') session.set_keyspace('system_traces') # use keyspace session.execute('USE system') session.execute('USE system_traces') # refresh schema cluster.refresh_schema_metadata() cluster.refresh_schema_metadata(max_schema_agreement_wait=0) assert_quiescent_pool_state(self, cluster) cluster.shutdown() @local def test_profile_load_balancing(self): """ Tests that profile load balancing policies are honored. @since 3.5 @jira_ticket PYTHON-569 @expected_result Execution Policy should be used when applicable. @test_category config_profiles """ query = "select release_version from system.local" node1 = ExecutionProfile( load_balancing_policy=HostFilterPolicy( RoundRobinPolicy(), lambda host: host.address == CASSANDRA_IP ) ) with TestCluster(execution_profiles={'node1': node1}, monitor_reporting_enabled=False) as cluster: session = cluster.connect(wait_for_all_pools=True) # default is DCA RR for all hosts expected_hosts = set(cluster.metadata.all_hosts()) queried_hosts = set() for _ in expected_hosts: rs = session.execute(query) queried_hosts.add(rs.response_future._current_host) self.assertEqual(queried_hosts, expected_hosts) # by name we should only hit the one expected_hosts = set(h for h in cluster.metadata.all_hosts() if h.address == CASSANDRA_IP) queried_hosts = set() for _ in cluster.metadata.all_hosts(): rs = session.execute(query, execution_profile='node1') queried_hosts.add(rs.response_future._current_host) self.assertEqual(queried_hosts, expected_hosts) # use a copied instance and override the row factory # assert last returned value can be accessed as a namedtuple so we can prove something different named_tuple_row = rs[0] self.assertIsInstance(named_tuple_row, tuple) self.assertTrue(named_tuple_row.release_version) tmp_profile = copy(node1) tmp_profile.row_factory = tuple_factory queried_hosts = set() for _ in cluster.metadata.all_hosts(): rs = session.execute(query, execution_profile=tmp_profile) queried_hosts.add(rs.response_future._current_host) self.assertEqual(queried_hosts, expected_hosts) tuple_row = rs[0] self.assertIsInstance(tuple_row, tuple) with self.assertRaises(AttributeError): tuple_row.release_version # make sure original profile is not impacted self.assertTrue(session.execute(query, execution_profile='node1')[0].release_version) def test_setting_lbp_legacy(self): cluster = TestCluster() self.addCleanup(cluster.shutdown) cluster.load_balancing_policy = RoundRobinPolicy() self.assertEqual( list(cluster.load_balancing_policy.make_query_plan()), [] ) cluster.connect() self.assertNotEqual( list(cluster.load_balancing_policy.make_query_plan()), [] ) def test_profile_lb_swap(self): """ Tests that profile load balancing policies are not shared Creates two LBP, runs a few queries, and validates that each LBP is execised seperately between EP's @since 3.5 @jira_ticket PYTHON-569 @expected_result LBP should not be shared. @test_category config_profiles """ query = "select release_version from system.local" rr1 = ExecutionProfile(load_balancing_policy=RoundRobinPolicy()) rr2 = ExecutionProfile(load_balancing_policy=RoundRobinPolicy()) exec_profiles = {'rr1': rr1, 'rr2': rr2} with TestCluster(execution_profiles=exec_profiles) as cluster: session = cluster.connect(wait_for_all_pools=True) # default is DCA RR for all hosts expected_hosts = set(cluster.metadata.all_hosts()) rr1_queried_hosts = set() rr2_queried_hosts = set() rs = session.execute(query, execution_profile='rr1') rr1_queried_hosts.add(rs.response_future._current_host) rs = session.execute(query, execution_profile='rr2') rr2_queried_hosts.add(rs.response_future._current_host) self.assertEqual(rr2_queried_hosts, rr1_queried_hosts) def test_ta_lbp(self): """ Test that execution profiles containing token aware LBP can be added @since 3.5 @jira_ticket PYTHON-569 @expected_result Queries can run @test_category config_profiles """ query = "select release_version from system.local" ta1 = ExecutionProfile() with TestCluster() as cluster: session = cluster.connect() cluster.add_execution_profile("ta1", ta1) rs = session.execute(query, execution_profile='ta1') def test_clone_shared_lbp(self): """ Tests that profile load balancing policies are shared on clone Creates one LBP clones it, and ensures that the LBP is shared between the two EP's @since 3.5 @jira_ticket PYTHON-569 @expected_result LBP is shared @test_category config_profiles """ query = "select release_version from system.local" rr1 = ExecutionProfile(load_balancing_policy=RoundRobinPolicy()) exec_profiles = {'rr1': rr1} with TestCluster(execution_profiles=exec_profiles) as cluster: session = cluster.connect(wait_for_all_pools=True) self.assertGreater(len(cluster.metadata.all_hosts()), 1, "We only have one host connected at this point") rr1_clone = session.execution_profile_clone_update('rr1', row_factory=tuple_factory) cluster.add_execution_profile("rr1_clone", rr1_clone) rr1_queried_hosts = set() rr1_clone_queried_hosts = set() rs = session.execute(query, execution_profile='rr1') rr1_queried_hosts.add(rs.response_future._current_host) rs = session.execute(query, execution_profile='rr1_clone') rr1_clone_queried_hosts.add(rs.response_future._current_host) self.assertNotEqual(rr1_clone_queried_hosts, rr1_queried_hosts) def test_missing_exec_prof(self): """ Tests to verify that using an unknown profile raises a ValueError @since 3.5 @jira_ticket PYTHON-569 @expected_result ValueError @test_category config_profiles """ query = "select release_version from system.local" rr1 = ExecutionProfile(load_balancing_policy=RoundRobinPolicy()) rr2 = ExecutionProfile(load_balancing_policy=RoundRobinPolicy()) exec_profiles = {'rr1': rr1, 'rr2': rr2} with TestCluster(execution_profiles=exec_profiles) as cluster: session = cluster.connect() with self.assertRaises(ValueError): session.execute(query, execution_profile='rr3') @local def test_profile_pool_management(self): """ Tests that changes to execution profiles correctly impact our cluster's pooling @since 3.5 @jira_ticket PYTHON-569 @expected_result pools should be correctly updated as EP's are added and removed @test_category config_profiles """ node1 = ExecutionProfile( load_balancing_policy=HostFilterPolicy( RoundRobinPolicy(), lambda host: host.address == "127.0.0.1" ) ) node2 = ExecutionProfile( load_balancing_policy=HostFilterPolicy( RoundRobinPolicy(), lambda host: host.address == "127.0.0.2" ) ) with TestCluster(execution_profiles={EXEC_PROFILE_DEFAULT: node1, 'node2': node2}) as cluster: session = cluster.connect(wait_for_all_pools=True) pools = session.get_pool_state() # there are more hosts, but we connected to the ones in the lbp aggregate self.assertGreater(len(cluster.metadata.all_hosts()), 2) self.assertEqual(set(h.address for h in pools), set(('127.0.0.1', '127.0.0.2'))) # dynamically update pools on add node3 = ExecutionProfile( load_balancing_policy=HostFilterPolicy( RoundRobinPolicy(), lambda host: host.address == "127.0.0.3" ) ) cluster.add_execution_profile('node3', node3) pools = session.get_pool_state() self.assertEqual(set(h.address for h in pools), set(('127.0.0.1', '127.0.0.2', '127.0.0.3'))) @local def test_add_profile_timeout(self): """ Tests that EP Timeouts are honored. @since 3.5 @jira_ticket PYTHON-569 @expected_result EP timeouts should override defaults @test_category config_profiles """ max_retry_count = 10 for i in range(max_retry_count): node1 = ExecutionProfile( load_balancing_policy=HostFilterPolicy( RoundRobinPolicy(), lambda host: host.address == "127.0.0.1" ) ) with TestCluster(execution_profiles={EXEC_PROFILE_DEFAULT: node1}) as cluster: session = cluster.connect(wait_for_all_pools=True) pools = session.get_pool_state() self.assertGreater(len(cluster.metadata.all_hosts()), 2) self.assertEqual(set(h.address for h in pools), set(('127.0.0.1',))) node2 = ExecutionProfile( load_balancing_policy=HostFilterPolicy( RoundRobinPolicy(), lambda host: host.address in ["127.0.0.2", "127.0.0.3"] ) ) start = time.time() try: self.assertRaises(cassandra.OperationTimedOut, cluster.add_execution_profile, 'profile_{0}'.format(i), node2, pool_wait_timeout=sys.float_info.min) break except AssertionError: end = time.time() self.assertAlmostEqual(start, end, 1) else: raise Exception("add_execution_profile didn't timeout after {0} retries".format(max_retry_count)) @notwindows def test_execute_query_timeout(self): with TestCluster() as cluster: session = cluster.connect(wait_for_all_pools=True) query = "SELECT * FROM system.local" # default is passed down default_profile = cluster.profile_manager.profiles[EXEC_PROFILE_DEFAULT] rs = session.execute(query) self.assertEqual(rs.response_future.timeout, default_profile.request_timeout) # tiny timeout times out as expected tmp_profile = copy(default_profile) tmp_profile.request_timeout = sys.float_info.min max_retry_count = 10 for _ in range(max_retry_count): start = time.time() try: with self.assertRaises(cassandra.OperationTimedOut): session.execute(query, execution_profile=tmp_profile) break except: import traceback traceback.print_exc() end = time.time() self.assertAlmostEqual(start, end, 1) else: raise Exception("session.execute didn't time out in {0} tries".format(max_retry_count)) def test_replicas_are_queried(self): """ Test that replicas are queried first for TokenAwarePolicy. A table with RF 1 is created. All the queries should go to that replica when TokenAwarePolicy is used. Then using HostFilterPolicy the replica is excluded from the considered hosts. By checking the trace we verify that there are no more replicas. @since 3.5 @jira_ticket PYTHON-653 @expected_result the replicas are queried for HostFilterPolicy @test_category metadata """ queried_hosts = set() tap_profile = ExecutionProfile( load_balancing_policy=TokenAwarePolicy(RoundRobinPolicy()) ) with TestCluster(execution_profiles={EXEC_PROFILE_DEFAULT: tap_profile}) as cluster: session = cluster.connect(wait_for_all_pools=True) session.execute(''' CREATE TABLE test1rf.table_with_big_key ( k1 int, k2 int, k3 int, k4 int, PRIMARY KEY((k1, k2, k3), k4))''') prepared = session.prepare("""SELECT * from test1rf.table_with_big_key WHERE k1 = ? AND k2 = ? AND k3 = ? AND k4 = ?""") for i in range(10): result = session.execute(prepared, (i, i, i, i), trace=True) trace = result.response_future.get_query_trace(query_cl=ConsistencyLevel.ALL) queried_hosts = self._assert_replica_queried(trace, only_replicas=True) last_i = i hfp_profile = ExecutionProfile( load_balancing_policy=HostFilterPolicy(RoundRobinPolicy(), predicate=lambda host: host.address != only_replica) ) only_replica = queried_hosts.pop() log = logging.getLogger(__name__) log.info("The only replica found was: {}".format(only_replica)) available_hosts = [host for host in ["127.0.0.1", "127.0.0.2", "127.0.0.3"] if host != only_replica] with TestCluster(contact_points=available_hosts, execution_profiles={EXEC_PROFILE_DEFAULT: hfp_profile}) as cluster: session = cluster.connect(wait_for_all_pools=True) prepared = session.prepare("""SELECT * from test1rf.table_with_big_key WHERE k1 = ? AND k2 = ? AND k3 = ? AND k4 = ?""") for _ in range(10): result = session.execute(prepared, (last_i, last_i, last_i, last_i), trace=True) trace = result.response_future.get_query_trace(query_cl=ConsistencyLevel.ALL) self._assert_replica_queried(trace, only_replicas=False) session.execute('''DROP TABLE test1rf.table_with_big_key''') @unittest.skip @greaterthanorequalcass30 @lessthanorequalcass40 def test_compact_option(self): """ Test the driver can connect with the no_compact option and the results are as expected. This test is very similar to the corresponding dtest @since 3.12 @jira_ticket PYTHON-366 @expected_result only one hosts' metadata will be populated @test_category connection """ nc_cluster = TestCluster(no_compact=True) nc_session = nc_cluster.connect() cluster = TestCluster(no_compact=False) session = cluster.connect() self.addCleanup(cluster.shutdown) self.addCleanup(nc_cluster.shutdown) nc_session.set_keyspace("test3rf") session.set_keyspace("test3rf") nc_session.execute( "CREATE TABLE IF NOT EXISTS compact_table (k int PRIMARY KEY, v1 int, v2 int) WITH COMPACT STORAGE;") for i in range(1, 5): nc_session.execute( "INSERT INTO compact_table (k, column1, v1, v2, value) VALUES " "({i}, 'a{i}', {i}, {i}, textAsBlob('b{i}'))".format(i=i)) nc_session.execute( "INSERT INTO compact_table (k, column1, v1, v2, value) VALUES " "({i}, 'a{i}{i}', {i}{i}, {i}{i}, textAsBlob('b{i}{i}'))".format(i=i)) nc_results = nc_session.execute("SELECT * FROM compact_table") self.assertEqual( set(nc_results.current_rows), {(1, u'a1', 11, 11, 'b1'), (1, u'a11', 11, 11, 'b11'), (2, u'a2', 22, 22, 'b2'), (2, u'a22', 22, 22, 'b22'), (3, u'a3', 33, 33, 'b3'), (3, u'a33', 33, 33, 'b33'), (4, u'a4', 44, 44, 'b4'), (4, u'a44', 44, 44, 'b44')}) results = session.execute("SELECT * FROM compact_table") self.assertEqual( set(results.current_rows), {(1, 11, 11), (2, 22, 22), (3, 33, 33), (4, 44, 44)}) def _assert_replica_queried(self, trace, only_replicas=True): queried_hosts = set() for row in trace.events: queried_hosts.add(row.source) if only_replicas: self.assertEqual(len(queried_hosts), 1, "The hosts queried where {}".format(queried_hosts)) else: self.assertGreater(len(queried_hosts), 1, "The host queried was {}".format(queried_hosts)) return queried_hosts def _check_trace(self, trace): self.assertIsNotNone(trace.request_type) self.assertIsNotNone(trace.duration) self.assertIsNotNone(trace.started_at) self.assertIsNotNone(trace.coordinator) self.assertIsNotNone(trace.events) class LocalHostAdressTranslator(AddressTranslator): def __init__(self, addr_map=None): self.addr_map = addr_map def translate(self, addr): new_addr = self.addr_map.get(addr) return new_addr @local class TestAddressTranslation(unittest.TestCase): def test_address_translator_basic(self): """ Test host address translation Uses a custom Address Translator to map all ip back to one. Validates AddressTranslator invocation by ensuring that only meta data associated with single host is populated @since 3.3 @jira_ticket PYTHON-69 @expected_result only one hosts' metadata will be populated @test_category metadata """ lh_ad = LocalHostAdressTranslator({'127.0.0.1': '127.0.0.1', '127.0.0.2': '127.0.0.1', '127.0.0.3': '127.0.0.1'}) c = TestCluster(address_translator=lh_ad) c.connect() self.assertEqual(len(c.metadata.all_hosts()), 1) c.shutdown() def test_address_translator_with_mixed_nodes(self): """ Test host address translation Uses a custom Address Translator to map ip's of non control_connection nodes to each other Validates AddressTranslator invocation by ensuring that metadata for mapped hosts is also mapped @since 3.3 @jira_ticket PYTHON-69 @expected_result metadata for crossed hosts will also be crossed @test_category metadata """ adder_map = {'127.0.0.1': '127.0.0.1', '127.0.0.2': '127.0.0.3', '127.0.0.3': '127.0.0.2'} lh_ad = LocalHostAdressTranslator(adder_map) c = TestCluster(address_translator=lh_ad) c.connect() for host in c.metadata.all_hosts(): self.assertEqual(adder_map.get(host.address), host.broadcast_address) c.shutdown() @local class ContextManagementTest(unittest.TestCase): load_balancing_policy = HostFilterPolicy( RoundRobinPolicy(), lambda host: host.address == CASSANDRA_IP ) cluster_kwargs = {'execution_profiles': {EXEC_PROFILE_DEFAULT: ExecutionProfile(load_balancing_policy= load_balancing_policy)}, 'schema_metadata_enabled': False, 'token_metadata_enabled': False} def test_no_connect(self): """ Test cluster context without connecting. @since 3.4 @jira_ticket PYTHON-521 @expected_result context should still be valid @test_category configuration """ with TestCluster() as cluster: self.assertFalse(cluster.is_shutdown) self.assertTrue(cluster.is_shutdown) def test_simple_nested(self): """ Test cluster and session contexts nested in one another. @since 3.4 @jira_ticket PYTHON-521 @expected_result cluster/session should be crated and shutdown appropriately. @test_category configuration """ with TestCluster(**self.cluster_kwargs) as cluster: with cluster.connect() as session: self.assertFalse(cluster.is_shutdown) self.assertFalse(session.is_shutdown) self.assertTrue(session.execute('select release_version from system.local')[0]) self.assertTrue(session.is_shutdown) self.assertTrue(cluster.is_shutdown) def test_cluster_no_session(self): """ Test cluster context without session context. @since 3.4 @jira_ticket PYTHON-521 @expected_result Session should be created correctly. Cluster should shutdown outside of context @test_category configuration """ with TestCluster(**self.cluster_kwargs) as cluster: session = cluster.connect() self.assertFalse(cluster.is_shutdown) self.assertFalse(session.is_shutdown) self.assertTrue(session.execute('select release_version from system.local')[0]) self.assertTrue(session.is_shutdown) self.assertTrue(cluster.is_shutdown) def test_session_no_cluster(self): """ Test session context without cluster context. @since 3.4 @jira_ticket PYTHON-521 @expected_result session should be created correctly. Session should shutdown correctly outside of context @test_category configuration """ cluster = TestCluster(**self.cluster_kwargs) unmanaged_session = cluster.connect() with cluster.connect() as session: self.assertFalse(cluster.is_shutdown) self.assertFalse(session.is_shutdown) self.assertFalse(unmanaged_session.is_shutdown) self.assertTrue(session.execute('select release_version from system.local')[0]) self.assertTrue(session.is_shutdown) self.assertFalse(cluster.is_shutdown) self.assertFalse(unmanaged_session.is_shutdown) unmanaged_session.shutdown() self.assertTrue(unmanaged_session.is_shutdown) self.assertFalse(cluster.is_shutdown) cluster.shutdown() self.assertTrue(cluster.is_shutdown) class HostStateTest(unittest.TestCase): def test_down_event_with_active_connection(self): """ Test to ensure that on down calls to clusters with connections still active don't result in a host being marked down. The second part of the test kills the connection then invokes on_down, and ensures the state changes for host's metadata. @since 3.7 @jira_ticket PYTHON-498 @expected_result host should never be toggled down while a connection is active. @test_category connection """ with TestCluster() as cluster: session = cluster.connect(wait_for_all_pools=True) random_host = cluster.metadata.all_hosts()[0] cluster.on_down(random_host, False) for _ in range(10): new_host = cluster.metadata.all_hosts()[0] self.assertTrue(new_host.is_up, "Host was not up on iteration {0}".format(_)) time.sleep(.01) pool = session._pools.get(random_host) pool.shutdown() cluster.on_down(random_host, False) was_marked_down = False for _ in range(20): new_host = cluster.metadata.all_hosts()[0] if not new_host.is_up: was_marked_down = True break time.sleep(.01) self.assertTrue(was_marked_down) @local class DontPrepareOnIgnoredHostsTest(unittest.TestCase): ignored_addresses = ['127.0.0.3'] ignore_node_3_policy = IgnoredHostPolicy(ignored_addresses) def test_prepare_on_ignored_hosts(self): cluster = TestCluster( execution_profiles={EXEC_PROFILE_DEFAULT: ExecutionProfile(load_balancing_policy=self.ignore_node_3_policy)} ) session = cluster.connect() cluster.reprepare_on_up, cluster.prepare_on_all_hosts = True, False hosts = cluster.metadata.all_hosts() session.execute("CREATE KEYSPACE clustertests " "WITH replication = " "{'class': 'SimpleStrategy', 'replication_factor': '1'}") session.execute("CREATE TABLE clustertests.tab (a text, PRIMARY KEY (a))") # assign to an unused variable so cluster._prepared_statements retains # reference _ = session.prepare("INSERT INTO clustertests.tab (a) VALUES ('a')") # noqa cluster.connection_factory = Mock(wraps=cluster.connection_factory) unignored_address = '127.0.0.1' unignored_host = next(h for h in hosts if h.address == unignored_address) ignored_host = next(h for h in hosts if h.address in self.ignored_addresses) unignored_host.is_up = ignored_host.is_up = False cluster.on_up(unignored_host) cluster.on_up(ignored_host) # the length of mock_calls will vary, but all should use the unignored # address for c in cluster.connection_factory.mock_calls: # PYTHON-1287 # # Cluster._prepare_all_queries() will call connection_factory _without_ the # on_orphaned_stream_released arg introduced in commit # 387150acc365b6cf1daaee58c62db13e4929099a. The reconnect handler for the # downed node _will_ add this arg when it tries to rebuild it's conn pool, and # whether this occurs while running this test amounts to a race condition. So # to cover this case we assert one of two call styles here... the key is that # the _only_ address we should see is the unignored_address. self.assertTrue( \ c == call(DefaultEndPoint(unignored_address)) or \ c == call(DefaultEndPoint(unignored_address), on_orphaned_stream_released=ANY)) cluster.shutdown() @protocolv6 class BetaProtocolTest(unittest.TestCase): @protocolv6 def test_invalid_protocol_version_beta_option(self): """ Test cluster connection with protocol v6 and beta flag not set @since 3.7.0 @jira_ticket PYTHON-614, PYTHON-1232 @expected_result client shouldn't connect with V6 and no beta flag set @test_category connection """ cluster = TestCluster(protocol_version=cassandra.ProtocolVersion.V6, allow_beta_protocol_version=False) try: with self.assertRaises(NoHostAvailable): cluster.connect() except Exception as e: self.fail("Unexpected error encountered {0}".format(e.message)) @protocolv6 def test_valid_protocol_version_beta_options_connect(self): """ Test cluster connection with protocol version 5 and beta flag set @since 3.7.0 @jira_ticket PYTHON-614, PYTHON-1232 @expected_result client should connect with protocol v6 and beta flag set. @test_category connection """ cluster = Cluster(protocol_version=cassandra.ProtocolVersion.V6, allow_beta_protocol_version=True) session = cluster.connect() self.assertEqual(cluster.protocol_version, cassandra.ProtocolVersion.V6) self.assertTrue(session.execute("select release_version from system.local")[0]) cluster.shutdown() class DeprecationWarningTest(unittest.TestCase): def test_deprecation_warnings_legacy_parameters(self): """ Tests the deprecation warning has been added when using legacy parameters @since 3.13 @jira_ticket PYTHON-877 @expected_result the deprecation warning is emitted @test_category logs """ with warnings.catch_warnings(record=True) as w: TestCluster(load_balancing_policy=RoundRobinPolicy()) self.assertEqual(len(w), 1) self.assertIn("Legacy execution parameters will be removed in 4.0. Consider using execution profiles.", str(w[0].message)) def test_deprecation_warnings_meta_refreshed(self): """ Tests the deprecation warning has been added when enabling metadata refreshment @since 3.13 @jira_ticket PYTHON-890 @expected_result the deprecation warning is emitted @test_category logs """ with warnings.catch_warnings(record=True) as w: cluster = TestCluster() cluster.set_meta_refresh_enabled(True) self.assertEqual(len(w), 1) self.assertIn("Cluster.set_meta_refresh_enabled is deprecated and will be removed in 4.0.", str(w[0].message)) def test_deprecation_warning_default_consistency_level(self): """ Tests the deprecation warning has been added when enabling session the default consistency level to session @since 3.14 @jira_ticket PYTHON-935 @expected_result the deprecation warning is emitted @test_category logs """ with warnings.catch_warnings(record=True) as w: cluster = TestCluster() session = cluster.connect() session.default_consistency_level = ConsistencyLevel.ONE self.assertEqual(len(w), 1) self.assertIn("Setting the consistency level at the session level will be removed in 4.0", str(w[0].message))