path: root/test/orm/test_cycles.py

"""Tests cyclical mapper relationships.

We might want to try an automated generate of much of this, all combos of
T1<->T2, with o2m or m2o between them, and a third T3 with o2m/m2o to one/both
T1/T2.

"""
from test.lib import testing
from sqlalchemy import Integer, String, ForeignKey
from test.lib.schema import Table, Column
from sqlalchemy.orm import mapper, relationship, backref, \
                            create_session, sessionmaker
from test.lib.testing import eq_
from test.lib.assertsql import RegexSQL, ExactSQL, CompiledSQL, AllOf
from test.lib import fixtures


class SelfReferentialTest(fixtures.MappedTest):
    """A self-referential mapper with an additional list of child objects."""

    @classmethod
    def define_tables(cls, metadata):
        Table('t1', metadata,
              Column('c1', Integer, primary_key=True,
                     test_needs_autoincrement=True),
              Column('parent_c1', Integer, ForeignKey('t1.c1')),
              Column('data', String(20)))
        Table('t2', metadata,
              Column('c1', Integer, primary_key=True,
                     test_needs_autoincrement=True),
              Column('c1id', Integer, ForeignKey('t1.c1')),
              Column('data', String(20)))

    @classmethod
    def setup_classes(cls):
        class C1(cls.Basic):
            def __init__(self, data=None):
                self.data = data

        class C2(cls.Basic):
            def __init__(self, data=None):
                self.data = data

    def test_single(self):
        C1, t1 = self.classes.C1, self.tables.t1

        mapper(C1, t1, properties = {
            'c1s':relationship(C1, cascade="all"),
            'parent':relationship(C1,
                              primaryjoin=t1.c.parent_c1 == t1.c.c1,
                              remote_side=t1.c.c1,
                              lazy='select',
                              uselist=False)})
        a = C1('head c1')
        a.c1s.append(C1('another c1'))

        sess = create_session( )
        sess.add(a)
        sess.flush()
        sess.delete(a)
        sess.flush()

    def test_many_to_one_only(self):
        """

        test that the circular dependency sort can assemble a many-to-one
        dependency processor when only the object on the "many" side is
        actually in the list of modified objects.

        """

        C1, t1 = self.classes.C1, self.tables.t1

        mapper(C1, t1, properties={
            'parent':relationship(C1,
                              primaryjoin=t1.c.parent_c1 == t1.c.c1,
                              remote_side=t1.c.c1)})

        c1 = C1()

        sess = create_session()
        sess.add(c1)
        sess.flush()
        sess.expunge_all()
        c1 = sess.query(C1).get(c1.c1)
        c2 = C1()
        c2.parent = c1
        sess.add(c2)
        sess.flush()
        assert c2.parent_c1==c1.c1

    def test_cycle(self):
        C2, C1, t2, t1 = (self.classes.C2,
                                self.classes.C1,
                                self.tables.t2,
                                self.tables.t1)

        mapper(C1, t1, properties = {
            'c1s' : relationship(C1, cascade="all"),
            'c2s' : relationship(mapper(C2, t2), cascade="all, delete-orphan")})

        a = C1('head c1')
        a.c1s.append(C1('child1'))
        a.c1s.append(C1('child2'))
        a.c1s[0].c1s.append(C1('subchild1'))
        a.c1s[0].c1s.append(C1('subchild2'))
        a.c1s[1].c2s.append(C2('child2 data1'))
        a.c1s[1].c2s.append(C2('child2 data2'))
        sess = create_session( )
        sess.add(a)
        sess.flush()

        sess.delete(a)
        sess.flush()

    def test_setnull_ondelete(self):
        C1, t1 = self.classes.C1, self.tables.t1

        mapper(C1, t1, properties={
            'children':relationship(C1)
        })

        sess = create_session()
        c1 = C1()
        c2 = C1()
        c1.children.append(c2)
        sess.add(c1)
        sess.flush()
        assert c2.parent_c1 == c1.c1

        sess.delete(c1)
        sess.flush()
        assert c2.parent_c1 is None

        sess.expire_all()
        assert c2.parent_c1 is None

