test_scylla_cloud: add new tests for using cloud config bundle

* those test are using CCM sni_proxy code

Author: fruch

.github/workflows/integration-tests.yml

@@ -19,5 +19,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
+        ./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
         # can't run this, cause only 2 cpus on github actions: tests/integration/standard/test_shard_aware.py

cassandra/cluster.py

@@ -1160,14 +1160,15 @@ def __init__(self,
             self.connection_class = connection_class
 
         if scylla_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 "
+            if contact_points is not _NOT_SET or ssl_context or ssl_options:
+                raise ValueError("contact_points, ssl_context, and ssl_options "
                                  "cannot be specified with a scylla cloud configuration")
 
             uses_twisted = TwistedConnection and issubclass(self.connection_class, TwistedConnection)
             uses_eventlet = EventletConnection and issubclass(self.connection_class, EventletConnection)
 
-            scylla_cloud_config = CloudConfiguration.create(scylla_cloud, pyopenssl=uses_twisted or uses_eventlet)
+            scylla_cloud_config = CloudConfiguration.create(scylla_cloud, pyopenssl=uses_twisted or uses_eventlet,
+                                                            endpoint_factory=endpoint_factory)
             ssl_context = scylla_cloud_config.ssl_context
             endpoint_factory = scylla_cloud_config.endpoint_factory
             contact_points = scylla_cloud_config.contact_points

cassandra/scylla/cloud.py

@@ -59,7 +59,7 @@ class CloudConfiguration:
     ssl_context: SSLContext
     skip_tls_verify: bool
 
-    def __init__(self, configuration_file, pyopenssl=False):
+    def __init__(self, configuration_file, pyopenssl=False, endpoint_factory=None):
         cloud_config = yaml.safe_load(open(configuration_file))
 
         self.current_context = cloud_config['contexts'][cloud_config['currentContext']]
@@ -69,9 +69,13 @@ def __init__(self, configuration_file, pyopenssl=False):
         self.skip_tls_verify = self.auth_info.get('insecureSkipTLSVerify', False)
         self.ssl_context = self.create_pyopenssl_context() if pyopenssl else self.create_ssl_context()
 
-        proxy_address, port, node_domain = self.get_server(self.data_centers[self.current_context['datacenterName']],
-                                                           keys_order=['testServer', 'server'])
-        self.endpoint_factory = SniEndPointFactory(proxy_address, port=int(port), node_domain=node_domain)
+        proxy_address, port, node_domain = self.get_server(self.data_centers[self.current_context['datacenterName']])
+
+        if not endpoint_factory:
+            endpoint_factory = SniEndPointFactory(proxy_address, port=int(port), node_domain=node_domain)
+        else:
+            assert isinstance(endpoint_factory, SniEndPointFactory)
+        self.endpoint_factory = endpoint_factory
 
         username, password = self.auth_info.get('username'), self.auth_info.get('password')
         if username and password:
@@ -86,17 +90,14 @@ def contact_points(self):
             _contact_points.append(self.endpoint_factory.create_from_sni(address))
         return _contact_points
 
-    def get_server(self, data_center, keys_order=None):
-        keys_order = keys_order or ['server']
-        for key in keys_order:
-            address = data_center.get(key, '')
-            if not address:
-                continue
-            address = address.split(":")
-            port = nth(address, 1, default=443)
-            address = nth(address, 0)
-            node_domain = data_center.get('nodeDomain')
-            return address, port, node_domain
+    def get_server(self, data_center):
+        address = data_center.get('server')
+        address = address.split(":")
+        port = nth(address, 1, default=443)
+        address = nth(address, 0)
+        node_domain = data_center.get('nodeDomain')
+        assert address and port and node_domain, "server or nodeDomain are missing"
+        return address, port, node_domain
 
     def create_ssl_context(self):
         ssl_context = ssl.SSLContext(protocol=ssl.PROTOCOL_SSLv23)
@@ -138,5 +139,5 @@ def create_pyopenssl_context(self):
         return ssl_context
 
     @classmethod
-    def create(cls, configuration_file, pyopenssl=False):
-        return cls(configuration_file, pyopenssl)
+    def create(cls, configuration_file, pyopenssl=False, endpoint_factory=None):
+        return cls(configuration_file, pyopenssl=pyopenssl, endpoint_factory=endpoint_factory)

tests/integration/standard/test_scylla_cloud.py

@@ -0,0 +1,82 @@
+import os.path
+from unittest import TestCase
+from ccmlib.utils.ssl_utils import generate_ssl_stores
+from ccmlib.utils.sni_proxy import refresh_certs, get_cluster_info, start_sni_proxy, create_cloud_config
+
+from tests.integration import use_cluster
+from cassandra.cluster import Cluster, TwistedConnection
+from cassandra.connection import SniEndPointFactory
+from cassandra.io.asyncorereactor import AsyncoreConnection
+from cassandra.io.libevreactor import LibevConnection
+from cassandra.io.geventreactor import GeventConnection
+from cassandra.io.eventletreactor import EventletConnection
+from cassandra.io.asyncioreactor import AsyncioConnection
+
+supported_connection_classes = [AsyncoreConnection, LibevConnection, TwistedConnection]
+# need to run them with specific configuration like `gevent.monkey.patch_all()` or under async functions
+unsupported_connection_classes = [GeventConnection, AsyncioConnection, EventletConnection]
+
+
+class ScyllaCloudConfigTests(TestCase):
+    def start_cluster_with_proxy(self):
+        ccm_cluster = self.ccm_cluster
+        generate_ssl_stores(ccm_cluster.get_path())
+        ssl_port = 9142
+        sni_port = 443
+        ccm_cluster.set_configuration_options(dict(
+            client_encryption_options=
+            dict(require_client_auth=True,
+                 truststore=os.path.join(ccm_cluster.get_path(), 'ccm_node.cer'),
+                 certificate=os.path.join(ccm_cluster.get_path(), 'ccm_node.pem'),
+                 keyfile=os.path.join(ccm_cluster.get_path(), 'ccm_node.key'),
+                 enabled=True),
+            native_transport_port_ssl=ssl_port))
+
+        ccm_cluster._update_config()
+
+        ccm_cluster.start(wait_for_binary_proto=True)
+
+        nodes_info = get_cluster_info(ccm_cluster, port=ssl_port)
+        refresh_certs(ccm_cluster, nodes_info)
+
+        docker_id, listen_address, listen_port = \
+            start_sni_proxy(ccm_cluster.get_path(), nodes_info=nodes_info, listen_port=sni_port)
+        ccm_cluster.sni_proxy_docker_id = docker_id
+        ccm_cluster.sni_proxy_listen_port = listen_port
+        ccm_cluster._update_config()
+
+        config_data_yaml, config_path_yaml = create_cloud_config(ccm_cluster.get_path(), listen_port)
+
+        endpoint_factory = SniEndPointFactory(listen_address, port=int(listen_port),
+                                              node_domain="cluster-id.scylla.com")
+
+        return config_data_yaml, config_path_yaml, endpoint_factory
+
+    def test_1_node_cluster(self):
+        self.ccm_cluster = use_cluster("sni_proxy", [1], start=False)
+        config_data_yaml, config_path_yaml, endpoint_factory = self.start_cluster_with_proxy()
+
+        for config in [config_path_yaml, config_data_yaml]:
+            for connection_class in supported_connection_classes:
+                cluster = Cluster(scylla_cloud=config, connection_class=connection_class,
+                                  endpoint_factory=endpoint_factory)
+                with cluster.connect() as session:
+                    res = session.execute("SELECT * FROM system.local")
+                    assert res.all()
+
+                    assert len(cluster.metadata._hosts) == 1
+                    assert len(cluster.metadata._host_id_by_endpoint) == 1
+
+    def test_3_node_cluster(self):
+        self.ccm_cluster = use_cluster("sni_proxy", [3], start=False)
+        config_data_yaml, config_path_yaml, endpoint_factory = self.start_cluster_with_proxy()
+
+        for config in [config_path_yaml, config_data_yaml]:
+            for connection_class in supported_connection_classes:
+                cluster = Cluster(scylla_cloud=config, connection_class=connection_class,
+                                  endpoint_factory=endpoint_factory)
+                with cluster.connect() as session:
+                    res = session.execute("SELECT * FROM system.local")
+                    assert res.all()
+                    assert len(cluster.metadata._hosts) == 3
+                    assert len(cluster.metadata._host_id_by_endpoint) == 3