Merge branch 'master' into ngdg_master_ft

Author: aboudreault

tests/integration/__init__.py

@@ -126,7 +126,7 @@ def _get_cass_version_from_dse(dse_version):
             cass_ver = '4.0.0.2349'
         else:
             cass_ver = '4.0.0.' + ''.join(dse_version.split('.'))
-    elif dse_version.startswith('6.7'):
+    elif Version(dse_version) >= Version('6.7'):
         if dse_version == '6.7.0':
             cass_ver = "4.0.0.67"
         else:
@@ -490,7 +490,13 @@ def use_cluster(cluster_name, nodes, ipformat=None, start=True, workloads=None,
         if CCM_CLUSTER:
             log.debug("Using external CCM cluster {0}".format(CCM_CLUSTER.name))
         else:
-            log.debug("Using unnamed external cluster")
+            ccm_path = os.getenv("CCM_PATH", None)
+            ccm_name = os.getenv("CCM_NAME", None)
+            if ccm_path and ccm_name:
+                CCM_CLUSTER = CCMClusterFactory.load(ccm_path, ccm_name)
+                log.debug("Using external CCM cluster {0}".format(CCM_CLUSTER.name))
+            else:
+                log.debug("Using unnamed external cluster")
         if set_keyspace and start:
             setup_keyspace(ipformat=ipformat, wait=False)
         return

tests/integration/cqlengine/query/test_named.py

@@ -335,7 +335,7 @@ def test_named_table_with_mv(self):
                         SELECT * FROM {0}.scores
                         WHERE game IS NOT NULL AND score IS NOT NULL AND user IS NOT NULL AND year IS NOT NULL AND month IS NOT NULL AND day IS NOT NULL
                         PRIMARY KEY (game, score, user, year, month, day)
-                        WITH CLUSTERING ORDER BY (score DESC)""".format(ks)
+                        WITH CLUSTERING ORDER BY (score DESC, user DESC, year DESC, month DESC, day DESC)""".format(ks)
 
         self.session.execute(create_mv_alltime)
 

tests/integration/long/test_failure_types.py

@@ -16,6 +16,8 @@
 import sys
 import traceback
 import time
+
+from ccmlib.dse_cluster import DseCluster
 from mock import Mock
 
 from cassandra.policies import HostFilterPolicy, RoundRobinPolicy
@@ -29,7 +31,7 @@
 from tests.integration import (
     use_singledc, PROTOCOL_VERSION, get_cluster, setup_keyspace, remove_cluster,
     get_node, start_cluster_wait_for_up, requiresmallclockgranularity,
-)
+    local)
 
 
 try:
@@ -40,6 +42,7 @@
 log = logging.getLogger(__name__)
 
 