class SelfReferentialNoPKTest(fixtures.MappedTest):
    """A self-referential relationship that joins on a column other than the primary key column"""

    @classmethod
    def define_tables(cls, metadata):
        Table('item', metadata,
           Column('id', Integer, primary_key=True, test_needs_autoincrement=True),
           Column('uuid', String(32), unique=True, nullable=False),
           Column('parent_uuid', String(32), ForeignKey('item.uuid'),
                  nullable=True))

    @classmethod
    def setup_classes(cls):
        class TT(cls.Basic):
            def __init__(self):
                self.uuid = hex(id(self))

    @classmethod
    def setup_mappers(cls):
        item, TT = cls.tables.item, cls.classes.TT

        mapper(TT, item, properties={
            'children': relationship(
                TT,
                remote_side=[item.c.parent_uuid],
                backref=backref('parent', remote_side=[item.c.uuid]))})

    def test_basic(self):
        TT = self.classes.TT

        t1 = TT()
        t1.children.append(TT())
        t1.children.append(TT())

        s = create_session()
        s.add(t1)
        s.flush()
        s.expunge_all()
        t = s.query(TT).filter_by(id=t1.id).one()
        eq_(t.children[0].parent_uuid, t1.uuid)

    def test_lazy_clause(self):
        TT = self.classes.TT

        s = create_session()
        t1 = TT()
        t2 = TT()
        t1.children.append(t2)
        s.add(t1)
        s.flush()
        s.expunge_all()

        t = s.query(TT).filter_by(id=t2.id).one()
        eq_(t.uuid, t2.uuid)
        eq_(t.parent.uuid, t1.uuid)


class InheritTestOne(fixtures.MappedTest):
    @classmethod
    def define_tables(cls, metadata):
        Table("parent", metadata,
            Column("id", Integer, primary_key=True, test_needs_autoincrement=True),
            Column("parent_data", String(50)),
            Column("type", String(10)))

        Table("child1", metadata,
              Column("id", Integer, ForeignKey("parent.id"), primary_key=True),
              Column("child1_data", String(50)))

        Table("child2", metadata,
            Column("id", Integer, ForeignKey("parent.id"), primary_key=True),
            Column("child1_id", Integer, ForeignKey("child1.id"),
                   nullable=False),
            Column("child2_data", String(50)))

    @classmethod
    def setup_classes(cls):
        class Parent(cls.Basic):
            pass

        class Child1(Parent):
            pass

        class Child2(Parent):
            pass

    @classmethod
    def setup_mappers(cls):
        child1, child2, parent, Parent, Child1, Child2 = (cls.tables.child1,
                                cls.tables.child2,
                                cls.tables.parent,
                                cls.classes.Parent,
                                cls.classes.Child1,
                                cls.classes.Child2)

        mapper(Parent, parent)
        mapper(Child1, child1, inherits=Parent)
        mapper(Child2, child2, inherits=Parent, properties=dict(
            child1=relationship(Child1,
                            primaryjoin=child2.c.child1_id == child1.c.id)))

    def test_many_to_one_only(self):
        """test similar to SelfReferentialTest.testmanytooneonly"""

        Child1, Child2 = self.classes.Child1, self.classes.Child2


        session = create_session()

        c1 = Child1()
        c1.child1_data = "qwerty"
        session.add(c1)
        session.flush()
        session.expunge_all()

        c1 = session.query(Child1).filter_by(child1_data="qwerty").one()
        c2 = Child2()
        c2.child1 = c1
        c2.child2_data = "asdfgh"
        session.add(c2)

        # the flush will fail if the UOW does not set up a many-to-one DP
        # attached to a task corresponding to c1, since "child1_id" is not
        # nullable
        session.flush()


class InheritTestTwo(fixtures.MappedTest):
    """

    The fix in BiDirectionalManyToOneTest raised this issue, regarding the
    'circular sort' containing UOWTasks that were still polymorphic, which
    could create duplicate entries in the final sort

    """

    @classmethod
    def define_tables(cls, metadata):
        Table('a', metadata,
            Column('id', Integer, primary_key=True, test_needs_autoincrement=True),
            Column('data', String(30)),
            Column('cid', Integer, ForeignKey('c.id')))

        Table('b', metadata,
            Column('id', Integer, ForeignKey("a.id"), primary_key=True),
            Column('data', String(30)))

        Table('c', metadata,
            Column('id', Integer, primary_key=True, test_needs_autoincrement=True),
            Column('data', String(30)),
            Column('aid', Integer,
                   ForeignKey('a.id', use_alter=True, name="foo")))

    @classmethod
    def setup_classes(cls):
        class A(cls.Basic):
            pass

        class B(A):
            pass

        class C(cls.Basic):
            pass

    def test_flush(self):
        a, A, c, b, C, B = (self.tables.a,
                                self.classes.A,
                                self.tables.c,
                                self.tables.b,
                                self.classes.C,
                                self.classes.B)

        mapper(A, a, properties={
            'cs':relationship(C, primaryjoin=a.c.cid==c.c.id)})

        mapper(B, b, inherits=A, inherit_condition=b.c.id == a.c.id)

        mapper(C, c, properties={
            'arel':relationship(A, primaryjoin=a.c.id == c.c.aid)})

        sess = create_session()
        bobj = B()
        sess.add(bobj)
        cobj = C()
        sess.add(cobj)
        sess.flush()


