ci: run integration tests against the latest Cassandra

.github/workflows/integration-tests.yml

@@ -1,5 +1,8 @@
 name: Integration tests
 
+permissions:
+  contents: read
+
 on:
   push:
     branches:
@@ -94,3 +97,56 @@ jobs:
           export PYTHON_GIL=0
         fi
         uv run pytest -v tests/integration/standard/ tests/integration/cqlengine/
+
+  tests-cassandra:
+    name: test cassandra ${{ matrix.event_loop_manager }} (${{ matrix.python-version }})
+    if: "!contains(github.event.pull_request.labels.*.name, 'disable-integration-tests')"
+    runs-on: ubuntu-24.04
+    timeout-minutes: 90
+    strategy:
+      fail-fast: false
+      matrix:
+        python-version: ["3.15"]
+        event_loop_manager: ["libev", "asyncio"]
+
+    steps:
+    - uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v7.0.0
+      with:
+        persist-credentials: false
+
+    # Cassandra 5.0 needs JDK 11+
+    - name: Set up JDK 11
+      uses: actions/setup-java@b843ebf9b913146b38da2eaa7aeddf76074ae1ad # v6.1.0
+      with:
+        java-version: 11
+        distribution: temurin
+
+    - name: Install libev
+      run: sudo apt-get install libev4 libev-dev
+
+    - name: Install uv
+      uses: astral-sh/setup-uv@11f9893b081a58869d3b5fccaea48c9e9e46f990 # v8.3.2
+      with:
+        python-version: ${{ matrix.python-version }}
+        enable-cache: true
+
+    - name: Build driver
+      run: uv sync
+
+    - name: Get latest Cassandra version
+      run: |
+        version=$(git ls-remote --tags --refs https://github.com/apache/cassandra 'cassandra-*' \
+          | sed 's|.*refs/tags/cassandra-||' | grep -E '^[0-9]+\.[0-9]+\.[0-9]+$' | sort -V | tail -1)
+        echo "CASSANDRA_VERSION=$version" >> "$GITHUB_ENV"
+
+    - name: Cache Cassandra download
+      uses: actions/cache@55cc8345863c7cc4c66a329aec7e433d2d1c52a9 # v6.1.0
+      with:
+        path: ~/.ccm/repository
+        key: cassandra-${{ env.CASSANDRA_VERSION }}-${{ runner.os }}
+
+    # no PROTOCOL_VERSION: use the harness default, v5 on Cassandra 4.0+, the only job covering v5
+    - name: Test with pytest
+      env:
+        EVENT_LOOP_MANAGER: ${{ matrix.event_loop_manager }}
+      run: uv run pytest -v tests/integration/standard/

tests/integration/__init__.py

@@ -146,6 +146,17 @@ def cmd_line_args_to_dict(env_var):
 CASSANDRA_VERSION = Version(mcv_string) if mcv_string else cassandra_version
 CCM_VERSION = mcv_string if mcv_string else cv_string
 
+if not SCYLLA_VERSION and CCMClusterFactory and CASSANDRA_VERSION.release[:2] >= (5, 0):
+    # scylla-ccm runs every Cassandra node with Java 8 (Node.get_env), but Cassandra 5.0 needs Java 11+.
+    # Use JAVA_HOME from the environment instead. Drop this once ccm picks the JDK by Cassandra version.
+    from ccmlib.node import Node as CCMNode
+
+    def _get_env_with_java_home(self):
+        update_conf = not self._Node__conf_updated
+        self._Node__conf_updated = True
+        return common.make_cassandra_env(self.get_install_dir(), self.get_path(), update_conf)
+    CCMNode.get_env = _get_env_with_java_home
+
 CASSANDRA_IP = os.getenv('CLUSTER_IP', '127.0.0.1')
 CASSANDRA_DIR = os.getenv('CASSANDRA_DIR', None)
 
@@ -309,6 +320,10 @@ def _has_vector_type():
     _has_vector_type(),
     'Cassandra >= 5.0 or Scylla >= 2025.4 required')
 xfail_scylla = lambda reason, *args, **kwargs: pytest.mark.xfail(SCYLLA_VERSION is not None, reason=reason, *args, **kwargs)
