<s> from uuid import uuid4 <EOL> import warnings <EOL> from cassandra . cqlengine import columns , CQLEngineException <EOL> from cassandra . cqlengine . models import Model , ModelException , ModelDefinitionException , ColumnQueryEvaluator <EOL> from cassandra . cqlengine . query import ModelQuerySet , DMLQuery <EOL> from tests . integration . cqlengine . base import BaseCassEngTestCase <EOL> class TestModelClassFunction ( BaseCassEngTestCase ) : <EOL> """<STR_LIT>""" <EOL> def test_column_attributes_handled_correctly ( self ) : <EOL> """<STR_LIT>""" <EOL> class TestModel ( Model ) : <EOL> id = columns . UUID ( primary_key = True , default = lambda : uuid4 ( ) ) <EOL> text = columns . Text ( ) <EOL> self . assertHasAttr ( TestModel , '<STR_LIT>' ) <EOL> self . assertHasAttr ( TestModel , '<STR_LIT:id>' ) <EOL> self . assertHasAttr ( TestModel , '<STR_LIT:text>' ) <EOL> inst = TestModel ( ) <EOL> self . assertHasAttr ( inst , '<STR_LIT:id>' ) <EOL> self . assertHasAttr ( inst , '<STR_LIT:text>' ) <EOL> self . assertIsNone ( inst . id ) <EOL> self . assertIsNone ( inst . text ) <EOL> def test_db_map ( self ) : <EOL> """<STR_LIT>""" <EOL> class WildDBNames ( Model ) : <EOL> id = columns . UUID ( primary_key = True , default = lambda : uuid4 ( ) ) <EOL> content = columns . Text ( db_field = '<STR_LIT>' ) <EOL> numbers = columns . Integer ( db_field = '<STR_LIT>' ) <EOL> db_map = WildDBNames . _db_map <EOL> self . assertEqual ( db_map [ '<STR_LIT>' ] , '<STR_LIT:content>' ) <EOL> self . assertEqual ( db_map [ '<STR_LIT>' ] , '<STR_LIT>' ) <EOL> def test_attempting_to_make_duplicate_column_names_fails ( self ) : <EOL> """<STR_LIT>""" <EOL> with self . assertRaisesRegexp ( ModelException , r"<STR_LIT>" ) : <EOL> class BadNames ( Model ) : <EOL> words = columns . Text ( primary_key = True ) <EOL> content = columns . Text ( db_field = '<STR_LIT>' ) <EOL> def test_column_ordering_is_preserved ( self ) : <EOL> """<STR_LIT>""" <EOL> class Stuff ( Model ) : <EOL> id = columns . UUID ( primary_key = True , default = lambda : uuid4 ( ) ) <EOL> words = columns . Text ( ) <EOL> content = columns . Text ( ) <EOL> numbers = columns . Integer ( ) <EOL> self . assertEqual ( [ x for x in Stuff . _columns . keys ( ) ] , [ '<STR_LIT:id>' , '<STR_LIT>' , '<STR_LIT:content>' , '<STR_LIT>' ] ) <EOL> def test_exception_raised_when_creating_class_without_pk ( self ) : <EOL> with self . assertRaises ( ModelDefinitionException ) : <EOL> class TestModel ( Model ) : <EOL> count = columns . Integer ( ) <EOL> text = columns . Text ( required = False ) <EOL> def test_value_managers_are_keeping_model_instances_isolated ( self ) : <EOL> """<STR_LIT>""" <EOL> class Stuff ( Model ) : <EOL> id = columns . UUID ( primary_key = True , default = lambda : uuid4 ( ) ) <EOL> num = columns . Integer ( ) <EOL> inst1 = Stuff ( num = <NUM_LIT:5> ) <EOL> inst2 = Stuff ( num = <NUM_LIT:7> ) <EOL> self . assertNotEqual ( inst1 . num , inst2 . num ) <EOL> self . assertEqual ( inst1 . num , <NUM_LIT:5> ) <EOL> self . assertEqual ( inst2 . num , <NUM_LIT:7> ) <EOL> def test_superclass_fields_are_inherited ( self ) : <EOL> """<STR_LIT>""" <EOL> class TestModel ( Model ) : <EOL> id = columns . UUID ( primary_key = True , default = lambda : uuid4 ( ) ) <EOL> text = columns . Text ( ) <EOL> class InheritedModel ( TestModel ) : <EOL> numbers = columns . Integer ( ) <EOL> assert '<STR_LIT:text>' in InheritedModel . _columns <EOL> assert '<STR_LIT>' in InheritedModel . _columns <EOL> def test_column_family_name_generation ( self ) : <EOL> """<STR_LIT>""" <EOL> class TestModel ( Model ) : <EOL> id = columns . UUID ( primary_key = True , default = lambda : uuid4 ( ) ) <EOL> text = columns . Text ( ) <EOL> assert TestModel . column_family_name ( include_keyspace = False ) == '<STR_LIT>' <EOL> def test_partition_keys ( self ) : <EOL> """<STR_LIT>""" <EOL> class ModelWithPartitionKeys ( Model ) : <EOL> id = columns . UUID ( primary_key = True , default = lambda : uuid4 ( ) ) <EOL> c1 = columns . Text ( primary_key = True ) <EOL> p1 = columns . Text ( partition_key = True ) <EOL> p2 = columns . Text ( partition_key = True ) <EOL> cols = ModelWithPartitionKeys . _columns <EOL> self . assertTrue ( cols [ '<STR_LIT>' ] . primary_key ) <EOL> self . assertFalse ( cols [ '<STR_LIT>' ] . partition_key ) <EOL> self . assertTrue ( cols [ '<STR_LIT>' ] . primary_key ) <EOL> self . assertTrue ( cols [ '<STR_LIT>' ] . partition_key ) <EOL> self . assertTrue ( cols [ '<STR_LIT>' ] . primary_key ) <EOL> self . assertTrue ( cols [ '<STR_LIT>' ] . partition_key ) <EOL> obj = ModelWithPartitionKeys ( p1 = '<STR_LIT:a>' , p2 = '<STR_LIT:b>' ) <EOL> self . assertEqual ( obj . pk , ( '<STR_LIT:a>' , '<STR_LIT:b>' ) ) <EOL> def test_del_attribute_is_assigned_properly ( self ) : <EOL> """<STR_LIT>""" <EOL> class DelModel ( Model ) : <EOL> id = columns . UUID ( primary_key = True , default = lambda : uuid4 ( ) ) <EOL> key = columns . Integer ( primary_key = True ) <EOL> data = columns . Integer ( required = False ) <EOL> model = DelModel ( key = <NUM_LIT:4> , data = <NUM_LIT:5> ) <EOL> del model . data <EOL> with self . assertRaises ( AttributeError ) : <EOL> del model . key <EOL> def test_does_not_exist_exceptions_are_not_shared_between_model ( self ) : <EOL> """<STR_LIT>""" <EOL> class Model1 ( Model ) : <EOL> id = columns . UUID ( primary_key = True , default = lambda : uuid4 ( ) ) <EOL> class Model2 ( Model ) : <EOL> id = columns . UUID ( primary_key = True , default = lambda : uuid4 ( ) ) <EOL> try : <EOL> raise Model1 . DoesNotExist <EOL> except Model2 . DoesNotExist : <EOL> assert False , "<STR_LIT>" <EOL> except Model1 . DoesNotExist : <EOL> pass <EOL> def test_does_not_exist_inherits_from_superclass ( self ) : <EOL> """<STR_LIT>""" <EOL> class Model1 ( Model ) : <EOL> id = columns . UUID ( primary_key = True , default = lambda : uuid4 ( ) ) <EOL> class Model2 ( Model1 ) : <EOL> pass <EOL> try : <EOL> raise Model2 . DoesNotExist <EOL> except Model1 . DoesNotExist : <EOL> pass <EOL> except Exception : <EOL> assert False , "<STR_LIT>" <EOL> def test_abstract_model_keyspace_warning_is_skipped ( self ) : <EOL> with warnings . catch_warnings ( record = True ) as warn : <EOL> class NoKeyspace ( Model ) : <EOL> __abstract__ = True <EOL> key = columns . UUID ( primary_key = True ) <EOL> self . assertEqual ( len ( warn ) , <NUM_LIT:0> ) <EOL> class TestManualTableNaming ( BaseCassEngTestCase ) : <EOL> class RenamedTest ( Model ) : <EOL> __keyspace__ = '<STR_LIT>' <EOL> __table_name__ = '<STR_LIT>' <EOL> id = columns . UUID ( primary_key = True ) <EOL> data = columns . Text ( ) <EOL> def test_proper_table_naming ( self ) : <EOL> assert self . RenamedTest . column_family_name ( include_keyspace = False ) == '<STR_LIT>' <EOL> assert self . RenamedTest . column_family_name ( include_keyspace = True ) == '<STR_LIT>' <EOL> class TestManualTableNamingCaseSensitive ( BaseCassEngTestCase ) : <EOL> class RenamedCaseInsensitiveTest ( Model ) : <EOL> __keyspace__ = '<STR_LIT>' <EOL> __table_name__ = '<STR_LIT>' <EOL> id = columns . UUID ( primary_key = True ) <EOL> class RenamedCaseSensitiveTest ( Model ) : <EOL> __keyspace__ = '<STR_LIT>' <EOL> __table_name__ = '<STR_LIT>' <EOL> __table_name_case_sensitive__ = True <EOL> id = columns . UUID ( primary_key = True ) <EOL> def test_proper_table_naming_case_insensitive ( self ) : <EOL> """<STR_LIT>""" <EOL> self . assertEqual ( self . RenamedCaseInsensitiveTest . column_family_name ( include_keyspace = False ) , '<STR_LIT>' ) <EOL> self . assertEqual ( self . RenamedCaseInsensitiveTest . column_family_name ( include_keyspace = True ) , '<STR_LIT>' ) <EOL> def test_proper_table_naming_case_sensitive ( self ) : <EOL> """<STR_LIT>""" <EOL> self . assertEqual ( self . RenamedCaseSensitiveTest . column_family_name ( include_keyspace = False ) , '<STR_LIT>' ) <EOL> self . assertEqual ( self . RenamedCaseSensitiveTest . column_family_name ( include_keyspace = True ) , '<STR_LIT>' ) <EOL> class AbstractModel ( Model ) : <EOL> __abstract__ = True <EOL> class ConcreteModel ( AbstractModel ) : <EOL> pkey = columns . Integer ( primary_key = True ) <EOL> data = columns . Integer ( ) <EOL> class AbstractModelWithCol ( Model ) : <EOL> __abstract__ = True <EOL> pkey = columns . Integer ( primary_key = True ) <EOL> class ConcreteModelWithCol ( AbstractModelWithCol ) : <EOL> data = columns . Integer ( ) <EOL> class AbstractModelWithFullCols ( Model ) : <EOL> __abstract__ = True <EOL> pkey = columns . Integer ( primary_key = True ) <EOL> data = columns . Integer ( ) <EOL> class TestAbstractModelClasses ( BaseCassEngTestCase ) : <EOL> def test_id_field_is_not_created ( self ) : <EOL> """<STR_LIT>""" <EOL> assert not hasattr ( AbstractModel , '<STR_LIT:id>' ) <EOL> assert not hasattr ( AbstractModelWithCol , '<STR_LIT:id>' ) <EOL> def test_id_field_is_not_created_on_subclass ( self ) : <EOL> assert not hasattr ( ConcreteModel , '<STR_LIT:id>' ) <EOL> def test_abstract_attribute_is_not_inherited ( self ) : <EOL> """<STR_LIT>""" <EOL> assert not ConcreteModel . __abstract__ <EOL> assert not ConcreteModelWithCol . __abstract__ <EOL> def test_attempting_to_save_abstract_model_fails ( self ) : <EOL> """<STR_LIT>""" <EOL> with self . assertRaises ( CQLEngineException ) : <EOL> AbstractModelWithFullCols . create ( pkey = <NUM_LIT:1> , data = <NUM_LIT:2> ) <EOL> def test_attempting_to_create_abstract_table_fails ( self ) : <EOL> """<STR_LIT>""" <EOL> from cassandra . cqlengine . management import sync_table <EOL> with self . assertRaises ( CQLEngineException ) : <EOL> sync_table ( AbstractModelWithFullCols ) <EOL> def test_attempting_query_on_abstract_model_fails ( self ) : <EOL> """<STR_LIT>""" <EOL> with self . assertRaises ( CQLEngineException ) : <EOL> iter ( AbstractModelWithFullCols . objects ( pkey = <NUM_LIT:5> ) ) . next ( ) <EOL> def test_abstract_columns_are_inherited ( self ) : <EOL> """<STR_LIT>""" <EOL> assert hasattr ( ConcreteModelWithCol , '<STR_LIT>' ) <EOL> assert isinstance ( ConcreteModelWithCol . pkey , ColumnQueryEvaluator ) <EOL> assert isinstance ( ConcreteModelWithCol . _columns [ '<STR_LIT>' ] , columns . Column ) <EOL> def test_concrete_class_table_creation_cycle ( self ) : <EOL> """<STR_LIT>""" <EOL> from cassandra . cqlengine . management import sync_table , drop_table <EOL> sync_table ( ConcreteModelWithCol ) <EOL> w1 = ConcreteModelWithCol . create ( pkey = <NUM_LIT:5> , data = <NUM_LIT:6> ) <EOL> w2 = ConcreteModelWithCol . create ( pkey = <NUM_LIT:6> , data = <NUM_LIT:7> ) <EOL> r1 = ConcreteModelWithCol . get ( pkey = <NUM_LIT:5> ) <EOL> r2 = ConcreteModelWithCol . get ( pkey = <NUM_LIT:6> ) <EOL> assert w1 . pkey == r1 . pkey <EOL> assert w1 . data == r1 . data <EOL> assert w2 . pkey == r2 . pkey <EOL> assert w2 . data == r2 . data <EOL> drop_table ( ConcreteModelWithCol ) <EOL> class TestCustomQuerySet ( BaseCassEngTestCase ) : <EOL> """<STR_LIT>""" <EOL> class TestException ( Exception ) : pass <EOL> def test_overriding_queryset ( self ) : <EOL> class QSet ( ModelQuerySet ) : <EOL> def create ( iself , ** kwargs ) : <EOL> raise self . TestException <EOL> class CQModel ( Model ) : <EOL> __queryset__ = QSet <EOL> part = columns . UUID ( primary_key = True ) <EOL> data = columns . Text ( ) <EOL> with self . assertRaises ( self . TestException ) : <EOL> CQModel . create ( part = uuid4 ( ) , data = '<STR_LIT:s>' ) <EOL> def test_overriding_dmlqueryset ( self ) : <EOL> class DMLQ ( DMLQuery ) : <EOL> def save ( iself ) : <EOL> raise self . TestException <EOL> class CDQModel ( Model ) : <EOL> __dmlquery__ = DMLQ <EOL> part = columns . UUID ( primary_key = True ) <EOL> data = columns . Text ( ) <EOL> with self . assertRaises ( self . TestException ) : <EOL> CDQModel ( ) . save ( ) <EOL> class TestCachedLengthIsNotCarriedToSubclasses ( BaseCassEngTestCase ) : <EOL> def test_subclassing ( self ) : <EOL> length = len ( ConcreteModelWithCol ( ) ) <EOL> class AlreadyLoadedTest ( ConcreteModelWithCol ) : <EOL> new_field = columns . Integer ( ) <EOL> self . assertGreater ( len ( AlreadyLoadedTest ( ) ) , length ) </s>