class BiDirectionalManyToOneTest(fixtures.MappedTest):
    run_define_tables = 'each'

    @classmethod
    def define_tables(cls, metadata):
        Table('t1', metadata,
            Column('id', Integer, primary_key=True, test_needs_autoincrement=True),
            Column('data', String(30)),
            Column('t2id', Integer, ForeignKey('t2.id')))
        Table('t2', metadata,
            Column('id', Integer, primary_key=True, test_needs_autoincrement=True),
            Column('data', String(30)),
            Column('t1id', Integer,
                   ForeignKey('t1.id', use_alter=True, name="foo_fk")))
        Table('t3', metadata,
            Column('id', Integer, primary_key=True, test_needs_autoincrement=True),
            Column('data', String(30)),
            Column('t1id', Integer, ForeignKey('t1.id'), nullable=False),
            Column('t2id', Integer, ForeignKey('t2.id'), nullable=False))

    @classmethod
    def setup_classes(cls):
        class T1(cls.Basic):
            pass
        class T2(cls.Basic):
            pass
        class T3(cls.Basic):
            pass

    @classmethod
    def setup_mappers(cls):
        t2, T2, T3, t1, t3, T1 = (cls.tables.t2,
                                cls.classes.T2,
                                cls.classes.T3,
                                cls.tables.t1,
                                cls.tables.t3,
                                cls.classes.T1)

        mapper(T1, t1, properties={
            't2':relationship(T2, primaryjoin=t1.c.t2id == t2.c.id)})
        mapper(T2, t2, properties={
            't1':relationship(T1, primaryjoin=t2.c.t1id == t1.c.id)})
        mapper(T3, t3, properties={
            't1':relationship(T1),
            't2':relationship(T2)})

    def test_reflush(self):
        T2, T3, T1 = (self.classes.T2,
                                self.classes.T3,
                                self.classes.T1)

        o1 = T1()
        o1.t2 = T2()
        sess = create_session()
        sess.add(o1)
        sess.flush()

        # the bug here is that the dependency sort comes up with T1/T2 in a
        # cycle, but there are no T1/T2 objects to be saved.  therefore no
        # "cyclical subtree" gets generated, and one or the other of T1/T2
        # gets lost, and processors on T3 dont fire off.  the test will then
        # fail because the FK's on T3 are not nullable.
        o3 = T3()
        o3.t1 = o1
        o3.t2 = o1.t2
        sess.add(o3)
        sess.flush()

    def test_reflush_2(self):
        """A variant on test_reflush()"""

        T2, T3, T1 = (self.classes.T2,
                                self.classes.T3,
                                self.classes.T1)

        o1 = T1()
        o1.t2 = T2()
        sess = create_session()
        sess.add(o1)
        sess.flush()

        # in this case, T1, T2, and T3 tasks will all be in the cyclical
        # tree normally.  the dependency processors for T3 are part of the
        # 'extradeps' collection so they all get assembled into the tree
        # as well.
        o1a = T1()
        o2a = T2()
        sess.add(o1a)
        sess.add(o2a)
        o3b = T3()
        o3b.t1 = o1a
        o3b.t2 = o2a
        sess.add(o3b)

        o3 = T3()
        o3.t1 = o1
        o3.t2 = o1.t2
        sess.add(o3)
        sess.flush()


class BiDirectionalOneToManyTest(fixtures.MappedTest):
    """tests two mappers with a one-to-many relationship to each other."""

    run_define_tables = 'each'

    @classmethod
    def define_tables(cls, metadata):
        Table('t1', metadata,
              Column('c1', Integer, primary_key=True, test_needs_autoincrement=True),
              Column('c2', Integer, ForeignKey('t2.c1')))

        Table('t2', metadata,
              Column('c1', Integer, primary_key=True, test_needs_autoincrement=True),
              Column('c2', Integer,
                     ForeignKey('t1.c1', use_alter=True, name='t1c1_fk')))

    @classmethod
    def setup_classes(cls):
        class C1(cls.Basic):
            pass

        class C2(cls.Basic):
            pass

    def test_cycle(self):
        C2, C1, t2, t1 = (self.classes.C2,
                                self.classes.C1,
                                self.tables.t2,
                                self.tables.t1)

        mapper(C2, t2, properties={
            'c1s': relationship(C1,
                            primaryjoin=t2.c.c1 == t1.c.c2,
                            uselist=True)})
        mapper(C1, t1, properties={
            'c2s': relationship(C2,
                            primaryjoin=t1.c.c1 == t2.c.c2,
                            uselist=True)})

        a = C1()
        b = C2()
        c = C1()
        d = C2()
        e = C2()
        f = C2()
        a.c2s.append(b)
        d.c1s.append(c)
        b.c1s.append(c)
        sess = create_session()
        sess.add_all((a, b, c, d, e, f))
        sess.flush()