+scylla_only = pytest.mark.skipif(SCYLLA_VERSION is None, reason='Scylla only test')
+# system.local is read with a fixed column list since #528, without broadcast_port / rpc_port
+xfail_cassandra_local_ports = pytest.mark.xfail(SCYLLA_VERSION is None and CASSANDRA_VERSION >= Version('4-a'), strict=True,
+                                                reason='broadcast_port / broadcast_rpc_port of the local host are not read')
 incorrect_test = lambda reason='This test seems to be incorrect and should be fixed', *args, **kwargs: pytest.mark.xfail(reason=reason, *args, **kwargs)
 
 pypy = unittest.skipUnless(platform.python_implementation() == "PyPy", "Test is skipped unless it's on PyPy")
@@ -1028,5 +1043,6 @@ def _get_config_val(self, k, v):
         return v
 
     def set_configuration_options(self, values=None, *args, **kwargs):
-        new_values = {self._get_config_key(k, str(v)):self._get_config_val(k, str(v)) for (k,v) in values.items()}
+        # keep v's type: str(True) is written as the YAML string 'True', which Cassandra refuses to start with
+        new_values = {self._get_config_key(k, v): self._get_config_val(k, v) for (k, v) in values.items()}
         super(Cassandra41CCMCluster, self).set_configuration_options(values=new_values, *args, **kwargs)

tests/integration/standard/test_authentication.py

@@ -22,7 +22,7 @@
 from cassandra.auth import PlainTextAuthProvider, SASLClient, SaslAuthProvider
 
 from tests.integration import use_singledc, get_cluster, remove_cluster, PROTOCOL_VERSION, \
-    CASSANDRA_IP, CASSANDRA_VERSION, start_cluster_wait_for_up, TestCluster
+    CASSANDRA_IP, CASSANDRA_VERSION, start_cluster_wait_for_up, TestCluster, SCYLLA_VERSION
 from tests.integration.util import assert_quiescent_pool_state
 
 import unittest
@@ -47,9 +47,11 @@ def setup_module():
         config_options = {
             'authenticator': 'PasswordAuthenticator',
             'authorizer': 'CassandraAuthorizer',
-            'auth_superuser_name': 'cassandra',
-            'auth_superuser_salted_password': '$6$x7IFjiX5VCpvNiFk$2IfjTvSyGL7zerpV.wbY7mJjaRCrJ/68dtT3UpT.sSmNYz1bPjtn3mH.kJKFvaZ2T4SbVeBijjmwGjcb83LlV/'
         }
+        if SCYLLA_VERSION:
+            # Scylla only options, Cassandra creates the cassandra/cassandra superuser by default
+            config_options['auth_superuser_name'] = 'cassandra'
+            config_options['auth_superuser_salted_password'] = '$6$x7IFjiX5VCpvNiFk$2IfjTvSyGL7zerpV.wbY7mJjaRCrJ/68dtT3UpT.sSmNYz1bPjtn3mH.kJKFvaZ2T4SbVeBijjmwGjcb83LlV/'
         ccm_cluster.set_configuration_options(config_options)
         log.debug("Starting ccm test cluster with %s", config_options)
         start_cluster_wait_for_up(ccm_cluster)

tests/integration/standard/test_client_routes.py

@@ -47,7 +47,10 @@
     use_cluster,
     wait_for_node_socket,
     skip_scylla_version_lt,
+    scylla_only,
 )
+
+pytestmark = scylla_only  # system.client_routes and CLIENT_ROUTES_CHANGE are Scylla only
 from tests.tcp_proxy import TcpProxy
 from tests.util import wait_until_not_raised
 

tests/integration/standard/test_cluster.py

@@ -43,7 +43,7 @@
 from tests.integration import use_cluster, get_server_versions, CASSANDRA_VERSION, \
     execute_until_pass, execute_with_long_wait_retry, get_node, MockLoggingHandler, get_unsupported_lower_protocol, \
     get_unsupported_upper_protocol, local, CASSANDRA_IP, greaterthanorequalcass30, \
-    lessthanorequalcass40, TestCluster, PROTOCOL_VERSION, incorrect_test
+    lessthanorequalcass40, TestCluster, PROTOCOL_VERSION, incorrect_test, SCYLLA_VERSION
 from tests.integration.util import assert_quiescent_pool_state
 from tests.util import assertListEqual
 import sys
