Fix cqlengine tests

.github/workflows/integration-tests-python2.yml

@@ -21,5 +21,5 @@ jobs:
 
     - name: Test with pytest
       run: |
-        ./ci/run_integration_test.sh tests/integration/standard/test_authentication.py tests/integration/standard/test_cluster.py tests/integration/standard/test_concurrent.py tests/integration/standard/test_connection.py tests/integration/standard/test_control_connection.py tests/integration/standard/test_custom_payload.py tests/integration/standard/test_custom_protocol_handler.py tests/integration/standard/test_cython_protocol_handlers.py tests/integration/standard/test_scylla_cloud.py tests/integration/standard/test_use_keyspace.py tests/integration/standard/test_ip_change.py
+        ./ci/run_integration_test.sh tests/integration/standard/test_authentication.py tests/integration/standard/test_cluster.py tests/integration/standard/test_concurrent.py tests/integration/standard/test_connection.py tests/integration/standard/test_control_connection.py tests/integration/standard/test_custom_payload.py tests/integration/standard/test_custom_protocol_handler.py tests/integration/standard/test_cython_protocol_handlers.py tests/integration/standard/test_scylla_cloud.py tests/integration/standard/test_use_keyspace.py tests/integration/standard/test_ip_change.py tests/integration/cqlengine/
         # can't run this, cause only 2 cpus on github actions: tests/integration/standard/test_shard_aware.py

.github/workflows/integration-tests.yml

@@ -21,5 +21,5 @@ jobs:
 
     - name: Test with pytest
       run: |
-        ./ci/run_integration_test.sh tests/integration/standard/test_authentication.py tests/integration/standard/test_cluster.py tests/integration/standard/test_concurrent.py tests/integration/standard/test_connection.py tests/integration/standard/test_control_connection.py tests/integration/standard/test_custom_payload.py tests/integration/standard/test_custom_protocol_handler.py tests/integration/standard/test_cython_protocol_handlers.py tests/integration/standard/test_scylla_cloud.py tests/integration/standard/test_use_keyspace.py tests/integration/standard/test_ip_change.py 
+        ./ci/run_integration_test.sh tests/integration/standard/test_authentication.py tests/integration/standard/test_cluster.py tests/integration/standard/test_concurrent.py tests/integration/standard/test_connection.py tests/integration/standard/test_control_connection.py tests/integration/standard/test_custom_payload.py tests/integration/standard/test_custom_protocol_handler.py tests/integration/standard/test_cython_protocol_handlers.py tests/integration/standard/test_scylla_cloud.py tests/integration/standard/test_use_keyspace.py tests/integration/standard/test_ip_change.py tests/integration/cqlengine/
         # can't run this, cause only 2 cpus on github actions: tests/integration/standard/test_shard_aware.py

cassandra/cluster.py

@@ -29,6 +29,7 @@
 import logging
 from warnings import warn
 from random import random
+import re
 import six
 from six.moves import filter, range, queue as Queue
 import socket
@@ -5349,6 +5350,8 @@ def cancel_continuous_paging(self):
         except AttributeError:
             raise DriverException("Attempted to cancel paging with no active session. This is only for requests with ContinuousdPagingOptions.")
 