class BiDirectionalOneToManyTest2(fixtures.MappedTest):
    """Two mappers with a one-to-many relationship to each other,
    with a second one-to-many on one of the mappers"""

    run_define_tables = 'each'

    @classmethod
    def define_tables(cls, metadata):
        Table('t1', metadata,
              Column('c1', Integer, primary_key=True, test_needs_autoincrement=True),
              Column('c2', Integer, ForeignKey('t2.c1')),
              test_needs_autoincrement=True)

        Table('t2', metadata,
              Column('c1', Integer, primary_key=True, test_needs_autoincrement=True),
              Column('c2', Integer,
                     ForeignKey('t1.c1', use_alter=True, name='t1c1_fq')),
              test_needs_autoincrement=True)

        Table('t1_data', metadata,
              Column('c1', Integer, primary_key=True, test_needs_autoincrement=True),
              Column('t1id', Integer, ForeignKey('t1.c1')),
              Column('data', String(20)),
              test_needs_autoincrement=True)

    @classmethod
    def setup_classes(cls):
        class C1(cls.Basic):
            pass

        class C2(cls.Basic):
            pass

        class C1Data(cls.Basic):
            pass

    @classmethod
    def setup_mappers(cls):
        t2, t1, C1Data, t1_data, C2, C1 = (cls.tables.t2,
                                cls.tables.t1,
                                cls.classes.C1Data,
                                cls.tables.t1_data,
                                cls.classes.C2,
                                cls.classes.C1)

        mapper(C2, t2, properties={
            'c1s': relationship(C1,
                            primaryjoin=t2.c.c1 == t1.c.c2,
                            uselist=True)})
        mapper(C1, t1, properties={
            'c2s': relationship(C2,
                             primaryjoin=t1.c.c1 == t2.c.c2,
                             uselist=True),
            'data': relationship(mapper(C1Data, t1_data))})

    def test_cycle(self):
        C2, C1, C1Data = (self.classes.C2,
                                self.classes.C1,
                                self.classes.C1Data)

        a = C1()
        b = C2()
        c = C1()
        d = C2()
        e = C2()
        f = C2()
        a.c2s.append(b)
        d.c1s.append(c)
        b.c1s.append(c)
        a.data.append(C1Data(data='c1data1'))
        a.data.append(C1Data(data='c1data2'))
        c.data.append(C1Data(data='c1data3'))
        sess = create_session()
        sess.add_all((a, b, c, d, e, f))
        sess.flush()

        sess.delete(d)
        sess.delete(c)
        sess.flush()

