from sqlalchemy.testing import eq_, assert_raises, assert_raises_message, is_
from sqlalchemy.testing import fixtures
from sqlalchemy import Integer, String, ForeignKey, or_, exc, \
    select, func, Boolean, case, text, column
from sqlalchemy.orm import mapper, relationship, backref, Session, \
    joinedload, synonym, query
from sqlalchemy import testing
from sqlalchemy.testing import mock

from sqlalchemy.testing.schema import Table, Column


class UpdateDeleteTest(fixtures.MappedTest):
    __backend__ = True

    @classmethod
    def define_tables(cls, metadata):
        Table('users', metadata,
              Column('id', Integer, primary_key=True,
                     test_needs_autoincrement=True),
              Column('name', String(32)),
              Column('age_int', Integer))
        Table(
            "addresses", metadata,
            Column('id', Integer, primary_key=True),
            Column('user_id', ForeignKey('users.id'))
        )

    @classmethod
    def setup_classes(cls):
        class User(cls.Comparable):
            pass

        class Address(cls.Comparable):
            pass

    @classmethod
    def insert_data(cls):
        users = cls.tables.users

        users.insert().execute([
            dict(id=1, name='john', age_int=25),
            dict(id=2, name='jack', age_int=47),
            dict(id=3, name='jill', age_int=29),
            dict(id=4, name='jane', age_int=37),
        ])

    @classmethod
    def setup_mappers(cls):
        User = cls.classes.User
        users = cls.tables.users

        Address = cls.classes.Address
        addresses = cls.tables.addresses

        mapper(User, users, properties={
            'age': users.c.age_int,
            'addresses': relationship(Address)
        })
        mapper(Address, addresses)

    def test_illegal_eval(self):
        User = self.classes.User
        s = Session()
        assert_raises_message(
            exc.ArgumentError,
            "Valid strategies for session synchronization "
            "are 'evaluate', 'fetch', False",
            s.query(User).update,
            {},
            synchronize_session="fake"
        )

    def test_illegal_operations(self):
        User = self.classes.User
        Address = self.classes.Address

        s = Session()

        for q, mname in (
            (s.query(User).limit(2), r"limit\(\)"),
            (s.query(User).offset(2), r"offset\(\)"),
            (s.query(User).limit(2).offset(2), r"limit\(\)"),
            (s.query(User).order_by(User.id), r"order_by\(\)"),
            (s.query(User).group_by(User.id), r"group_by\(\)"),
            (s.query(User).distinct(), r"distinct\(\)"),
            (s.query(User).join(User.addresses),
                r"join\(\), outerjoin\(\), select_from\(\), or from_self\(\)"),
            (s.query(User).outerjoin(User.addresses),
                r"join\(\), outerjoin\(\), select_from\(\), or from_self\(\)"),
            (s.query(User).select_from(Address),
                r"join\(\), outerjoin\(\), select_from\(\), or from_self\(\)"),
            (s.query(User).from_self(),
                r"join\(\), outerjoin\(\), select_from\(\), or from_self\(\)"),
        ):
            assert_raises_message(
                exc.InvalidRequestError,
                r"Can't call Query.update\(\) or Query.delete\(\) when "
                "%s has been called" % mname,
                q.update,
                {'name': 'ed'})
            assert_raises_message(
                exc.InvalidRequestError,
                r"Can't call Query.update\(\) or Query.delete\(\) when "
                "%s has been called" % mname,
                q.delete)

    def test_evaluate_clauseelement(self):
        User = self.classes.User

        class Thing(object):
            def __clause_element__(self):
                return User.name.__clause_element__()

        s = Session()
        jill = s.query(User).get(3)
        s.query(User).update(
            {Thing(): 'moonbeam'},
            synchronize_session='evaluate')
        eq_(jill.name, 'moonbeam')

    def test_evaluate_invalid(self):
        User = self.classes.User

        class Thing(object):
            def __clause_element__(self):
                return 5

        s = Session()

        assert_raises_message(
            exc.InvalidRequestError,
            "Invalid expression type: 5",
            s.query(User).update, {Thing(): 'moonbeam'},
            synchronize_session='evaluate'
        )

    def test_evaluate_unmapped_col(self):
        User = self.classes.User

        s = Session()
        jill = s.query(User).get(3)
        s.query(User).update(
            {column('name'): 'moonbeam'},
            synchronize_session='evaluate')
        eq_(jill.name, 'jill')
        s.expire(jill)
        eq_(jill.name, 'moonbeam')

    def test_evaluate_synonym_string(self):
        class Foo(object):
            pass
        mapper(Foo, self.tables.users, properties={
            'uname': synonym("name", )
        })

        s = Session()
        jill = s.query(Foo).get(3)
        s.query(Foo).update(
            {'uname': 'moonbeam'},
            synchronize_session='evaluate')
        eq_(jill.uname, 'moonbeam')

    def test_evaluate_synonym_attr(self):
        class Foo(object):
            pass
        mapper(Foo, self.tables.users, properties={
            'uname': synonym("name", )
        })

        s = Session()
        jill = s.query(Foo).get(3)
        s.query(Foo).update(
            {Foo.uname: 'moonbeam'},
            synchronize_session='evaluate')
        eq_(jill.uname, 'moonbeam')

    def test_evaluate_double_synonym_attr(self):
        class Foo(object):
            pass
        mapper(Foo, self.tables.users, properties={
            'uname': synonym("name"),
            'ufoo': synonym('uname')
        })

        s = Session()
        jill = s.query(Foo).get(3)
        s.query(Foo).update(
            {Foo.ufoo: 'moonbeam'},
            synchronize_session='evaluate')
        eq_(jill.ufoo, 'moonbeam')

    def test_evaluate_hybrid_attr(self):
        from sqlalchemy.ext.hybrid import hybrid_property

        class Foo(object):
            @hybrid_property
            def uname(self):
                return self.name

        mapper(Foo, self.tables.users)

        s = Session()
        jill = s.query(Foo).get(3)
        s.query(Foo).update(
            {Foo.uname: 'moonbeam'},
            synchronize_session='evaluate')
        eq_(jill.uname, 'moonbeam')

    def test_delete(self):
        User = self.classes.User

        sess = Session()

        john, jack, jill, jane = sess.query(User).order_by(User.id).all()
        sess.query(User).filter(
            or_(User.name == 'john', User.name == 'jill')).delete()

        assert john not in sess and jill not in sess

        eq_(sess.query(User).order_by(User.id).all(), [jack, jane])

    def test_delete_against_metadata(self):
        User = self.classes.User
        users = self.tables.users

        sess = Session()
        sess.query(users).delete(synchronize_session=False)
        eq_(sess.query(User).count(), 0)

    def test_delete_with_bindparams(self):
        User = self.classes.User

        sess = Session()

        john, jack, jill, jane = sess.query(User).order_by(User.id).all()
        sess.query(User).filter(text('name = :name')).params(
            name='john').delete('fetch')
        assert john not in sess

        eq_(sess.query(User).order_by(User.id).all(), [jack, jill, jane])

    def test_delete_rollback(self):
        User = self.classes.User

        sess = Session()
        john, jack, jill, jane = sess.query(User).order_by(User.id).all()
        sess.query(User).filter(
            or_(User.name == 'john', User.name == 'jill')).\
            delete(synchronize_session='evaluate')
        assert john not in sess and jill not in sess
        sess.rollback()
        assert john in sess and jill in sess

    def test_delete_rollback_with_fetch(self):
        User = self.classes.User

        sess = Session()
        john, jack, jill, jane = sess.query(User).order_by(User.id).all()
        sess.query(User).filter(
            or_(User.name == 'john', User.name == 'jill')).\
            delete(synchronize_session='fetch')
        assert john not in sess and jill not in sess
        sess.rollback()
        assert john in sess and jill in sess

    def test_delete_without_session_sync(self):
        User = self.classes.User

        sess = Session()

        john, jack, jill, jane = sess.query(User).order_by(User.id).all()
        sess.query(User).filter(
            or_(User.name == 'john', User.name == 'jill')).\
            delete(synchronize_session=False)

        assert john in sess and jill in sess

        eq_(sess.query(User).order_by(User.id).all(), [jack, jane])

    def test_delete_with_fetch_strategy(self):
        User = self.classes.User

        sess = Session()

        john, jack, jill, jane = sess.query(User).order_by(User.id).all()
        sess.query(User).filter(
            or_(User.name == 'john', User.name == 'jill')).\
            delete(synchronize_session='fetch')

        assert john not in sess and jill not in sess

        eq_(sess.query(User).order_by(User.id).all(), [jack, jane])

    @testing.fails_on('mysql', 'FIXME: unknown')
    def test_delete_invalid_evaluation(self):
        User = self.classes.User

        sess = Session()

        john, jack, jill, jane = sess.query(User).order_by(User.id).all()

        assert_raises(exc.InvalidRequestError,
                      sess.query(User).
                      filter(
                          User.name == select([func.max(User.name)])).delete,
                      synchronize_session='evaluate'
                      )

        sess.query(User).filter(User.name == select([func.max(User.name)])).\
            delete(synchronize_session='fetch')

        assert john not in sess

        eq_(sess.query(User).order_by(User.id).all(), [jack, jill, jane])

    def test_update(self):
        User, users = self.classes.User, self.tables.users

        sess = Session()

        john, jack, jill, jane = sess.query(User).order_by(User.id).all()
        sess.query(User).filter(User.age > 29).\
            update({'age': User.age - 10}, synchronize_session='evaluate')

        eq_([john.age, jack.age, jill.age, jane.age], [25, 37, 29, 27])
        eq_(sess.query(User.age).order_by(
            User.id).all(), list(zip([25, 37, 29, 27])))

        sess.query(User).filter(User.age > 29).\
            update({User.age: User.age - 10}, synchronize_session='evaluate')
        eq_([john.age, jack.age, jill.age, jane.age], [25, 27, 29, 27])
        eq_(sess.query(User.age).order_by(
            User.id).all(), list(zip([25, 27, 29, 27])))

        sess.query(User).filter(User.age > 27).\
            update(
                {users.c.age_int: User.age - 10},
                synchronize_session='evaluate')
        eq_([john.age, jack.age, jill.age, jane.age], [25, 27, 19, 27])
        eq_(sess.query(User.age).order_by(
            User.id).all(), list(zip([25, 27, 19, 27])))

        sess.query(User).filter(User.age == 25).\
            update({User.age: User.age - 10}, synchronize_session='fetch')
        eq_([john.age, jack.age, jill.age, jane.age], [15, 27, 19, 27])
        eq_(sess.query(User.age).order_by(
            User.id).all(), list(zip([15, 27, 19, 27])))

    def test_update_against_table_col(self):
        User, users = self.classes.User, self.tables.users

        sess = Session()
        john, jack, jill, jane = sess.query(User).order_by(User.id).all()
        eq_([john.age, jack.age, jill.age, jane.age], [25, 47, 29, 37])
        sess.query(User).filter(User.age > 27).\
            update(
                {users.c.age_int: User.age - 10},
                synchronize_session='evaluate')
        eq_([john.age, jack.age, jill.age, jane.age], [25, 37, 19, 27])

    def test_update_against_metadata(self):
        User, users = self.classes.User, self.tables.users

        sess = Session()

        sess.query(users).update(
            {users.c.age_int: 29}, synchronize_session=False)
        eq_(sess.query(User.age).order_by(
            User.id).all(), list(zip([29, 29, 29, 29])))

    def test_update_with_bindparams(self):
        User = self.classes.User

        sess = Session()

        john, jack, jill, jane = sess.query(User).order_by(User.id).all()

        sess.query(User).filter(text('age_int > :x')).params(x=29).\
            update({'age': User.age - 10}, synchronize_session='fetch')

        eq_([john.age, jack.age, jill.age, jane.age], [25, 37, 29, 27])
        eq_(sess.query(User.age).order_by(
            User.id).all(), list(zip([25, 37, 29, 27])))

    def test_update_without_load(self):
        User = self.classes.User

        sess = Session()

        sess.query(User).filter(User.id == 3).\
            update({'age': 44}, synchronize_session='fetch')
        eq_(sess.query(User.age).order_by(
            User.id).all(), list(zip([25, 47, 44, 37])))

    def test_update_changes_resets_dirty(self):
        User = self.classes.User

        sess = Session(autoflush=False)

        john, jack, jill, jane = sess.query(User).order_by(User.id).all()

        john.age = 50
        jack.age = 37

        # autoflush is false.  therefore our '50' and '37' are getting
        # blown away by this operation.

        sess.query(User).filter(User.age > 29).\
            update({'age': User.age - 10}, synchronize_session='evaluate')

        for x in (john, jack, jill, jane):
            assert not sess.is_modified(x)

        eq_([john.age, jack.age, jill.age, jane.age], [25, 37, 29, 27])

        john.age = 25
        assert john in sess.dirty
        assert jack in sess.dirty
        assert jill not in sess.dirty
        assert not sess.is_modified(john)
        assert not sess.is_modified(jack)

    def test_update_changes_with_autoflush(self):
        User = self.classes.User

        sess = Session()

        john, jack, jill, jane = sess.query(User).order_by(User.id).all()

        john.age = 50
        jack.age = 37

        sess.query(User).filter(User.age > 29).\
            update({'age': User.age - 10}, synchronize_session='evaluate')

        for x in (john, jack, jill, jane):
            assert not sess.is_modified(x)

        eq_([john.age, jack.age, jill.age, jane.age], [40, 27, 29, 27])

        john.age = 25
        assert john in sess.dirty
        assert jack not in sess.dirty
        assert jill not in sess.dirty
        assert sess.is_modified(john)
        assert not sess.is_modified(jack)

    def test_update_with_expire_strategy(self):
        User = self.classes.User

        sess = Session()

        john, jack, jill, jane = sess.query(User).order_by(User.id).all()
        sess.query(User).filter(User.age > 29).\
            update({'age': User.age - 10}, synchronize_session='fetch')

        eq_([john.age, jack.age, jill.age, jane.age], [25, 37, 29, 27])
        eq_(sess.query(User.age).order_by(
            User.id).all(), list(zip([25, 37, 29, 27])))

    @testing.fails_if(lambda: not testing.db.dialect.supports_sane_rowcount)
    def test_update_returns_rowcount(self):
        User = self.classes.User

        sess = Session()

        rowcount = sess.query(User).filter(
            User.age > 29).update({'age': User.age + 0})
        eq_(rowcount, 2)

        rowcount = sess.query(User).filter(
            User.age > 29).update({'age': User.age - 10})
        eq_(rowcount, 2)

    @testing.fails_if(lambda: not testing.db.dialect.supports_sane_rowcount)
    def test_delete_returns_rowcount(self):
        User = self.classes.User

        sess = Session()

        rowcount = sess.query(User).filter(User.age > 26).\
            delete(synchronize_session=False)
        eq_(rowcount, 3)

    def test_update_all(self):
        User = self.classes.User

        sess = Session()

        john, jack, jill, jane = sess.query(User).order_by(User.id).all()
        sess.query(User).update({'age': 42}, synchronize_session='evaluate')

        eq_([john.age, jack.age, jill.age, jane.age], [42, 42, 42, 42])
        eq_(sess.query(User.age).order_by(
            User.id).all(), list(zip([42, 42, 42, 42])))

    def test_delete_all(self):
        User = self.classes.User

        sess = Session()

        john, jack, jill, jane = sess.query(User).order_by(User.id).all()
        sess.query(User).delete(synchronize_session='evaluate')

        assert not (
            john in sess or jack in sess or jill in sess or jane in sess)
        eq_(sess.query(User).count(), 0)

    def test_autoflush_before_evaluate_update(self):
        User = self.classes.User

        sess = Session()
        john = sess.query(User).filter_by(name='john').one()
        john.name = 'j2'

        sess.query(User).filter_by(name='j2').\
            update({'age': 42},
                   synchronize_session='evaluate')
        eq_(john.age, 42)

    def test_autoflush_before_fetch_update(self):
        User = self.classes.User

        sess = Session()
        john = sess.query(User).filter_by(name='john').one()
        john.name = 'j2'

        sess.query(User).filter_by(name='j2').\
            update({'age': 42},
                   synchronize_session='fetch')
        eq_(john.age, 42)

    def test_autoflush_before_evaluate_delete(self):
        User = self.classes.User

        sess = Session()
        john = sess.query(User).filter_by(name='john').one()
        john.name = 'j2'

        sess.query(User).filter_by(name='j2').\
            delete(
                synchronize_session='evaluate')
        assert john not in sess

    def test_autoflush_before_fetch_delete(self):
        User = self.classes.User

        sess = Session()
        john = sess.query(User).filter_by(name='john').one()
        john.name = 'j2'

        sess.query(User).filter_by(name='j2').\
            delete(
                synchronize_session='fetch')
        assert john not in sess

    def test_evaluate_before_update(self):
        User = self.classes.User

        sess = Session()
        john = sess.query(User).filter_by(name='john').one()
        sess.expire(john, ['age'])

        # eval must be before the update.  otherwise
        # we eval john, age has been expired and doesn't
        # match the new value coming in
        sess.query(User).filter_by(name='john').filter_by(age=25).\
            update({'name': 'j2', 'age': 40},
                   synchronize_session='evaluate')
        eq_(john.name, 'j2')
        eq_(john.age, 40)

    def test_fetch_before_update(self):
        User = self.classes.User

        sess = Session()
        john = sess.query(User).filter_by(name='john').one()
        sess.expire(john, ['age'])

        sess.query(User).filter_by(name='john').filter_by(age=25).\
            update({'name': 'j2', 'age': 40},
                   synchronize_session='fetch')
        eq_(john.name, 'j2')
        eq_(john.age, 40)

    def test_evaluate_before_delete(self):
        User = self.classes.User

        sess = Session()
        john = sess.query(User).filter_by(name='john').one()
        sess.expire(john, ['age'])

        sess.query(User).filter_by(name='john').\
            filter_by(age=25).\
            delete(
                synchronize_session='evaluate')
        assert john not in sess

    def test_fetch_before_delete(self):
        User = self.classes.User

        sess = Session()
        john = sess.query(User).filter_by(name='john').one()
        sess.expire(john, ['age'])

        sess.query(User).filter_by(name='john').\
            filter_by(age=25).\
            delete(
                synchronize_session='fetch')
        assert john not in sess

    def test_update_unordered_dict(self):
        User = self.classes.User
        session = Session()

        # Do an update using unordered dict and check that the parameters used
        # are ordered in table order
        with mock.patch.object(session, "execute") as exec_:
            session.query(User).filter(User.id == 15).update(
                {'name': 'foob', 'id': 123})
            # Confirm that parameters are a dict instead of tuple or list
            params_type = type(exec_.mock_calls[0][1][0].parameters)
            is_(params_type, dict)

    def test_update_preserve_parameter_order(self):
        User = self.classes.User
        session = Session()

        # Do update using a tuple and check that order is preserved
        with mock.patch.object(session, "execute") as exec_:
            session.query(User).filter(User.id == 15).update(
                (('id', 123), ('name', 'foob')),
                update_args={"preserve_parameter_order": True})
            cols = [c.key
                    for c in exec_.mock_calls[0][1][0]._parameter_ordering]
            eq_(['id', 'name'], cols)

        # Now invert the order and use a list instead, and check that order is
        # also preserved
        with mock.patch.object(session, "execute") as exec_:
            session.query(User).filter(User.id == 15).update(
                [('name', 'foob'), ('id', 123)],
                update_args={"preserve_parameter_order": True})
            cols = [c.key
                    for c in exec_.mock_calls[0][1][0]._parameter_ordering]
            eq_(['name', 'id'], cols)


