- starting to groom the branch for its inclusion

- one-to-many relationships now maintain a list of positive
parent-child associations within the flush, preventing
previous parents marked as deleted from cascading a
delete or NULL foreign key set on those child objects,
despite the end-user not removing the child from the old
association. [ticket:1764]
- re-established Preprocess as unique on their arguments,
as they were definitely duped in inheritance scenarios
- added a "memo" feature to UOWTransaction which represents the usual
pattern of using the .attributes collection
- added the test case from [ticket:1081] into perf/

3 files changed, 238 insertions, 8 deletions

diff --git a/test/orm/test_cascade.py b/test/orm/test_cascade.py
index a7152ecc1..7b07898a5 100644
--- a/test/orm/test_cascade.py
+++ b/test/orm/test_cascade.py
@@ -1156,7 +1156,6 @@ class UnsavedOrphansTest3(_base.MappedTest):
         assert c not in s, "Should expunge customer when both parents are gone"
 
         
-        
 class DoubleParentOrphanTest(_base.MappedTest):
     """test orphan detection for an entity with two parent relationships"""
 
@@ -1276,6 +1275,151 @@ class CollectionAssignmentOrphanTest(_base.MappedTest):
         eq_(sess.query(A).get(a1.id),
             A(name='a1', bs=[B(name='b1'), B(name='b2'), B(name='b3')]))
 