class OneToManyManyToOneTest(fixtures.MappedTest):
    """

    Tests two mappers, one has a one-to-many on the other mapper, the other
    has a separate many-to-one relationship to the first.  two tests will have
    a row for each item that is dependent on the other.  without the
    "post_update" flag, such relationships raise an exception when
    dependencies are sorted.

    """
    run_define_tables = 'each'

    @classmethod
    def define_tables(cls, metadata):
        Table('ball', metadata,
              Column('id', Integer, primary_key=True, test_needs_autoincrement=True),
              Column('person_id', Integer,
                     ForeignKey('person.id', use_alter=True, name='fk_person_id')),
              Column('data', String(30)))

        Table('person', metadata,
              Column('id', Integer, primary_key=True, test_needs_autoincrement=True),
              Column('favorite_ball_id', Integer, ForeignKey('ball.id')),
              Column('data', String(30)))

    @classmethod
    def setup_classes(cls):
        class Person(cls.Basic):
            pass

        class Ball(cls.Basic):
            pass

    def test_cycle(self):
        """
        This test has a peculiar aspect in that it doesnt create as many
        dependent relationships as the other tests, and revealed a small
        glitch in the circular dependency sorting.

        """

        person, ball, Ball, Person = (self.tables.person,
                                self.tables.ball,
                                self.classes.Ball,
                                self.classes.Person)

        mapper(Ball, ball)
        mapper(Person, person, properties=dict(
            balls=relationship(Ball,
                           primaryjoin=ball.c.person_id == person.c.id,
                           remote_side=ball.c.person_id),
            favorite=relationship(Ball,
                              primaryjoin=person.c.favorite_ball_id == ball.c.id,
                              remote_side=ball.c.id)))

        b = Ball()
        p = Person()
        p.balls.append(b)
        sess = create_session()
        sess.add(p)
        sess.flush()

    def test_post_update_m2o(self):
        """A cycle between two rows, with a post_update on the many-to-one"""

        person, ball, Ball, Person = (self.tables.person,
                                self.tables.ball,
                                self.classes.Ball,
                                self.classes.Person)

        mapper(Ball, ball)
        mapper(Person, person, properties=dict(
            balls=relationship(Ball,
                           primaryjoin=ball.c.person_id == person.c.id,
                           remote_side=ball.c.person_id,
                           post_update=False,
                           cascade="all, delete-orphan"),
            favorite=relationship(Ball,
                              primaryjoin=person.c.favorite_ball_id == ball.c.id,
                              remote_side=person.c.favorite_ball_id,
                              post_update=True)))

        b = Ball(data='some data')
        p = Person(data='some data')
        p.balls.append(b)
        p.balls.append(Ball(data='some data'))
        p.balls.append(Ball(data='some data'))
        p.balls.append(Ball(data='some data'))
        p.favorite = b
        sess = create_session()
        sess.add(b)
        sess.add(p)

        self.assert_sql_execution(
            testing.db,
            sess.flush,
            RegexSQL("^INSERT INTO person", {'data':'some data'}),
            RegexSQL("^INSERT INTO ball", lambda c: {'person_id':p.id, 'data':'some data'}),
            RegexSQL("^INSERT INTO ball", lambda c: {'person_id':p.id, 'data':'some data'}),
            RegexSQL("^INSERT INTO ball", lambda c: {'person_id':p.id, 'data':'some data'}),
            RegexSQL("^INSERT INTO ball", lambda c: {'person_id':p.id, 'data':'some data'}),
            ExactSQL("UPDATE person SET favorite_ball_id=:favorite_ball_id "
                        "WHERE person.id = :person_id",
                        lambda ctx:{'favorite_ball_id':p.favorite.id, 'person_id':p.id}
             ),
        )

        sess.delete(p)

        self.assert_sql_execution(
            testing.db,
            sess.flush,
            ExactSQL("UPDATE person SET favorite_ball_id=:favorite_ball_id "
                "WHERE person.id = :person_id",
                lambda ctx: {'person_id': p.id, 'favorite_ball_id': None}),
            ExactSQL("DELETE FROM ball WHERE ball.id = :id", None), # lambda ctx:[{'id': 1L}, {'id': 4L}, {'id': 3L}, {'id': 2L}])
            ExactSQL("DELETE FROM person WHERE person.id = :id", lambda ctx:[{'id': p.id}])
        )

    def test_post_update_backref(self):
        """test bidirectional post_update."""

        person, ball, Ball, Person = (self.tables.person,
                                self.tables.ball,
                                self.classes.Ball,
                                self.classes.Person)


        mapper(Ball, ball)
        mapper(Person, person, properties=dict(
            balls=relationship(Ball,
                           primaryjoin=ball.c.person_id == person.c.id,
                           remote_side=ball.c.person_id, post_update=True,
                           backref=backref('person', post_update=True)
                           ),
           favorite=relationship(Ball,
                             primaryjoin=person.c.favorite_ball_id == ball.c.id,
                             remote_side=person.c.favorite_ball_id)

            ))

        sess = sessionmaker()()
        p1 = Person(data='p1')
        p2 = Person(data='p2')
        p3 = Person(data='p3')

        b1 = Ball(data='b1')

        b1.person = p1
        sess.add_all([p1, p2, p3])
        sess.commit()

        # switch here.  the post_update
        # on ball.person can't get tripped up
        # by the fact that there's a "reverse" prop.
        b1.person = p2
        sess.commit()
        eq_(
            p2, b1.person
        )

        # do it the other way
        p3.balls.append(b1)
        sess.commit()
        eq_(
            p3, b1.person
        )



    def test_post_update_o2m(self):
        """A cycle between two rows, with a post_update on the one-to-many"""

        person, ball, Ball, Person = (self.tables.person,
                                self.tables.ball,
                                self.classes.Ball,
                                self.classes.Person)


        mapper(Ball, ball)
        mapper(Person, person, properties=dict(
            balls=relationship(Ball,
                           primaryjoin=ball.c.person_id == person.c.id,
                           remote_side=ball.c.person_id,
                           cascade="all, delete-orphan",
                           post_update=True,
                           backref='person'),
            favorite=relationship(Ball,
                              primaryjoin=person.c.favorite_ball_id == ball.c.id,
                              remote_side=person.c.favorite_ball_id)))

        b = Ball(data='some data')
        p = Person(data='some data')
        p.balls.append(b)
        b2 = Ball(data='some data')
        p.balls.append(b2)
        b3 = Ball(data='some data')
        p.balls.append(b3)
        b4 = Ball(data='some data')
        p.balls.append(b4)
        p.favorite = b
        sess = create_session()
        sess.add_all((b,p,b2,b3,b4))

        self.assert_sql_execution(
            testing.db,
            sess.flush,
            CompiledSQL("INSERT INTO ball (person_id, data) "
             "VALUES (:person_id, :data)",
             {'person_id':None, 'data':'some data'}),

            CompiledSQL("INSERT INTO ball (person_id, data) "
             "VALUES (:person_id, :data)",
             {'person_id':None, 'data':'some data'}),

            CompiledSQL("INSERT INTO ball (person_id, data) "
             "VALUES (:person_id, :data)",
             {'person_id':None, 'data':'some data'}),

            CompiledSQL("INSERT INTO ball (person_id, data) "
             "VALUES (:person_id, :data)",
             {'person_id':None, 'data':'some data'}),

            CompiledSQL("INSERT INTO person (favorite_ball_id, data) "
             "VALUES (:favorite_ball_id, :data)",
             lambda ctx:{'favorite_ball_id':b.id, 'data':'some data'}),

             CompiledSQL("UPDATE ball SET person_id=:person_id "
              "WHERE ball.id = :ball_id",
              lambda ctx:[
                {'person_id':p.id,'ball_id':b.id},
                {'person_id':p.id,'ball_id':b2.id},
                {'person_id':p.id,'ball_id':b3.id},
                {'person_id':p.id,'ball_id':b4.id}
                ]
            ),

        )

        sess.delete(p)

        self.assert_sql_execution(testing.db, sess.flush,
            CompiledSQL("UPDATE ball SET person_id=:person_id "
                "WHERE ball.id = :ball_id",
                lambda ctx:[
                    {'person_id': None, 'ball_id': b.id},
                    {'person_id': None, 'ball_id': b2.id},
                    {'person_id': None, 'ball_id': b3.id},
                    {'person_id': None, 'ball_id': b4.id}
                ]
            ),
            CompiledSQL("DELETE FROM person WHERE person.id = :id",
             lambda ctx:[{'id':p.id}]),

            CompiledSQL("DELETE FROM ball WHERE ball.id = :id",
             lambda ctx:[{'id': b.id},
                         {'id': b2.id},
                         {'id': b3.id},
                         {'id': b4.id}])
        )