+@local
 def setup_module():
     """
     We need some custom setup for this module. All unit tests in this module
@@ -52,7 +55,7 @@ def setup_module():
         use_singledc(start=False)
         ccm_cluster = get_cluster()
         ccm_cluster.stop()
-        config_options = {'tombstone_failure_threshold': 2000, 'tombstone_warn_threshold': 1000}
+        config_options = {'guardrails.tombstone_failure_threshold': 2000, 'guardrails.tombstone_warn_threshold': 1000}
         ccm_cluster.set_configuration_options(config_options)
         start_cluster_wait_for_up(ccm_cluster)
         setup_keyspace()
@@ -252,7 +255,7 @@ def test_tombstone_overflow_read_failure(self):
         parameters = [(x,) for x in range(3000)]
         self.execute_concurrent_args_helper(self.session, statement, parameters)
 
-        statement = self.session.prepare("DELETE v1 FROM test3rf.test2 WHERE k = 1 AND v0 =?")
+        statement = self.session.prepare("DELETE FROM test3rf.test2 WHERE k = 1 AND v0 =?")
         parameters = [(x,) for x in range(2001)]
         self.execute_concurrent_args_helper(self.session, statement, parameters)
 

tests/integration/simulacron/test_cluster.py

@@ -16,14 +16,20 @@
 except ImportError:
     import unittest  # noqa
 
-from tests.integration.simulacron import SimulacronCluster
-from tests.integration import (requiressimulacron, PROTOCOL_VERSION)
-from tests.integration.simulacron.utils import prime_query
+import logging
+from packaging.version import Version
+
+import cassandra
+from tests.integration.simulacron import SimulacronCluster, SimulacronBase
+from tests.integration import (requiressimulacron, PROTOCOL_VERSION, DSE_VERSION, MockLoggingHandler)
+from tests.integration.simulacron.utils import prime_query, start_and_prime_singledc
 
 from cassandra import (WriteTimeout, WriteType,
                        ConsistencyLevel, UnresolvableContactPoints)
-from cassandra.cluster import Cluster
+from cassandra.cluster import Cluster, ControlConnection
+
 
+PROTOCOL_VERSION = min(4, PROTOCOL_VERSION if (DSE_VERSION is None or DSE_VERSION >= Version('5.0')) else 3)
 
 @requiressimulacron
 class ClusterTests(SimulacronCluster):
@@ -78,3 +84,30 @@ def test_connection_with_only_unresolvable_contact_points(self):
             self.cluster = Cluster(['dns.invalid'],
                                    protocol_version=PROTOCOL_VERSION,
                                    compression=False)
+
+
+@requiressimulacron
+class DuplicateRpcTest(SimulacronCluster):
+    connect = False
+
+    def test_duplicate(self):
+        mock_handler = MockLoggingHandler()
+        logger = logging.getLogger(cassandra.cluster.__name__)
+        logger.addHandler(mock_handler)
+        address_column = "native_transport_address" if DSE_VERSION and DSE_VERSION > Version("6.0") else "rpc_address"
+        rows = [
+            {"peer": "127.0.0.1", "data_center": "dc", "host_id": "dontcare1", "rack": "rack1",
+             "release_version": "3.11.4", address_column: "127.0.0.1", "schema_version": "dontcare", "tokens": "1"},
+            {"peer": "127.0.0.2", "data_center": "dc", "host_id": "dontcare2", "rack": "rack1",
+             "release_version": "3.11.4", address_column: "127.0.0.2", "schema_version": "dontcare", "tokens": "2"},
+        ]
+        prime_query(ControlConnection._SELECT_PEERS, rows=rows)
+
+        cluster = Cluster(protocol_version=PROTOCOL_VERSION, compression=False)
+        session = cluster.connect(wait_for_all_pools=True)
+
+        warnings = mock_handler.messages.get("warning")
+        self.assertEqual(len(warnings), 1)
+        self.assertTrue('multiple hosts with the same endpoint' in warnings[0])
+        logger.removeHandler(mock_handler)
+        cluster.shutdown()

tests/integration/standard/test_cluster.py

@@ -27,7 +27,7 @@
 from packaging.version import Version
 
 import cassandra
-from cassandra.cluster import Cluster, NoHostAvailable, ExecutionProfile, EXEC_PROFILE_DEFAULT
+from cassandra.cluster import Cluster, NoHostAvailable, ExecutionProfile, EXEC_PROFILE_DEFAULT, ControlConnection
 from cassandra.concurrent import execute_concurrent
 from cassandra.policies import (RoundRobinPolicy, ExponentialReconnectionPolicy,
                                 RetryPolicy, SimpleConvictionPolicy, HostDistance,
@@ -502,79 +502,70 @@ def test_refresh_schema_type(self):
     @local
     @notwindows
     def test_refresh_schema_no_wait(self):
-        contact_points = [CASSANDRA_IP]
-        with Cluster(protocol_version=PROTOCOL_VERSION, max_schema_agreement_wait=10,
-                     contact_points=contact_points,
-                     execution_profiles=
-                     {EXEC_PROFILE_DEFAULT: ExecutionProfile(load_balancing_policy=
-                             HostFilterPolicy(
-                                 RoundRobinPolicy(), lambda host: host.address == CASSANDRA_IP
-                             ))}) as cluster:
-            session = cluster.connect()
-
-            schema_ver = session.execute("SELECT schema_version FROM system.local WHERE key='local'")[0][0]
-            new_schema_ver = uuid4()
-            session.execute("UPDATE system.local SET schema_version=%s WHERE key='local'", (new_schema_ver,))
-
-            try:
-                agreement_timeout = 1
-
-                # cluster agreement wait exceeded
-                c = Cluster(protocol_version=PROTOCOL_VERSION, 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.assertRaisesRegexp(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 = Cluster(protocol_version=PROTOCOL_VERSION, 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.assertRaisesRegexp(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()
-            finally:
-                # TODO once fixed this connect call
-                session = cluster.connect()
-                session.execute("UPDATE system.local SET schema_version=%s WHERE key='local'", (schema_ver,))
-
+        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 = Cluster(protocol_version=PROTOCOL_VERSION, 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.assertRaisesRegexp(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 = Cluster(protocol_version=PROTOCOL_VERSION, 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.assertRaisesRegexp(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):
         """
