def test_autoflush_expressions(self):
        """test that an expression which is dependent on object state is
        evaluated after the session autoflushes.   This is the lambda
        inside of strategies.py lazy_clause.

        """

        users, Address, addresses, User = (self.tables.users,
                                self.classes.Address,
                                self.tables.addresses,
                                self.classes.User)

        mapper(User, users, properties={
            'addresses': relationship(Address, backref="user")})
        mapper(Address, addresses)

        sess = create_session(autoflush=True, autocommit=False)
        u = User(name='ed', addresses=[Address(email_address='foo')])
        sess.add(u)
        eq_(sess.query(Address).filter(Address.user == u).one(),
            Address(email_address='foo'))

        # still works after "u" is garbage collected
        sess.commit()
        sess.close()
        u = sess.query(User).get(u.id)
        q = sess.query(Address).filter(Address.user == u)
        del u
        gc_collect()
        eq_(q.one(), Address(email_address='foo'))

    def test_defaults_second_table(self):
        users, addresses = self.tables.users, self.tables.addresses

        values = {
            addresses.c.email_address: users.c.name,
            users.c.name: 'ed2'
        }

        ret = testing.db.execute(
            addresses.update().
                values(values).
                where(users.c.id == addresses.c.user_id).
                where(users.c.name == 'ed'))

        eq_(set(ret.prefetch_cols()), set([users.c.some_update]))

        expected = [
            (2, 8, 'ed'),
            (3, 8, 'ed'),
            (4, 9, '[email protected]')]
        self._assert_addresses(addresses, expected)

        expected = [
            (8, 'ed2', 'im the update'),
            (9, 'fred', 'value')]
        self._assert_users(users, expected)

    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_attached_as_schema(self):
        cx = testing.db.connect()
        try:
            cx.execute('ATTACH DATABASE ":memory:" AS  test_schema')
            dialect = cx.dialect
            assert dialect.get_table_names(cx, 'test_schema') == []
            meta = MetaData(cx)
            Table('created', meta, Column('id', Integer),
                  schema='test_schema')
            alt_master = Table('sqlite_master', meta, autoload=True,
                               schema='test_schema')
            meta.create_all(cx)
            eq_(dialect.get_table_names(cx, 'test_schema'), ['created'])
            assert len(alt_master.c) > 0
            meta.clear()
            reflected = Table('created', meta, autoload=True,
                              schema='test_schema')
            assert len(reflected.c) == 1
            cx.execute(reflected.insert(), dict(id=1))
            r = cx.execute(reflected.select()).fetchall()
            assert list(r) == [(1, )]
            cx.execute(reflected.update(), dict(id=2))
            r = cx.execute(reflected.select()).fetchall()
            assert list(r) == [(2, )]
            cx.execute(reflected.delete(reflected.c.id == 2))
            r = cx.execute(reflected.select()).fetchall()
            assert list(r) == []

            # note that sqlite_master is cleared, above

            meta.drop_all()
            assert dialect.get_table_names(cx, 'test_schema') == []
        finally:
            cx.execute('DETACH DATABASE test_schema')

    def test_boolean(self):
        "Exercise type specification for boolean type."

        columns = [
            # column type, args, kwargs, expected ddl
            (Boolean, [], {},
             'BIT'),
        ]

        metadata = MetaData()
        table_args = ['test_mssql_boolean', metadata]
        for index, spec in enumerate(columns):
            type_, args, kw, res = spec
            table_args.append(
                Column('c%s' % index, type_(*args, **kw), nullable=None))

        boolean_table = Table(*table_args)
        dialect = mssql.dialect()
        gen = dialect.ddl_compiler(dialect, schema.CreateTable(boolean_table))

        for col in boolean_table.c:
            index = int(col.name[1:])
            testing.eq_(
                gen.get_column_specification(col),
                "%s %s" % (col.name, columns[index][3]))
            self.assert_(repr(col))

 def test_with_transaction(self):
     conn = self.engine.connect()
     trans = conn.begin()
     eq_(conn.execute(select([1])).scalar(), 1)
     assert not conn.closed
     self.engine.test_shutdown()
     _assert_invalidated(conn.execute, select([1]))
     assert not conn.closed
     assert conn.invalidated
     assert trans.is_active
     assert_raises_message(
         tsa.exc.StatementError,
         "Can't reconnect until invalid transaction is rolled back",
         conn.execute, select([1]))
     assert trans.is_active
     assert_raises_message(
         tsa.exc.InvalidRequestError,
         "Can't reconnect until invalid transaction is rolled back",
         trans.commit
     )
     assert trans.is_active
     trans.rollback()
     assert not trans.is_active
     assert conn.invalidated
     eq_(conn.execute(select([1])).scalar(), 1)
     assert not conn.invalidated

    def _test_round_trip(self, fixture, warnings=False):
        from sqlalchemy import inspect
        conn = testing.db.connect()
        for from_, to_ in self._fixture_as_string(fixture):
            inspector = inspect(conn)
            conn.execute("CREATE TABLE foo (data %s)" % from_)
            try:
                if warnings:
                    def go():
                        return inspector.get_columns("foo")[0]
                    col_info = testing.assert_warnings(go,
                                    ["Could not instantiate"], regex=True)
                else:
                    col_info = inspector.get_columns("foo")[0]
                expected_type = type(to_)
                is_(type(col_info['type']), expected_type)

                # test args
                for attr in ("scale", "precision", "length"):
                    if getattr(to_, attr, None) is not None:
                        eq_(
                            getattr(col_info['type'], attr),
                            getattr(to_, attr, None)
                        )
            finally:
                conn.execute("DROP TABLE foo")

    def test_selective_relationships(self):
        sub, base_mtom, Related, Base, related, sub_mtom, base, Sub = (self.tables.sub,
                                self.tables.base_mtom,
                                self.classes.Related,
                                self.classes.Base,
                                self.tables.related,
                                self.tables.sub_mtom,
                                self.tables.base,
                                self.classes.Sub)

        mapper(Base, base, properties={'related': relationship(Related,
               secondary=base_mtom, backref='bases',
               order_by=related.c.id)})
        mapper(Sub, sub, inherits=Base, concrete=True,
               properties={'related': relationship(Related,
               secondary=sub_mtom, backref='subs',
               order_by=related.c.id)})
        mapper(Related, related)
        sess = sessionmaker()()
        b1, s1, r1, r2, r3 = Base(), Sub(), Related(), Related(), \
            Related()
        b1.related.append(r1)
        b1.related.append(r2)
        s1.related.append(r2)
        s1.related.append(r3)
        sess.add_all([b1, s1])
        sess.commit()
        eq_(s1.related, [r2, r3])
        eq_(b1.related, [r1, r2])

 def test_time_result_processor(self):
     eq_(
         mysql.TIME().result_processor(None, None)(
             datetime.timedelta(seconds=35, minutes=517,
                                microseconds=450)),
         datetime.time(8, 37, 35, 450)
     )

    def test_set_composite_attrs_via_selectable(self):
        Values, CustomValues, values, Descriptions, descriptions = (self.classes.Values,
                                self.classes.CustomValues,
                                self.tables.values,
                                self.classes.Descriptions,
                                self.tables.descriptions)

        session = Session()
        d = Descriptions(
            custom_descriptions = CustomValues('Color', 'Number'),
            values =[
                Values(custom_values = CustomValues('Red', '5')),
                Values(custom_values=CustomValues('Blue', '1'))
            ]
        )

        session.add(d)
        session.commit()
        eq_(
            testing.db.execute(descriptions.select()).fetchall(),
            [(1, 'Color', 'Number')]
        )
        eq_(
            testing.db.execute(values.select()).fetchall(),
            [(1, 1, 'Red', '5'), (2, 1, 'Blue', '1')]
        )

 def test_basic(self):
     pjoin = polymorphic_union({'manager': managers_table, 'engineer'
                               : engineers_table}, 'type', 'pjoin')
     employee_mapper = mapper(Employee, pjoin,
                              polymorphic_on=pjoin.c.type)
     manager_mapper = mapper(Manager, managers_table,
                             inherits=employee_mapper,
                             concrete=True,
                             polymorphic_identity='manager')
     engineer_mapper = mapper(Engineer, engineers_table,
                              inherits=employee_mapper,
                              concrete=True,
                              polymorphic_identity='engineer')
     session = create_session()
     session.add(Manager('Tom', 'knows how to manage things'))
     session.add(Engineer('Kurt', 'knows how to hack'))
     session.flush()
     session.expunge_all()
     assert set([repr(x) for x in session.query(Employee)]) \
         == set(['Engineer Kurt knows how to hack',
                'Manager Tom knows how to manage things'])
     assert set([repr(x) for x in session.query(Manager)]) \
         == set(['Manager Tom knows how to manage things'])
     assert set([repr(x) for x in session.query(Engineer)]) \
         == set(['Engineer Kurt knows how to hack'])
     manager = session.query(Manager).one()
     session.expire(manager, ['manager_data'])
     eq_(manager.manager_data, 'knows how to manage things')

 def test_determine_local_remote_pairs_m2m(self):
     joincond = self._join_fixture_m2m()
     eq_(
         joincond.local_remote_pairs,
         [(self.m2mleft.c.id, self.m2msecondary.c.lid),
         (self.m2mright.c.id, self.m2msecondary.c.rid)]
     )

 def test_determine_remote_columns_m2o_composite_selfref(self):
     joincond = self._join_fixture_m2o_composite_selfref()
     eq_(
         joincond.remote_columns,
         set([self.composite_selfref.c.id,
             self.composite_selfref.c.group_id])
     )

 def test_determine_remote_columns_compound_2(self):
     joincond = self._join_fixture_compound_expression_2(
                             support_sync=False)
     eq_(
         joincond.remote_columns,
         set([self.right.c.x, self.right.c.y])
     )

    def test_invocation_systemwide_loaders(self):
        baked.bake_lazy_loaders()
        try:
            User, Address = self._o2m_fixture()

            sess = Session()
            q = sess.query(User).options(lazyload(User.addresses))
            with mock.patch.object(BakedLazyLoader, "_emit_lazyload") as el:
                u1 = q.first()
                u1.addresses
                # invoked
                is_(
                    el.mock_calls[0][1][1],
                    u1._sa_instance_state
                )
        finally:
            baked.unbake_lazy_loaders()

        clear_mappers()
        User, Address = self._o2m_fixture()
        sess = Session()
        q = sess.query(User).options(lazyload(User.addresses))

        with mock.patch.object(BakedLazyLoader, "_emit_lazyload") as el:
            u1 = q.first()
            u1.addresses
            # not invoked
            eq_(el.mock_calls, [])

    def test_limit_offset_with_correlated_order_by(self):
        t1 = table('t1', column('x', Integer), column('y', Integer))
        t2 = table('t2', column('x', Integer), column('y', Integer))

        order_by = select([t2.c.y]).where(t1.c.x == t2.c.x).as_scalar()
        s = select([t1]).where(t1.c.x == 5).order_by(order_by) \
            .limit(10).offset(20)

        self.assert_compile(
            s,
            "SELECT anon_1.x, anon_1.y "
            "FROM (SELECT t1.x AS x, t1.y AS y, "
            "ROW_NUMBER() OVER (ORDER BY "
            "(SELECT t2.y FROM t2 WHERE t1.x = t2.x)"
            ") AS mssql_rn "
            "FROM t1 "
            "WHERE t1.x = :x_1) AS anon_1 "
            "WHERE mssql_rn > :param_1 AND mssql_rn <= :param_2 + :param_1",
            checkparams={'param_1': 20, 'param_2': 10, 'x_1': 5}
        )

        c = s.compile(dialect=mssql.MSDialect())
        eq_(len(c._result_columns), 2)
        assert t1.c.x in set(c._create_result_map()['x'][1])
        assert t1.c.y in set(c._create_result_map()['y'][1])

    def _test_queuepool(self, q, dispose=True):
        conn = q.connect()
        conn.close()
        conn = None

        conn = q.connect()
        conn.close()
        conn = None

        conn = q.connect()
        conn = None
        del conn
        lazy_gc()
        q.dispose()

        eq_(
            [buf.msg for buf in self.buf.buffer],
            [
                "Created new connection %r",
                "Connection %r checked out from pool",
                "Connection %r being returned to pool",
                "Connection %s rollback-on-return%s",
                "Connection %r checked out from pool",
                "Connection %r being returned to pool",
                "Connection %s rollback-on-return%s",
                "Connection %r checked out from pool",
                "Connection %r being returned to pool",
                "Connection %s rollback-on-return%s",
                "Closing connection %r",
            ]
            + (["Pool disposed. %s"] if dispose else []),
        )

        def go():
            sess = create_session()
            a1 = A(col2="a1")
            a2 = A(col2="a2")
            a3 = A(col2="a3")
            a1.bs.append(B(col2="b1"))
            a1.bs.append(B(col2="b2"))
            a3.bs.append(B(col2="b3"))
            for x in [a1, a2, a3]:
                sess.add(x)
            sess.flush()
            sess.expunge_all()

            alist = sess.query(A).order_by(A.col1).all()
            eq_(
                [
                    A(col2="a1", bs=[B(col2="b1"), B(col2="b2")]),
                    A(col2="a2", bs=[]),
                    A(col2="a3", bs=[B(col2="b3")]),
                ],
                alist,
            )

            for a in alist:
                sess.delete(a)
            sess.flush()

    def test_basic(self):
        for threadlocal in False, True:
            engine = engines.reconnecting_engine(
                options={'pool_threadlocal': threadlocal})

            conn = engine.contextual_connect()
            eq_(conn.execute(select([1])).scalar(), 1)
            conn.close()

            # set the pool recycle down to 1.
            # we aren't doing this inline with the
            # engine create since cx_oracle takes way
            # too long to create the 1st connection and don't
            # want to build a huge delay into this test.

            engine.pool._recycle = 1

            # kill the DB connection
            engine.test_shutdown()

            # wait until past the recycle period
            time.sleep(2)

            # can connect, no exception
            conn = engine.contextual_connect()
            eq_(conn.execute(select([1])).scalar(), 1)
            conn.close()

    def test_boolean(self):
        """Test that the boolean only treats 1 as True

        """

        meta = MetaData(testing.db)
        t = Table('bool_table', meta, Column('id', Integer,
                  primary_key=True), Column('boo',
                  Boolean(create_constraint=False)))
        try:
            meta.create_all()
            testing.db.execute("INSERT INTO bool_table (id, boo) "
                               "VALUES (1, 'false');")
            testing.db.execute("INSERT INTO bool_table (id, boo) "
                               "VALUES (2, 'true');")
            testing.db.execute("INSERT INTO bool_table (id, boo) "
                               "VALUES (3, '1');")
            testing.db.execute("INSERT INTO bool_table (id, boo) "
                               "VALUES (4, '0');")
            testing.db.execute('INSERT INTO bool_table (id, boo) '
                               'VALUES (5, 1);')
            testing.db.execute('INSERT INTO bool_table (id, boo) '
                               'VALUES (6, 0);')
            eq_(t.select(t.c.boo).order_by(t.c.id).execute().fetchall(),
                [(3, True), (5, True)])
        finally:
            meta.drop_all()