class SelfReferentialPostUpdateTest(fixtures.MappedTest):
    """Post_update on a single self-referential mapper.


    """

    @classmethod
    def define_tables(cls, metadata):
        Table('node', metadata,
              Column('id', Integer, primary_key=True,
                     test_needs_autoincrement=True),
              Column('path', String(50), nullable=False),
              Column('parent_id', Integer,
                     ForeignKey('node.id'), nullable=True),
              Column('prev_sibling_id', Integer,
                     ForeignKey('node.id'), nullable=True),
              Column('next_sibling_id', Integer,
                     ForeignKey('node.id'), nullable=True))

    @classmethod
    def setup_classes(cls):
        class Node(cls.Basic):
            def __init__(self, path=''):
                self.path = path

    def test_one(self):
        """Post_update only fires off when needed.

        This test case used to produce many superfluous update statements,
        particularly upon delete

        """

        node, Node = self.tables.node, self.classes.Node


        mapper(Node, node, properties={
            'children': relationship(
                Node,
                primaryjoin=node.c.id==node.c.parent_id,
                cascade="all",
                backref=backref("parent", remote_side=node.c.id)
            ),
            'prev_sibling': relationship(
                Node,
                primaryjoin=node.c.prev_sibling_id==node.c.id,
                remote_side=node.c.id,
                uselist=False),
            'next_sibling': relationship(
                Node,
                primaryjoin=node.c.next_sibling_id==node.c.id,
                remote_side=node.c.id,
                uselist=False,
                post_update=True)})

        session = create_session()

        def append_child(parent, child):
            if parent.children:
                parent.children[-1].next_sibling = child
                child.prev_sibling = parent.children[-1]
            parent.children.append(child)

        def remove_child(parent, child):
            child.parent = None
            node = child.next_sibling
            node.prev_sibling = child.prev_sibling
            child.prev_sibling.next_sibling = node
            session.delete(child)
        root = Node('root')

        about = Node('about')
        cats = Node('cats')
        stories = Node('stories')
        bruce = Node('bruce')

        append_child(root, about)
        assert(about.prev_sibling is None)
        append_child(root, cats)
        assert(cats.prev_sibling is about)
        assert(cats.next_sibling is None)
        assert(about.next_sibling is cats)
        assert(about.prev_sibling is None)
        append_child(root, stories)
        append_child(root, bruce)
        session.add(root)
        session.flush()

        remove_child(root, cats)

        # pre-trigger lazy loader on 'cats' to make the test easier
        cats.children
        self.assert_sql_execution(
            testing.db,
            session.flush,
            AllOf(
            CompiledSQL("UPDATE node SET prev_sibling_id=:prev_sibling_id "
             "WHERE node.id = :node_id",
             lambda ctx:{'prev_sibling_id':about.id, 'node_id':stories.id}),

            CompiledSQL("UPDATE node SET next_sibling_id=:next_sibling_id "
             "WHERE node.id = :node_id",
             lambda ctx:{'next_sibling_id':stories.id, 'node_id':about.id}),

            CompiledSQL("UPDATE node SET next_sibling_id=:next_sibling_id "
             "WHERE node.id = :node_id",
             lambda ctx:{'next_sibling_id':None, 'node_id':cats.id}),
            ),

            CompiledSQL("DELETE FROM node WHERE node.id = :id",
             lambda ctx:[{'id':cats.id}])
        )

        session.delete(root)

        self.assert_sql_execution(
            testing.db,
            session.flush,
            CompiledSQL("UPDATE node SET next_sibling_id=:next_sibling_id "
                "WHERE node.id = :node_id",
                lambda ctx: [
                            {'node_id': about.id, 'next_sibling_id': None},
                            {'node_id': stories.id, 'next_sibling_id': None}
                        ]
            ),
            AllOf(
                CompiledSQL("DELETE FROM node WHERE node.id = :id",
                    lambda ctx:{'id':about.id}
                ),
                CompiledSQL("DELETE FROM node WHERE node.id = :id",
                    lambda ctx:{'id':stories.id}
                ),
                CompiledSQL("DELETE FROM node WHERE node.id = :id",
                    lambda ctx:{'id':bruce.id}
                ),
            ),
            CompiledSQL("DELETE FROM node WHERE node.id = :id",
                lambda ctx:{'id':root.id}
            ),
        )
        about = Node('about')
        cats = Node('cats')
        about.next_sibling = cats
        cats.prev_sibling = about
        session.add(about)
        session.flush()
        session.delete(about)
        cats.prev_sibling = None
        session.flush()

