<s> from stdnet import odm , ManyToManyError <EOL> from stdnet . utils import test <EOL> from examples . models import Role , Profile <EOL> from examples . m2m import Composite , Element , CompositeElement <EOL> class TestManyToManyBase ( object ) : <EOL> models = ( Role , Profile ) <EOL> def addsome ( self , role1 = '<STR_LIT>' , role2 = '<STR_LIT>' ) : <EOL> models = self . mapper <EOL> session = models . session ( ) <EOL> with session . begin ( ) as t : <EOL> profile = t . add ( models . profile ( name = '<STR_LIT>' ) ) <EOL> profile2 = t . add ( models . profile ( name = '<STR_LIT>' ) ) <EOL> profile3 = t . add ( models . profile ( name = '<STR_LIT>' ) ) <EOL> role1 = t . add ( models . role ( name = role1 ) ) <EOL> role2 = t . add ( models . role ( name = role2 ) ) <EOL> yield t . on_result <EOL> with session . begin ( ) as t : <EOL> pr1 = profile . roles . add ( role1 ) <EOL> pr2 = profile . roles . add ( role2 ) <EOL> yield t . on_result <EOL> self . assertEqual ( len ( t . saved ) , <NUM_LIT:1> ) <EOL> self . assertEqual ( len ( list ( t . saved . values ( ) ) [ <NUM_LIT:0> ] ) , <NUM_LIT:2> ) <EOL> t1 = yield role1 . profiles . throughquery ( ) . load_related ( '<STR_LIT>' ) . all ( ) <EOL> t2 = yield role2 . profiles . throughquery ( ) . load_related ( '<STR_LIT>' ) . all ( ) <EOL> self . assertEqual ( len ( t1 ) , <NUM_LIT:1> ) <EOL> self . assertEqual ( len ( t2 ) , <NUM_LIT:1> ) <EOL> self . assertEqual ( t1 [ <NUM_LIT:0> ] . role , role1 ) <EOL> self . assertEqual ( t2 [ <NUM_LIT:0> ] . role , role2 ) <EOL> p1 = yield role1 . profiles . query ( ) . all ( ) <EOL> p2 = yield role2 . profiles . query ( ) . all ( ) <EOL> self . assertEqual ( len ( p1 ) , <NUM_LIT:1> ) <EOL> self . assertEqual ( len ( p2 ) , <NUM_LIT:1> ) <EOL> self . assertEqual ( p1 [ <NUM_LIT:0> ] , profile ) <EOL> self . assertEqual ( p2 [ <NUM_LIT:0> ] , profile ) <EOL> t1 = yield profile . roles . throughquery ( ) . all ( ) <EOL> self . assertEqual ( len ( t1 ) , <NUM_LIT:2> ) <EOL> p1 , p2 = yield self . multi_async ( ( t1 [ <NUM_LIT:0> ] . profile , t1 [ <NUM_LIT:1> ] . profile ) ) <EOL> self . assertEqual ( p1 , profile ) <EOL> self . assertEqual ( p2 , profile ) <EOL> t1 = yield profile . roles . throughquery ( ) . load_related ( '<STR_LIT>' ) . all ( ) <EOL> self . assertEqual ( len ( t1 ) , <NUM_LIT:2> ) <EOL> self . assertEqual ( t1 [ <NUM_LIT:0> ] . profile , profile ) <EOL> self . assertEqual ( t1 [ <NUM_LIT:1> ] . profile , profile ) <EOL> r = yield profile . roles . query ( ) . all ( ) <EOL> self . assertEqual ( len ( r ) , <NUM_LIT:2> ) <EOL> self . assertEqual ( set ( r ) , set ( ( role1 , role2 ) ) ) <EOL> yield role1 , role2 <EOL> class TestManyToMany ( TestManyToManyBase , test . TestCase ) : <EOL> def test_meta ( self ) : <EOL> self . assertEqual ( Profile . _meta . manytomany , [ '<STR_LIT>' ] ) <EOL> roles = Profile . roles <EOL> self . assertEqual ( roles . model . _meta . name , '<STR_LIT>' ) <EOL> self . assertEqual ( roles . relmodel , Profile ) <EOL> self . assertEqual ( roles . name_relmodel , '<STR_LIT>' ) <EOL> self . assertEqual ( roles . formodel , Role ) <EOL> profiles = Role . profiles <EOL> self . assertEqual ( profiles . model . _meta . name , '<STR_LIT>' ) <EOL> self . assertEqual ( profiles . relmodel , Role ) <EOL> self . assertEqual ( profiles . formodel , Profile ) <EOL> self . assertEqual ( profiles . name_relmodel , '<STR_LIT>' ) <EOL> through = roles . model <EOL> self . assertEqual ( through , profiles . model ) <EOL> self . assertEqual ( len ( through . _meta . dfields ) , <NUM_LIT:3> ) <EOL> def test_meta_instance ( self ) : <EOL> p = Profile ( ) <EOL> self . assertEqual ( p . roles . formodel , Role ) <EOL> self . assertEqual ( p . roles . related_instance , p ) <EOL> yield self . addsome ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> role = yield self . query ( Role ) . get ( name = '<STR_LIT>' ) <EOL> self . assertEqual ( role . profiles . formodel , Profile ) <EOL> self . assertEqual ( role . profiles . related_instance , role ) <EOL> def testQuery ( self ) : <EOL> yield self . addsome ( '<STR_LIT>' , '<STR_LIT:foo>' ) <EOL> role = yield self . query ( Role ) . get ( name = '<STR_LIT>' ) <EOL> profiles = role . profiles . query ( ) <EOL> self . assertEqual ( profiles . model , Profile ) <EOL> self . assertEqual ( profiles . session , role . session ) <EOL> def test_throughquery ( self ) : <EOL> yield self . addsome ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> role = yield self . query ( Role ) . get ( name = '<STR_LIT>' ) <EOL> query = role . profiles . throughquery ( ) <EOL> self . assertEqual ( query . model , role . profiles . model ) <EOL> self . assertEqual ( query . session , role . session ) <EOL> def test_multiple_add ( self ) : <EOL> yield self . addsome ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> role = yield self . query ( Role ) . get ( name = '<STR_LIT>' ) <EOL> profiles = yield role . profiles . query ( ) . all ( ) <EOL> self . assertEqual ( len ( profiles ) , <NUM_LIT:1> ) <EOL> profile = profiles [ <NUM_LIT:0> ] <EOL> yield role . profiles . add ( profile ) <EOL> profiles = yield role . profiles . query ( ) . all ( ) <EOL> self . assertEqual ( len ( profiles ) , <NUM_LIT:1> ) <EOL> self . assertEqual ( profile , profiles [ <NUM_LIT:0> ] ) <EOL> class TestManyToManyAddDelete ( TestManyToManyBase , test . TestWrite ) : <EOL> def testAdd ( self ) : <EOL> return self . addsome ( ) <EOL> def testDelete1 ( self ) : <EOL> role1 , role2 = yield self . addsome ( '<STR_LIT>' , '<STR_LIT:foo>' ) <EOL> session = self . session ( ) <EOL> profiles = yield role1 . profiles . query ( ) . all ( ) <EOL> self . assertEqual ( len ( profiles ) , <NUM_LIT:1> ) <EOL> profile = profiles [ <NUM_LIT:0> ] <EOL> yield self . async . assertEqual ( profile . roles . query ( ) . count ( ) , <NUM_LIT:2> ) <EOL> yield session . delete ( profile ) <EOL> role1 , role2 = yield session . query ( Role ) . filter ( name = ( '<STR_LIT>' , '<STR_LIT:foo>' ) ) . all ( ) <EOL> yield self . async . assertEqual ( role1 . profiles . query ( ) . count ( ) , <NUM_LIT:0> ) <EOL> yield self . async . assertEqual ( role2 . profiles . query ( ) . count ( ) , <NUM_LIT:0> ) <EOL> def testDelete2 ( self ) : <EOL> yield self . addsome ( ) <EOL> session = self . session ( ) <EOL> roles = session . query ( Role ) <EOL> yield self . async . assertEqual ( roles . count ( ) , <NUM_LIT:2> ) <EOL> yield roles . delete ( ) <EOL> yield self . async . assertEqual ( session . query ( Role ) . count ( ) , <NUM_LIT:0> ) <EOL> profile = yield session . query ( Profile ) . get ( id = <NUM_LIT:1> ) <EOL> yield self . async . assertEqual ( profile . roles . query ( ) . count ( ) , <NUM_LIT:0> ) <EOL> yield session . delete ( profile ) <EOL> def test_remove ( self ) : <EOL> session = self . session ( ) <EOL> with session . begin ( ) as t : <EOL> p1 = t . add ( Profile ( name = '<STR_LIT>' ) ) <EOL> p2 = t . add ( Profile ( name = '<STR_LIT>' ) ) <EOL> yield t . on_result <EOL> role , created = yield session . update_or_create ( Role , name = '<STR_LIT>' ) <EOL> self . assertTrue ( created ) <EOL> role , created = yield session . update_or_create ( Role , name = '<STR_LIT>' ) <EOL> self . assertFalse ( created ) <EOL> self . assertTrue ( role . id ) <EOL> with p1 . session . begin ( ) as t : <EOL> p1 . roles . add ( role ) <EOL> p2 . roles . add ( role ) <EOL> self . assertEqual ( len ( t . session . dirty ) , <NUM_LIT:2> ) <EOL> yield t . on_result <EOL> profiles = role . profiles . query ( ) <EOL> yield self . async . assertEqual ( profiles . count ( ) , <NUM_LIT:2> ) <EOL> p2 . roles . add ( role ) <EOL> profiles = role . profiles . query ( ) <EOL> yield self . async . assertEqual ( profiles . count ( ) , <NUM_LIT:2> ) <EOL> yield p2 . roles . remove ( role ) <EOL> profiles = role . profiles . query ( ) <EOL> yield self . async . assertEqual ( profiles . count ( ) , <NUM_LIT:1> ) <EOL> yield p1 . roles . remove ( role ) <EOL> profiles = role . profiles . query ( ) <EOL> yield self . async . assertEqual ( profiles . count ( ) , <NUM_LIT:0> ) <EOL> class TestRegisteredThroughModel ( TestManyToManyBase , test . TestCase ) : <EOL> def testMeta ( self ) : <EOL> models = self . mapper <EOL> through = Profile . roles . model <EOL> self . assertTrue ( through in models ) <EOL> objects = models [ through ] <EOL> name = through . __name__ <EOL> self . assertEqual ( name , '<STR_LIT>' ) <EOL> self . assertEqual ( objects . backend , models [ Profile ] . backend ) <EOL> self . assertEqual ( objects . backend , models [ Role ] . backend ) <EOL> self . assertEqual ( through . role . field . model , through ) <EOL> self . assertEqual ( through . profile . field . model , through ) <EOL> pk = through . pk ( ) <EOL> self . assertTrue ( isinstance ( pk , odm . CompositeIdField ) ) <EOL> self . assertEqual ( pk . fields [ <NUM_LIT:0> ] . relmodel , Profile ) <EOL> self . assertEqual ( pk . fields [ <NUM_LIT:1> ] . relmodel , Role ) <EOL> def test_class_add ( self ) : <EOL> self . assertRaises ( ManyToManyError , Profile . roles . add , Role ( name = '<STR_LIT:foo>' ) ) <EOL> self . assertRaises ( ManyToManyError , Role . profiles . add , Profile ( ) ) <EOL> def test_through_query ( self ) : <EOL> m = self . mapper <EOL> p1 , p2 , p3 = yield self . multi_async ( ( m . profile . new ( name = '<STR_LIT>' ) , <EOL> m . profile . new ( name = '<STR_LIT>' ) , <EOL> m . profile . new ( name = '<STR_LIT>' ) ) ) <EOL> r1 , r2 = yield self . multi_async ( ( m . role . new ( name = '<STR_LIT>' ) , <EOL> m . role . new ( name = '<STR_LIT:foo>' ) ) ) <EOL> pr1 , pr2 = yield self . multi_async ( ( p1 . roles . add ( r1 ) , p2 . roles . add ( r1 ) ) ) <EOL> yield self . async . assertEqual ( pr1 . role , r1 ) <EOL> yield self . async . assertEqual ( pr2 . role , r1 ) <EOL> class TestManyToManyThrough ( test . TestCase ) : <EOL> models = ( Composite , Element , CompositeElement ) <EOL> def testMetaComposite ( self ) : <EOL> meta = Composite . _meta <EOL> m2m = None <EOL> for field in meta . fields : <EOL> if field . name == '<STR_LIT>' : <EOL> m2m = field <EOL> self . assertTrue ( isinstance ( m2m , odm . ManyToManyField ) ) <EOL> self . assertFalse ( '<STR_LIT>' in meta . dfields ) <EOL> self . assertEqual ( m2m . through , CompositeElement ) <EOL> self . assertTrue ( '<STR_LIT>' in meta . related ) <EOL> manager = Composite . elements <EOL> self . assertEqual ( manager . model , CompositeElement ) <EOL> self . assertEqual ( manager . relmodel , Composite ) <EOL> self . assertEqual ( manager . formodel , Element ) <EOL> self . assertEqual ( len ( CompositeElement . _meta . indices ) , <NUM_LIT:2> ) <EOL> def testMetaElement ( self ) : <EOL> meta = Element . _meta <EOL> self . assertTrue ( '<STR_LIT>' in meta . related ) <EOL> manager = Element . composites <EOL> self . assertEqual ( manager . model , CompositeElement ) <EOL> self . assertEqual ( manager . relmodel , Element ) <EOL> self . assertEqual ( manager . formodel , Composite ) <EOL> def testAdd ( self ) : <EOL> session = self . session ( ) <EOL> with session . begin ( ) as t : <EOL> c = t . add ( Composite ( name = '<STR_LIT:test>' ) ) <EOL> e1 = t . add ( Element ( name = '<STR_LIT:foo>' ) ) <EOL> e2 = t . add ( Element ( name = '<STR_LIT>' ) ) <EOL> yield t . on_result <EOL> yield c . elements . add ( e1 , weight = <NUM_LIT> ) <EOL> yield c . elements . add ( e2 , weight = - <NUM_LIT:1> ) <EOL> elems = yield c . elements . throughquery ( ) . all ( ) <EOL> for elem in elems : <EOL> self . assertTrue ( elem . weight ) </s>