+class O2MConflictTest(_base.MappedTest):
+    """test that O2M dependency detects a change in parent, does the
+    right thing, and even updates the collection/attribute.
+    
+    """
+    
+    @classmethod
+    def define_tables(cls, metadata):
+        Table("parent", metadata,
+            Column("id", Integer, primary_key=True, test_needs_autoincrement=True)
+        )
+        Table("child", metadata,
+            Column("id", Integer, primary_key=True, test_needs_autoincrement=True),
+            Column('parent_id', Integer, ForeignKey('parent.id'), nullable=False)
+        )
+    
+    @classmethod
+    def setup_classes(cls):
+        class Parent(_base.ComparableEntity):
+            pass
+        class Child(_base.ComparableEntity):
+            pass
+    
+    @testing.resolve_artifact_names
+    def _do_delete_old_test(self):
+        sess = create_session()
+        
+        p1, p2, c1 = Parent(), Parent(), Child()
+        if Parent.child.property.uselist:
+            p1.child.append(c1)
+        else:
+            p1.child = c1
+        sess.add_all([p1, c1])
+        sess.flush()
+        
+        sess.delete(p1)
+        
+        if Parent.child.property.uselist:
+            p2.child.append(c1)
+        else:
+            p2.child = c1
+        sess.add(p2)
+
+        sess.flush()
+        eq_(sess.query(Child).filter(Child.parent_id==p2.id).all(), [c1])
+
+    @testing.resolve_artifact_names
+    def _do_move_test(self):
+        sess = create_session()
+
+        p1, p2, c1 = Parent(), Parent(), Child()
+        if Parent.child.property.uselist:
+            p1.child.append(c1)
+        else:
+            p1.child = c1
+        sess.add_all([p1, c1])
+        sess.flush()
+
+        if Parent.child.property.uselist:
+            p2.child.append(c1)
+        else:
+            p2.child = c1
+        sess.add(p2)
+
+        sess.flush()
+        eq_(sess.query(Child).filter(Child.parent_id==p2.id).all(), [c1])
+        
+    @testing.resolve_artifact_names
+    def test_o2o_delete_old(self):
+        mapper(Parent, parent, properties={
+            'child':relationship(Child, uselist=False)
+        })
+        mapper(Child, child)
+        self._do_delete_old_test()
+        self._do_move_test()
+
+    @testing.resolve_artifact_names
+    def test_o2m_delete_old(self):
+        mapper(Parent, parent, properties={
+            'child':relationship(Child, uselist=True)
+        })
+        mapper(Child, child)
+        self._do_delete_old_test()
+        self._do_move_test()
+
+    @testing.resolve_artifact_names
+    def test_o2o_backref_delete_old(self):
+        mapper(Parent, parent, properties={
+            'child':relationship(Child, uselist=False, backref='parent')
+        })
+        mapper(Child, child)
+        self._do_delete_old_test()
+        self._do_move_test()
+        
+    @testing.resolve_artifact_names
+    def test_o2o_delcascade_delete_old(self):
+        mapper(Parent, parent, properties={
+            'child':relationship(Child, uselist=False, cascade="all, delete")
+        })
+        mapper(Child, child)
+        self._do_delete_old_test()
+        self._do_move_test()
+
+    @testing.resolve_artifact_names
+    def test_o2o_delorphan_delete_old(self):
+        mapper(Parent, parent, properties={
+            'child':relationship(Child, uselist=False, cascade="all, delete, delete-orphan")
+        })
+        mapper(Child, child)
+        self._do_delete_old_test()
+        self._do_move_test()
+
+    @testing.resolve_artifact_names
+    def test_o2o_delorphan_backref_delete_old(self):
+        mapper(Parent, parent, properties={
+            'child':relationship(Child, uselist=False, 
+                                        cascade="all, delete, delete-orphan", 
+                                        backref='parent')
+        })
+        mapper(Child, child)
+        self._do_delete_old_test()
+        self._do_move_test()
+
+    @testing.resolve_artifact_names
+    def test_o2o_backref_delorphan_delete_old(self):
+        mapper(Parent, parent)
+        mapper(Child, child, properties = {
+            'parent' : relationship(Parent, uselist=False, single_parent=True, 
+                                backref=backref('child', uselist=False), 
+                                cascade="all,delete,delete-orphan")
+        })
+        self._do_delete_old_test()
+        self._do_move_test()
+
+    @testing.resolve_artifact_names
+    def test_o2m_backref_delorphan_delete_old(self):
+        mapper(Parent, parent)
+        mapper(Child, child, properties = {
+            'parent' : relationship(Parent, uselist=False, single_parent=True, 
+                                backref=backref('child', uselist=True), 
+                                cascade="all,delete,delete-orphan")
+        })
+        self._do_delete_old_test()
+        self._do_move_test()
+        
 
 class PartialFlushTest(_base.MappedTest):
     """test cascade behavior as it relates to object lists passed to flush().
diff --git a/test/orm/test_unitofworkv2.py b/test/orm/test_unitofworkv2.py
index 7fef87d33..e28537b00 100644
--- a/test/orm/test_unitofworkv2.py
+++ b/test/orm/test_unitofworkv2.py
@@ -14,10 +14,7 @@ from test.orm._fixtures import keywords, addresses, Base, Keyword,  \
             composite_pk_table, CompositePk
 
 class AssertsUOW(object):
-    def _assert_uow_size(self,
-        session, 
-        expected
-    ):
+    def _get_test_uow(self, session):
         uow = unitofwork.UOWTransaction(session)
         deleted = set(session._deleted)
         new = set(session._new)
@@ -26,6 +23,13 @@ class AssertsUOW(object):
             uow.register_object(s)
         for d in deleted:
             uow.register_object(d, isdelete=True)
+        return uow
+        
+    def _assert_uow_size(self,
+        session, 
+        expected
+    ):
+        uow = self._get_test_uow(session)
         postsort_actions = uow._generate_actions()
         print postsort_actions
         eq_(len(postsort_actions), expected, postsort_actions)
@@ -33,8 +37,6 @@ class AssertsUOW(object):
 class UOWTest(_fixtures.FixtureTest, testing.AssertsExecutionResults, AssertsUOW):
     run_inserts = None
 
-    
-
 class RudimentaryFlushTest(UOWTest):
 
     def test_one_to_many_save(self):
@@ -215,7 +217,7 @@ class RudimentaryFlushTest(UOWTest):
                     {'id':u1.id}
                 ),
         )
-
+    
     def test_m2o_flush_size(self):
         mapper(User, users)
         mapper(Address, addresses, properties={
diff --git a/test/perf/large_flush.py b/test/perf/large_flush.py
new file mode 100644
index 000000000..431a28944
--- /dev/null
+++ b/test/perf/large_flush.py
@@ -0,0 +1,84 @@
+import sqlalchemy as sa
+from sqlalchemy import create_engine, MetaData, orm
+from sqlalchemy import Column, ForeignKey
+from sqlalchemy import Integer, String
+from sqlalchemy.orm import mapper
+from sqlalchemy.test import profiling
+
+class Object(object):
+    pass
+
+class Q(Object):
+    pass
+
+class A(Object):
+    pass
+
+class C(Object):
+    pass
+
+class WC(C):
+    pass
+
+engine = create_engine('sqlite:///:memory:', echo=True)
+
+sm = orm.sessionmaker(bind=engine)
+
+SA_Session = orm.scoped_session(sm)
+
+SA_Metadata = MetaData()
+
+object_table =  sa.Table('Object',
+                          SA_Metadata,
+                          Column('ObjectID', Integer,primary_key=True),
+                          Column('Type', String(1), nullable=False))
+
+q_table = sa.Table('Q',
+                   SA_Metadata,
+                   Column('QID', Integer, ForeignKey('Object.ObjectID'),primary_key=True))
+
+c_table = sa.Table('C',
+                   SA_Metadata,
+                   Column('CID', Integer, ForeignKey('Object.ObjectID'),primary_key=True))
+
+wc_table = sa.Table('WC',
+                    SA_Metadata,
+                    Column('WCID', Integer, ForeignKey('C.CID'), primary_key=True))
+
+a_table = sa.Table('A',
+                   SA_Metadata,
+                   Column('AID', Integer, ForeignKey('Object.ObjectID'),primary_key=True),
+                   Column('QID', Integer, ForeignKey('Q.QID')),
+                   Column('CID', Integer, ForeignKey('C.CID')))
+
+mapper(Object, object_table, polymorphic_on=object_table.c.Type, polymorphic_identity='O')
+
+mapper(Q, q_table, inherits=Object, polymorphic_identity='Q')
+mapper(C, c_table, inherits=Object, polymorphic_identity='C')
+mapper(WC, wc_table, inherits=C, polymorphic_identity='W')
+
+mapper(A, a_table, inherits=Object, polymorphic_identity='A',
+       properties = {
+                     'Q' : orm.relation(Q,primaryjoin=a_table.c.QID==q_table.c.QID,
+                                        backref='As'
+                                        ),
+                     'C' : orm.relation(C,primaryjoin=a_table.c.CID==c_table.c.CID,
+                                        backref='A',
+                                        uselist=False)
+                     }
+       )
+
+SA_Metadata.create_all(engine)
+
+@profiling.profiled('large_flush', always=True, sort=['file'])
+def generate_error():
+    q = Q()
+    for j in range(100): #at 306 the error does not pop out (depending on recursion depth)
+        a = A()
+        a.Q = q
+        a.C = WC()
+
+    SA_Session.add(q)
+    SA_Session.commit() #here the error pops out
+
+generate_error()
\ No newline at end of file