class SelfReferentialPostUpdateTest2(fixtures.MappedTest):

    @classmethod
    def define_tables(cls, metadata):
        Table("a_table", metadata,
              Column("id", Integer(), primary_key=True, test_needs_autoincrement=True),
              Column("fui", String(128)),
              Column("b", Integer(), ForeignKey("a_table.id")))

    @classmethod
    def setup_classes(cls):
        class A(cls.Basic):
            pass

    def test_one(self):
        """
        Test that post_update remembers to be involved in update operations as
        well, since it replaces the normal dependency processing completely
        [ticket:413]

        """

        A, a_table = self.classes.A, self.tables.a_table


        mapper(A, a_table, properties={
            'foo': relationship(A,
                            remote_side=[a_table.c.id],
                            post_update=True)})

        session = create_session()

        f1 = A(fui="f1")
        session.add(f1)
        session.flush()

        f2 = A(fui="f2", foo=f1)

        # at this point f1 is already inserted.  but we need post_update
        # to fire off anyway
        session.add(f2)
        session.flush()
        session.expunge_all()

        f1 = session.query(A).get(f1.id)
        f2 = session.query(A).get(f2.id)
        assert f2.foo is f1


class SelfReferentialPostUpdateTest3(fixtures.MappedTest):
    @classmethod
    def define_tables(cls, metadata):
        Table('parent', metadata,
              Column('id', Integer, primary_key=True,
                     test_needs_autoincrement=True),
              Column('name', String(50), nullable=False),
              Column('child_id', Integer,
                     ForeignKey('child.id', use_alter=True, name='c1'), nullable=True))

        Table('child', metadata,
           Column('id', Integer, primary_key=True,
                  test_needs_autoincrement=True),
           Column('name', String(50), nullable=False),
           Column('child_id', Integer,
                  ForeignKey('child.id')),
           Column('parent_id', Integer,
                  ForeignKey('parent.id'), nullable=True))

    @classmethod
    def setup_classes(cls):
        class Parent(cls.Basic):
            def __init__(self, name=''):
                self.name = name

        class Child(cls.Basic):
            def __init__(self, name=''):
                self.name = name

    def test_one(self):
        Child, Parent, parent, child = (self.classes.Child,
                                self.classes.Parent,
                                self.tables.parent,
                                self.tables.child)

        mapper(Parent, parent, properties={
            'children':relationship(Child, primaryjoin=parent.c.id==child.c.parent_id),
            'child':relationship(Child, primaryjoin=parent.c.child_id==child.c.id, post_update=True)
        })
        mapper(Child, child, properties={
            'parent':relationship(Child, remote_side=child.c.id)
        })

        session = create_session()
        p1 = Parent('p1')
        c1 = Child('c1')
        c2 = Child('c2')
        p1.children =[c1, c2]
        c2.parent = c1
        p1.child = c2

        session.add_all([p1, c1, c2])
        session.flush()

        p2 = Parent('p2')
        c3 = Child('c3')
        p2.children = [c3]
        p2.child = c3
        session.add(p2)

        session.delete(c2)
        p1.children.remove(c2)
        p1.child = None
        session.flush()

        p2.child = None
        session.flush()

