Merge branch 'oss-next' into ngdg_master_ft

Author: aboudreault

CHANGELOG.rst

@@ -130,6 +130,7 @@ Bug Fixes
 * Call ConnectionException with correct kwargs (PYTHON-1117)
 * Can't connect to clusters built from source because version parsing doesn't handle 'x.y-SNAPSHOT' (PYTHON-1118)
 * Discovered node doesn´t honor the configured Cluster port on connection (PYTHON-1127)
+* Exception when use pk__token__gt filter In python 3.7 (PYTHON-1121)
 
 Other
 -----

build.yaml

@@ -288,8 +288,8 @@ build:
         EVENT_LOOP_MANAGER=$EVENT_LOOP_MANAGER CASSANDRA_DIR=$CCM_INSTALL_DIR DSE_VERSION=$DSE_VERSION ADS_HOME=$HOME/ VERIFY_CYTHON=$FORCE_CYTHON nosetests -s -v --logging-format="[%(levelname)s] %(asctime)s %(thread)d: %(message)s" --with-ignore-docstrings --with-xunit --xunit-file=dse_results.xml tests/integration/advanced/ || true
       fi
 
-      echo "==========RUNNING ADVANCED AND CLOUD TESTS=========="
-      EVENT_LOOP_MANAGER=$EVENT_LOOP_MANAGER CLOUD_PROXY_PATH="$HOME/proxy/" CASSANDRA_VERSION=$CCM_CASSANDRA_VERSION 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=advanced_results.xml tests/integration/advanced/ || true
+      echo "==========RUNNING CLOUD TESTS=========="
+      EVENT_LOOP_MANAGER=$EVENT_LOOP_MANAGER CLOUD_PROXY_PATH="$HOME/proxy/" CASSANDRA_VERSION=$CCM_CASSANDRA_VERSION 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=advanced_results.xml tests/integration/cloud/ || true
 
       if [ -z "$EXCLUDE_LONG" ]; then
         echo "==========RUNNING LONG INTEGRATION TESTS=========="

cassandra/cluster.py

@@ -38,6 +38,12 @@
 
 import weakref
 from weakref import WeakValueDictionary
+
+try:
+    from cassandra.io.twistedreactor import TwistedConnection
+except ImportError:
+    TwistedConnection = None
+
 try:
     from weakref import WeakSet
 except ImportError:
@@ -47,7 +53,7 @@
                        OperationTimedOut, UnsupportedOperation,
                        SchemaTargetType, DriverException, ProtocolVersion,
                        UnresolvableContactPoints)
-from cassandra.auth import _proxy_execute_key
+from cassandra.auth import _proxy_execute_key, PlainTextAuthProvider
 from cassandra.connection import (ConnectionException, ConnectionShutdown,
                                   ConnectionHeartbeat, ProtocolVersionUnsupported,
                                   EndPoint, DefaultEndPoint, DefaultEndPointFactory,
@@ -1094,13 +1100,18 @@ def __init__(self,
 
         Any of the mutable Cluster attributes may be set as keyword arguments to the constructor.
         """
+        if connection_class is not None:
+            self.connection_class = connection_class
 
         if cloud is not None:
             if contact_points is not _NOT_SET or endpoint_factory or ssl_context or ssl_options:
                 raise ValueError("contact_points, endpoint_factory, ssl_context, and ssl_options "
                                  "cannot be specified with a cloud configuration")
 
-            cloud_config = dscloud.get_cloud_config(cloud)
+            cloud_config = dscloud.get_cloud_config(
+                cloud,
+                create_pyopenssl_context=self.connection_class is TwistedConnection
+            )
 
             ssl_context = cloud_config.ssl_context
             ssl_options = {'check_hostname': True}
@@ -1192,9 +1203,6 @@ def __init__(self,
                 raise TypeError("address_translator should not be a class, it should be an instance of that class")
             self.address_translator = address_translator
 
-        if connection_class is not None:
-            self.connection_class = connection_class
-
         if timestamp_generator is not None:
             if not callable(timestamp_generator):
                 raise ValueError("timestamp_generator must be callable")

cassandra/datastax/cloud/__init__.py

@@ -75,34 +75,37 @@ def from_dict(cls, d):
         return c
 
 
-def get_cloud_config(cloud_config):
+def get_cloud_config(cloud_config, create_pyopenssl_context=False):
     if not _HAS_SSL:
         raise DriverException("A Python installation with SSL is required to connect to a cloud cluster.")
 
     if 'secure_connect_bundle' not in cloud_config:
         raise ValueError("The cloud config doesn't have a secure_connect_bundle specified.")
 
     try:
-        config = read_cloud_config_from_zip(cloud_config)
+        config = read_cloud_config_from_zip(cloud_config, create_pyopenssl_context)
     except BadZipFile:
         raise ValueError("Unable to open the zip file for the cloud config. Check your secure connect bundle.")
 
-    return read_metadata_info(config, cloud_config)
+    config = read_metadata_info(config, cloud_config)
+    if create_pyopenssl_context:
+        config.ssl_context = config.pyopenssl_context
+    return config
 
 
-def read_cloud_config_from_zip(cloud_config):
+def read_cloud_config_from_zip(cloud_config, create_pyopenssl_context):
     secure_bundle = cloud_config['secure_connect_bundle']
     with ZipFile(secure_bundle) as zipfile:
         base_dir = os.path.dirname(secure_bundle)
         tmp_dir = tempfile.mkdtemp(dir=base_dir)
         try:
             zipfile.extractall(path=tmp_dir)
-            return parse_cloud_config(os.path.join(tmp_dir, 'config.json'), cloud_config)
+            return parse_cloud_config(os.path.join(tmp_dir, 'config.json'), cloud_config, create_pyopenssl_context)
         finally:
             shutil.rmtree(tmp_dir)
 
 
-def parse_cloud_config(path, cloud_config):
+def parse_cloud_config(path, cloud_config, create_pyopenssl_context):
     with open(path, 'r') as stream:
         data = json.load(stream)
 
@@ -116,7 +119,11 @@ def parse_cloud_config(path, cloud_config):
         ca_cert_location = os.path.join(config_dir, 'ca.crt')
         cert_location = os.path.join(config_dir, 'cert')
         key_location = os.path.join(config_dir, 'key')
+        # Regardless of if we create a pyopenssl context, we still need the builtin one
+        # to connect to the metadata service
         config.ssl_context = _ssl_context_from_cert(ca_cert_location, cert_location, key_location)
+        if create_pyopenssl_context:
+            config.pyopenssl_context = _pyopenssl_context_from_cert(ca_cert_location, cert_location, key_location)
 
     return config
 
@@ -165,3 +172,17 @@ def _ssl_context_from_cert(ca_cert_location, cert_location, key_location):
     ssl_context.load_cert_chain(certfile=cert_location, keyfile=key_location)
 
     return ssl_context
+
+
+def _pyopenssl_context_from_cert(ca_cert_location, cert_location, key_location):
+    try:
+        from OpenSSL import SSL
+    except ImportError:
+        return None
+    ssl_context = SSL.Context(SSL.TLSv1_METHOD)
+    ssl_context.set_verify(SSL.VERIFY_PEER, callback=lambda _1, _2, _3, _4, ok: ok)
+    ssl_context.use_certificate_file(cert_location)
+    ssl_context.use_privatekey_file(key_location)
+    ssl_context.load_verify_locations(ca_cert_location)
+
+    return ssl_context