| # Licensed to the Apache Software Foundation (ASF) under one |
| # or more contributor license agreements. See the NOTICE file |
| # distributed with this work for additional information |
| # regarding copyright ownership. The ASF licenses this file |
| # to you 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)) |
