| # 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 |
| from cassandra.datastax.cloud import parse_metadata_info |
| from cassandra.query import SimpleStatement |
| from cassandra.cqlengine import connection |
| from cassandra.cqlengine.management import sync_table, create_keyspace_simple |
| from cassandra.cqlengine.models import Model |
| from cassandra.cqlengine import columns |
| |
| import unittest |
| |
| from ssl import SSLContext, PROTOCOL_TLS |
| |
| from cassandra import DriverException, ConsistencyLevel, InvalidRequest |
| from cassandra.cluster import NoHostAvailable, ExecutionProfile, Cluster, _execution_profile_to_string |
| from cassandra.connection import SniEndPoint |
| from cassandra.auth import PlainTextAuthProvider |
| from cassandra.policies import TokenAwarePolicy, DCAwareRoundRobinPolicy, ConstantReconnectionPolicy |
| |
| from unittest.mock import patch |
| |
| from tests.integration import requirescloudproxy |
| from tests.util import wait_until_not_raised |
| from tests.integration.cloud import CloudProxyCluster, CLOUD_PROXY_SERVER |
| |
| DISALLOWED_CONSISTENCIES = [ |
| ConsistencyLevel.ANY, |
| ConsistencyLevel.ONE, |
| ConsistencyLevel.LOCAL_ONE |
| ] |
| |
| |
| @requirescloudproxy |
| class CloudTests(CloudProxyCluster): |
| def hosts_up(self): |
| return [h for h in self.cluster.metadata.all_hosts() if h.is_up] |
| |
| def test_resolve_and_connect(self): |
| self.connect(self.creds) |
| |
| self.assertEqual(len(self.hosts_up()), 3) |
| for host in self.cluster.metadata.all_hosts(): |
| self.assertTrue(host.is_up) |
| self.assertIsInstance(host.endpoint, SniEndPoint) |
| self.assertEqual(str(host.endpoint), "{}:{}:{}".format( |
| host.endpoint.address, host.endpoint.port, host.host_id)) |
| self.assertIn(host.endpoint._resolved_address, ("127.0.0.1", '::1')) |
| |
| def test_match_system_local(self): |
| self.connect(self.creds) |
| |
| self.assertEqual(len(self.hosts_up()), 3) |
| for host in self.cluster.metadata.all_hosts(): |
| row = self.session.execute('SELECT * FROM system.local', host=host).one() |
| self.assertEqual(row.host_id, host.host_id) |
| self.assertEqual(row.rpc_address, host.broadcast_rpc_address) |
| |
| def test_set_auth_provider(self): |
| self.connect(self.creds) |
| self.assertIsInstance(self.cluster.auth_provider, PlainTextAuthProvider) |
| self.assertEqual(self.cluster.auth_provider.username, 'user1') |
| self.assertEqual(self.cluster.auth_provider.password, 'user1') |
| |
| def test_support_leaving_the_auth_unset(self): |
| with self.assertRaises(NoHostAvailable): |
| self.connect(self.creds_no_auth) |
| self.assertIsNone(self.cluster.auth_provider) |
| |
| def test_support_overriding_auth_provider(self): |
| try: |
| self.connect(self.creds, auth_provider=PlainTextAuthProvider('invalid', 'invalid')) |
| except: |
| pass # this will fail soon when sni_single_endpoint is updated |
| self.assertIsInstance(self.cluster.auth_provider, PlainTextAuthProvider) |
| self.assertEqual(self.cluster.auth_provider.username, 'invalid') |
| self.assertEqual(self.cluster.auth_provider.password, 'invalid') |
| |
| def test_error_overriding_ssl_context(self): |
| with self.assertRaises(ValueError) as cm: |
| self.connect(self.creds, ssl_context=SSLContext(PROTOCOL_TLS)) |
| |
| self.assertIn('cannot be specified with a cloud configuration', str(cm.exception)) |
| |
| def test_error_overriding_ssl_options(self): |
| with self.assertRaises(ValueError) as cm: |
| self.connect(self.creds, ssl_options={'check_hostname': True}) |
| |
| self.assertIn('cannot be specified with a cloud configuration', str(cm.exception)) |
| |
| def _bad_hostname_metadata(self, config, http_data): |
| config = parse_metadata_info(config, http_data) |
| config.sni_host = "127.0.0.1" |
| return config |
| |
| def test_verify_hostname(self): |
| with patch('cassandra.datastax.cloud.parse_metadata_info', wraps=self._bad_hostname_metadata): |
| with self.assertRaises(NoHostAvailable) as e: |
| self.connect(self.creds) |
| self.assertIn("hostname", str(e.exception).lower()) |
| |
| def test_error_when_bundle_doesnt_exist(self): |
| try: |
| self.connect('/invalid/path/file.zip') |
| except Exception as e: |
| self.assertIsInstance(e, FileNotFoundError) |
| |
| def test_load_balancing_policy_is_dcawaretokenlbp(self): |
| self.connect(self.creds) |
| self.assertIsInstance(self.cluster.profile_manager.default.load_balancing_policy, |
| TokenAwarePolicy) |
| self.assertIsInstance(self.cluster.profile_manager.default.load_balancing_policy._child_policy, |
| DCAwareRoundRobinPolicy) |
| |
| def test_resolve_and_reconnect_on_node_down(self): |
| |
| self.connect(self.creds, |
| idle_heartbeat_interval=1, idle_heartbeat_timeout=1, |
| reconnection_policy=ConstantReconnectionPolicy(120)) |
| |
| self.assertEqual(len(self.hosts_up()), 3) |
| CLOUD_PROXY_SERVER.stop_node(1) |
| wait_until_not_raised( |
| lambda: self.assertEqual(len(self.hosts_up()), 2), |
| 0.02, 250) |
| |
| host = [h for h in self.cluster.metadata.all_hosts() if not h.is_up][0] |
| with patch.object(SniEndPoint, "resolve", wraps=host.endpoint.resolve) as mocked_resolve: |
| CLOUD_PROXY_SERVER.start_node(1) |
| wait_until_not_raised( |
| lambda: self.assertEqual(len(self.hosts_up()), 3), |
| 0.02, 250) |
| mocked_resolve.assert_called() |
| |
| def test_metadata_unreachable(self): |
| with self.assertRaises(DriverException) as cm: |
| self.connect(self.creds_unreachable, connect_timeout=1) |
| |
| self.assertIn('Unable to connect to the metadata service', str(cm.exception)) |
| |
| def test_metadata_ssl_error(self): |
| with self.assertRaises(DriverException) as cm: |
| self.connect(self.creds_invalid_ca) |
| |
| self.assertIn('Unable to connect to the metadata', str(cm.exception)) |
| |
| def test_default_consistency(self): |
| self.connect(self.creds) |
| self.assertEqual(self.session.default_consistency_level, ConsistencyLevel.LOCAL_QUORUM) |
| # Verify EXEC_PROFILE_DEFAULT, EXEC_PROFILE_GRAPH_DEFAULT, |
| # EXEC_PROFILE_GRAPH_SYSTEM_DEFAULT, EXEC_PROFILE_GRAPH_ANALYTICS_DEFAULT |
| for ep_key in self.cluster.profile_manager.profiles.keys(): |
| ep = self.cluster.profile_manager.profiles[ep_key] |
| self.assertEqual( |
| ep.consistency_level, |
| ConsistencyLevel.LOCAL_QUORUM, |
| "Expecting LOCAL QUORUM for profile {}, but got {} instead".format( |
| _execution_profile_to_string(ep_key), ConsistencyLevel.value_to_name[ep.consistency_level] |
| )) |
| |
| def test_default_consistency_of_execution_profiles(self): |
| cloud_config = {'secure_connect_bundle': self.creds} |
| self.cluster = Cluster(cloud=cloud_config, protocol_version=4, execution_profiles={ |
| 'pre_create_default_ep': ExecutionProfile(), |
| 'pre_create_changed_ep': ExecutionProfile( |
| consistency_level=ConsistencyLevel.LOCAL_ONE, |
| ), |
| }) |
| self.cluster.add_execution_profile('pre_connect_default_ep', ExecutionProfile()) |
| self.cluster.add_execution_profile( |
| 'pre_connect_changed_ep', |
| ExecutionProfile( |
| consistency_level=ConsistencyLevel.LOCAL_ONE, |
| ) |
| ) |
| session = self.cluster.connect(wait_for_all_pools=True) |
| |
| self.cluster.add_execution_profile('post_connect_default_ep', ExecutionProfile()) |
| self.cluster.add_execution_profile( |
| 'post_connect_changed_ep', |
| ExecutionProfile( |
| consistency_level=ConsistencyLevel.LOCAL_ONE, |
| ) |
| ) |
| |
| for default in ['pre_create_default_ep', 'pre_connect_default_ep', 'post_connect_default_ep']: |
| cl = self.cluster.profile_manager.profiles[default].consistency_level |
| self.assertEqual( |
| cl, ConsistencyLevel.LOCAL_QUORUM, |
| "Expecting LOCAL QUORUM for profile {}, but got {} instead".format(default, cl) |
| ) |
| for changed in ['pre_create_changed_ep', 'pre_connect_changed_ep', 'post_connect_changed_ep']: |
| cl = self.cluster.profile_manager.profiles[changed].consistency_level |
| self.assertEqual( |
| cl, ConsistencyLevel.LOCAL_ONE, |
| "Expecting LOCAL ONE for profile {}, but got {} instead".format(default, cl) |
| ) |
| |
| def test_consistency_guardrails(self): |
| self.connect(self.creds) |
| self.session.execute( |
| "CREATE KEYSPACE IF NOT EXISTS test_consistency_guardrails " |
| "with replication={'class': 'SimpleStrategy', 'replication_factor': 1}" |
| ) |
| self.session.execute("CREATE TABLE IF NOT EXISTS test_consistency_guardrails.guardrails (id int primary key)") |
| for consistency in DISALLOWED_CONSISTENCIES: |
| statement = SimpleStatement( |
| "INSERT INTO test_consistency_guardrails.guardrails (id) values (1)", |
| consistency_level=consistency |
| ) |
| with self.assertRaises(InvalidRequest) as e: |
| self.session.execute(statement) |
| self.assertIn('not allowed for Write Consistency Level', str(e.exception)) |
| |
| # Sanity check to make sure we can do a normal insert |
| statement = SimpleStatement( |
| "INSERT INTO test_consistency_guardrails.guardrails (id) values (1)", |
| consistency_level=ConsistencyLevel.LOCAL_QUORUM |
| ) |
| try: |
| self.session.execute(statement) |
| except InvalidRequest: |
| self.fail("InvalidRequest was incorrectly raised for write query at LOCAL QUORUM!") |
| |
| def test_cqlengine_can_connect(self): |
| class TestModel(Model): |
| id = columns.Integer(primary_key=True) |
| val = columns.Text() |
| |
| connection.setup(None, "test", cloud={'secure_connect_bundle': self.creds}) |
| create_keyspace_simple('test', 1) |
| sync_table(TestModel) |
| TestModel.objects.create(id=42, value='test') |
| self.assertEqual(len(TestModel.objects.all()), 1) |
