test_class_construction.py
No OneTemporary
Actions

Size

14 KB

Subscribers

None

test_class_construction.py
View Options

	# 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 uuid import uuid4
	import warnings

	from cassandra.cqlengine import columns, CQLEngineException
	from cassandra.cqlengine.models import Model, ModelException, ModelDefinitionException, ColumnQueryEvaluator
	from cassandra.cqlengine.query import ModelQuerySet, DMLQuery

	from tests.integration.cqlengine.base import BaseCassEngTestCase


	class TestModelClassFunction(BaseCassEngTestCase):
	"""
	Tests verifying the behavior of the Model metaclass
	"""

	def test_column_attributes_handled_correctly(self):
	"""
	Tests that column attributes are moved to a _columns dict
	and replaced with simple value attributes
	"""

	class TestModel(Model):

	id = columns.UUID(primary_key=True, default=lambda:uuid4())
	text = columns.Text()

	# check class attibutes
	self.assertHasAttr(TestModel, '_columns')
	self.assertHasAttr(TestModel, 'id')
	self.assertHasAttr(TestModel, 'text')

	# check instance attributes
	inst = TestModel()
	self.assertHasAttr(inst, 'id')
	self.assertHasAttr(inst, 'text')
	self.assertIsNotNone(inst.id)
	self.assertIsNone(inst.text)

	def test_values_on_instantiation(self):
	"""
	Tests defaults and user-provided values on instantiation.
	"""

	class TestPerson(Model):
	first_name = columns.Text(primary_key=True, default='kevin')
	last_name = columns.Text(default='deldycke')

	# Check that defaults are available at instantiation.
	inst1 = TestPerson()
	self.assertHasAttr(inst1, 'first_name')
	self.assertHasAttr(inst1, 'last_name')
	self.assertEqual(inst1.first_name, 'kevin')
	self.assertEqual(inst1.last_name, 'deldycke')

	# Check that values on instantiation overrides defaults.
	inst2 = TestPerson(first_name='bob', last_name='joe')
	self.assertEqual(inst2.first_name, 'bob')
	self.assertEqual(inst2.last_name, 'joe')

	def test_db_map(self):
	"""
	Tests that the db_map is properly defined
	-the db_map allows columns
	"""
	class WildDBNames(Model):

	id = columns.UUID(primary_key=True, default=lambda:uuid4())
	content = columns.Text(db_field='words_and_whatnot')
	numbers = columns.Integer(db_field='integers_etc')

	db_map = WildDBNames._db_map
	self.assertEqual(db_map['words_and_whatnot'], 'content')
	self.assertEqual(db_map['integers_etc'], 'numbers')

	def test_attempting_to_make_duplicate_column_names_fails(self):
	"""
	Tests that trying to create conflicting db column names will fail
	"""

	with self.assertRaisesRegexp(ModelException, r".*more than once$"):
	class BadNames(Model):
	words = columns.Text(primary_key=True)
	content = columns.Text(db_field='words')

	def test_column_ordering_is_preserved(self):
	"""
	Tests that the _columns dics retains the ordering of the class definition
	"""

	class Stuff(Model):

	id = columns.UUID(primary_key=True, default=lambda:uuid4())
	words = columns.Text()
	content = columns.Text()
	numbers = columns.Integer()

	self.assertEqual([x for x in Stuff._columns.keys()], ['id', 'words', 'content', 'numbers'])

	def test_exception_raised_when_creating_class_without_pk(self):
	with self.assertRaises(ModelDefinitionException):
	class TestModel(Model):

	count = columns.Integer()
	text = columns.Text(required=False)

	def test_value_managers_are_keeping_model_instances_isolated(self):
	"""
	Tests that instance value managers are isolated from other instances
	"""
	class Stuff(Model):

	id = columns.UUID(primary_key=True, default=lambda:uuid4())
	num = columns.Integer()

	inst1 = Stuff(num=5)
	inst2 = Stuff(num=7)

	self.assertNotEqual(inst1.num, inst2.num)
	self.assertEqual(inst1.num, 5)
	self.assertEqual(inst2.num, 7)

	def test_superclass_fields_are_inherited(self):
	"""
	Tests that fields defined on the super class are inherited properly
	"""
	class TestModel(Model):

	id = columns.UUID(primary_key=True, default=lambda:uuid4())
	text = columns.Text()

	class InheritedModel(TestModel):
	numbers = columns.Integer()

	assert 'text' in InheritedModel._columns
	assert 'numbers' in InheritedModel._columns

	def test_column_family_name_generation(self):
	""" Tests that auto column family name generation works as expected """
	class TestModel(Model):

	id = columns.UUID(primary_key=True, default=lambda:uuid4())
	text = columns.Text()

	assert TestModel.column_family_name(include_keyspace=False) == 'test_model'

	def test_partition_keys(self):
	"""
	Test compound partition key definition
	"""
	class ModelWithPartitionKeys(Model):

	id = columns.UUID(primary_key=True, default=lambda:uuid4())
	c1 = columns.Text(primary_key=True)
	p1 = columns.Text(partition_key=True)
	p2 = columns.Text(partition_key=True)

	cols = ModelWithPartitionKeys._columns

	self.assertTrue(cols['c1'].primary_key)
	self.assertFalse(cols['c1'].partition_key)

	self.assertTrue(cols['p1'].primary_key)
	self.assertTrue(cols['p1'].partition_key)
	self.assertTrue(cols['p2'].primary_key)
	self.assertTrue(cols['p2'].partition_key)

	obj = ModelWithPartitionKeys(p1='a', p2='b')
	self.assertEqual(obj.pk, ('a', 'b'))

	def test_del_attribute_is_assigned_properly(self):
	""" Tests that columns that can be deleted have the del attribute """
	class DelModel(Model):

	id = columns.UUID(primary_key=True, default=lambda:uuid4())
	key = columns.Integer(primary_key=True)
	data = columns.Integer(required=False)

	model = DelModel(key=4, data=5)
	del model.data
	with self.assertRaises(AttributeError):
	del model.key

	def test_does_not_exist_exceptions_are_not_shared_between_model(self):
	""" Tests that DoesNotExist exceptions are not the same exception between models """

	class Model1(Model):

	id = columns.UUID(primary_key=True, default=lambda:uuid4())

	class Model2(Model):

	id = columns.UUID(primary_key=True, default=lambda:uuid4())

	try:
	raise Model1.DoesNotExist
	except Model2.DoesNotExist:
	assert False, "Model1 exception should not be caught by Model2"
	except Model1.DoesNotExist:
	# expected
	pass

	def test_does_not_exist_inherits_from_superclass(self):
	""" Tests that a DoesNotExist exception can be caught by it's parent class DoesNotExist """
	class Model1(Model):

	id = columns.UUID(primary_key=True, default=lambda:uuid4())

	class Model2(Model1):
	pass

	try:
	raise Model2.DoesNotExist
	except Model1.DoesNotExist:
	# expected
	pass
	except Exception:
	assert False, "Model2 exception should not be caught by Model1"

	def test_abstract_model_keyspace_warning_is_skipped(self):
	with warnings.catch_warnings(record=True) as warn:
	class NoKeyspace(Model):
	__abstract__ = True
	key = columns.UUID(primary_key=True)

	self.assertEqual(len(warn), 0)


	class TestManualTableNaming(BaseCassEngTestCase):

	class RenamedTest(Model):
	__keyspace__ = 'whatever'
	__table_name__ = 'manual_name'

	id = columns.UUID(primary_key=True)
	data = columns.Text()

	def test_proper_table_naming(self):
	assert self.RenamedTest.column_family_name(include_keyspace=False) == 'manual_name'
	assert self.RenamedTest.column_family_name(include_keyspace=True) == 'whatever.manual_name'


	class TestManualTableNamingCaseSensitive(BaseCassEngTestCase):

	class RenamedCaseInsensitiveTest(Model):
	__keyspace__ = 'whatever'
	__table_name__ = 'Manual_Name'

	id = columns.UUID(primary_key=True)

	class RenamedCaseSensitiveTest(Model):
	__keyspace__ = 'whatever'
	__table_name__ = 'Manual_Name'
	__table_name_case_sensitive__ = True

	id = columns.UUID(primary_key=True)

	def test_proper_table_naming_case_insensitive(self):
	"""
	Test to ensure case senstivity is not honored by default honored

	@since 3.1
	@jira_ticket PYTHON-337
	@expected_result table_names arel lowercase

	@test_category object_mapper
	"""
	self.assertEqual(self.RenamedCaseInsensitiveTest.column_family_name(include_keyspace=False), 'manual_name')
	self.assertEqual(self.RenamedCaseInsensitiveTest.column_family_name(include_keyspace=True), 'whatever.manual_name')

	def test_proper_table_naming_case_sensitive(self):
	"""
	Test to ensure case is honored when the flag is correctly set.

	@since 3.1
	@jira_ticket PYTHON-337
	@expected_result table_name case is honored.

	@test_category object_mapper
	"""

	self.assertEqual(self.RenamedCaseSensitiveTest.column_family_name(include_keyspace=False), '"Manual_Name"')
	self.assertEqual(self.RenamedCaseSensitiveTest.column_family_name(include_keyspace=True), 'whatever."Manual_Name"')


	class AbstractModel(Model):
	__abstract__ = True


	class ConcreteModel(AbstractModel):
	pkey = columns.Integer(primary_key=True)
	data = columns.Integer()


	class AbstractModelWithCol(Model):

	__abstract__ = True
	pkey = columns.Integer(primary_key=True)


	class ConcreteModelWithCol(AbstractModelWithCol):
	data = columns.Integer()


	class AbstractModelWithFullCols(Model):
	__abstract__ = True

	pkey = columns.Integer(primary_key=True)
	data = columns.Integer()


	class TestAbstractModelClasses(BaseCassEngTestCase):

	def test_id_field_is_not_created(self):
	""" Tests that an id field is not automatically generated on abstract classes """
	assert not hasattr(AbstractModel, 'id')
	assert not hasattr(AbstractModelWithCol, 'id')

	def test_id_field_is_not_created_on_subclass(self):
	assert not hasattr(ConcreteModel, 'id')

	def test_abstract_attribute_is_not_inherited(self):
	""" Tests that __abstract__ attribute is not inherited """
	assert not ConcreteModel.__abstract__
	assert not ConcreteModelWithCol.__abstract__

	def test_attempting_to_save_abstract_model_fails(self):
	""" Attempting to save a model from an abstract model should fail """
	with self.assertRaises(CQLEngineException):
	AbstractModelWithFullCols.create(pkey=1, data=2)

	def test_attempting_to_create_abstract_table_fails(self):
	""" Attempting to create a table from an abstract model should fail """
	from cassandra.cqlengine.management import sync_table
	with self.assertRaises(CQLEngineException):
	sync_table(AbstractModelWithFullCols)

	def test_attempting_query_on_abstract_model_fails(self):
	""" Tests attempting to execute query with an abstract model fails """
	with self.assertRaises(CQLEngineException):
	iter(AbstractModelWithFullCols.objects(pkey=5)).next()

	def test_abstract_columns_are_inherited(self):
	""" Tests that columns defined in the abstract class are inherited into the concrete class """
	assert hasattr(ConcreteModelWithCol, 'pkey')
	assert isinstance(ConcreteModelWithCol.pkey, ColumnQueryEvaluator)
	assert isinstance(ConcreteModelWithCol._columns['pkey'], columns.Column)

	def test_concrete_class_table_creation_cycle(self):
	""" Tests that models with inherited abstract classes can be created, and have io performed """
	from cassandra.cqlengine.management import sync_table, drop_table
	sync_table(ConcreteModelWithCol)

	w1 = ConcreteModelWithCol.create(pkey=5, data=6)
	w2 = ConcreteModelWithCol.create(pkey=6, data=7)

	r1 = ConcreteModelWithCol.get(pkey=5)
	r2 = ConcreteModelWithCol.get(pkey=6)

	assert w1.pkey == r1.pkey
	assert w1.data == r1.data
	assert w2.pkey == r2.pkey
	assert w2.data == r2.data

	drop_table(ConcreteModelWithCol)


	class TestCustomQuerySet(BaseCassEngTestCase):
	""" Tests overriding the default queryset class """

	class TestException(Exception): pass

	def test_overriding_queryset(self):

	class QSet(ModelQuerySet):
	def create(iself, **kwargs):
	raise self.TestException

	class CQModel(Model):
	__queryset__ = QSet

	part = columns.UUID(primary_key=True)
	data = columns.Text()

	with self.assertRaises(self.TestException):
	CQModel.create(part=uuid4(), data='s')

	def test_overriding_dmlqueryset(self):

	class DMLQ(DMLQuery):
	def save(iself):
	raise self.TestException

	class CDQModel(Model):

	__dmlquery__ = DMLQ
	part = columns.UUID(primary_key=True)
	data = columns.Text()

	with self.assertRaises(self.TestException):
	CDQModel().save()


	class TestCachedLengthIsNotCarriedToSubclasses(BaseCassEngTestCase):
	def test_subclassing(self):

	length = len(ConcreteModelWithCol())

	class AlreadyLoadedTest(ConcreteModelWithCol):
	new_field = columns.Integer()

	self.assertGreater(len(AlreadyLoadedTest()), length)

File Metadata

Mime Type: text/x-python
Expires: Jun 4 2025, 6:46 PM (14 w, 1 d ago)
Storage Engine: blob
Storage Format: Raw Data
Storage Handle: 3398745

test_class_construction.pyNo OneTemporaryActions

test_class_construction.pyView Options

File Metadata

Event Timeline

test_class_construction.py
No OneTemporary
Actions

test_class_construction.py
View Options