@@ -1254,7 +1254,8 @@ def test_replicas_are_queried(self):
             session.execute("DROP KEYSPACE IF EXISTS {}".format(ks_name))
             session.execute(
                 "CREATE KEYSPACE {} WITH replication = {{'class': 'NetworkTopologyStrategy', "
-                "'replication_factor': '1'}} AND tablets = {{'enabled': false}}".format(ks_name)
+                "'replication_factor': '1'}}".format(ks_name)
+                + (" AND tablets = {'enabled': false}" if SCYLLA_VERSION else "")
             )
             session.execute('''
                     CREATE TABLE {}.table_with_big_key (

tests/integration/standard/test_control_connection.py

@@ -24,7 +24,7 @@
 
 from cassandra.protocol import ConfigurationException
 from tests.integration import use_singledc, PROTOCOL_VERSION, TestCluster, greaterthanorequalcass40, \
-    xfail_scylla_version_lt
+    xfail_scylla_version_lt, scylla_only, xfail_cassandra_local_ports
 from tests.integration.datatype_utils import update_datatypes
 
 
@@ -109,6 +109,7 @@ def test_get_control_connection_host(self):
         assert new_host1 != new_host2
 
     @greaterthanorequalcass40
+    @xfail_cassandra_local_ports
     def test_control_connection_port_discovery(self):
         """
         Test to validate that the correct port is discovered when peersV2 is used (C* 4.0+).
@@ -134,6 +135,7 @@ def test_control_connection_port_discovery(self):
             assert 9042 == host.broadcast_rpc_port
             assert 7000 == host.broadcast_port
 
+    @scylla_only
     @xfail_scylla_version_lt(reason='scylladb/scylladb#26992 - system.client_routes is not yet supported',
                              scylla_version="2026.1.0")
     def test_client_routes_change_event(self):

tests/integration/standard/test_custom_protocol_handler.py

@@ -134,7 +134,6 @@ def test_protocol_divergence_v4_fail_by_flag_uses_int(self):
         self._protocol_divergence_fail_by_flag_uses_int(ProtocolVersion.V4, uses_int_query_flag=False,
                                                         int_flag=True)
 
-    @unittest.expectedFailure
     @greaterthanorequalcass40
     def test_protocol_v5_uses_flag_int(self):
         """
@@ -148,7 +147,6 @@ def test_protocol_v5_uses_flag_int(self):
         self._protocol_divergence_fail_by_flag_uses_int(ProtocolVersion.V5, uses_int_query_flag=True, beta=True,
                                                         int_flag=True)
 
-    @unittest.expectedFailure
     @greaterthanorequalcass40
     def test_protocol_divergence_v5_fail_by_flag_uses_int(self):
         """

tests/integration/standard/test_driver_config.py

@@ -24,7 +24,7 @@
                                 ConstantSpeculativeExecutionPolicy, FallthroughRetryPolicy)
 from tests.driver_config_schema import validate_report
 from tests.integration import (TestCluster, get_client_options, use_single_node, remove_cluster,
-                               xfail_scylla_version_lt)
+                               xfail_scylla_version_lt, scylla_only)
 
 
 def setup_module():
@@ -313,6 +313,7 @@ def test_the_reported_configuration_survives_the_round_trip(self):
         finally:
             cluster.shutdown()
 
+    @scylla_only
     def test_the_server_side_timeout_is_reported_against_scylla(self):
         """
         USING TIMEOUT is a ScyllaDB extension, so this key is reported only on a

tests/integration/standard/test_exit_without_shutdown.py

@@ -17,7 +17,10 @@
 import sys
 
 from tests.integration import use_cluster, get_cluster, get_node, start_cluster_wait_for_up, \
-    libevtest, PROTOCOL_VERSION, CASSANDRA_IP
+    libevtest, PROTOCOL_VERSION, CASSANDRA_IP, scylla_only
+
+# the many-shard node comes from ScyllaNode.set_smp(); ccm's Cassandra nodes have no shards
+pytestmark = scylla_only
 
 # GH-1133: `cqlsh -e` drops its Session without shutdown(). When a GC pass
 # collects it, Session.__del__ closes all its connections on whatever thread

tests/integration/standard/test_ip_change.py

@@ -5,7 +5,7 @@
 from cassandra.cluster import ExecutionProfile
 from cassandra.policies import WhiteListRoundRobinPolicy
 
-from tests.integration import use_cluster, get_node, get_cluster, local, TestCluster
+from tests.integration import use_cluster, get_node, get_cluster, local, TestCluster, SCYLLA_VERSION
 from tests.util import wait_until_not_raised
 
 LOGGER = logging.getLogger(__name__)
@@ -46,7 +46,10 @@ def test_change_address_during_live_session(self):
         LOGGER.debug("Change IP address for node3")
         ip_prefix = get_cluster().get_ipprefix()
         new_ip = f'{ip_prefix}33'
-        node3.set_configuration_options(values={'listen_address': new_ip, 'rpc_address': new_ip, 'api_address': new_ip})
+        new_addresses = {'listen_address': new_ip, 'rpc_address': new_ip}
+        if SCYLLA_VERSION:
+            new_addresses['api_address'] = new_ip
+        node3.set_configuration_options(values=new_addresses)
         node3.network_interfaces = {k: (new_ip, v[1]) for k, v in node3.network_interfaces.items()}
         LOGGER.debug(f"Start node3 again with ip address {new_ip}")
         node3.start(wait_for_binary_proto=True)

tests/integration/standard/test_metadata.py

@@ -37,7 +37,8 @@
                                get_supported_protocol_versions, greaterthancass20,
                                greaterthancass21, lessthancass40,
                                TestCluster, requires_java_udf, requires_composite_type,
-                               requires_collection_indexes, SCYLLA_VERSION, xfail_scylla, requirescompactstorage, get_tablets_disabled_ddl_suffix, execute_with_long_wait_retry)
+                               requires_collection_indexes, SCYLLA_VERSION, xfail_scylla, requirescompactstorage, get_tablets_disabled_ddl_suffix, execute_with_long_wait_retry,
+                               scylla_only, xfail_cassandra_local_ports)
 
 from tests.util import wait_until, assertRegex, assertDictEqual, assertListEqual, assert_startswith_diff
 
@@ -50,6 +51,7 @@ def setup_module():
 
 class HostMetaDataTests(BasicExistingKeyspaceUnitTestCase):
     @local
+    @xfail_cassandra_local_ports
     def test_host_addresses(self):
         """
         Check to ensure that the broadcast_address, broadcast_rpc_address,
@@ -1267,7 +1269,6 @@ def test_already_exists_exceptions(self):
         cluster.shutdown()
 
     @local
-    @pytest.mark.xfail(reason='AssertionError: \'RAC1\' != \'r1\' - probably a bug in driver or in Scylla')
     def test_replicas(self):
         """
         Ensure cluster.metadata.get_replicas return correctly when not attached to keyspace
@@ -1283,7 +1284,7 @@ def test_replicas(self):
         assert list(cluster.metadata.get_replicas('test3rf', b'key')) != []
         host = list(cluster.metadata.get_replicas('test3rf', b'key'))[0]
         assert host.datacenter == 'dc1'
-        assert host.rack == 'r1'
+        assert host.rack == ('RAC1' if SCYLLA_VERSION else 'r1')
         cluster.shutdown()
 
     def test_token_map(self):
@@ -1323,6 +1324,8 @@ def test_token(self):
         cluster.shutdown()
 
 
+# metadata timeout (USING TIMEOUT) is Scylla only
+@scylla_only
 class TestMetadataTimeout:
     """
     Test of TokenMap creation and other behavior.

tests/integration/standard/test_metrics.py

@@ -15,7 +15,7 @@
 import time
 
 from cassandra.connection import ConnectionShutdown
-from cassandra.policies import HostFilterPolicy, RoundRobinPolicy, FallthroughRetryPolicy
+from cassandra.policies import HostFilterPolicy, RoundRobinPolicy, FallthroughRetryPolicy, ConstantReconnectionPolicy
 
 import unittest
 
@@ -41,7 +41,10 @@ class MetricsTests(unittest.TestCase):
 
     def setUp(self):
         contact_point = ['127.0.0.2']
+        # constant reconnects: test_connection_error restarts the whole cluster, and the default
+        # exponential backoff can push the next reconnect past its wait for the cluster
         self.cluster = TestCluster(contact_points=contact_point, metrics_enabled=True,
+                                   reconnection_policy=ConstantReconnectionPolicy(1),
                                    execution_profiles=
                                    {EXEC_PROFILE_DEFAULT:
                                        ExecutionProfile(
@@ -79,7 +82,7 @@ def test_connection_error(self):
             # Wait for the cluster to come back up for the next test
             wait_until_not_raised(
                 lambda: self.session.execute("SELECT key FROM system.local WHERE key='local'"),
-                delay=0.5, max_attempts=30)
+                delay=0.5, max_attempts=120)
 
         assert self.cluster.metrics.stats.connection_errors > 0
 

tests/integration/standard/test_prepared_statements.py

@@ -24,7 +24,7 @@
 from cassandra import ConsistencyLevel, ProtocolVersion
 from cassandra.query import PreparedStatement, UNSET_VALUE
 from tests.integration import (get_server_versions, greaterthanorequalcass40,
-    BasicSharedKeyspaceUnitTestCase)
+    BasicSharedKeyspaceUnitTestCase, scylla_only)
 
 import logging
 import pytest
@@ -406,6 +406,7 @@ def test_raise_error_on_prepared_statement_execution_dropped_table(self):
         with pytest.raises(InvalidRequest):
             self.session.execute(prepared, [0])
 
+    @scylla_only  # LWT flag in PREPARED metadata is a Scylla protocol extension
     def test_recognize_lwt_query(self):
         self.session.execute("CREATE TABLE IF NOT EXISTS preparedtests.bound_statement_test (a int PRIMARY KEY, b int)")
         # Prepare a non-LWT statement

tests/integration/standard/test_query.py

@@ -1020,6 +1020,9 @@ def test_no_connection_refused_on_timeout(self):
                         # actually returned a WriteTimeout/WriteFailure.
                         received_client_timeout = True
                     continue
+                if exception_type == "ErrorMessage" and result.code == 0x1700:
+                    # Cassandra 4.0+ CAS_WRITE_UNKNOWN: "CAS operation result is unknown"
+                    continue
 
                 pytest.fail("Unexpected exception %s: %s" % (exception_type, result))
 
@@ -1681,6 +1684,8 @@ def test_reprepare_after_host_is_down(self):
             wait_until(
                 lambda: mock_handler.get_message_count('debug', 'Preparing all known prepared statements') >= 1,
                 delay=0.5, max_attempts=20)
+            # statements are prepared while the pool opens, before the host is marked up
+            wait_until(lambda: self.cluster.metadata.get_host('127.0.0.1').is_up, delay=0.5, max_attempts=20)
 
             results = self.session.execute(prepared_statement, (1,), execution_profile="only_first")
             assert results.one() == (1, )

tests/integration/standard/test_rack_aware_policy.py

@@ -4,13 +4,16 @@
 from cassandra.cluster import Cluster
 from cassandra.policies import ConstantReconnectionPolicy, RackAwareRoundRobinPolicy
 
-from tests.integration import PROTOCOL_VERSION, get_cluster, use_multidc
+from tests.integration import PROTOCOL_VERSION, get_cluster, use_multidc, scylla_only
 
 LOGGER = logging.getLogger(__name__)
 
 def setup_module():
     use_multidc({'DC1': {'RC1': 2, 'RC2': 2}, 'DC2': {'RC1': 3}})
 
+# Cassandra fails to start this topology: "Token allocation failed: the number of racks 2 in datacenter DC1
+# is lower than its replication factor 3"
+@scylla_only
 class RackAwareRoundRobinPolicyTests(unittest.TestCase):
     @classmethod
     def setup_class(cls):

tests/integration/standard/test_rate_limit_exceeded.py

@@ -4,14 +4,15 @@
 from cassandra.cluster import Cluster
 from cassandra.policies import ConstantReconnectionPolicy, RoundRobinPolicy, TokenAwarePolicy
 
-from tests.integration import PROTOCOL_VERSION, use_singledc
+from tests.integration import PROTOCOL_VERSION, use_singledc, scylla_only
 import pytest
 
 LOGGER = logging.getLogger(__name__)
 
 def setup_module():
     use_singledc()
 
+@scylla_only
 class TestRateLimitExceededException(unittest.TestCase):
     @classmethod
     def setup_class(cls):