def test_pre_ping_db_is_restarted(self):
        engine = engines.reconnecting_engine(
            options={"pool_pre_ping": True}
        )

        conn = engine.connect()
        eq_(conn.execute(select([1])).scalar(), 1)
        stale_connection = conn.connection.connection
        conn.close()

        engine.test_shutdown()
        engine.test_restart()

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

        def exercise_stale_connection():
            curs = stale_connection.cursor()
            curs.execute("select 1")

        assert_raises(
            engine.dialect.dbapi.Error,
            exercise_stale_connection
        )

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

    def test_pre_ping_db_stays_shutdown(self):
        engine = engines.reconnecting_engine(options={"pool_pre_ping": True})

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

        engine.test_shutdown(stop=True)

        assert_raises(exc.DBAPIError, engine.connect)

 def test_null_pool(self):
     engine = engines.reconnecting_engine(options=dict(poolclass=pool.NullPool))
     conn = engine.connect()
     eq_(conn.execute(select([1])).scalar(), 1)
     assert not conn.closed
     engine.test_shutdown()
     _assert_invalidated(conn.execute, select([1]))
     assert not conn.closed
     assert conn.invalidated
     eq_(conn.execute(select([1])).scalar(), 1)
     assert not conn.invalidated

 def test_null_pool(self):
     engine = \
         engines.reconnecting_engine(options=dict(poolclass=pool.NullPool))
     conn = engine.connect()
     eq_(conn.execute(select([1])).scalar(), 1)
     assert not conn.closed
     engine.test_shutdown()
     try:
         conn.execute(select([1]))
         assert False
     except tsa.exc.DBAPIError, e:
         if not e.connection_invalidated:
             raise

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

 def setup(self):
     self.engine = engines.reconnecting_engine()

 def setup(self):
     global engine
     engine = engines.reconnecting_engine()