class UpdateDeleteIgnoresLoadersTest(fixtures.MappedTest):

    @classmethod
    def define_tables(cls, metadata):
        Table('users', metadata,
              Column('id', Integer, primary_key=True,
                     test_needs_autoincrement=True),
              Column('name', String(32)),
              Column('age', Integer))

        Table('documents', metadata,
              Column('id', Integer, primary_key=True,
                     test_needs_autoincrement=True),
              Column('user_id', None, ForeignKey('users.id')),
              Column('title', String(32)))

    @classmethod
    def setup_classes(cls):
        class User(cls.Comparable):
            pass

        class Document(cls.Comparable):
            pass

    @classmethod
    def insert_data(cls):
        users = cls.tables.users

        users.insert().execute([
            dict(id=1, name='john', age=25),
            dict(id=2, name='jack', age=47),
            dict(id=3, name='jill', age=29),
            dict(id=4, name='jane', age=37),
        ])

        documents = cls.tables.documents

        documents.insert().execute([
            dict(id=1, user_id=1, title='foo'),
            dict(id=2, user_id=1, title='bar'),
            dict(id=3, user_id=2, title='baz'),
        ])

    @classmethod
    def setup_mappers(cls):
        documents, Document, User, users = (cls.tables.documents,
                                            cls.classes.Document,
                                            cls.classes.User,
                                            cls.tables.users)

        mapper(User, users)
        mapper(Document, documents, properties={
            'user': relationship(User, lazy='joined',
                                 backref=backref('documents', lazy='select'))
        })

    def test_update_with_eager_relationships(self):
        Document = self.classes.Document

        sess = Session()

        foo, bar, baz = sess.query(Document).order_by(Document.id).all()
        sess.query(Document).filter(Document.user_id == 1).\
            update({'title': Document.title + Document.title},
                   synchronize_session='fetch')

        eq_([foo.title, bar.title, baz.title], ['foofoo', 'barbar', 'baz'])
        eq_(sess.query(Document.title).order_by(Document.id).all(),
            list(zip(['foofoo', 'barbar', 'baz'])))

    def test_update_with_explicit_joinedload(self):
        User = self.classes.User

        sess = Session()

        john, jack, jill, jane = sess.query(User).order_by(User.id).all()
        sess.query(User).options(
            joinedload(User.documents)).filter(User.age > 29).\
            update({'age': User.age - 10}, synchronize_session='fetch')

        eq_([john.age, jack.age, jill.age, jane.age], [25, 37, 29, 27])
        eq_(sess.query(User.age).order_by(
            User.id).all(), list(zip([25, 37, 29, 27])))

    def test_delete_with_eager_relationships(self):
        Document = self.classes.Document

        sess = Session()

        sess.query(Document).filter(Document.user_id == 1).\
            delete(synchronize_session=False)

        eq_(sess.query(Document.title).all(), list(zip(['baz'])))