class PostUpdateBatchingTest(fixtures.MappedTest):
    """test that lots of post update cols batch together into a single UPDATE."""

    @classmethod
    def define_tables(cls, metadata):
        Table('parent', metadata,
              Column('id', Integer, primary_key=True,
                     test_needs_autoincrement=True),
              Column('name', String(50), nullable=False),
              Column('c1_id', Integer,
                     ForeignKey('child1.id', use_alter=True, name='c1'), nullable=True),
              Column('c2_id', Integer,
                    ForeignKey('child2.id', use_alter=True, name='c2'), nullable=True),
              Column('c3_id', Integer,
                       ForeignKey('child3.id', use_alter=True, name='c3'), nullable=True)
            )

        Table('child1', metadata,
           Column('id', Integer, primary_key=True,
                  test_needs_autoincrement=True),
           Column('name', String(50), nullable=False),
           Column('parent_id', Integer,
                  ForeignKey('parent.id'), nullable=False))

        Table('child2', metadata,
             Column('id', Integer, primary_key=True,
                    test_needs_autoincrement=True),
             Column('name', String(50), nullable=False),
             Column('parent_id', Integer,
                    ForeignKey('parent.id'), nullable=False))

        Table('child3', metadata,
           Column('id', Integer, primary_key=True,
                  test_needs_autoincrement=True),
           Column('name', String(50), nullable=False),
           Column('parent_id', Integer,
                  ForeignKey('parent.id'), nullable=False))

    @classmethod
    def setup_classes(cls):
        class Parent(cls.Basic):
            def __init__(self, name=''):
                self.name = name
        class Child1(cls.Basic):
            def __init__(self, name=''):
                self.name = name
        class Child2(cls.Basic):
            def __init__(self, name=''):
                self.name = name
        class Child3(cls.Basic):
            def __init__(self, name=''):
                self.name = name

    def test_one(self):
        child1, child2, child3, Parent, parent, Child1, Child2, Child3 = (self.tables.child1,
                                self.tables.child2,
                                self.tables.child3,
                                self.classes.Parent,
                                self.tables.parent,
                                self.classes.Child1,
                                self.classes.Child2,
                                self.classes.Child3)

        mapper(Parent, parent, properties={
            'c1s':relationship(Child1, primaryjoin=child1.c.parent_id==parent.c.id),
            'c2s':relationship(Child2, primaryjoin=child2.c.parent_id==parent.c.id),
            'c3s':relationship(Child3, primaryjoin=child3.c.parent_id==parent.c.id),

            'c1':relationship(Child1, primaryjoin=child1.c.id==parent.c.c1_id, post_update=True),
            'c2':relationship(Child2, primaryjoin=child2.c.id==parent.c.c2_id, post_update=True),
            'c3':relationship(Child3, primaryjoin=child3.c.id==parent.c.c3_id, post_update=True),
        })
        mapper(Child1, child1)
        mapper(Child2, child2)
        mapper(Child3, child3)

        sess = create_session()

        p1 = Parent('p1')
        c11, c12, c13 = Child1('c1'), Child1('c2'), Child1('c3')
        c21, c22, c23 = Child2('c1'), Child2('c2'), Child2('c3')
        c31, c32, c33 = Child3('c1'), Child3('c2'), Child3('c3')

        p1.c1s = [c11, c12, c13]
        p1.c2s = [c21, c22, c23]
        p1.c3s = [c31, c32, c33]
        sess.add(p1)
        sess.flush()

        p1.c1 = c12
        p1.c2 = c23
        p1.c3 = c31

        self.assert_sql_execution(
            testing.db,
            sess.flush,
            CompiledSQL(
                "UPDATE parent SET c1_id=:c1_id, c2_id=:c2_id, "
                "c3_id=:c3_id WHERE parent.id = :parent_id",
                lambda ctx: {'c2_id': c23.id, 'parent_id': p1.id, 'c1_id': c12.id, 'c3_id': c31.id}
            )
        )

        p1.c1 = p1.c2 = p1.c3 = None

        self.assert_sql_execution(
            testing.db,
            sess.flush,
            CompiledSQL(
                "UPDATE parent SET c1_id=:c1_id, c2_id=:c2_id, "
                "c3_id=:c3_id WHERE parent.id = :parent_id",
                lambda ctx: {'c2_id': None, 'parent_id': p1.id, 'c1_id': None, 'c3_id': None}
            )
        )