<s> from playhouse . tests . base import compiler <EOL> from playhouse . tests . base import database_initializer <EOL> from playhouse . tests . base import ModelTestCase <EOL> from playhouse . tests . base import PeeweeTestCase <EOL> from playhouse . tests . base import skip_if <EOL> from playhouse . tests . base import test_db <EOL> from playhouse . tests . models import * <EOL> class TestForeignKeyToNonPrimaryKey ( ModelTestCase ) : <EOL> requires = [ Package , PackageItem ] <EOL> def setUp ( self ) : <EOL> super ( TestForeignKeyToNonPrimaryKey , self ) . setUp ( ) <EOL> for barcode in [ '<STR_LIT>' , '<STR_LIT>' ] : <EOL> Package . create ( barcode = barcode ) <EOL> for i in range ( <NUM_LIT:2> ) : <EOL> PackageItem . create ( <EOL> package = barcode , <EOL> title = '<STR_LIT>' % ( barcode , i ) ) <EOL> def test_fk_resolution ( self ) : <EOL> pi = PackageItem . get ( PackageItem . title == '<STR_LIT>' ) <EOL> self . assertEqual ( pi . _data [ '<STR_LIT>' ] , '<STR_LIT>' ) <EOL> self . assertEqual ( pi . package , Package . get ( Package . barcode == '<STR_LIT>' ) ) <EOL> def test_select_generation ( self ) : <EOL> p = Package . get ( Package . barcode == '<STR_LIT>' ) <EOL> self . assertEqual ( <EOL> [ item . title for item in p . items . order_by ( PackageItem . title ) ] , <EOL> [ '<STR_LIT>' , '<STR_LIT>' ] ) <EOL> class TestMultipleForeignKey ( ModelTestCase ) : <EOL> requires = [ Manufacturer , Component , Computer ] <EOL> test_values = [ <EOL> [ '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ] , <EOL> [ '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ] , <EOL> ] <EOL> def setUp ( self ) : <EOL> super ( TestMultipleForeignKey , self ) . setUp ( ) <EOL> intel = Manufacturer . create ( name = '<STR_LIT>' ) <EOL> amd = Manufacturer . create ( name = '<STR_LIT>' ) <EOL> kingston = Manufacturer . create ( name = '<STR_LIT>' ) <EOL> for hard_drive , memory , processor in self . test_values : <EOL> c = Computer . create ( <EOL> hard_drive = Component . create ( name = hard_drive ) , <EOL> memory = Component . create ( name = memory , manufacturer = kingston ) , <EOL> processor = Component . create ( name = processor , manufacturer = intel ) ) <EOL> c . processor . manufacturer = amd <EOL> c . processor . save ( ) <EOL> def test_multi_join ( self ) : <EOL> HDD = Component . alias ( ) <EOL> HDDMf = Manufacturer . alias ( ) <EOL> Memory = Component . alias ( ) <EOL> MemoryMf = Manufacturer . alias ( ) <EOL> Processor = Component . alias ( ) <EOL> ProcessorMf = Manufacturer . alias ( ) <EOL> query = ( Computer <EOL> . select ( <EOL> Computer , <EOL> HDD , <EOL> Memory , <EOL> Processor , <EOL> HDDMf , <EOL> MemoryMf , <EOL> ProcessorMf ) <EOL> . join ( HDD , on = ( <EOL> Computer . hard_drive == HDD . id ) . alias ( '<STR_LIT>' ) ) <EOL> . join ( <EOL> HDDMf , <EOL> JOIN . LEFT_OUTER , <EOL> on = ( HDD . manufacturer == HDDMf . id ) ) <EOL> . switch ( Computer ) <EOL> . join ( Memory , on = ( <EOL> Computer . memory == Memory . id ) . alias ( '<STR_LIT>' ) ) <EOL> . join ( <EOL> MemoryMf , <EOL> JOIN . LEFT_OUTER , <EOL> on = ( Memory . manufacturer == MemoryMf . id ) ) <EOL> . switch ( Computer ) <EOL> . join ( Processor , on = ( <EOL> Computer . processor == Processor . id ) . alias ( '<STR_LIT>' ) ) <EOL> . join ( <EOL> ProcessorMf , <EOL> JOIN . LEFT_OUTER , <EOL> on = ( Processor . manufacturer == ProcessorMf . id ) ) <EOL> . order_by ( Computer . id ) ) <EOL> with self . assertQueryCount ( <NUM_LIT:1> ) : <EOL> vals = [ ] <EOL> manufacturers = [ ] <EOL> for computer in query : <EOL> components = [ <EOL> computer . hard_drive , <EOL> computer . memory , <EOL> computer . processor ] <EOL> vals . append ( [ component . name for component in components ] ) <EOL> for component in components : <EOL> if component . manufacturer : <EOL> manufacturers . append ( component . manufacturer . name ) <EOL> else : <EOL> manufacturers . append ( None ) <EOL> self . assertEqual ( vals , self . test_values ) <EOL> self . assertEqual ( manufacturers , [ <EOL> None , '<STR_LIT>' , '<STR_LIT>' , <EOL> None , '<STR_LIT>' , '<STR_LIT>' , <EOL> ] ) <EOL> class TestMultipleForeignKeysJoining ( ModelTestCase ) : <EOL> requires = [ User , Relationship ] <EOL> def test_multiple_fks ( self ) : <EOL> a = User . create ( username = '<STR_LIT:a>' ) <EOL> b = User . create ( username = '<STR_LIT:b>' ) <EOL> c = User . create ( username = '<STR_LIT:c>' ) <EOL> self . assertEqual ( list ( a . relationships ) , [ ] ) <EOL> self . assertEqual ( list ( a . related_to ) , [ ] ) <EOL> r_ab = Relationship . create ( from_user = a , to_user = b ) <EOL> self . assertEqual ( list ( a . relationships ) , [ r_ab ] ) <EOL> self . assertEqual ( list ( a . related_to ) , [ ] ) <EOL> self . assertEqual ( list ( b . relationships ) , [ ] ) <EOL> self . assertEqual ( list ( b . related_to ) , [ r_ab ] ) <EOL> r_bc = Relationship . create ( from_user = b , to_user = c ) <EOL> following = User . select ( ) . join ( <EOL> Relationship , on = Relationship . to_user <EOL> ) . where ( Relationship . from_user == a ) <EOL> self . assertEqual ( list ( following ) , [ b ] ) <EOL> followers = User . select ( ) . join ( <EOL> Relationship , on = Relationship . from_user <EOL> ) . where ( Relationship . to_user == a . id ) <EOL> self . assertEqual ( list ( followers ) , [ ] ) <EOL> following = User . select ( ) . join ( <EOL> Relationship , on = Relationship . to_user <EOL> ) . where ( Relationship . from_user == b . id ) <EOL> self . assertEqual ( list ( following ) , [ c ] ) <EOL> followers = User . select ( ) . join ( <EOL> Relationship , on = Relationship . from_user <EOL> ) . where ( Relationship . to_user == b . id ) <EOL> self . assertEqual ( list ( followers ) , [ a ] ) <EOL> following = User . select ( ) . join ( <EOL> Relationship , on = Relationship . to_user <EOL> ) . where ( Relationship . from_user == c . id ) <EOL> self . assertEqual ( list ( following ) , [ ] ) <EOL> followers = User . select ( ) . join ( <EOL> Relationship , on = Relationship . from_user <EOL> ) . where ( Relationship . to_user == c . id ) <EOL> self . assertEqual ( list ( followers ) , [ b ] ) <EOL> class TestCompositePrimaryKey ( ModelTestCase ) : <EOL> requires = [ Tag , Post , TagPostThrough , CompositeKeyModel , User , UserThing ] <EOL> def setUp ( self ) : <EOL> super ( TestCompositePrimaryKey , self ) . setUp ( ) <EOL> tags = [ Tag . create ( tag = '<STR_LIT>' % i ) for i in range ( <NUM_LIT:1> , <NUM_LIT:4> ) ] <EOL> posts = [ Post . create ( title = '<STR_LIT>' % i ) for i in range ( <NUM_LIT:1> , <NUM_LIT:4> ) ] <EOL> p12 = Post . create ( title = '<STR_LIT>' ) <EOL> for t , p in zip ( tags , posts ) : <EOL> TagPostThrough . create ( tag = t , post = p ) <EOL> TagPostThrough . create ( tag = tags [ <NUM_LIT:0> ] , post = p12 ) <EOL> TagPostThrough . create ( tag = tags [ <NUM_LIT:1> ] , post = p12 ) <EOL> def test_create_table_query ( self ) : <EOL> query , params = compiler . create_table ( TagPostThrough ) <EOL> self . assertEqual ( <EOL> query , <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> '<STR_LIT:)>' ) <EOL> def test_get_set_id ( self ) : <EOL> tpt = ( TagPostThrough <EOL> . select ( ) <EOL> . join ( Tag ) <EOL> . switch ( TagPostThrough ) <EOL> . join ( Post ) <EOL> . order_by ( Tag . tag , Post . title ) ) . get ( ) <EOL> self . assertEqual ( tpt . tag . tag , '<STR_LIT>' ) <EOL> self . assertEqual ( tpt . post . title , '<STR_LIT>' ) <EOL> tag = Tag . select ( ) . where ( Tag . tag == '<STR_LIT>' ) . get ( ) <EOL> post = Post . select ( ) . where ( Post . title == '<STR_LIT>' ) . get ( ) <EOL> self . assertEqual ( tpt . _get_pk_value ( ) , ( tag , post ) ) <EOL> tpt . _set_pk_value ( None ) <EOL> self . assertEqual ( tpt . _get_pk_value ( ) , ( tag , post ) ) <EOL> def test_querying ( self ) : <EOL> posts = ( Post . select ( ) <EOL> . join ( TagPostThrough ) <EOL> . join ( Tag ) <EOL> . where ( Tag . tag == '<STR_LIT>' ) <EOL> . order_by ( Post . title ) ) <EOL> self . assertEqual ( [ p . title for p in posts ] , [ '<STR_LIT>' , '<STR_LIT>' ] ) <EOL> tags = ( Tag . select ( ) <EOL> . join ( TagPostThrough ) <EOL> . join ( Post ) <EOL> . where ( Post . title == '<STR_LIT>' ) <EOL> . order_by ( Tag . tag ) ) <EOL> self . assertEqual ( [ t . tag for t in tags ] , [ '<STR_LIT>' , '<STR_LIT>' ] ) <EOL> def test_composite_key_model ( self ) : <EOL> CKM = CompositeKeyModel <EOL> values = [ <EOL> ( '<STR_LIT:a>' , <NUM_LIT:1> , <NUM_LIT:1.0> ) , <EOL> ( '<STR_LIT:a>' , <NUM_LIT:2> , <NUM_LIT> ) , <EOL> ( '<STR_LIT:b>' , <NUM_LIT:1> , <NUM_LIT:1.0> ) , <EOL> ( '<STR_LIT:b>' , <NUM_LIT:2> , <NUM_LIT> ) ] <EOL> c1 , c2 , c3 , c4 = [ <EOL> CKM . create ( f1 = f1 , f2 = f2 , f3 = f3 ) for f1 , f2 , f3 in values ] <EOL> CKM . update ( f3 = <NUM_LIT> ) . where ( ( CKM . f1 == '<STR_LIT:a>' ) & ( CKM . f2 == <NUM_LIT:2> ) ) . execute ( ) <EOL> c = CKM . get ( ( CKM . f1 == '<STR_LIT:a>' ) & ( CKM . f2 == <NUM_LIT:2> ) ) <EOL> self . assertEqual ( c . f3 , <NUM_LIT> ) <EOL> c3 . f3 = <NUM_LIT> <EOL> c3 . save ( ) <EOL> c = CKM . get ( ( CKM . f1 == '<STR_LIT:b>' ) & ( CKM . f2 == <NUM_LIT:1> ) ) <EOL> self . assertEqual ( c . f3 , <NUM_LIT> ) <EOL> query = CKM . select ( ) . where ( CKM . f3 == <NUM_LIT> ) <EOL> self . assertEqual ( query . wrapped_count ( ) , <NUM_LIT:1> ) <EOL> c4 . f1 = '<STR_LIT:c>' <EOL> c4 . save ( ) <EOL> self . assertRaises ( <EOL> CKM . DoesNotExist , CKM . get , ( CKM . f1 == '<STR_LIT:c>' ) & ( CKM . f2 == <NUM_LIT:2> ) ) <EOL> def test_count_composite_key ( self ) : <EOL> CKM = CompositeKeyModel <EOL> values = [ <EOL> ( '<STR_LIT:a>' , <NUM_LIT:1> , <NUM_LIT:1.0> ) , <EOL> ( '<STR_LIT:a>' , <NUM_LIT:2> , <NUM_LIT> ) , <EOL> ( '<STR_LIT:b>' , <NUM_LIT:1> , <NUM_LIT:1.0> ) , <EOL> ( '<STR_LIT:b>' , <NUM_LIT:2> , <NUM_LIT:1.0> ) ] <EOL> for f1 , f2 , f3 in values : <EOL> CKM . create ( f1 = f1 , f2 = f2 , f3 = f3 ) <EOL> self . assertEqual ( CKM . select ( ) . wrapped_count ( ) , <NUM_LIT:4> ) <EOL> self . assertEqual ( CKM . select ( ) . count ( ) , <NUM_LIT:4> ) <EOL> self . assertTrue ( CKM . select ( ) . where ( <EOL> ( CKM . f1 == '<STR_LIT:a>' ) & <EOL> ( CKM . f2 == <NUM_LIT:1> ) ) . exists ( ) ) <EOL> self . assertFalse ( CKM . select ( ) . where ( <EOL> ( CKM . f1 == '<STR_LIT:a>' ) & <EOL> ( CKM . f2 == <NUM_LIT:3> ) ) . exists ( ) ) <EOL> def test_delete_instance ( self ) : <EOL> u1 , u2 = [ User . create ( username = '<STR_LIT>' % i ) for i in range ( <NUM_LIT:2> ) ] <EOL> ut1 = UserThing . create ( thing = '<STR_LIT>' , user = u1 ) <EOL> ut2 = UserThing . create ( thing = '<STR_LIT>' , user = u1 ) <EOL> ut3 = UserThing . create ( thing = '<STR_LIT>' , user = u2 ) <EOL> ut4 = UserThing . create ( thing = '<STR_LIT>' , user = u2 ) <EOL> res = ut1 . delete_instance ( ) <EOL> self . assertEqual ( res , <NUM_LIT:1> ) <EOL> self . assertEqual ( <EOL> [ x . thing for x in UserThing . select ( ) . order_by ( UserThing . thing ) ] , <EOL> [ '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ] ) <EOL> class TestForeignKeyNonPrimaryKeyCreateTable ( PeeweeTestCase ) : <EOL> def test_create_table ( self ) : <EOL> class A ( TestModel ) : <EOL> cf = CharField ( max_length = <NUM_LIT:100> , unique = True ) <EOL> df = DecimalField ( <EOL> max_digits = <NUM_LIT:4> , <EOL> decimal_places = <NUM_LIT:2> , <EOL> auto_round = True , <EOL> unique = True ) <EOL> class CF ( TestModel ) : <EOL> a = ForeignKeyField ( A , to_field = '<STR_LIT>' ) <EOL> class DF ( TestModel ) : <EOL> a = ForeignKeyField ( A , to_field = '<STR_LIT>' ) <EOL> cf_create , _ = compiler . create_table ( CF ) <EOL> self . assertEqual ( <EOL> cf_create , <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' ) <EOL> df_create , _ = compiler . create_table ( DF ) <EOL> self . assertEqual ( <EOL> df_create , <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' ) <EOL> class TestDeferredForeignKey ( ModelTestCase ) : <EOL> def setUp ( self ) : <EOL> super ( TestDeferredForeignKey , self ) . setUp ( ) <EOL> Snippet . drop_table ( True ) <EOL> Language . drop_table ( True ) <EOL> Language . create_table ( ) <EOL> Snippet . create_table ( ) <EOL> def tearDown ( self ) : <EOL> super ( TestDeferredForeignKey , self ) . tearDown ( ) <EOL> Snippet . drop_table ( True ) <EOL> Language . drop_table ( True ) <EOL> def test_field_definitions ( self ) : <EOL> self . assertEqual ( Snippet . _meta . fields [ '<STR_LIT>' ] . rel_model , Language ) <EOL> self . assertEqual ( Language . _meta . fields [ '<STR_LIT>' ] . rel_model , <EOL> Snippet ) <EOL> def test_create_table_query ( self ) : <EOL> query , params = compiler . create_table ( Snippet ) <EOL> self . assertEqual ( <EOL> query , <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> '<STR_LIT:)>' ) <EOL> query , params = compiler . create_table ( Language ) <EOL> self . assertEqual ( <EOL> query , <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' ) <EOL> def test_storage_retrieval ( self ) : <EOL> python = Language . create ( name = '<STR_LIT>' ) <EOL> javascript = Language . create ( name = '<STR_LIT>' ) <EOL> p1 = Snippet . create ( code = "<STR_LIT>" , language = python ) <EOL> p2 = Snippet . create ( code = "<STR_LIT>" , language = python ) <EOL> j1 = Snippet . create ( code = "<STR_LIT>" , language = javascript ) <EOL> self . assertEqual ( Snippet . get ( Snippet . id == p1 . id ) . language , python ) <EOL> self . assertEqual ( Snippet . get ( Snippet . id == j1 . id ) . language , javascript ) <EOL> python . selected_snippet = p2 <EOL> python . save ( ) <EOL> self . assertEqual ( <EOL> Language . get ( Language . id == python . id ) . selected_snippet , p2 ) <EOL> self . assertEqual ( <EOL> Language . get ( Language . id == javascript . id ) . selected_snippet , None ) <EOL> class TestSQLiteDeferredForeignKey ( PeeweeTestCase ) : <EOL> def test_doc_example ( self ) : <EOL> db = database_initializer . get_in_memory_database ( ) <EOL> TweetDeferred = DeferredRelation ( ) <EOL> class Base ( Model ) : <EOL> class Meta : <EOL> database = db <EOL> class User ( Base ) : <EOL> username = CharField ( ) <EOL> favorite_tweet = ForeignKeyField ( TweetDeferred , null = True ) <EOL> class Tweet ( Base ) : <EOL> user = ForeignKeyField ( User ) <EOL> message = TextField ( ) <EOL> TweetDeferred . set_model ( Tweet ) <EOL> with db . transaction ( ) : <EOL> User . create_table ( ) <EOL> Tweet . create_table ( ) <EOL> self . assertRaises ( <EOL> OperationalError , <EOL> lambda : db . create_foreign_key ( User , User . favorite_tweet ) ) <EOL> @ skip_if ( lambda : not test_db . foreign_keys ) <EOL> class TestForeignKeyConstraints ( ModelTestCase ) : <EOL> requires = [ User , Blog ] <EOL> def test_constraint_exists ( self ) : <EOL> max_id = User . select ( fn . Max ( User . id ) ) . scalar ( ) or <NUM_LIT:0> <EOL> def will_fail ( ) : <EOL> with test_db . transaction ( ) as txn : <EOL> Blog . create ( user = max_id + <NUM_LIT:1> , title = '<STR_LIT>' ) <EOL> self . assertRaises ( IntegrityError , will_fail ) <EOL> def test_constraint_creation ( self ) : <EOL> class FKC_a ( TestModel ) : <EOL> name = CharField ( ) <EOL> fkc_deferred = DeferredRelation ( ) <EOL> class FKC_b ( TestModel ) : <EOL> fkc_a = ForeignKeyField ( fkc_deferred ) <EOL> fkc_deferred . set_model ( FKC_a ) <EOL> with test_db . transaction ( ) as txn : <EOL> FKC_b . drop_table ( True ) <EOL> FKC_a . drop_table ( True ) <EOL> FKC_a . create_table ( ) <EOL> FKC_b . create_table ( ) <EOL> fb = FKC_b . create ( fkc_a = - <NUM_LIT:1000> ) <EOL> fb . delete_instance ( ) <EOL> test_db . create_foreign_key ( FKC_b , FKC_b . fkc_a ) <EOL> def _trigger_exc ( ) : <EOL> with test_db . savepoint ( ) as s1 : <EOL> fb = FKC_b . create ( fkc_a = - <NUM_LIT:1000> ) <EOL> self . assertRaises ( IntegrityError , _trigger_exc ) <EOL> fa = FKC_a . create ( name = '<STR_LIT>' ) <EOL> fb = FKC_b . create ( fkc_a = fa ) <EOL> txn . rollback ( ) </s>
