<s> from sqlalchemy import * <EOL> from sqlalchemy . orm import * <EOL> from sqlalchemy . ext . declarative import declarative_base <EOL> from sqlalchemy . testing import eq_ , AssertsExecutionResults , assert_raises <EOL> from sqlalchemy import testing <EOL> from sqlalchemy . testing import fixtures <EOL> from sqlalchemy . orm . attributes import instance_state <EOL> from sqlalchemy . orm . exc import FlushError <EOL> from sqlalchemy . testing . schema import Table , Column <EOL> engine = testing . db <EOL> class FlushOnPendingTest ( AssertsExecutionResults , fixtures . TestBase ) : <EOL> def setUp ( self ) : <EOL> global Parent , Child , Base <EOL> Base = declarative_base ( ) <EOL> class Parent ( Base ) : <EOL> __tablename__ = '<STR_LIT>' <EOL> id = Column ( Integer , primary_key = True , test_needs_autoincrement = True ) <EOL> name = Column ( String ( <NUM_LIT:50> ) , nullable = False ) <EOL> children = relationship ( "<STR_LIT>" , load_on_pending = True ) <EOL> class Child ( Base ) : <EOL> __tablename__ = '<STR_LIT>' <EOL> id = Column ( Integer , primary_key = True , test_needs_autoincrement = True ) <EOL> parent_id = Column ( Integer , ForeignKey ( '<STR_LIT>' ) ) <EOL> Base . metadata . create_all ( engine ) <EOL> def tearDown ( self ) : <EOL> Base . metadata . drop_all ( engine ) <EOL> def test_annoying_autoflush_one ( self ) : <EOL> sess = Session ( engine ) <EOL> p1 = Parent ( ) <EOL> sess . add ( p1 ) <EOL> p1 . children = [ ] <EOL> def test_annoying_autoflush_two ( self ) : <EOL> sess = Session ( engine ) <EOL> p1 = Parent ( ) <EOL> sess . add ( p1 ) <EOL> assert p1 . children == [ ] <EOL> def test_dont_load_if_no_keys ( self ) : <EOL> sess = Session ( engine ) <EOL> p1 = Parent ( ) <EOL> sess . add ( p1 ) <EOL> def go ( ) : <EOL> assert p1 . children == [ ] <EOL> self . assert_sql_count ( testing . db , go , <NUM_LIT:0> ) <EOL> class LoadOnFKsTest ( AssertsExecutionResults , fixtures . TestBase ) : <EOL> def setUp ( self ) : <EOL> global Parent , Child , Base <EOL> Base = declarative_base ( ) <EOL> class Parent ( Base ) : <EOL> __tablename__ = '<STR_LIT>' <EOL> __table_args__ = { '<STR_LIT>' : '<STR_LIT>' } <EOL> id = Column ( Integer , primary_key = True , test_needs_autoincrement = True ) <EOL> class Child ( Base ) : <EOL> __tablename__ = '<STR_LIT>' <EOL> __table_args__ = { '<STR_LIT>' : '<STR_LIT>' } <EOL> id = Column ( Integer , primary_key = True , test_needs_autoincrement = True ) <EOL> parent_id = Column ( Integer , ForeignKey ( '<STR_LIT>' ) ) <EOL> parent = relationship ( Parent , backref = backref ( "<STR_LIT>" ) ) <EOL> Base . metadata . create_all ( engine ) <EOL> global sess , p1 , p2 , c1 , c2 <EOL> sess = Session ( bind = engine ) <EOL> p1 = Parent ( ) <EOL> p2 = Parent ( ) <EOL> c1 , c2 = Child ( ) , Child ( ) <EOL> c1 . parent = p1 <EOL> sess . add_all ( [ p1 , p2 ] ) <EOL> assert c1 in sess <EOL> sess . commit ( ) <EOL> def tearDown ( self ) : <EOL> sess . rollback ( ) <EOL> Base . metadata . drop_all ( engine ) <EOL> def test_load_on_pending_disallows_backref_event ( self ) : <EOL> Child . parent . property . load_on_pending = True <EOL> sess . autoflush = False <EOL> c3 = Child ( ) <EOL> sess . add ( c3 ) <EOL> c3 . parent_id = p1 . id <EOL> c3 . parent = p1 <EOL> assert c3 not in p1 . children <EOL> def test_enable_rel_loading_disallows_backref_event ( self ) : <EOL> sess . autoflush = False <EOL> c3 = Child ( ) <EOL> sess . enable_relationship_loading ( c3 ) <EOL> c3 . parent_id = p1 . id <EOL> c3 . parent = p1 <EOL> assert c3 not in p1 . children <EOL> def test_load_on_persistent_allows_backref_event ( self ) : <EOL> Child . parent . property . load_on_pending = True <EOL> c3 = Child ( ) <EOL> sess . add ( c3 ) <EOL> c3 . parent_id = p1 . id <EOL> c3 . parent = p1 <EOL> assert c3 in p1 . children <EOL> def test_enable_rel_loading_on_persistent_disallows_backref_event ( self ) : <EOL> c3 = Child ( ) <EOL> sess . enable_relationship_loading ( c3 ) <EOL> c3 . parent_id = p1 . id <EOL> c3 . parent = p1 <EOL> assert c3 not in p1 . children <EOL> def test_no_load_on_pending_allows_backref_event ( self ) : <EOL> sess . autoflush = False <EOL> c3 = Child ( ) <EOL> sess . add ( c3 ) <EOL> c3 . parent_id = p1 . id <EOL> c3 . parent = p1 <EOL> assert c3 in p1 . children <EOL> def test_autoflush_on_pending ( self ) : <EOL> c3 = Child ( ) <EOL> sess . add ( c3 ) <EOL> c3 . parent_id = p1 . id <EOL> assert c3 . parent is None <EOL> def test_autoflush_on_pending ( self ) : <EOL> Child . parent . property . load_on_pending = True <EOL> c3 = Child ( ) <EOL> sess . add ( c3 ) <EOL> c3 . parent_id = p1 . id <EOL> assert c3 . parent is p1 <EOL> def test_collection_load_from_pending_populated ( self ) : <EOL> Parent . children . property . load_on_pending = True <EOL> p2 = Parent ( id = p1 . id ) <EOL> sess . add ( p2 ) <EOL> def go ( ) : <EOL> assert p2 . children <EOL> self . assert_sql_count ( testing . db , go , <NUM_LIT:1> ) <EOL> def test_collection_load_from_pending_no_sql ( self ) : <EOL> Parent . children . property . load_on_pending = True <EOL> p2 = Parent ( id = None ) <EOL> sess . add ( p2 ) <EOL> def go ( ) : <EOL> assert not p2 . children <EOL> self . assert_sql_count ( testing . db , go , <NUM_LIT:0> ) <EOL> def test_load_on_pending_with_set ( self ) : <EOL> Child . parent . property . load_on_pending = True <EOL> p1 . children <EOL> c3 = Child ( ) <EOL> sess . add ( c3 ) <EOL> c3 . parent_id = p1 . id <EOL> def go ( ) : <EOL> c3 . parent = p1 <EOL> self . assert_sql_count ( testing . db , go , <NUM_LIT:0> ) <EOL> def test_backref_doesnt_double ( self ) : <EOL> Child . parent . property . load_on_pending = True <EOL> sess . autoflush = False <EOL> p1 . children <EOL> c3 = Child ( ) <EOL> sess . add ( c3 ) <EOL> c3 . parent = p1 <EOL> c3 . parent = p1 <EOL> c3 . parent = p1 <EOL> c3 . parent = p1 <EOL> assert len ( p1 . children ) == <NUM_LIT:2> <EOL> def test_m2o_lazy_loader_on_persistent ( self ) : <EOL> """<STR_LIT>""" <EOL> for loadfk in ( True , False ) : <EOL> for loadrel in ( True , False ) : <EOL> for autoflush in ( True , False ) : <EOL> for manualflush in ( True , False ) : <EOL> for fake_autoexpire in ( True , False ) : <EOL> sess . autoflush = autoflush <EOL> if loadfk : <EOL> c1 . parent_id <EOL> if loadrel : <EOL> c1 . parent <EOL> c1 . parent_id = p2 . id <EOL> if manualflush : <EOL> sess . flush ( ) <EOL> if fake_autoexpire : <EOL> sess . expire ( c1 , [ '<STR_LIT>' ] ) <EOL> if loadrel and not fake_autoexpire : <EOL> assert c1 . parent is p1 <EOL> else : <EOL> assert c1 . parent is p2 <EOL> sess . rollback ( ) <EOL> def test_m2o_lazy_loader_on_pending ( self ) : <EOL> for loadonpending in ( False , True ) : <EOL> for autoflush in ( False , True ) : <EOL> for manualflush in ( False , True ) : <EOL> Child . parent . property . load_on_pending = loadonpending <EOL> sess . autoflush = autoflush <EOL> c2 = Child ( ) <EOL> sess . add ( c2 ) <EOL> c2 . parent_id = p2 . id <EOL> if manualflush : <EOL> sess . flush ( ) <EOL> if loadonpending or manualflush : <EOL> assert c2 . parent is p2 <EOL> else : <EOL> assert c2 . parent is None <EOL> sess . rollback ( ) <EOL> def test_m2o_lazy_loader_on_transient ( self ) : <EOL> for loadonpending in ( False , True ) : <EOL> for attach in ( False , True ) : <EOL> for autoflush in ( False , True ) : <EOL> for manualflush in ( False , True ) : <EOL> for enable_relationship_rel in ( False , True ) : <EOL> Child . parent . property . load_on_pending = loadonpending <EOL> sess . autoflush = autoflush <EOL> c2 = Child ( ) <EOL> if attach : <EOL> state = instance_state ( c2 ) <EOL> state . session_id = sess . hash_key <EOL> if enable_relationship_rel : <EOL> sess . enable_relationship_loading ( c2 ) <EOL> c2 . parent_id = p2 . id <EOL> if manualflush : <EOL> sess . flush ( ) <EOL> if ( loadonpending and attach ) or enable_relationship_rel : <EOL> assert c2 . parent is p2 <EOL> else : <EOL> assert c2 . parent is None <EOL> sess . rollback ( ) </s>
