| # 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 unittest import mock |
| import logging |
| from packaging.version import Version |
| from cassandra.cqlengine.connection import get_session, get_cluster |
| from cassandra.cqlengine import CQLEngineException |
| from cassandra.cqlengine import management |
| from cassandra.cqlengine.management import _get_table_metadata, sync_table, drop_table, sync_type |
| from cassandra.cqlengine.models import Model |
| from cassandra.cqlengine import columns |
| |
| from tests.integration import DSE_VERSION, PROTOCOL_VERSION, greaterthancass20, MockLoggingHandler, CASSANDRA_VERSION |
| from tests.integration.cqlengine.base import BaseCassEngTestCase |
| from tests.integration.cqlengine.query.test_queryset import TestModel |
| from cassandra.cqlengine.usertype import UserType |
| from tests.integration.cqlengine import DEFAULT_KEYSPACE |
| |
| |
| INCLUDE_REPAIR = not CASSANDRA_VERSION >= Version('4-a') # This should cover DSE 6.0+ |
| |
| |
| class KeyspaceManagementTest(BaseCassEngTestCase): |
| def test_create_drop_succeeeds(self): |
| cluster = get_cluster() |
| |
| keyspace_ss = 'test_ks_ss' |
| self.assertNotIn(keyspace_ss, cluster.metadata.keyspaces) |
| management.create_keyspace_simple(keyspace_ss, 2) |
| self.assertIn(keyspace_ss, cluster.metadata.keyspaces) |
| |
| management.drop_keyspace(keyspace_ss) |
| self.assertNotIn(keyspace_ss, cluster.metadata.keyspaces) |
| |
| keyspace_nts = 'test_ks_nts' |
| self.assertNotIn(keyspace_nts, cluster.metadata.keyspaces) |
| management.create_keyspace_network_topology(keyspace_nts, {'dc1': 1}) |
| self.assertIn(keyspace_nts, cluster.metadata.keyspaces) |
| |
| management.drop_keyspace(keyspace_nts) |
| self.assertNotIn(keyspace_nts, cluster.metadata.keyspaces) |
| |
| |
| class DropTableTest(BaseCassEngTestCase): |
| |
| def test_multiple_deletes_dont_fail(self): |
| sync_table(TestModel) |
| |
| drop_table(TestModel) |
| drop_table(TestModel) |
| |
| |
| class LowercaseKeyModel(Model): |
| |
| first_key = columns.Integer(primary_key=True) |
| second_key = columns.Integer(primary_key=True) |
| some_data = columns.Text() |
| |
| |
| class CapitalizedKeyModel(Model): |
| |
| firstKey = columns.Integer(primary_key=True) |
| secondKey = columns.Integer(primary_key=True) |
| someData = columns.Text() |
| |
| |
| class PrimaryKeysOnlyModel(Model): |
| |
| __table_name__ = "primary_keys_only" |
| __options__ = {'compaction': {'class': 'LeveledCompactionStrategy'}} |
| |
| first_key = columns.Integer(primary_key=True) |
| second_key = columns.Integer(primary_key=True) |
| |
| |
| class PrimaryKeysModelChanged(Model): |
| |
| __table_name__ = "primary_keys_only" |
| __options__ = {'compaction': {'class': 'LeveledCompactionStrategy'}} |
| |
| new_first_key = columns.Integer(primary_key=True) |
| second_key = columns.Integer(primary_key=True) |
| |
| |
| class PrimaryKeysModelTypeChanged(Model): |
| |
| __table_name__ = "primary_keys_only" |
| __options__ = {'compaction': {'class': 'LeveledCompactionStrategy'}} |
| |
| first_key = columns.Float(primary_key=True) |
| second_key = columns.Integer(primary_key=True) |
| |
| |
| class PrimaryKeysRemovedPk(Model): |
| |
| __table_name__ = "primary_keys_only" |
| __options__ = {'compaction': {'class': 'LeveledCompactionStrategy'}} |
| |
| second_key = columns.Integer(primary_key=True) |
| |
| |
| class PrimaryKeysAddedClusteringKey(Model): |
| |
| __table_name__ = "primary_keys_only" |
| __options__ = {'compaction': {'class': 'LeveledCompactionStrategy'}} |
| |
| new_first_key = columns.Float(primary_key=True) |
| second_key = columns.Integer(primary_key=True) |
| |
| |
| class CapitalizedKeyTest(BaseCassEngTestCase): |
| |
| def test_table_definition(self): |
| """ Tests that creating a table with capitalized column names succeeds """ |
| sync_table(LowercaseKeyModel) |
| sync_table(CapitalizedKeyModel) |
| |
| drop_table(LowercaseKeyModel) |
| drop_table(CapitalizedKeyModel) |
| |
| |
| class FirstModel(Model): |
| |
| __table_name__ = 'first_model' |
| first_key = columns.UUID(primary_key=True) |
| second_key = columns.UUID() |
| third_key = columns.Text() |
| |
| |
| class SecondModel(Model): |
| |
| __table_name__ = 'first_model' |
| first_key = columns.UUID(primary_key=True) |
| second_key = columns.UUID() |
| third_key = columns.Text() |
| fourth_key = columns.Text() |
| |
| |
| class ThirdModel(Model): |
| |
| __table_name__ = 'first_model' |
| first_key = columns.UUID(primary_key=True) |
| second_key = columns.UUID() |
| third_key = columns.Text() |
| # removed fourth key, but it should stay in the DB |
| blah = columns.Map(columns.Text, columns.Text) |
| |
| |
| class FourthModel(Model): |
| |
| __table_name__ = 'first_model' |
| first_key = columns.UUID(primary_key=True) |
| second_key = columns.UUID() |
| third_key = columns.Text() |
| # renamed model field, but map to existing column |
| renamed = columns.Map(columns.Text, columns.Text, db_field='blah') |
| |
| |
| class AddColumnTest(BaseCassEngTestCase): |
| def setUp(self): |
| drop_table(FirstModel) |
| |
| def test_add_column(self): |
| sync_table(FirstModel) |
| meta_columns = _get_table_metadata(FirstModel).columns |
| self.assertEqual(set(meta_columns), set(FirstModel._columns)) |
| |
| sync_table(SecondModel) |
| meta_columns = _get_table_metadata(FirstModel).columns |
| self.assertEqual(set(meta_columns), set(SecondModel._columns)) |
| |
| sync_table(ThirdModel) |
| meta_columns = _get_table_metadata(FirstModel).columns |
| self.assertEqual(len(meta_columns), 5) |
| self.assertEqual(len(ThirdModel._columns), 4) |
| self.assertIn('fourth_key', meta_columns) |
| self.assertNotIn('fourth_key', ThirdModel._columns) |
| self.assertIn('blah', ThirdModel._columns) |
| self.assertIn('blah', meta_columns) |
| |
| sync_table(FourthModel) |
| meta_columns = _get_table_metadata(FirstModel).columns |
| self.assertEqual(len(meta_columns), 5) |
| self.assertEqual(len(ThirdModel._columns), 4) |
| self.assertIn('fourth_key', meta_columns) |
| self.assertNotIn('fourth_key', FourthModel._columns) |
| self.assertIn('renamed', FourthModel._columns) |
| self.assertNotIn('renamed', meta_columns) |
| self.assertIn('blah', meta_columns) |
| |
| |
| class ModelWithTableProperties(Model): |
| |
| __options__ = {'bloom_filter_fp_chance': '0.76328', |
| 'comment': 'TxfguvBdzwROQALmQBOziRMbkqVGFjqcJfVhwGR', |
| 'gc_grace_seconds': '2063'} |
| |
| if INCLUDE_REPAIR: |
| __options__.update( |
| {'read_repair_chance': '0.17985', |
| 'dclocal_read_repair_chance': '0.50811'} |
| ) |
| |
| key = columns.UUID(primary_key=True) |
| |
| |
| class TablePropertiesTests(BaseCassEngTestCase): |
| |
| def setUp(self): |
| drop_table(ModelWithTableProperties) |
| |
| def test_set_table_properties(self): |
| |
| sync_table(ModelWithTableProperties) |
| expected = {'bloom_filter_fp_chance': 0.76328, |
| 'comment': 'TxfguvBdzwROQALmQBOziRMbkqVGFjqcJfVhwGR', |
| 'gc_grace_seconds': 2063, |
| # For some reason 'dclocal_read_repair_chance' in CQL is called |
| # just 'local_read_repair_chance' in the schema table. |
| # Source: https://issues.apache.org/jira/browse/CASSANDRA-6717 |
| # TODO: due to a bug in the native driver i'm not seeing the local read repair chance show up |
| # 'local_read_repair_chance': 0.50811, |
| } |
| if INCLUDE_REPAIR: |
| expected.update({'read_repair_chance': 0.17985}) |
| |
| options = management._get_table_metadata(ModelWithTableProperties).options |
| self.assertEqual(dict([(k, options.get(k)) for k in expected.keys()]), |
| expected) |
| |
| def test_table_property_update(self): |
| ModelWithTableProperties.__options__['bloom_filter_fp_chance'] = 0.66778 |
| ModelWithTableProperties.__options__['comment'] = 'xirAkRWZVVvsmzRvXamiEcQkshkUIDINVJZgLYSdnGHweiBrAiJdLJkVohdRy' |
| ModelWithTableProperties.__options__['gc_grace_seconds'] = 96362 |
| |
| if INCLUDE_REPAIR: |
| ModelWithTableProperties.__options__['read_repair_chance'] = 0.2989 |
| ModelWithTableProperties.__options__['dclocal_read_repair_chance'] = 0.12732 |
| |
| sync_table(ModelWithTableProperties) |
| |
| table_options = management._get_table_metadata(ModelWithTableProperties).options |
| |
| self.assertLessEqual(ModelWithTableProperties.__options__.items(), table_options.items()) |
| |
| def test_bogus_option_update(self): |
| sync_table(ModelWithTableProperties) |
| option = 'no way will this ever be an option' |
| try: |
| ModelWithTableProperties.__options__[option] = 'what was I thinking?' |
| self.assertRaisesRegex(KeyError, "Invalid table option.*%s.*" % option, sync_table, ModelWithTableProperties) |
| finally: |
| ModelWithTableProperties.__options__.pop(option, None) |
| |
| |
| class SyncTableTests(BaseCassEngTestCase): |
| |
| def setUp(self): |
| drop_table(PrimaryKeysOnlyModel) |
| |
| def test_sync_table_works_with_primary_keys_only_tables(self): |
| |
| sync_table(PrimaryKeysOnlyModel) |
| # blows up with DoesNotExist if table does not exist |
| table_meta = management._get_table_metadata(PrimaryKeysOnlyModel) |
| |
| self.assertIn('LeveledCompactionStrategy', table_meta.as_cql_query()) |
| |
| PrimaryKeysOnlyModel.__options__['compaction']['class'] = 'SizeTieredCompactionStrategy' |
| |
| sync_table(PrimaryKeysOnlyModel) |
| |
| table_meta = management._get_table_metadata(PrimaryKeysOnlyModel) |
| self.assertIn('SizeTieredCompactionStrategy', table_meta.as_cql_query()) |
| |
| def test_primary_key_validation(self): |
| """ |
| Test to ensure that changes to primary keys throw CQLEngineExceptions |
| |
| @since 3.2 |
| @jira_ticket PYTHON-532 |
| @expected_result Attempts to modify primary keys throw an exception |
| |
| @test_category object_mapper |
| """ |
| sync_table(PrimaryKeysOnlyModel) |
| self.assertRaises(CQLEngineException, sync_table, PrimaryKeysModelChanged) |
| self.assertRaises(CQLEngineException, sync_table, PrimaryKeysAddedClusteringKey) |
| self.assertRaises(CQLEngineException, sync_table, PrimaryKeysRemovedPk) |
| |
| |
| class IndexModel(Model): |
| |
| __table_name__ = 'index_model' |
| first_key = columns.UUID(primary_key=True) |
| second_key = columns.Text(index=True) |
| |
| |
| class IndexCaseSensitiveModel(Model): |
| |
| __table_name__ = 'IndexModel' |
| __table_name_case_sensitive__ = True |
| first_key = columns.UUID(primary_key=True) |
| second_key = columns.Text(index=True) |
| |
| |
| class BaseInconsistent(Model): |
| |
| __table_name__ = 'inconsistent' |
| first_key = columns.UUID(primary_key=True) |
| second_key = columns.Integer(index=True) |
| third_key = columns.Integer(index=True) |
| |
| |
| class ChangedInconsistent(Model): |
| |
| __table_name__ = 'inconsistent' |
| __table_name_case_sensitive__ = True |
| first_key = columns.UUID(primary_key=True) |
| second_key = columns.Text(index=True) |
| |
| |
| class BaseInconsistentType(UserType): |
| __type_name__ = 'type_inconsistent' |
| age = columns.Integer() |
| name = columns.Text() |
| |
| |
| class ChangedInconsistentType(UserType): |
| __type_name__ = 'type_inconsistent' |
| age = columns.Integer() |
| name = columns.Integer() |
| |
| |
| class InconsistentTable(BaseCassEngTestCase): |
| |
| def setUp(self): |
| drop_table(IndexModel) |
| |
| def test_sync_warnings(self): |
| """ |
| Test to insure when inconsistent changes are made to a table, or type as part of a sync call that the proper logging messages are surfaced |
| |
| @since 3.2 |
| @jira_ticket PYTHON-260 |
| @expected_result warnings are logged |
| |
| @test_category object_mapper |
| """ |
| mock_handler = MockLoggingHandler() |
| logger = logging.getLogger(management.__name__) |
| logger.addHandler(mock_handler) |
| sync_table(BaseInconsistent) |
| sync_table(ChangedInconsistent) |
| self.assertTrue('differing from the model type' in mock_handler.messages.get('warning')[0]) |
| if CASSANDRA_VERSION >= Version('2.1'): |
| sync_type(DEFAULT_KEYSPACE, BaseInconsistentType) |
| mock_handler.reset() |
| sync_type(DEFAULT_KEYSPACE, ChangedInconsistentType) |
| self.assertTrue('differing from the model user type' in mock_handler.messages.get('warning')[0]) |
| logger.removeHandler(mock_handler) |
| |
| |
| class TestIndexSetModel(Model): |
| partition = columns.UUID(primary_key=True) |
| int_set = columns.Set(columns.Integer, index=True) |
| int_list = columns.List(columns.Integer, index=True) |
| text_map = columns.Map(columns.Text, columns.DateTime, index=True) |
| mixed_tuple = columns.Tuple(columns.Text, columns.Integer, columns.Text, index=True) |
| |
| |
| class IndexTests(BaseCassEngTestCase): |
| |
| def setUp(self): |
| drop_table(IndexModel) |
| drop_table(IndexCaseSensitiveModel) |
| |
| def test_sync_index(self): |
| """ |
| Tests the default table creation, and ensures the table_name is created and surfaced correctly |
| in the table metadata |
| |
| @since 3.1 |
| @jira_ticket PYTHON-337 |
| @expected_result table_name is lower case |
| |
| @test_category object_mapper |
| """ |
| sync_table(IndexModel) |
| table_meta = management._get_table_metadata(IndexModel) |
| self.assertIsNotNone(management._get_index_name_by_column(table_meta, 'second_key')) |
| |
| # index already exists |
| sync_table(IndexModel) |
| table_meta = management._get_table_metadata(IndexModel) |
| self.assertIsNotNone(management._get_index_name_by_column(table_meta, 'second_key')) |
| |
| def test_sync_index_case_sensitive(self): |
| """ |
| Tests the default table creation, and ensures the table_name is created correctly and surfaced correctly |
| in table metadata |
| |
| @since 3.1 |
| @jira_ticket PYTHON-337 |
| @expected_result table_name is lower case |
| |
| @test_category object_mapper |
| """ |
| sync_table(IndexCaseSensitiveModel) |
| table_meta = management._get_table_metadata(IndexCaseSensitiveModel) |
| self.assertIsNotNone(management._get_index_name_by_column(table_meta, 'second_key')) |
| |
| # index already exists |
| sync_table(IndexCaseSensitiveModel) |
| table_meta = management._get_table_metadata(IndexCaseSensitiveModel) |
| self.assertIsNotNone(management._get_index_name_by_column(table_meta, 'second_key')) |
| |
| @greaterthancass20 |
| def test_sync_indexed_set(self): |
| """ |
| Tests that models that have container types with indices can be synced. |
| |
| @since 3.2 |
| @jira_ticket PYTHON-533 |
| @expected_result table_sync should complete without a server error. |
| |
| @test_category object_mapper |
| """ |
| sync_table(TestIndexSetModel) |
| table_meta = management._get_table_metadata(TestIndexSetModel) |
| self.assertIsNotNone(management._get_index_name_by_column(table_meta, 'int_set')) |
| self.assertIsNotNone(management._get_index_name_by_column(table_meta, 'int_list')) |
| self.assertIsNotNone(management._get_index_name_by_column(table_meta, 'text_map')) |
| self.assertIsNotNone(management._get_index_name_by_column(table_meta, 'mixed_tuple')) |
| |
| |
| class NonModelFailureTest(BaseCassEngTestCase): |
| class FakeModel(object): |
| pass |
| |
| def test_failure(self): |
| with self.assertRaises(CQLEngineException): |
| sync_table(self.FakeModel) |
| |
| |
| class StaticColumnTests(BaseCassEngTestCase): |
| def test_static_columns(self): |
| if PROTOCOL_VERSION < 2: |
| raise unittest.SkipTest("Native protocol 2+ required, currently using: {0}".format(PROTOCOL_VERSION)) |
| |
| class StaticModel(Model): |
| id = columns.Integer(primary_key=True) |
| c = columns.Integer(primary_key=True) |
| name = columns.Text(static=True) |
| |
| drop_table(StaticModel) |
| |
| session = get_session() |
| |
| with mock.patch.object(session, "execute", wraps=session.execute) as m: |
| sync_table(StaticModel) |
| |
| self.assertGreater(m.call_count, 0) |
| statement = m.call_args[0][0].query_string |
| self.assertIn('"name" text static', statement) |
| |
| # if we sync again, we should not apply an alter w/ a static |
| sync_table(StaticModel) |
| |
| with mock.patch.object(session, "execute", wraps=session.execute) as m2: |
| sync_table(StaticModel) |
| |
| self.assertEqual(len(m2.call_args_list), 0) |