+    batch_regex = re.compile('^\s*BEGIN\s+[a-zA-Z]*\s*BATCH')
+
     @property
     def was_applied(self):
         """
@@ -5363,7 +5366,8 @@ def was_applied(self):
         if self.response_future.row_factory not in (named_tuple_factory, dict_factory, tuple_factory):
             raise RuntimeError("Cannot determine LWT result with row factory %s" % (self.response_future.row_factory,))
 
-        is_batch_statement = isinstance(self.response_future.query, BatchStatement)
+        is_batch_statement = isinstance(self.response_future.query, BatchStatement) \
+                            or (isinstance(self.response_future.query, SimpleStatement) and self.batch_regex.match(self.response_future.query.query_string))
         if is_batch_statement and (not self.column_names or self.column_names[0] != "[applied]"):
             raise RuntimeError("No LWT were present in the BatchStatement")
 

cassandra/cqlengine/management.py

@@ -483,9 +483,17 @@ def _update_options(model, connection=None):
         else:
             try:
                 for k, v in value.items():
-                    if existing_value[k] != v:
-                        update_options[name] = value
-                        break
+                    # When creating table with compaction 'class': 'org.apache.cassandra.db.compaction.LeveledCompactionStrategy' in Scylla,
+                    # it will be silently changed to 'class': 'LeveledCompactionStrategy' - same for at least SizeTieredCompactionStrategy,
+                    # probably others too. We need to handle this case here.
+                    if k == 'class' and name == 'compaction':
+                        if existing_value[k] != v and existing_value[k] != v.split('.')[-1]:
+                            update_options[name] = value
+                            break
+                    else:
+                        if existing_value[k] != v:
+                            update_options[name] = value
+                            break
             except KeyError:
                 update_options[name] = value
 

tests/integration/__init__.py

@@ -337,6 +337,7 @@ def _id_and_mark(f):
 lessthenprotocolv4 = unittest.skipUnless(PROTOCOL_VERSION < 4, 'Protocol versions 4 or greater not supported')
 greaterthanprotocolv3 = unittest.skipUnless(PROTOCOL_VERSION >= 4, 'Protocol versions less than 4 are not supported')
 protocolv6 = unittest.skipUnless(6 in get_supported_protocol_versions(), 'Protocol versions less than 6 are not supported')
+
 greaterthancass20 = unittest.skipUnless(CASSANDRA_VERSION >= Version('2.1'), 'Cassandra version 2.1 or greater required')
 greaterthancass21 = unittest.skipUnless(CASSANDRA_VERSION >= Version('2.2'), 'Cassandra version 2.2 or greater required')
 greaterthanorequalcass30 = unittest.skipUnless(CASSANDRA_VERSION >= Version('3.0'), 'Cassandra version 3.0 or greater required')
@@ -348,6 +349,7 @@ def _id_and_mark(f):
 lessthanorequalcass40 = unittest.skipUnless(CASSANDRA_VERSION <= Version('4.0-a'), 'Cassandra version less or equal to 4.0 required')
 lessthancass40 = unittest.skipUnless(CASSANDRA_VERSION < Version('4.0-a'), 'Cassandra version less than 4.0 required')
 lessthancass30 = unittest.skipUnless(CASSANDRA_VERSION < Version('3.0'), 'Cassandra version less then 3.0 required')
+
 greaterthanorequaldse68 = unittest.skipUnless(DSE_VERSION and DSE_VERSION >= Version('6.8'), "DSE 6.8 or greater required for this test")
 greaterthanorequaldse67 = unittest.skipUnless(DSE_VERSION and DSE_VERSION >= Version('6.7'), "DSE 6.7 or greater required for this test")
 greaterthanorequaldse60 = unittest.skipUnless(DSE_VERSION and DSE_VERSION >= Version('6.0'), "DSE 6.0 or greater required for this test")
@@ -356,6 +358,9 @@ def _id_and_mark(f):
 lessthandse51 = unittest.skipUnless(DSE_VERSION and DSE_VERSION < Version('5.1'), "DSE version less than 5.1 required")
 lessthandse60 = unittest.skipUnless(DSE_VERSION and DSE_VERSION < Version('6.0'), "DSE version less than 6.0 required")
 
+requirescollectionindexes = unittest.skipUnless(SCYLLA_VERSION is None or Version(SCYLLA_VERSION.split(':')[1]) >= Version('5.2'), 'Test requires Scylla >= 5.2 or Cassandra')
+requirescustomindexes = unittest.skipUnless(SCYLLA_VERSION is None, 'Currently, Scylla does not support SASI or any other CUSTOM INDEX class.')
+
 pypy = unittest.skipUnless(platform.python_implementation() == "PyPy", "Test is skipped unless it's on PyPy")
 notpy3 = unittest.skipIf(sys.version_info >= (3, 0), "Test not applicable for Python 3.x runtime")
 requiresmallclockgranularity = unittest.skipIf("Windows" in platform.system() or "asyncore" in EVENT_LOOP_MANAGER,

tests/integration/cqlengine/__init__.py

@@ -13,35 +13,19 @@
 # limitations under the License.
 
 import os
-import warnings
 import unittest
 from cassandra import ConsistencyLevel
 
 from cassandra.cqlengine import connection
-from cassandra.cqlengine.management import create_keyspace_simple, drop_keyspace, CQLENG_ALLOW_SCHEMA_MANAGEMENT
 import cassandra
 
-from tests.integration import get_server_versions, use_single_node, PROTOCOL_VERSION, CASSANDRA_IP, ALLOW_BETA_PROTOCOL
+from tests.integration import get_server_versions, PROTOCOL_VERSION, CASSANDRA_IP, ALLOW_BETA_PROTOCOL
 
 DEFAULT_KEYSPACE = 'cqlengine_test'
 
 
 CQL_SKIP_EXECUTE = bool(os.getenv('CQL_SKIP_EXECUTE', False))
 
-
-def setup_package():
-    warnings.simplefilter('always')  # for testing warnings, make sure all are let through
-    os.environ[CQLENG_ALLOW_SCHEMA_MANAGEMENT] = '1'
-
-    use_single_node()
-
-    setup_connection(DEFAULT_KEYSPACE)
-    create_keyspace_simple(DEFAULT_KEYSPACE, 1)
-
-
-def teardown_package():
-    connection.unregister_connection("default")
-
 def is_prepend_reversed():
     # do we have https://issues.apache.org/jira/browse/CASSANDRA-8733 ?
     ver, _ = get_server_versions()

tests/integration/cqlengine/conftest.py

@@ -0,0 +1,54 @@
+# Copyright ScyllaDB, Inc.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+# Copyright DataStax, Inc.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+import warnings
+import os
+
+import pytest
+
+from cassandra.cqlengine import connection
+from cassandra.cqlengine.management import create_keyspace_simple, drop_keyspace, CQLENG_ALLOW_SCHEMA_MANAGEMENT
+from tests.integration import use_single_node
+
+from . import setup_connection, DEFAULT_KEYSPACE
+
+
+@pytest.fixture(scope='package', autouse=True)
+def cqlengine_fixture():
+    warnings.simplefilter('always')  # for testing warnings, make sure all are let through
+    os.environ[CQLENG_ALLOW_SCHEMA_MANAGEMENT] = '1'
+
+    use_single_node()
+
+    setup_connection(DEFAULT_KEYSPACE)
+    create_keyspace_simple(DEFAULT_KEYSPACE, 1)
+
+    yield
+
+    drop_keyspace(DEFAULT_KEYSPACE)
+    connection.unregister_connection("default")

tests/integration/cqlengine/management/test_compaction_settings.py

@@ -118,8 +118,16 @@ def _verify_options(self, table_meta, expected_options):
                 for subname, subvalue in value.items():
                     attr = "'%s': '%s'" % (subname, subvalue)
                     found_at = cql.find(attr, start)
-                    self.assertTrue(found_at > start)
-                    self.assertTrue(found_at < end)
+                    # When creating table with compaction 'class': 'org.apache.cassandra.db.compaction.LeveledCompactionStrategy' in Scylla,
+                    # it will be silently changed to 'class': 'LeveledCompactionStrategy' - same for at least SizeTieredCompactionStrategy,
+                    # probably others too. We need to handle this case here.
+                    if found_at == -1 and name == 'compaction' and subname == 'class':
+                        attr = "'%s': '%s'" % (subname, subvalue.split('.')[-1])
+                        found_at = cql.find(attr, start)
+                    else:
+
+                        self.assertTrue(found_at > start)
+                        self.assertTrue(found_at < end)
 
     def test_all_size_tiered_options(self):
         class AllSizeTieredOptionsModel(Model):

tests/integration/cqlengine/management/test_management.py

@@ -23,7 +23,7 @@
 from cassandra.cqlengine.models import Model
 from cassandra.cqlengine import columns
 
-from tests.integration import DSE_VERSION, PROTOCOL_VERSION, greaterthancass20, MockLoggingHandler, CASSANDRA_VERSION
+from tests.integration import DSE_VERSION, PROTOCOL_VERSION, greaterthancass20, requirescollectionindexes, MockLoggingHandler, CASSANDRA_VERSION
 from tests.integration.cqlengine.base import BaseCassEngTestCase
 from tests.integration.cqlengine.query.test_queryset import TestModel
 from cassandra.cqlengine.usertype import UserType
@@ -426,6 +426,7 @@ def test_sync_index_case_sensitive(self):
         self.assertIsNotNone(management._get_index_name_by_column(table_meta, 'second_key'))
 
     @greaterthancass20
+    @requirescollectionindexes
     def test_sync_indexed_set(self):
         """
         Tests that models that have container types with indices can be synced.