@@ -1480,52 +1471,6 @@ def test_prepare_on_ignored_hosts(self):
         cluster.shutdown()
 
 
-@local
-class DuplicateRpcTest(unittest.TestCase):
-
-    load_balancing_policy = HostFilterPolicy(RoundRobinPolicy(),
-                                             lambda host: host.address == "127.0.0.1")
-
-    def setUp(self):
-        self.cluster = Cluster(protocol_version=PROTOCOL_VERSION,
-                               execution_profiles={EXEC_PROFILE_DEFAULT: ExecutionProfile(load_balancing_policy=self.load_balancing_policy)})
-        self.session = self.cluster.connect()
-
-        self.address_column = "native_transport_address" if DSE_VERSION and DSE_VERSION >= Version("6.0") else "rpc_address"
-        self.session.execute("UPDATE system.peers SET {} = '127.0.0.1' WHERE peer='127.0.0.2'".
-                             format(self.address_column))
-
-    def tearDown(self):
-        self.session.execute("UPDATE system.peers SET {} = '127.0.0.2' WHERE peer='127.0.0.2'".
-                             format(self.address_column))
-        self.cluster.shutdown()
-
-    def test_duplicate(self):
-        """
-        Test duplicate RPC addresses.
-
-        Modifies the system.peers table to make hosts have the same rpc address. Ensures such hosts are filtered out and a message is logged
-
-        @since 3.4
-        @jira_ticket PYTHON-366
-        @expected_result only one hosts' metadata will be populated
-
-        @test_category metadata
-        """
-        mock_handler = MockLoggingHandler()
-        logger = logging.getLogger(cassandra.cluster.__name__)
-        logger.addHandler(mock_handler)
-        test_cluster = Cluster(protocol_version=PROTOCOL_VERSION,
-                               execution_profiles={EXEC_PROFILE_DEFAULT: ExecutionProfile(load_balancing_policy=self.load_balancing_policy)})
-
-        test_cluster.connect()
-        warnings = mock_handler.messages.get("warning")
-        self.assertEqual(len(warnings), 1)
-        self.assertTrue('multiple' in warnings[0])
-        logger.removeHandler(mock_handler)
-        test_cluster.shutdown()
-
-
 @protocolv5
 class BetaProtocolTest(unittest.TestCase):
 

tests/integration/standard/test_control_connection.py

@@ -14,6 +14,7 @@
 #
 #
 #
+from cassandra import InvalidRequest
 
 try:
     import unittest2 as unittest
@@ -43,7 +44,7 @@ def setUp(self):
     def tearDown(self):
         try:
             self.session.execute("DROP KEYSPACE keyspacetodrop ")
-        except (ConfigurationException):
+        except (ConfigurationException, InvalidRequest):
             # we already removed the keyspace.
             pass
         self.cluster.shutdown()