| # Licensed to the Apache Software Foundation (ASF) under one |
| # or more contributor license agreements. See the NOTICE file |
| # distributed with this work for additional information |
| # regarding copyright ownership. The ASF licenses this file |
| # to you 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 unittest |
| |
| from binascii import unhexlify |
| import logging |
| from unittest.mock import Mock |
| import os |
| import timeit |
| |
| import cassandra |
| from cassandra.cqltypes import strip_frozen |
| from cassandra.marshal import uint16_unpack, uint16_pack |
| from cassandra.metadata import (Murmur3Token, MD5Token, |
| BytesToken, ReplicationStrategy, |
| NetworkTopologyStrategy, SimpleStrategy, |
| LocalStrategy, protect_name, |
| protect_names, protect_value, is_valid_name, |
| UserType, KeyspaceMetadata, get_schema_parser, |
| _UnknownStrategy, ColumnMetadata, TableMetadata, |
| IndexMetadata, Function, Aggregate, |
| Metadata, TokenMap, ReplicationFactor) |
| from cassandra.policies import SimpleConvictionPolicy |
| from cassandra.pool import Host |
| |
| |
| log = logging.getLogger(__name__) |
| |
| |
| class ReplicationFactorTest(unittest.TestCase): |
| |
| def test_replication_factor_parsing(self): |
| rf = ReplicationFactor.create('3') |
| self.assertEqual(rf.all_replicas, 3) |
| self.assertEqual(rf.full_replicas, 3) |
| self.assertEqual(rf.transient_replicas, None) |
| self.assertEqual(str(rf), '3') |
| |
| rf = ReplicationFactor.create('3/1') |
| self.assertEqual(rf.all_replicas, 3) |
| self.assertEqual(rf.full_replicas, 2) |
| self.assertEqual(rf.transient_replicas, 1) |
| self.assertEqual(str(rf), '3/1') |
| |
| self.assertRaises(ValueError, ReplicationFactor.create, '3/') |
| self.assertRaises(ValueError, ReplicationFactor.create, 'a/1') |
| self.assertRaises(ValueError, ReplicationFactor.create, 'a') |
| self.assertRaises(ValueError, ReplicationFactor.create, '3/a') |
| |
| def test_replication_factor_equality(self): |
| self.assertEqual(ReplicationFactor.create('3/1'), ReplicationFactor.create('3/1')) |
| self.assertEqual(ReplicationFactor.create('3'), ReplicationFactor.create('3')) |
| self.assertNotEqual(ReplicationFactor.create('3'), ReplicationFactor.create('3/1')) |
| self.assertNotEqual(ReplicationFactor.create('3'), ReplicationFactor.create('3/1')) |
| |
| |
| |
| class StrategiesTest(unittest.TestCase): |
| |
| @classmethod |
| def setUpClass(cls): |
| "Hook method for setting up class fixture before running tests in the class." |
| if not hasattr(cls, 'assertItemsEqual'): |
| cls.assertItemsEqual = cls.assertCountEqual |
| |
| def test_replication_strategy(self): |
| """ |
| Basic code coverage testing that ensures different ReplicationStrategies |
| can be initiated using parameters correctly. |
| """ |
| |
| rs = ReplicationStrategy() |
| |
| self.assertEqual(rs.create('OldNetworkTopologyStrategy', None), _UnknownStrategy('OldNetworkTopologyStrategy', None)) |
| fake_options_map = {'options': 'map'} |
| uks = rs.create('OldNetworkTopologyStrategy', fake_options_map) |
| self.assertEqual(uks, _UnknownStrategy('OldNetworkTopologyStrategy', fake_options_map)) |
| self.assertEqual(uks.make_token_replica_map({}, []), {}) |
| |
| fake_options_map = {'dc1': '3'} |
| self.assertIsInstance(rs.create('NetworkTopologyStrategy', fake_options_map), NetworkTopologyStrategy) |
| self.assertEqual(rs.create('NetworkTopologyStrategy', fake_options_map).dc_replication_factors, |
| NetworkTopologyStrategy(fake_options_map).dc_replication_factors) |
| |
| fake_options_map = {'options': 'map'} |
| self.assertIsNone(rs.create('SimpleStrategy', fake_options_map)) |
| |
| fake_options_map = {'options': 'map'} |
| self.assertIsInstance(rs.create('LocalStrategy', fake_options_map), LocalStrategy) |
| |
| fake_options_map = {'options': 'map', 'replication_factor': 3} |
| self.assertIsInstance(rs.create('SimpleStrategy', fake_options_map), SimpleStrategy) |
| self.assertEqual(rs.create('SimpleStrategy', fake_options_map).replication_factor, |
| SimpleStrategy(fake_options_map).replication_factor) |
| |
| self.assertEqual(rs.create('xxxxxxxx', fake_options_map), _UnknownStrategy('xxxxxxxx', fake_options_map)) |
| |
| self.assertRaises(NotImplementedError, rs.make_token_replica_map, None, None) |
| self.assertRaises(NotImplementedError, rs.export_for_schema) |
| |
| def test_simple_replication_type_parsing(self): |
| """ Test equality between passing numeric and string replication factor for simple strategy """ |
| rs = ReplicationStrategy() |
| |
| simple_int = rs.create('SimpleStrategy', {'replication_factor': 3}) |
| simple_str = rs.create('SimpleStrategy', {'replication_factor': '3'}) |
| |
| self.assertEqual(simple_int.export_for_schema(), simple_str.export_for_schema()) |
| self.assertEqual(simple_int, simple_str) |
| |
| # make token replica map |
| ring = [MD5Token(0), MD5Token(1), MD5Token(2)] |
| hosts = [Host('dc1.{}'.format(host), SimpleConvictionPolicy) for host in range(3)] |
| token_to_host = dict(zip(ring, hosts)) |
| self.assertEqual( |
| simple_int.make_token_replica_map(token_to_host, ring), |
| simple_str.make_token_replica_map(token_to_host, ring) |
| ) |
| |
| def test_transient_replication_parsing(self): |
| """ Test that we can PARSE a transient replication factor for SimpleStrategy """ |
| rs = ReplicationStrategy() |
| |
| simple_transient = rs.create('SimpleStrategy', {'replication_factor': '3/1'}) |
| self.assertEqual(simple_transient.replication_factor_info, ReplicationFactor(3, 1)) |
| self.assertEqual(simple_transient.replication_factor, 2) |
| self.assertIn("'replication_factor': '3/1'", simple_transient.export_for_schema()) |
| |
| simple_str = rs.create('SimpleStrategy', {'replication_factor': '2'}) |
| self.assertNotEqual(simple_transient, simple_str) |
| |
| # make token replica map |
| ring = [MD5Token(0), MD5Token(1), MD5Token(2)] |
| hosts = [Host('dc1.{}'.format(host), SimpleConvictionPolicy) for host in range(3)] |
| token_to_host = dict(zip(ring, hosts)) |
| self.assertEqual( |
| simple_transient.make_token_replica_map(token_to_host, ring), |
| simple_str.make_token_replica_map(token_to_host, ring) |
| ) |
| |
| def test_nts_replication_parsing(self): |
| """ Test equality between passing numeric and string replication factor for NTS """ |
| rs = ReplicationStrategy() |
| |
| nts_int = rs.create('NetworkTopologyStrategy', {'dc1': 3, 'dc2': 5}) |
| nts_str = rs.create('NetworkTopologyStrategy', {'dc1': '3', 'dc2': '5'}) |
| |
| self.assertEqual(nts_int.dc_replication_factors['dc1'], 3) |
| self.assertEqual(nts_str.dc_replication_factors['dc1'], 3) |
| self.assertEqual(nts_int.dc_replication_factors_info['dc1'], ReplicationFactor(3)) |
| self.assertEqual(nts_str.dc_replication_factors_info['dc1'], ReplicationFactor(3)) |
| |
| self.assertEqual(nts_int.export_for_schema(), nts_str.export_for_schema()) |
| self.assertEqual(nts_int, nts_str) |
| |
| # make token replica map |
| ring = [MD5Token(0), MD5Token(1), MD5Token(2)] |
| hosts = [Host('dc1.{}'.format(host), SimpleConvictionPolicy) for host in range(3)] |
| token_to_host = dict(zip(ring, hosts)) |
| self.assertEqual( |
| nts_int.make_token_replica_map(token_to_host, ring), |
| nts_str.make_token_replica_map(token_to_host, ring) |
| ) |
| |
| def test_nts_transient_parsing(self): |
| """ Test that we can PARSE a transient replication factor for NTS """ |
| rs = ReplicationStrategy() |
| |
| nts_transient = rs.create('NetworkTopologyStrategy', {'dc1': '3/1', 'dc2': '5/1'}) |
| self.assertEqual(nts_transient.dc_replication_factors_info['dc1'], ReplicationFactor(3, 1)) |
| self.assertEqual(nts_transient.dc_replication_factors_info['dc2'], ReplicationFactor(5, 1)) |
| self.assertEqual(nts_transient.dc_replication_factors['dc1'], 2) |
| self.assertEqual(nts_transient.dc_replication_factors['dc2'], 4) |
| self.assertIn("'dc1': '3/1', 'dc2': '5/1'", nts_transient.export_for_schema()) |
| |
| nts_str = rs.create('NetworkTopologyStrategy', {'dc1': '3', 'dc2': '5'}) |
| self.assertNotEqual(nts_transient, nts_str) |
| |
| # make token replica map |
| ring = [MD5Token(0), MD5Token(1), MD5Token(2)] |
| hosts = [Host('dc1.{}'.format(host), SimpleConvictionPolicy) for host in range(3)] |
| token_to_host = dict(zip(ring, hosts)) |
| self.assertEqual( |
| nts_transient.make_token_replica_map(token_to_host, ring), |
| nts_str.make_token_replica_map(token_to_host, ring) |
| ) |
| |
| def test_nts_make_token_replica_map(self): |
| token_to_host_owner = {} |
| |
| dc1_1 = Host('dc1.1', SimpleConvictionPolicy) |
| dc1_2 = Host('dc1.2', SimpleConvictionPolicy) |
| dc1_3 = Host('dc1.3', SimpleConvictionPolicy) |
| for host in (dc1_1, dc1_2, dc1_3): |
| host.set_location_info('dc1', 'rack1') |
| token_to_host_owner[MD5Token(0)] = dc1_1 |
| token_to_host_owner[MD5Token(100)] = dc1_2 |
| token_to_host_owner[MD5Token(200)] = dc1_3 |
| |
| dc2_1 = Host('dc2.1', SimpleConvictionPolicy) |
| dc2_2 = Host('dc2.2', SimpleConvictionPolicy) |
| dc2_1.set_location_info('dc2', 'rack1') |
| dc2_2.set_location_info('dc2', 'rack1') |
| token_to_host_owner[MD5Token(1)] = dc2_1 |
| token_to_host_owner[MD5Token(101)] = dc2_2 |
| |
| dc3_1 = Host('dc3.1', SimpleConvictionPolicy) |
| dc3_1.set_location_info('dc3', 'rack3') |
| token_to_host_owner[MD5Token(2)] = dc3_1 |
| |
| ring = [MD5Token(0), |
| MD5Token(1), |
| MD5Token(2), |
| MD5Token(100), |
| MD5Token(101), |
| MD5Token(200)] |
| |
| nts = NetworkTopologyStrategy({'dc1': 2, 'dc2': 2, 'dc3': 1}) |
| replica_map = nts.make_token_replica_map(token_to_host_owner, ring) |
| |
| self.assertItemsEqual(replica_map[MD5Token(0)], (dc1_1, dc1_2, dc2_1, dc2_2, dc3_1)) |
| |
| def test_nts_token_performance(self): |
| """ |
| Tests to ensure that when rf exceeds the number of nodes available, that we dont' |
| needlessly iterate trying to construct tokens for nodes that don't exist. |
| |
| @since 3.7 |
| @jira_ticket PYTHON-379 |
| @expected_result timing with 1500 rf should be same/similar to 3rf if we have 3 nodes |
| |
| @test_category metadata |
| """ |
| |
| token_to_host_owner = {} |
| ring = [] |
| dc1hostnum = 3 |
| current_token = 0 |
| vnodes_per_host = 500 |
| for i in range(dc1hostnum): |
| |
| host = Host('dc1.{0}'.format(i), SimpleConvictionPolicy) |
| host.set_location_info('dc1', "rack1") |
| for vnode_num in range(vnodes_per_host): |
| md5_token = MD5Token(current_token+vnode_num) |
| token_to_host_owner[md5_token] = host |
| ring.append(md5_token) |
| current_token += 1000 |
| |
| nts = NetworkTopologyStrategy({'dc1': 3}) |
| start_time = timeit.default_timer() |
| nts.make_token_replica_map(token_to_host_owner, ring) |
| elapsed_base = timeit.default_timer() - start_time |
| |
| nts = NetworkTopologyStrategy({'dc1': 1500}) |
| start_time = timeit.default_timer() |
| nts.make_token_replica_map(token_to_host_owner, ring) |
| elapsed_bad = timeit.default_timer() - start_time |
| difference = elapsed_bad - elapsed_base |
| self.assertTrue(difference < 1 and difference > -1) |
| |
| def test_nts_make_token_replica_map_multi_rack(self): |
| token_to_host_owner = {} |
| |
| # (A) not enough distinct racks, first skipped is used |
| dc1_1 = Host('dc1.1', SimpleConvictionPolicy) |
| dc1_2 = Host('dc1.2', SimpleConvictionPolicy) |
| dc1_3 = Host('dc1.3', SimpleConvictionPolicy) |
| dc1_4 = Host('dc1.4', SimpleConvictionPolicy) |
| dc1_1.set_location_info('dc1', 'rack1') |
| dc1_2.set_location_info('dc1', 'rack1') |
| dc1_3.set_location_info('dc1', 'rack2') |
| dc1_4.set_location_info('dc1', 'rack2') |
| token_to_host_owner[MD5Token(0)] = dc1_1 |
| token_to_host_owner[MD5Token(100)] = dc1_2 |
| token_to_host_owner[MD5Token(200)] = dc1_3 |
| token_to_host_owner[MD5Token(300)] = dc1_4 |
| |
| # (B) distinct racks, but not contiguous |
| dc2_1 = Host('dc2.1', SimpleConvictionPolicy) |
| dc2_2 = Host('dc2.2', SimpleConvictionPolicy) |
| dc2_3 = Host('dc2.3', SimpleConvictionPolicy) |
| dc2_1.set_location_info('dc2', 'rack1') |
| dc2_2.set_location_info('dc2', 'rack1') |
| dc2_3.set_location_info('dc2', 'rack2') |
| token_to_host_owner[MD5Token(1)] = dc2_1 |
| token_to_host_owner[MD5Token(101)] = dc2_2 |
| token_to_host_owner[MD5Token(201)] = dc2_3 |
| |
| ring = [MD5Token(0), |
| MD5Token(1), |
| MD5Token(100), |
| MD5Token(101), |
| MD5Token(200), |
| MD5Token(201), |
| MD5Token(300)] |
| |
| nts = NetworkTopologyStrategy({'dc1': 3, 'dc2': 2}) |
| replica_map = nts.make_token_replica_map(token_to_host_owner, ring) |
| |
| token_replicas = replica_map[MD5Token(0)] |
| self.assertItemsEqual(token_replicas, (dc1_1, dc1_2, dc1_3, dc2_1, dc2_3)) |
| |
| def test_nts_make_token_replica_map_empty_dc(self): |
| host = Host('1', SimpleConvictionPolicy) |
| host.set_location_info('dc1', 'rack1') |
| token_to_host_owner = {MD5Token(0): host} |
| ring = [MD5Token(0)] |
| nts = NetworkTopologyStrategy({'dc1': 1, 'dc2': 0}) |
| |
| replica_map = nts.make_token_replica_map(token_to_host_owner, ring) |
| self.assertEqual(set(replica_map[MD5Token(0)]), set([host])) |
| |
| def test_nts_export_for_schema(self): |
| strategy = NetworkTopologyStrategy({'dc1': '1', 'dc2': '2'}) |
| self.assertEqual("{'class': 'NetworkTopologyStrategy', 'dc1': '1', 'dc2': '2'}", |
| strategy.export_for_schema()) |
| |
| def test_simple_strategy_make_token_replica_map(self): |
| host1 = Host('1', SimpleConvictionPolicy) |
| host2 = Host('2', SimpleConvictionPolicy) |
| host3 = Host('3', SimpleConvictionPolicy) |
| token_to_host_owner = { |
| MD5Token(0): host1, |
| MD5Token(100): host2, |
| MD5Token(200): host3 |
| } |
| ring = [MD5Token(0), MD5Token(100), MD5Token(200)] |
| |
| rf1_replicas = SimpleStrategy({'replication_factor': '1'}).make_token_replica_map(token_to_host_owner, ring) |
| self.assertItemsEqual(rf1_replicas[MD5Token(0)], [host1]) |
| self.assertItemsEqual(rf1_replicas[MD5Token(100)], [host2]) |
| self.assertItemsEqual(rf1_replicas[MD5Token(200)], [host3]) |
| |
| rf2_replicas = SimpleStrategy({'replication_factor': '2'}).make_token_replica_map(token_to_host_owner, ring) |
| self.assertItemsEqual(rf2_replicas[MD5Token(0)], [host1, host2]) |
| self.assertItemsEqual(rf2_replicas[MD5Token(100)], [host2, host3]) |
| self.assertItemsEqual(rf2_replicas[MD5Token(200)], [host3, host1]) |
| |
| rf3_replicas = SimpleStrategy({'replication_factor': '3'}).make_token_replica_map(token_to_host_owner, ring) |
| self.assertItemsEqual(rf3_replicas[MD5Token(0)], [host1, host2, host3]) |
| self.assertItemsEqual(rf3_replicas[MD5Token(100)], [host2, host3, host1]) |
| self.assertItemsEqual(rf3_replicas[MD5Token(200)], [host3, host1, host2]) |
| |
| def test_ss_equals(self): |
| self.assertNotEqual(SimpleStrategy({'replication_factor': '1'}), NetworkTopologyStrategy({'dc1': 2})) |
| |
| |
| class NameEscapingTest(unittest.TestCase): |
| |
| def test_protect_name(self): |
| """ |
| Test cassandra.metadata.protect_name output |
| """ |
| self.assertEqual(protect_name('tests'), 'tests') |
| self.assertEqual(protect_name('test\'s'), '"test\'s"') |
| self.assertEqual(protect_name('test\'s'), "\"test's\"") |
| self.assertEqual(protect_name('tests ?!@#$%^&*()'), '"tests ?!@#$%^&*()"') |
| self.assertEqual(protect_name('1'), '"1"') |
| self.assertEqual(protect_name('1test'), '"1test"') |
| |
| def test_protect_names(self): |
| """ |
| Test cassandra.metadata.protect_names output |
| """ |
| self.assertEqual(protect_names(['tests']), ['tests']) |
| self.assertEqual(protect_names( |
| [ |
| 'tests', |
| 'test\'s', |
| 'tests ?!@#$%^&*()', |
| '1' |
| ]), |
| [ |
| 'tests', |
| "\"test's\"", |
| '"tests ?!@#$%^&*()"', |
| '"1"' |
| ]) |
| |
| def test_protect_value(self): |
| """ |
| Test cassandra.metadata.protect_value output |
| """ |
| self.assertEqual(protect_value(True), "true") |
| self.assertEqual(protect_value(False), "false") |
| self.assertEqual(protect_value(3.14), '3.14') |
| self.assertEqual(protect_value(3), '3') |
| self.assertEqual(protect_value('test'), "'test'") |
| self.assertEqual(protect_value('test\'s'), "'test''s'") |
| self.assertEqual(protect_value(None), 'NULL') |
| |
| def test_is_valid_name(self): |
| """ |
| Test cassandra.metadata.is_valid_name output |
| """ |
| self.assertEqual(is_valid_name(None), False) |
| self.assertEqual(is_valid_name('test'), True) |
| self.assertEqual(is_valid_name('Test'), False) |
| self.assertEqual(is_valid_name('t_____1'), True) |
| self.assertEqual(is_valid_name('test1'), True) |
| self.assertEqual(is_valid_name('1test1'), False) |
| |
| invalid_keywords = cassandra.metadata.cql_keywords - cassandra.metadata.cql_keywords_unreserved |
| for keyword in invalid_keywords: |
| self.assertEqual(is_valid_name(keyword), False) |
| |
| |
| class GetReplicasTest(unittest.TestCase): |
| def _get_replicas(self, token_klass): |
| tokens = [token_klass(i) for i in range(0, (2 ** 127 - 1), 2 ** 125)] |
| hosts = [Host("ip%d" % i, SimpleConvictionPolicy) for i in range(len(tokens))] |
| token_to_primary_replica = dict(zip(tokens, hosts)) |
| keyspace = KeyspaceMetadata("ks", True, "SimpleStrategy", {"replication_factor": "1"}) |
| metadata = Mock(spec=Metadata, keyspaces={'ks': keyspace}) |
| token_map = TokenMap(token_klass, token_to_primary_replica, tokens, metadata) |
| |
| # tokens match node tokens exactly |
| for token, expected_host in zip(tokens, hosts): |
| replicas = token_map.get_replicas("ks", token) |
| self.assertEqual(set(replicas), {expected_host}) |
| |
| # shift the tokens back by one |
| for token, expected_host in zip(tokens, hosts): |
| replicas = token_map.get_replicas("ks", token_klass(token.value - 1)) |
| self.assertEqual(set(replicas), {expected_host}) |
| |
| # shift the tokens forward by one |
| for i, token in enumerate(tokens): |
| replicas = token_map.get_replicas("ks", token_klass(token.value + 1)) |
| expected_host = hosts[(i + 1) % len(hosts)] |
| self.assertEqual(set(replicas), {expected_host}) |
| |
| def test_murmur3_tokens(self): |
| self._get_replicas(Murmur3Token) |
| |
| def test_md5_tokens(self): |
| self._get_replicas(MD5Token) |
| |
| def test_bytes_tokens(self): |
| self._get_replicas(BytesToken) |
| |
| |
| class Murmur3TokensTest(unittest.TestCase): |
| |
| def test_murmur3_init(self): |
| murmur3_token = Murmur3Token(cassandra.metadata.MIN_LONG - 1) |
| self.assertEqual(str(murmur3_token), '<Murmur3Token: -9223372036854775809>') |
| |
| def test_python_vs_c(self): |
| from cassandra.murmur3 import _murmur3 as mm3_python |
| try: |
| from cassandra.cmurmur3 import murmur3 as mm3_c |
| |
| iterations = 100 |
| for _ in range(iterations): |
| for len in range(0, 32): # zero to one block plus full range of tail lengths |
| key = os.urandom(len) |
| self.assertEqual(mm3_python(key), mm3_c(key)) |
| |
| except ImportError: |
| raise unittest.SkipTest('The cmurmur3 extension is not available') |
| |
| def test_murmur3_python(self): |
| from cassandra.murmur3 import _murmur3 |
| self._verify_hash(_murmur3) |
| |
| def test_murmur3_c(self): |
| try: |
| from cassandra.cmurmur3 import murmur3 |
| self._verify_hash(murmur3) |
| except ImportError: |
| raise unittest.SkipTest('The cmurmur3 extension is not available') |
| |
| def _verify_hash(self, fn): |
| self.assertEqual(fn(b'123'), -7468325962851647638) |
| self.assertEqual(fn(b'\x00\xff\x10\xfa\x99' * 10), 5837342703291459765) |
| self.assertEqual(fn(b'\xfe' * 8), -8927430733708461935) |
| self.assertEqual(fn(b'\x10' * 8), 1446172840243228796) |
| self.assertEqual(fn(str(cassandra.metadata.MAX_LONG).encode()), 7162290910810015547) |
| |
| |
| class MD5TokensTest(unittest.TestCase): |
| |
| def test_md5_tokens(self): |
| md5_token = MD5Token(cassandra.metadata.MIN_LONG - 1) |
| self.assertEqual(md5_token.hash_fn('123'), 42767516990368493138776584305024125808) |
| self.assertEqual(md5_token.hash_fn(str(cassandra.metadata.MAX_LONG)), 28528976619278518853815276204542453639) |
| self.assertEqual(str(md5_token), '<MD5Token: %s>' % -9223372036854775809) |
| |
| |
| class BytesTokensTest(unittest.TestCase): |
| |
| def test_bytes_tokens(self): |
| bytes_token = BytesToken(unhexlify(b'01')) |
| self.assertEqual(bytes_token.value, b'\x01') |
| self.assertEqual(str(bytes_token), "<BytesToken: %s>" % bytes_token.value) |
| self.assertEqual(bytes_token.hash_fn('123'), '123') |
| self.assertEqual(bytes_token.hash_fn(123), 123) |
| self.assertEqual(bytes_token.hash_fn(str(cassandra.metadata.MAX_LONG)), str(cassandra.metadata.MAX_LONG)) |
| |
| def test_from_string(self): |
| from_unicode = BytesToken.from_string('0123456789abcdef') |
| from_bin = BytesToken.from_string(b'0123456789abcdef') |
| self.assertEqual(from_unicode, from_bin) |
| self.assertIsInstance(from_unicode.value, bytes) |
| self.assertIsInstance(from_bin.value, bytes) |
| |
| def test_comparison(self): |
| tok = BytesToken.from_string('0123456789abcdef') |
| token_high_order = uint16_unpack(tok.value[0:2]) |
| self.assertLess(BytesToken(uint16_pack(token_high_order - 1)), tok) |
| self.assertGreater(BytesToken(uint16_pack(token_high_order + 1)), tok) |
| |
| def test_comparison_unicode(self): |
| value = b'\'_-()"\xc2\xac' |
| t0 = BytesToken(value) |
| t1 = BytesToken.from_string('00') |
| self.assertGreater(t0, t1) |
| self.assertFalse(t0 < t1) |
| |
| |
| class KeyspaceMetadataTest(unittest.TestCase): |
| |
| def test_export_as_string_user_types(self): |
| keyspace_name = 'test' |
| keyspace = KeyspaceMetadata(keyspace_name, True, 'SimpleStrategy', dict(replication_factor=3)) |
| keyspace.user_types['a'] = UserType(keyspace_name, 'a', ['one', 'two'], ['c', 'int']) |
| keyspace.user_types['b'] = UserType(keyspace_name, 'b', ['one', 'two', 'three'], ['d', 'int', 'a']) |
| keyspace.user_types['c'] = UserType(keyspace_name, 'c', ['one'], ['int']) |
| keyspace.user_types['d'] = UserType(keyspace_name, 'd', ['one'], ['c']) |
| |
| self.assertEqual("""CREATE KEYSPACE test WITH replication = {'class': 'SimpleStrategy', 'replication_factor': '3'} AND durable_writes = true; |
| |
| CREATE TYPE test.c ( |
| one int |
| ); |
| |
| CREATE TYPE test.a ( |
| one c, |
| two int |
| ); |
| |
| CREATE TYPE test.d ( |
| one c |
| ); |
| |
| CREATE TYPE test.b ( |
| one d, |
| two int, |
| three a |
| );""", keyspace.export_as_string()) |
| |
| |
| class UserTypesTest(unittest.TestCase): |
| |
| def test_as_cql_query(self): |
| field_types = ['varint', 'ascii', 'frozen<tuple<varint, ascii>>'] |
| udt = UserType("ks1", "mytype", ["a", "b", "c"], field_types) |
| self.assertEqual("CREATE TYPE ks1.mytype (a varint, b ascii, c frozen<tuple<varint, ascii>>)", udt.as_cql_query(formatted=False)) |
| |
| self.assertEqual("""CREATE TYPE ks1.mytype ( |
| a varint, |
| b ascii, |
| c frozen<tuple<varint, ascii>> |
| );""", udt.export_as_string()) |
| |
| def test_as_cql_query_name_escaping(self): |
| udt = UserType("MyKeyspace", "MyType", ["AbA", "keyspace"], ['ascii', 'ascii']) |
| self.assertEqual('CREATE TYPE "MyKeyspace"."MyType" ("AbA" ascii, "keyspace" ascii)', udt.as_cql_query(formatted=False)) |
| |
| |
| class UserDefinedFunctionTest(unittest.TestCase): |
| def test_as_cql_query_removes_frozen(self): |
| func = Function( |
| "ks1", "myfunction", ["frozen<tuple<int, text>>"], ["a"], |
| "int", "java", "return 0;", True, False, False, False |
| ) |
| expected_result = ( |
| "CREATE FUNCTION ks1.myfunction(a tuple<int, text>) " |
| "CALLED ON NULL INPUT " |
| "RETURNS int " |
| "LANGUAGE java " |
| "AS $$return 0;$$" |
| ) |
| self.assertEqual(expected_result, func.as_cql_query(formatted=False)) |
| |
| |
| class UserDefinedAggregateTest(unittest.TestCase): |
| def test_as_cql_query_removes_frozen(self): |
| aggregate = Aggregate("ks1", "myaggregate", ["frozen<tuple<int>>"], "statefunc", "frozen<tuple<int>>", "finalfunc", "(0)", "tuple<int>", False) |
| expected_result = ( |
| "CREATE AGGREGATE ks1.myaggregate(tuple<int>) " |
| "SFUNC statefunc " |
| "STYPE tuple<int> " |
| "FINALFUNC finalfunc " |
| "INITCOND (0)" |
| ) |
| self.assertEqual(expected_result, aggregate.as_cql_query(formatted=False)) |
| |
| |
| class IndexTest(unittest.TestCase): |
| |
| def test_build_index_as_cql(self): |
| column_meta = Mock() |
| column_meta.name = 'column_name_here' |
| column_meta.table.name = 'table_name_here' |
| column_meta.table.keyspace_name = 'keyspace_name_here' |
| column_meta.table.columns = {column_meta.name: column_meta} |
| parser = get_schema_parser(Mock(), '2.1.0', None, 0.1) |
| |
| row = {'index_name': 'index_name_here', 'index_type': 'index_type_here'} |
| index_meta = parser._build_index_metadata(column_meta, row) |
| self.assertEqual(index_meta.as_cql_query(), |
| 'CREATE INDEX index_name_here ON keyspace_name_here.table_name_here (column_name_here)') |
| |
| row['index_options'] = '{ "class_name": "class_name_here" }' |
| row['index_type'] = 'CUSTOM' |
| index_meta = parser._build_index_metadata(column_meta, row) |
| self.assertEqual(index_meta.as_cql_query(), |
| "CREATE CUSTOM INDEX index_name_here ON keyspace_name_here.table_name_here (column_name_here) USING 'class_name_here'") |
| |
| |
| class UnicodeIdentifiersTests(unittest.TestCase): |
| """ |
| Exercise cql generation with unicode characters. Keyspace, Table, and Index names |
| cannot have special chars because C* names files by those identifiers, but they are |
| tested anyway. |
| |
| Looking for encoding errors like PYTHON-447 |
| """ |
| |
| name = b'\'_-()"\xc2\xac'.decode('utf-8') |
| |
| def test_keyspace_name(self): |
| km = KeyspaceMetadata(self.name, False, 'SimpleStrategy', {'replication_factor': 1}) |
| km.export_as_string() |
| |
| def test_table_name(self): |
| tm = TableMetadata(self.name, self.name) |
| tm.export_as_string() |
| |
| def test_column_name_single_partition(self): |
| tm = TableMetadata('ks', 'table') |
| cm = ColumnMetadata(tm, self.name, u'int') |
| tm.columns[cm.name] = cm |
| tm.partition_key.append(cm) |
| tm.export_as_string() |
| |
| def test_column_name_single_partition_single_clustering(self): |
| tm = TableMetadata('ks', 'table') |
| cm = ColumnMetadata(tm, self.name, u'int') |
| tm.columns[cm.name] = cm |
| tm.partition_key.append(cm) |
| cm = ColumnMetadata(tm, self.name + 'x', u'int') |
| tm.columns[cm.name] = cm |
| tm.clustering_key.append(cm) |
| tm.export_as_string() |
| |
| def test_column_name_multiple_partition(self): |
| tm = TableMetadata('ks', 'table') |
| cm = ColumnMetadata(tm, self.name, u'int') |
| tm.columns[cm.name] = cm |
| tm.partition_key.append(cm) |
| cm = ColumnMetadata(tm, self.name + 'x', u'int') |
| tm.columns[cm.name] = cm |
| tm.partition_key.append(cm) |
| tm.export_as_string() |
| |
| def test_index(self): |
| im = IndexMetadata(self.name, self.name, self.name, kind='', index_options={'target': self.name}) |
| log.debug(im.export_as_string()) |
| im = IndexMetadata(self.name, self.name, self.name, kind='CUSTOM', index_options={'target': self.name, 'class_name': 'Class'}) |
| log.debug(im.export_as_string()) |
| # PYTHON-1008 |
| im = IndexMetadata(self.name, self.name, self.name, kind='CUSTOM', index_options={'target': self.name, 'class_name': 'Class', 'delimiter': self.name}) |
| log.debug(im.export_as_string()) |
| |
| def test_function(self): |
| fm = Function(keyspace=self.name, name=self.name, |
| argument_types=(u'int', u'int'), |
| argument_names=(u'x', u'y'), |
| return_type=u'int', language=u'language', |
| body=self.name, called_on_null_input=False, |
| deterministic=True, |
| monotonic=False, monotonic_on=(u'x',)) |
| fm.export_as_string() |
| |
| def test_aggregate(self): |
| am = Aggregate(self.name, self.name, (u'text',), self.name, u'text', self.name, self.name, u'text', True) |
| am.export_as_string() |
| |
| def test_user_type(self): |
| um = UserType(self.name, self.name, [self.name, self.name], [u'int', u'text']) |
| um.export_as_string() |
| |
| |
| class FunctionToCQLTests(unittest.TestCase): |
| |
| base_vars = { |
| 'keyspace': 'ks_name', |
| 'name': 'function_name', |
| 'argument_types': (u'int', u'int'), |
| 'argument_names': (u'x', u'y'), |
| 'return_type': u'int', |
| 'language': u'language', |
| 'body': 'body', |
| 'called_on_null_input': False, |
| 'deterministic': True, |
| 'monotonic': False, |
| 'monotonic_on': () |
| } |
| |
| def _function_with_kwargs(self, **kwargs): |
| return Function(**dict(self.base_vars, |
| **kwargs) |
| ) |
| |
| def test_non_monotonic(self): |
| self.assertNotIn( |
| 'MONOTONIC', |
| self._function_with_kwargs( |
| monotonic=False, |
| monotonic_on=() |
| ).export_as_string() |
| ) |
| |
| def test_monotonic_all(self): |
| mono_function = self._function_with_kwargs( |
| monotonic=True, |
| monotonic_on=() |
| ) |
| self.assertIn( |
| 'MONOTONIC LANG', |
| mono_function.as_cql_query(formatted=False) |
| ) |
| self.assertIn( |
| 'MONOTONIC\n LANG', |
| mono_function.as_cql_query(formatted=True) |
| ) |
| |
| def test_monotonic_one(self): |
| mono_on_function = self._function_with_kwargs( |
| monotonic=False, |
| monotonic_on=('x',) |
| ) |
| self.assertIn( |
| 'MONOTONIC ON x LANG', |
| mono_on_function.as_cql_query(formatted=False) |
| ) |
| self.assertIn( |
| 'MONOTONIC ON x\n LANG', |
| mono_on_function.as_cql_query(formatted=True) |
| ) |
| |
| def test_nondeterministic(self): |
| self.assertNotIn( |
| 'DETERMINISTIC', |
| self._function_with_kwargs( |
| deterministic=False |
| ).as_cql_query(formatted=False) |
| ) |
| |
| def test_deterministic(self): |
| self.assertIn( |
| 'DETERMINISTIC', |
| self._function_with_kwargs( |
| deterministic=True |
| ).as_cql_query(formatted=False) |
| ) |
| self.assertIn( |
| 'DETERMINISTIC\n', |
| self._function_with_kwargs( |
| deterministic=True |
| ).as_cql_query(formatted=True) |
| ) |
| |
| |
| class AggregateToCQLTests(unittest.TestCase): |
| base_vars = { |
| 'keyspace': 'ks_name', |
| 'name': 'function_name', |
| 'argument_types': (u'int', u'int'), |
| 'state_func': 'funcname', |
| 'state_type': u'int', |
| 'return_type': u'int', |
| 'final_func': None, |
| 'initial_condition': '0', |
| 'deterministic': True |
| } |
| |
| def _aggregate_with_kwargs(self, **kwargs): |
| return Aggregate(**dict(self.base_vars, |
| **kwargs) |
| ) |
| |
| def test_nondeterministic(self): |
| self.assertNotIn( |
| 'DETERMINISTIC', |
| self._aggregate_with_kwargs( |
| deterministic=False |
| ).as_cql_query(formatted=True) |
| ) |
| |
| def test_deterministic(self): |
| for formatted in (True, False): |
| query = self._aggregate_with_kwargs( |
| deterministic=True |
| ).as_cql_query(formatted=formatted) |
| self.assertTrue(query.endswith('DETERMINISTIC'), |
| msg="'DETERMINISTIC' not found in {}".format(query) |
| ) |
| |
| |
| class HostsTests(unittest.TestCase): |
| def test_iterate_all_hosts_and_modify(self): |
| """ |
| PYTHON-572 |
| """ |
| metadata = Metadata() |
| metadata.add_or_return_host(Host('dc1.1', SimpleConvictionPolicy)) |
| metadata.add_or_return_host(Host('dc1.2', SimpleConvictionPolicy)) |
| |
| self.assertEqual(len(metadata.all_hosts()), 2) |
| |
| for host in metadata.all_hosts(): # this would previously raise in Py3 |
| metadata.remove_host(host) |
| |
| self.assertEqual(len(metadata.all_hosts()), 0) |
| |
| |
| class MetadataHelpersTest(unittest.TestCase): |
| """ For any helper functions that need unit tests """ |
| def test_strip_frozen(self): |
| self.longMessage = True |
| |
| argument_to_expected_results = [ |
| ('int', 'int'), |
| ('tuple<text>', 'tuple<text>'), |
| (r'map<"!@#$%^&*()[]\ frozen >>>", int>', r'map<"!@#$%^&*()[]\ frozen >>>", int>'), # A valid UDT name |
| ('frozen<tuple<text>>', 'tuple<text>'), |
| (r'frozen<map<"!@#$%^&*()[]\ frozen >>>", int>>', r'map<"!@#$%^&*()[]\ frozen >>>", int>'), |
| ('frozen<map<frozen<tuple<int, frozen<list<text>>, int>>, frozen<map<int, frozen<tuple<int>>>>>>', |
| 'map<tuple<int, list<text>, int>, map<int, tuple<int>>>'), |
| ] |
| for argument, expected_result in argument_to_expected_results: |
| result = strip_frozen(argument) |
| self.assertEqual(result, expected_result, "strip_frozen() arg: {}".format(argument)) |