class UpdateDeleteFromTest(fixtures.MappedTest):

    @classmethod
    def define_tables(cls, metadata):
        Table('users', metadata,
              Column('id', Integer, primary_key=True),
              Column('samename', String(10)),
              )
        Table('documents', metadata,
              Column('id', Integer, primary_key=True),
              Column('user_id', None, ForeignKey('users.id')),
              Column('title', String(32)),
              Column('flag', Boolean),
              Column('samename', String(10)),
              )

    @classmethod
    def setup_classes(cls):
        class User(cls.Comparable):
            pass

        class Document(cls.Comparable):
            pass

    @classmethod
    def insert_data(cls):
        users = cls.tables.users

        users.insert().execute([
            dict(id=1, ),
            dict(id=2, ),
            dict(id=3, ),
            dict(id=4, ),
        ])

        documents = cls.tables.documents

        documents.insert().execute([
            dict(id=1, user_id=1, title='foo'),
            dict(id=2, user_id=1, title='bar'),
            dict(id=3, user_id=2, title='baz'),
            dict(id=4, user_id=2, title='hoho'),
            dict(id=5, user_id=3, title='lala'),
            dict(id=6, user_id=3, title='bleh'),
        ])

    @classmethod
    def setup_mappers(cls):
        documents, Document, User, users = (cls.tables.documents,
                                            cls.classes.Document,
                                            cls.classes.User,
                                            cls.tables.users)

        mapper(User, users)
        mapper(Document, documents, properties={
            'user': relationship(User, backref='documents')
        })

    @testing.requires.update_from
    def test_update_from_joined_subq_test(self):
        Document = self.classes.Document
        s = Session()

        subq = s.query(func.max(Document.title).label('title')).\
            group_by(Document.user_id).subquery()

        s.query(Document).filter(Document.title == subq.c.title).\
            update({'flag': True}, synchronize_session=False)

        eq_(
            set(s.query(Document.id, Document.flag)),
            set([
                (1, True), (2, None),
                (3, None), (4, True),
                (5, True), (6, None)])
        )

    def test_no_eval_against_multi_table_criteria(self):
        User = self.classes.User
        Document = self.classes.Document

        s = Session()

        q = s.query(User).filter(User.id == Document.user_id)
        assert_raises_message(
            exc.InvalidRequestError,
            "Could not evaluate current criteria in Python.",
            q.update,
            {"name": "ed"}
        )

    @testing.requires.update_where_target_in_subquery
    def test_update_using_in(self):
        Document = self.classes.Document
        s = Session()

        subq = s.query(func.max(Document.title).label('title')).\
            group_by(Document.user_id).subquery()

        s.query(Document).filter(Document.title.in_(subq)).\
            update({'flag': True}, synchronize_session=False)

        eq_(
            set(s.query(Document.id, Document.flag)),
            set([
                (1, True), (2, None),
                (3, None), (4, True),
                (5, True), (6, None)])
        )

    @testing.requires.update_where_target_in_subquery
    @testing.requires.standalone_binds
    def test_update_using_case(self):
        Document = self.classes.Document
        s = Session()

        subq = s.query(func.max(Document.title).label('title')).\
            group_by(Document.user_id).subquery()

        # this would work with Firebird if you do literal_column('1')
        # instead
        case_stmt = case([(Document.title.in_(subq), True)], else_=False)
        s.query(Document).update(
            {'flag': case_stmt}, synchronize_session=False)

        eq_(
            set(s.query(Document.id, Document.flag)),
            set([
                (1, True), (2, False),
                (3, False), (4, True),
                (5, True), (6, False)])
        )

    @testing.only_on('mysql', 'Multi table update')
    def test_update_from_multitable_same_names(self):
        Document = self.classes.Document
        User = self.classes.User

        s = Session()

        s.query(Document).\
            filter(User.id == Document.user_id).\
            filter(User.id == 2).update({
                Document.samename: 'd_samename',
                User.samename: 'u_samename'
            }, synchronize_session=False)
        eq_(
            s.query(User.id, Document.samename, User.samename).
            filter(User.id == Document.user_id).
            order_by(User.id).all(),
            [
                (1, None, None),
                (1, None, None),
                (2, 'd_samename', 'u_samename'),
                (2, 'd_samename', 'u_samename'),
                (3, None, None),
                (3, None, None),
            ]
        )