tests/integration/cqlengine/query/test_named.py

@@ -27,7 +27,7 @@
 from tests.integration.cqlengine.query.test_queryset import BaseQuerySetUsage
 
 
-from tests.integration import BasicSharedKeyspaceUnitTestCase, greaterthanorequalcass30
+from tests.integration import BasicSharedKeyspaceUnitTestCase, greaterthanorequalcass30, requirescollectionindexes
 
 
 class TestQuerySetOperation(BaseCassEngTestCase):
@@ -118,7 +118,7 @@ def test_query_expression_where_clause_generation(self):
         self.assertIsInstance(where.operator, GreaterThanOrEqualOperator)
         self.assertEqual(where.value, 1)
 
-
+@requirescollectionindexes
 class TestQuerySetCountSelectionAndIteration(BaseQuerySetUsage):
 
     @classmethod

tests/integration/cqlengine/query/test_queryset.py

@@ -39,7 +39,7 @@
 from cassandra.util import uuid_from_time
 from cassandra.cqlengine.connection import get_session
 from tests.integration import PROTOCOL_VERSION, CASSANDRA_VERSION, greaterthancass20, greaterthancass21, \
-    greaterthanorequalcass30, TestCluster
+    greaterthanorequalcass30, TestCluster, requirescollectionindexes
 from tests.integration.cqlengine import execute_count, DEFAULT_KEYSPACE
 
 
