Merge pull request datastax#986 from datastax/python-1020_fix-vtables-against-dse-6.0.X

PYTHON-1020 fix vtables against dse 6.0.x

Author: mambocab

CHANGELOG.rst

@@ -5,6 +5,7 @@ Bug Fixes
 ---------
 * Improve and fix socket error-catching code in nonblocking-socket reactors (PYTHON-1024)
 * Non-ASCII characters in schema break CQL string generation (PYTHON-1008)
+* Fix OSS driver's virtual table support against DSE 6.0.X and future server releases (PYTHON-1020)
 
 Other
 -----

build.yaml

@@ -118,6 +118,7 @@ cassandra:
   - '3.0'
   - '3.11'
   - 'test-dse'
+  - 'dse-6.7'
 
 env:
   CYTHON:
@@ -159,6 +160,13 @@ build:
         exit 0
       fi
 
+      if [[ $CCM_IS_DSE == 'true' ]]; then
+        # We only use a DSE version for unreleased DSE versions, so we only need to run the smoke tests here
+        echo "CCM_IS_DSE: $CCM_IS_DSE"
+        echo "==========RUNNING SMOKE TESTS==========="
+        EVENT_LOOP_MANAGER=$EVENT_LOOP_MANAGER CCM_ARGS="$CCM_ARGS" CASSANDRA_VERSION=$CCM_CASSANDRA_VERSION DSE_VERSION='6.7.0' MAPPED_CASSANDRA_VERSION=$MAPPED_CASSANDRA_VERSION VERIFY_CYTHON=$FORCE_CYTHON nosetests -s -v --logging-format="[%(levelname)s] %(asctime)s %(thread)d: %(message)s" --with-ignore-docstrings --with-xunit --xunit-file=standard_results.xml tests/integration/standard/test_dse.py || true
+        exit 0
+      fi
 
       # Run the unit tests, this is not done in travis because
       # it takes too much time for the whole matrix to build with cython

cassandra/metadata.py

@@ -1667,8 +1667,30 @@ def __init__(self, connection, timeout):
         self.connection = connection
         self.timeout = timeout
 
-    def _handle_results(self, success, result):
-        if success:
+    def _handle_results(self, success, result, expected_failures=tuple()):
+        """
+        Given a bool and a ResultSet (the form returned per result from
+        Connection.wait_for_responses), return a dictionary containing the
+        results. Used to process results from asynchronous queries to system
+        tables.
+
+        ``expected_failures`` will usually be used to allow callers to ignore
+        ``InvalidRequest`` errors caused by a missing system keyspace. For
+        example, some DSE versions report a 4.X server version, but do not have
+        virtual tables. Thus, running against 4.X servers, SchemaParserV4 uses
+        expected_failures to make a best-effort attempt to read those
+        keyspaces, but treat them as empty if they're not found.
+
+        :param success: A boolean representing whether or not the query
+        succeeded
+        :param result: The resultset in question.
+        :expected_failures: An Exception class or an iterable thereof. If the
+        query failed, but raised an instance of an expected failure class, this
+        will ignore the failure and return an empty list.
+        """
+        if not success and isinstance(result, expected_failures):
+            return []
+        elif success:
             return dict_factory(*result.results) if result else []
         else:
             raise result
@@ -1784,11 +1806,9 @@ def get_table(self, keyspaces, keyspace, table):
         table_result = self._handle_results(cf_success, cf_result)
         col_result = self._handle_results(col_success, col_result)
 
-        # handle the triggers table not existing in Cassandra 1.2
-        if not triggers_success and isinstance(triggers_result, InvalidRequest):
-            triggers_result = []
-        else:
-            triggers_result = self._handle_results(triggers_success, triggers_result)
+        # the triggers table doesn't exist in C* 1.2
+        triggers_result = self._handle_results(triggers_success, triggers_result,
+                                               expected_failures=InvalidRequest)
 
         if table_result:
             return self._build_table_metadata(table_result[0], col_result, triggers_result)
@@ -2531,12 +2551,21 @@ def _query_all(self):
         self.indexes_result = self._handle_results(indexes_success, indexes_result)
         self.views_result = self._handle_results(views_success, views_result)
         # V4-only results
-        self.virtual_keyspaces_result = self._handle_results(virtual_ks_success,
-                                                             virtual_ks_result)
-        self.virtual_tables_result = self._handle_results(virtual_table_success,
-                                                          virtual_table_result)
-        self.virtual_columns_result = self._handle_results(virtual_column_success,
-                                                           virtual_column_result)
+        # These tables don't exist in some DSE versions reporting 4.X so we can
+        # ignore them if we got an error
+        self.virtual_keyspaces_result = self._handle_results(
+            virtual_ks_success, virtual_ks_result,
+            expected_failures=InvalidRequest
+        )
+        self.virtual_tables_result = self._handle_results(
+            virtual_table_success, virtual_table_result,
+            expected_failures=InvalidRequest
+        )
+        self.virtual_columns_result = self._handle_results(
+            virtual_column_success, virtual_column_result,
+            expected_failures=InvalidRequest
+        )
+
         self._aggregate_results()
 
     def _aggregate_results(self):
@@ -2722,9 +2751,7 @@ def export_as_string(self):
 
 def get_schema_parser(connection, server_version, timeout):
     server_major_version = int(server_version.split('.')[0])
-    # check for DSE version
-    has_build_version = len(server_version.split('.')) > 3
-    if server_major_version >= 4 and not has_build_version:
+    if server_major_version >= 4:
         return SchemaParserV4(connection, timeout)
     if server_major_version >= 3:
         return SchemaParserV3(connection, timeout)

tests/integration/standard/test_dse.py

@@ -18,6 +18,7 @@
 
 from cassandra.cluster import Cluster
 from tests import notwindows
+from tests.unit.cython.utils import notcython
 from tests.integration import (execute_until_pass,
                                execute_with_long_wait_retry, use_cluster)
 
@@ -27,8 +28,12 @@
     import unittest  # noqa
 
 
+CCM_IS_DSE = (os.environ.get('CCM_IS_DSE', None) == 'true')
+
+
 @unittest.skipIf(os.environ.get('CCM_ARGS', None), 'environment has custom CCM_ARGS; skipping')
 @notwindows
+@notcython  # no need to double up on this test; also __default__ setting doesn't work
 class DseCCMClusterTest(unittest.TestCase):
     """
     This class can be executed setting the DSE_VERSION variable, for example:
@@ -42,6 +47,10 @@ def test_dse_5x(self):
     def test_dse_60(self):
         self._test_basic(Version('6.0.2'))
 
+    @unittest.skipUnless(CCM_IS_DSE, 'DSE version unavailable')
+    def test_dse_67(self):
+        self._test_basic(Version('6.7.0'))
+
     def _test_basic(self, dse_version):
         """
         Test basic connection and usage
@@ -52,7 +61,8 @@ def _test_basic(self, dse_version):
         use_cluster(cluster_name=cluster_name, nodes=[3],
                     dse_cluster=True, dse_options={}, dse_version=dse_version)
 
-        cluster = Cluster()
+        cluster = Cluster(
+            allow_beta_protocol_version=(dse_version >= Version('6.7.0')))
         session = cluster.connect()
         result = execute_until_pass(
             session,