class ExpressionUpdateTest(fixtures.MappedTest):

    @classmethod
    def define_tables(cls, metadata):
        data = Table('data', metadata,
                     Column('id', Integer, primary_key=True,
                            test_needs_autoincrement=True),
                     Column('counter', Integer, nullable=False, default=0)
                     )

    @classmethod
    def setup_classes(cls):
        class Data(cls.Comparable):
            pass

    @classmethod
    def setup_mappers(cls):
        data = cls.tables.data
        mapper(cls.classes.Data, data, properties={'cnt': data.c.counter})

    @testing.provide_metadata
    def test_update_attr_names(self):
        Data = self.classes.Data

        d1 = Data()
        sess = Session()
        sess.add(d1)
        sess.commit()
        eq_(d1.cnt, 0)

        sess.query(Data).update({Data.cnt: Data.cnt + 1})
        sess.flush()

        eq_(d1.cnt, 1)

        sess.query(Data).update({Data.cnt: Data.cnt + 1}, 'fetch')
        sess.flush()

        eq_(d1.cnt, 2)
        sess.close()

    def test_update_args(self):
        Data = self.classes.Data
        session = testing.mock.Mock(wraps=Session())
        update_args = {"mysql_limit": 1}
        query.Query(Data, session).update({Data.cnt: Data.cnt + 1},
                                          update_args=update_args)
        eq_(session.execute.call_count, 1)
        args, kwargs = session.execute.call_args
        eq_(len(args), 1)
        update_stmt = args[0]
        eq_(update_stmt.dialect_kwargs, update_args)


