def test_unknown_types(self):
        meta = MetaData(testing.db)
        t = Table("test", meta,
            Column('foo', sa.DateTime))

        import sys
        dialect_module = sys.modules[testing.db.dialect.__module__]

        # we're relying on the presence of "ischema_names" in the
        # dialect module, else we can't test this.  we need to be able
        # to get the dialect to not be aware of some type so we temporarily
        # monkeypatch.  not sure what a better way for this could be,
        # except for an established dialect hook or dialect-specific tests
        if not hasattr(dialect_module, 'ischema_names'):
            return

        ischema_names = dialect_module.ischema_names
        t.create()
        dialect_module.ischema_names = {}
        try:
            m2 = MetaData(testing.db)
            assert_raises(tsa.exc.SAWarning, Table, "test", m2, autoload=True)

            @testing.emits_warning('Did not recognize type')
            def warns():
                m3 = MetaData(testing.db)
                t3 = Table("test", m3, autoload=True)
                assert t3.c.foo.type.__class__ == sa.types.NullType

        finally:
            dialect_module.ischema_names = ischema_names
            t.drop()

    def test_tometadata_with_schema(self):
        meta = MetaData()

        table = Table(
            "mytable",
            meta,
            Column("myid", Integer, primary_key=True),
            Column("name", String(40), nullable=True),
            Column("description", String(30), CheckConstraint("description='hi'")),
            UniqueConstraint("name"),
            test_needs_fk=True,
        )

        table2 = Table(
            "othertable",
            meta,
            Column("id", Integer, primary_key=True),
            Column("myid", Integer, ForeignKey("mytable.myid")),
            test_needs_fk=True,
        )

        meta2 = MetaData()
        table_c = table.tometadata(meta2, schema="someschema")
        table2_c = table2.tometadata(meta2, schema="someschema")

        eq_(str(table_c.join(table2_c).onclause), str(table_c.c.myid == table2_c.c.myid))
        eq_(str(table_c.join(table2_c).onclause), "someschema.mytable.myid = someschema.othertable.myid")

    def test_tometadata_strip_schema(self):
        meta = MetaData()

        table = Table('mytable', meta,
            Column('myid', Integer, primary_key=True),
            Column('name', String(40), nullable=True),
            Column('description', String(30),
                        CheckConstraint("description='hi'")),
            UniqueConstraint('name'),
            test_needs_fk=True,
        )

        table2 = Table('othertable', meta,
            Column('id', Integer, primary_key=True),
            Column('myid', Integer, ForeignKey('mytable.myid')),
            test_needs_fk=True,
            )

        meta2 = MetaData()
        table_c = table.tometadata(meta2, schema=None)
        table2_c = table2.tometadata(meta2, schema=None)

        eq_(str(table_c.join(table2_c).onclause), str(table_c.c.myid
            == table2_c.c.myid))
        eq_(str(table_c.join(table2_c).onclause),
            'mytable.myid = othertable.myid')

 def test_direct_quoting(self):
     m = MetaData(testing.db)
     t = Table("weird_casing", m, autoload=True)
     self.assert_compile(
         t.select(),
         'SELECT weird_casing.col1, weird_casing."Col2", weird_casing."col3" FROM weird_casing'
     )

    def test_override_keys(self):
        """test that columns can be overridden with a 'key', 
        and that ForeignKey targeting during reflection still works."""
        

        meta = MetaData(testing.db)
        a1 = Table('a', meta,
            Column('x', sa.Integer, primary_key=True),
            Column('z', sa.Integer),
            test_needs_fk=True
        )
        b1 = Table('b', meta,
            Column('y', sa.Integer, sa.ForeignKey('a.x')),
            test_needs_fk=True
        )
        meta.create_all()
        try:
            m2 = MetaData(testing.db)
            a2 = Table('a', m2, Column('x', sa.Integer, primary_key=True, key='x1'), autoload=True)
            b2 = Table('b', m2, autoload=True)
            
            assert a2.join(b2).onclause.compare(a2.c.x1==b2.c.y)
            assert b2.c.y.references(a2.c.x1)
        finally:
            meta.drop_all()

    def test_passive_override(self):
        """
        Primarily for postgres, tests that when we get a primary key column
        back from reflecting a table which has a default value on it, we
        pre-execute that DefaultClause upon insert, even though DefaultClause
        says "let the database execute this", because in postgres we must have
        all the primary key values in memory before insert; otherwise we can't
        locate the just inserted row.

        """
        # TODO: move this to dialect/postgres
        try:
            meta = MetaData(testing.db)
            testing.db.execute("""
             CREATE TABLE speedy_users
             (
                 speedy_user_id   SERIAL     PRIMARY KEY,

                 user_name        VARCHAR    NOT NULL,
                 user_password    VARCHAR    NOT NULL
             );
            """, None)

            t = Table("speedy_users", meta, autoload=True)
            t.insert().execute(user_name='user', user_password='lala')
            l = t.select().execute().fetchall()
            eq_(l, [(1, 'user', 'lala')])
        finally:
            testing.db.execute("drop table speedy_users", None)

 def setup_class(cls):
     global counters, metadata
     metadata = MetaData()
     counters = Table('forupdate_counters', metadata,
                      Column('counter_id', INT, primary_key=True),
                      Column('counter_value', INT),
                      test_needs_acid=True)
     counters.create(testing.db)

 def setup(self):
     meta = MetaData(testing.db)
     global table, seq
     seq = Sequence('tid_seq')
     table = Table('tables', meta,
                 Column('id', Integer, seq, primary_key=True),
                 Column('data', String(50))
             )
     table.create(checkfirst=True)

 def test_metadata_connect(self):
     metadata = MetaData()
     t1 = Table("table1", metadata, Column("col1", Integer, primary_key=True), Column("col2", String(20)))
     metadata.bind = testing.db
     metadata.create_all()
     try:
         assert t1.count().scalar() == 0
     finally:
         metadata.drop_all()

    def test_autoincrement_single_col(self):
        single = Table('single', self.metadata,
                       Column('id', Integer, primary_key=True))
        single.create()

        r = single.insert().execute()
        id_ = r.last_inserted_ids()[0]
        assert id_ is not None
        eq_(1, sa.select([func.count(sa.text('*'))], from_obj=single).scalar())

 def setup_class(cls):
     global users, metadata, tlengine
     tlengine = create_engine(testing.db.url, strategy='threadlocal')
     metadata = MetaData()
     users = Table('query_users', metadata,
         Column('user_id', INT, Sequence('query_users_id_seq', optional=True), primary_key=True),
         Column('user_name', VARCHAR(20)),
         test_needs_acid=True,
     )
     users.create(tlengine)

 def setup(self):
     global meta, table, engine
     engine = engines.reconnecting_engine()
     meta = MetaData(engine)
     table = Table('sometable', meta,
         Column('id', Integer, primary_key=True),
         Column('name', String(50)))
     meta.create_all()
     table.insert().execute(
         [{'id':i, 'name':'row %d' % i} for i in range(1, 100)]
     )

    def test_append_constraint_unique(self):
        meta = MetaData()

        users = Table('users', meta, Column('id', sa.Integer))
        addresses = Table('addresses', meta, Column('id', sa.Integer), Column('user_id', sa.Integer))

        fk = sa.ForeignKeyConstraint(['user_id'],[users.c.id])

        addresses.append_constraint(fk)
        addresses.append_constraint(fk)
        assert len(addresses.c.user_id.foreign_keys) == 1
        assert addresses.constraints == set([addresses.primary_key, fk])

    def test_override_existing_fk(self):
        """test that you can override columns and specify new foreign keys to other reflected tables,
        on columns which *do* already have that foreign key, and that the FK is not duped.
        """

        meta = MetaData(testing.db)
        users = Table('users', meta,
            Column('id', sa.Integer, primary_key=True),
            Column('name', sa.String(30)),
            test_needs_fk=True)
        addresses = Table('addresses', meta,
            Column('id', sa.Integer, primary_key=True),
            Column('user_id', sa.Integer, sa.ForeignKey('users.id')),
            test_needs_fk=True)

        meta.create_all()
        try:
            meta2 = MetaData(testing.db)
            a2 = Table('addresses', meta2,
                Column('user_id', sa.Integer, sa.ForeignKey('users.id')),
                autoload=True)
            u2 = Table('users', meta2, autoload=True)

            s = sa.select([a2])
            assert s.c.user_id
            assert len(a2.foreign_keys) == 1
            assert len(a2.c.user_id.foreign_keys) == 1
            assert len(a2.constraints) == 2
            assert [c.parent for c in a2.foreign_keys] == [a2.c.user_id]
            assert [c.parent for c in a2.c.user_id.foreign_keys] == [a2.c.user_id]
            assert list(a2.c.user_id.foreign_keys)[0].parent is a2.c.user_id
            assert u2.join(a2).onclause == u2.c.id==a2.c.user_id

            meta2 = MetaData(testing.db)
            u2 = Table('users', meta2,
                Column('id', sa.Integer, primary_key=True),
                autoload=True)
            a2 = Table('addresses', meta2,
                Column('id', sa.Integer, primary_key=True),
                Column('user_id', sa.Integer, sa.ForeignKey('users.id')),
                autoload=True)

            s = sa.select([a2])
            assert s.c.user_id
            assert len(a2.foreign_keys) == 1
            assert len(a2.c.user_id.foreign_keys) == 1
            assert len(a2.constraints) == 2
            assert [c.parent for c in a2.foreign_keys] == [a2.c.user_id]
            assert [c.parent for c in a2.c.user_id.foreign_keys] == [a2.c.user_id]
            assert list(a2.c.user_id.foreign_keys)[0].parent is a2.c.user_id
            assert u2.join(a2).onclause == u2.c.id==a2.c.user_id
        finally:
            meta.drop_all()

    def test_tometadata_kwargs(self):
        meta = MetaData()

        table = Table('mytable', meta,
            Column('myid', Integer, primary_key=True),
            mysql_engine='InnoDB',
        )

        meta2 = MetaData()
        table_c = table.tometadata(meta2)

        eq_(table.kwargs,table_c.kwargs)

 def test_empty_insert(self):
     metadata = MetaData(testing.db)
     t1 = Table('t1', metadata,
             Column('is_true', Boolean, server_default=('1')))
     metadata.create_all()
     
     try:
         result = t1.insert().execute()
         eq_(1, select([func.count(text('*'))], from_obj=t1).scalar())
         eq_(True, t1.select().scalar())
     finally:
         metadata.drop_all()

    def test_non_autoincrement(self):
        # sqlite INT primary keys can be non-unique! (only for ints)
        nonai = Table("nonaitest", self.metadata,
            Column('id', Integer, autoincrement=False, primary_key=True),
            Column('data', String(20)))
        nonai.create()


        try:
            # postgres + mysql strict will fail on first row,
            # mysql in legacy mode fails on second row
            nonai.insert().execute(data='row 1')
            nonai.insert().execute(data='row 2')
            assert False
        except sa.exc.SQLError, e:
            assert True

    def test_fk_no_such_target_col_error(self):
        meta = MetaData()
        a = Table('a', meta, Column('a', Integer))
        b = Table('b', meta, Column('b', Integer))
        a.append_constraint(
            ForeignKeyConstraint(['a'], ['b.x'])
        )

        def go():
            list(a.c.a.foreign_keys)[0].column
        assert_raises_message(
            exc.NoReferencedColumnError,
            "Could not create ForeignKey 'b.x' on "
            "table 'a': table 'b' has no column named 'x'",
            go
        )

    def test_tometadata_already_there(self):

        meta1 = MetaData()
        table1 = Table('mytable', meta1,
            Column('myid', Integer, primary_key=True),
        )
        meta2 = MetaData()
        table2 = Table('mytable', meta2,
            Column('yourid', Integer, primary_key=True),
        )

        meta3 = MetaData()

        table_c = table1.tometadata(meta2)
        table_d = table2.tometadata(meta2)

        # d'oh!
        assert table_c is table_d

    def test_rollback_deadlock(self):
        """test that returning connections to the pool clears any object locks."""
        conn1 = testing.db.connect()
        conn2 = testing.db.connect()
        users = Table('deadlock_users', metadata,
            Column('user_id', INT, primary_key = True),
            Column('user_name', VARCHAR(20)),
            test_needs_acid=True,
        )
        users.create(conn1)
        conn1.execute("select * from deadlock_users")
        conn1.close()

        # without auto-rollback in the connection pool's return() logic, this
        # deadlocks in PostgreSQL, because conn1 is returned to the pool but
        # still has a lock on "deadlock_users".
        # comment out the rollback in pool/ConnectionFairy._close() to see !
        users.drop(conn2)
        conn2.close()