@@ -384,7 +384,7 @@ def tearDownClass(cls):
         drop_table(CustomIndexedTestModel)
         drop_table(TestMultiClusteringModel)
 
-
+@requirescollectionindexes
 class TestQuerySetCountSelectionAndIteration(BaseQuerySetUsage):
 
     @execute_count(2)
@@ -558,7 +558,7 @@ class NonEqualityFilteringModel(Model):
     num = qa.count()
     assert num == 1, num
 
-
+@requirescollectionindexes
 class TestQuerySetDistinct(BaseQuerySetUsage):
 
     @execute_count(1)
@@ -597,6 +597,7 @@ def test_distinct_with_explicit_count(self):
         self.assertEqual(q.count(), 2)
 
 
+@requirescollectionindexes
 class TestQuerySetOrdering(BaseQuerySetUsage):
     @execute_count(2)
     def test_order_by_success_case(self):
@@ -645,6 +646,7 @@ def test_ordering_on_multiple_clustering_columns(self):
         assert [r.three for r in results] == [1, 2, 3, 4, 5]
 
 
+@requirescollectionindexes
 class TestQuerySetSlicing(BaseQuerySetUsage):
 
     @execute_count(1)
@@ -699,6 +701,7 @@ def test_negative_slicing(self):
             self.assertEqual(model.attempt_id, expect)
 
 