class InheritTest(fixtures.DeclarativeMappedTest):

    run_inserts = 'each'

    run_deletes = 'each'
    __backend__ = True

    @classmethod
    def setup_classes(cls):
        Base = cls.DeclarativeBasic

        class Person(Base):
            __tablename__ = 'person'
            id = Column(
                Integer, primary_key=True, test_needs_autoincrement=True)
            type = Column(String(50))
            name = Column(String(50))

        class Engineer(Person):
            __tablename__ = 'engineer'
            id = Column(Integer, ForeignKey('person.id'), primary_key=True)
            engineer_name = Column(String(50))

        class Manager(Person):
            __tablename__ = 'manager'
            id = Column(Integer, ForeignKey('person.id'), primary_key=True)
            manager_name = Column(String(50))

    @classmethod
    def insert_data(cls):
        Engineer, Person, Manager = cls.classes.Engineer, \
            cls.classes.Person, cls.classes.Manager
        s = Session(testing.db)
        s.add_all([
            Engineer(name='e1', engineer_name='e1'),
            Manager(name='m1', manager_name='m1'),
            Engineer(name='e2', engineer_name='e2'),
            Person(name='p1'),
        ])
        s.commit()

    def test_illegal_metadata(self):
        person = self.classes.Person.__table__
        engineer = self.classes.Engineer.__table__

        sess = Session()
        assert_raises_message(
            exc.InvalidRequestError,
            "This operation requires only one Table or entity be "
            "specified as the target.",
            sess.query(person.join(engineer)).update, {}
        )

    def test_update_subtable_only(self):
        Engineer = self.classes.Engineer
        s = Session(testing.db)
        s.query(Engineer).update({'engineer_name': 'e5'})

        eq_(
            s.query(Engineer.engineer_name).all(),
            [('e5', ), ('e5', )]
        )

    @testing.requires.update_from
    def test_update_from(self):
        Engineer = self.classes.Engineer
        Person = self.classes.Person
        s = Session(testing.db)
        s.query(Engineer).filter(Engineer.id == Person.id).\
            filter(Person.name == 'e2').update({'engineer_name': 'e5'})

        eq_(
            set(s.query(Person.name, Engineer.engineer_name)),
            set([('e1', 'e1', ), ('e2', 'e5')])
        )

    @testing.only_on('mysql', 'Multi table update')
    def test_update_from_multitable(self):
        Engineer = self.classes.Engineer
        Person = self.classes.Person
        s = Session(testing.db)
        s.query(Engineer).filter(Engineer.id == Person.id).\
            filter(Person.name == 'e2').update({Person.name: 'e22',
                                                Engineer.engineer_name: 'e55'})

        eq_(
            set(s.query(Person.name, Engineer.engineer_name)),
            set([('e1', 'e1', ), ('e22', 'e55')])
        )