+@requirescollectionindexes
 class TestQuerySetValidation(BaseQuerySetUsage):
 
     def test_primary_key_or_index_must_be_specified(self):
@@ -780,6 +783,7 @@ def test_custom_indexed_field_can_be_queried(self):
         list(CustomIndexedTestModel.objects.filter(test_id=1, description='test'))
 
 
+@requirescollectionindexes
 class TestQuerySetDelete(BaseQuerySetUsage):
 
     @execute_count(9)
@@ -938,6 +942,7 @@ def test_success_case(self):
         assert '4' in datas
 
 
+@requirescollectionindexes
 class TestInOperator(BaseQuerySetUsage):
     @execute_count(1)
     def test_kwarg_success_case(self):
@@ -998,6 +1003,7 @@ class bool_model2(Model):
 
 
 @greaterthancass20
+@requirescollectionindexes
 class TestContainsOperator(BaseQuerySetUsage):
 
     @execute_count(6)
@@ -1063,6 +1069,7 @@ def test_query_expression_success_case(self):
             self.assertEqual(q.count(), 0)
 
 
+@requirescollectionindexes
 class TestValuesList(BaseQuerySetUsage):
 
     @execute_count(2)
@@ -1075,6 +1082,7 @@ def test_values_list(self):
         assert item == 10
 
 
+@requirescollectionindexes
 class TestObjectsProperty(BaseQuerySetUsage):
     @execute_count(1)
     def test_objects_property_returns_fresh_queryset(self):
@@ -1105,6 +1113,7 @@ class PagingTest(Model):
         assert len(results) == 2
 
 
+@requirescollectionindexes
 class ModelQuerySetTimeoutTestCase(BaseQuerySetUsage):
     def test_default_timeout(self):
         with mock.patch.object(Session, 'execute') as mock_execute:
@@ -1122,6 +1131,7 @@ def test_none_timeout(self):
             self.assertEqual(mock_execute.call_args[-1]['timeout'], None)
 
 
+@requirescollectionindexes
 class DMLQueryTimeoutTestCase(BaseQuerySetUsage):
     def setUp(self):
         self.model = TestModel(test_id=1, attempt_id=1, description='timeout test')

tests/integration/cqlengine/statements/test_base_statement.py

@@ -26,7 +26,7 @@
 
 from tests.integration.cqlengine.base import BaseCassEngTestCase, TestQueryUpdateModel
 from tests.integration.cqlengine import DEFAULT_KEYSPACE
-from tests.integration import greaterthanorequalcass3_10, TestCluster
+from tests.integration import greaterthanorequalcass3_10, requirescustomindexes, TestCluster
 
 from cassandra.cqlengine.connection import execute
 
@@ -102,6 +102,7 @@ def test_insert_statement_execute(self):
         self.assertEqual(TestQueryUpdateModel.objects.count(), 0)
 
     @greaterthanorequalcass3_10
+    @requirescustomindexes
     def test_like_operator(self):
         """
         Test to verify the like operator works appropriately

tests/integration/cqlengine/test_batch_query.py

@@ -218,6 +218,7 @@ def my_callback(*args, **kwargs):
             call_history.append(args)
 
         with warnings.catch_warnings(record=True) as w:
+            warnings.simplefilter("always")
             with BatchQuery() as batch:
                 batch.add_callback(my_callback)
                 batch.execute()
@@ -243,6 +244,7 @@ def my_callback(*args, **kwargs):
 
         with patch('cassandra.cqlengine.query.BatchQuery.warn_multiple_exec', False):
             with warnings.catch_warnings(record=True) as w:
+                warnings.simplefilter("always")
                 with BatchQuery() as batch:
                     batch.add_callback(my_callback)
                     batch.execute()