import itertools
import time
from unittest.mock import call
from unittest.mock import Mock

import sqlalchemy as tsa
from sqlalchemy import create_engine
from sqlalchemy import event
from sqlalchemy import exc
from sqlalchemy import Integer
from sqlalchemy import MetaData
from sqlalchemy import pool
from sqlalchemy import select
from sqlalchemy import String
from sqlalchemy import testing
from sqlalchemy import util
from sqlalchemy.engine import default
from sqlalchemy.testing import assert_raises
from sqlalchemy.testing import assert_raises_message
from sqlalchemy.testing import assert_raises_message_context_ok
from sqlalchemy.testing import assert_warns_message
from sqlalchemy.testing import engines
from sqlalchemy.testing import eq_
from sqlalchemy.testing import expect_raises
from sqlalchemy.testing import expect_raises_message
from sqlalchemy.testing import fixtures
from sqlalchemy.testing import is_
from sqlalchemy.testing import is_false
from sqlalchemy.testing import is_true
from sqlalchemy.testing import mock
from sqlalchemy.testing import ne_
from sqlalchemy.testing.engines import DBAPIProxyConnection
from sqlalchemy.testing.engines import DBAPIProxyCursor
from sqlalchemy.testing.engines import testing_engine
from sqlalchemy.testing.schema import Column
from sqlalchemy.testing.schema import Table
from sqlalchemy.testing.util import gc_collect


class MockError(Exception):
    pass


class MockDisconnect(MockError):
    pass


class MockExitIsh(BaseException):
    pass


def mock_connection():
    def mock_cursor():
        def execute(*args, **kwargs):
            if conn.explode == "execute":
                raise MockDisconnect("Lost the DB connection on execute")
            elif conn.explode == "interrupt":
                conn.explode = "explode_no_disconnect"
                raise MockExitIsh("Keyboard / greenlet / etc interruption")
            elif conn.explode == "interrupt_dont_break":
                conn.explode = None
                raise MockExitIsh("Keyboard / greenlet / etc interruption")
            elif conn.explode in (
                "execute_no_disconnect",
                "explode_no_disconnect",
            ):
                raise MockError(
                    "something broke on execute but we didn't lose the "
                    "connection"
                )
            elif conn.explode in (
                "rollback",
                "rollback_no_disconnect",
                "explode_no_disconnect",
            ):
                raise MockError(
                    "something broke on execute but we didn't lose the "
                    "connection"
                )
            elif args and "SELECT" in args[0]:
                cursor.description = [("foo", None, None, None, None, None)]
            else:
                return

        def close():
            cursor.fetchall = cursor.fetchone = Mock(
                side_effect=MockError("cursor closed")
            )

        cursor = Mock(
            execute=Mock(side_effect=execute), close=Mock(side_effect=close)
        )
        return cursor

    def cursor():
        while True:
            yield mock_cursor()

    def rollback():
        if conn.explode == "rollback":
            raise MockDisconnect("Lost the DB connection on rollback")
        if conn.explode == "rollback_no_disconnect":
            raise MockError(
                "something broke on rollback but we didn't lose the "
                "connection"
            )
        else:
            return

    def commit():
        if conn.explode == "commit":
            raise MockDisconnect("Lost the DB connection on commit")
        elif conn.explode == "commit_no_disconnect":
            raise MockError(
                "something broke on commit but we didn't lose the "
                "connection"
            )
        else:
            return

    conn = Mock(
        rollback=Mock(side_effect=rollback),
        commit=Mock(side_effect=commit),
        cursor=Mock(side_effect=cursor()),
    )
    return conn


def MockDBAPI():
    connections = []
    stopped = [False]

    def connect():
        while True:
            if stopped[0]:
                raise MockDisconnect("database is stopped")
            conn = mock_connection()
            connections.append(conn)
            yield conn

    def shutdown(explode="execute", stop=False):
        stopped[0] = stop
        for c in connections:
            c.explode = explode

    def restart():
        stopped[0] = False
        connections[:] = []

    def dispose():
        stopped[0] = False
        for c in connections:
            c.explode = None
        connections[:] = []

    return Mock(
        connect=Mock(side_effect=connect()),
        shutdown=Mock(side_effect=shutdown),
        dispose=Mock(side_effect=dispose),
        restart=Mock(side_effect=restart),
        paramstyle="named",
        connections=connections,
        Error=MockError,
    )


class PrePingMockTest(fixtures.TestBase):
    def setup_test(self):
        self.dbapi = MockDBAPI()

    def _pool_fixture(self, pre_ping, setup_disconnect=True, pool_kw=None):
        dialect = default.DefaultDialect()
        dialect.dbapi = self.dbapi
        _pool = pool.QueuePool(
            creator=lambda: self.dbapi.connect("foo.db"),
            pre_ping=pre_ping,
            dialect=dialect,
            **(pool_kw if pool_kw else {}),
        )

        if setup_disconnect:
            dialect.is_disconnect = lambda e, conn, cursor: isinstance(
                e, MockDisconnect
            )
        return _pool

    def teardown_test(self):
        self.dbapi.dispose()

    def test_ping_not_on_first_connect(self):
        pool = self._pool_fixture(
            pre_ping=True, pool_kw=dict(pool_size=1, max_overflow=0)
        )

        conn = pool.connect()
        dbapi_conn = conn.dbapi_connection
        eq_(dbapi_conn.mock_calls, [])
        conn.close()

        # no ping, so no cursor() call.
        eq_(dbapi_conn.mock_calls, [call.rollback()])

        conn = pool.connect()
        is_(conn.dbapi_connection, dbapi_conn)

        # ping, so cursor() call.
        eq_(dbapi_conn.mock_calls, [call.rollback(), call.cursor()])

        conn.close()

        conn = pool.connect()
        is_(conn.dbapi_connection, dbapi_conn)

        # ping, so cursor() call.
        eq_(
            dbapi_conn.mock_calls,
            [call.rollback(), call.cursor(), call.rollback(), call.cursor()],
        )

        conn.close()

    def test_ping_not_on_reconnect(self):
        pool = self._pool_fixture(
            pre_ping=True, pool_kw=dict(pool_size=1, max_overflow=0)
        )

        conn = pool.connect()
        dbapi_conn = conn.dbapi_connection
        conn_rec = conn._connection_record
        eq_(dbapi_conn.mock_calls, [])
        conn.close()

        conn = pool.connect()
        is_(conn.dbapi_connection, dbapi_conn)
        # ping, so cursor() call.
        eq_(dbapi_conn.mock_calls, [call.rollback(), call.cursor()])

        conn.invalidate()

        is_(conn.dbapi_connection, None)

        # connect again, make sure we're on the same connection record
        conn = pool.connect()
        is_(conn._connection_record, conn_rec)

        # no ping
        dbapi_conn = conn.dbapi_connection
        eq_(dbapi_conn.mock_calls, [])

    def test_connect_across_restart(self):
        pool = self._pool_fixture(pre_ping=True)

        conn = pool.connect()
        stale_connection = conn.dbapi_connection
        conn.close()

        self.dbapi.shutdown("execute")
        self.dbapi.restart()

        conn = pool.connect()
        cursor = conn.cursor()
        cursor.execute("hi")

        stale_cursor = stale_connection.cursor()
        assert_raises(MockDisconnect, stale_cursor.execute, "hi")

    def test_handle_error_sets_disconnect(self):
        pool = self._pool_fixture(pre_ping=True, setup_disconnect=False)

        @event.listens_for(pool._dialect, "handle_error")
        def setup_disconnect(ctx):
            assert isinstance(ctx.sqlalchemy_exception, exc.DBAPIError)
            assert isinstance(ctx.original_exception, MockDisconnect)
            ctx.is_disconnect = True

        conn = pool.connect()
        stale_connection = conn.dbapi_connection
        conn.close()

        self.dbapi.shutdown("execute")
        self.dbapi.restart()

        conn = pool.connect()
        cursor = conn.cursor()
        cursor.execute("hi")

        stale_cursor = stale_connection.cursor()
        assert_raises(MockDisconnect, stale_cursor.execute, "hi")

    def test_raise_db_is_stopped(self):
        pool = self._pool_fixture(pre_ping=True)

        conn = pool.connect()
        conn.close()

        self.dbapi.shutdown("execute", stop=True)

        assert_raises_message_context_ok(
            MockDisconnect, "database is stopped", pool.connect
        )

    def test_waits_til_exec_wo_ping_db_is_stopped(self):
        pool = self._pool_fixture(pre_ping=False)

        conn = pool.connect()
        conn.close()

        self.dbapi.shutdown("execute", stop=True)

        conn = pool.connect()

        cursor = conn.cursor()
        assert_raises_message(
            MockDisconnect,
            "Lost the DB connection on execute",
            cursor.execute,
            "foo",
        )

    def test_waits_til_exec_wo_ping_db_is_restarted(self):
        pool = self._pool_fixture(pre_ping=False)

        conn = pool.connect()
        conn.close()

        self.dbapi.shutdown("execute", stop=True)
        self.dbapi.restart()

        conn = pool.connect()

        cursor = conn.cursor()
        assert_raises_message(
            MockDisconnect,
            "Lost the DB connection on execute",
            cursor.execute,
            "foo",
        )

    @testing.requires.predictable_gc
    def test_pre_ping_weakref_finalizer(self):
        pool = self._pool_fixture(pre_ping=True)

        conn = pool.connect()
        old_dbapi_conn = conn.dbapi_connection
        conn.close()

        # no cursor() because no pre ping
        eq_(old_dbapi_conn.mock_calls, [call.rollback()])

        conn = pool.connect()
        conn.close()

        # connect again, we see pre-ping
        eq_(
            old_dbapi_conn.mock_calls,
            [call.rollback(), call.cursor(), call.rollback()],
        )

        self.dbapi.shutdown("execute", stop=True)
        self.dbapi.restart()

        conn = pool.connect()
        dbapi_conn = conn.dbapi_connection
        del conn
        gc_collect()

        # new connection was reset on return appropriately
        eq_(dbapi_conn.mock_calls, [call.rollback()])

        # old connection was just closed - did not get an
        # erroneous reset on return
        eq_(
            old_dbapi_conn.mock_calls,
            [
                call.rollback(),
                call.cursor(),
                call.rollback(),
                call.cursor(),
                call.close(),
            ],
        )


class MockReconnectTest(fixtures.TestBase):
    def setup_test(self):
        self.dbapi = MockDBAPI()

        self.db = testing_engine(
            "postgresql+psycopg2://foo:bar@localhost/test",
            options=dict(module=self.dbapi, _initialize=False),
        )

        self.mock_connect = call(
            host="localhost", password="bar", user="foo", dbname="test"
        )
        # monkeypatch disconnect checker
        self.db.dialect.is_disconnect = lambda e, conn, cursor: isinstance(
            e, MockDisconnect
        )

    def teardown_test(self):
        self.dbapi.dispose()

    def test_reconnect(self):
        """test that an 'is_disconnect' condition will invalidate the
        connection, and additionally dispose the previous connection
        pool and recreate."""

        # make a connection

        conn = self.db.connect()

        # connection works

        conn.execute(select(1))

        # create a second connection within the pool, which we'll ensure
        # also goes away

        conn2 = self.db.connect()
        conn2.close()

        # two connections opened total now

        assert len(self.dbapi.connections) == 2

        # set it to fail

        self.dbapi.shutdown()

        # force windows monotonic timer to definitely increment
        time.sleep(0.5)

        # close on DBAPI connection occurs here, as it is detected
        # as invalid.
        assert_raises(tsa.exc.DBAPIError, conn.execute, select(1))

        # assert was invalidated

        assert not conn.closed
        assert conn.invalidated

        # close shouldn't break

        conn.close()

        # ensure one connection closed...
        eq_(
            [c.close.mock_calls for c in self.dbapi.connections],
            [[call()], []],
        )

        conn = self.db.connect()

        eq_(
            [c.close.mock_calls for c in self.dbapi.connections],
            [[call()], [call()], []],
        )

        conn.execute(select(1))
        conn.close()

        eq_(
            [c.close.mock_calls for c in self.dbapi.connections],
            [[call()], [call()], []],
        )

    def test_invalidate_on_execute_trans(self):
        conn = self.db.connect()
        trans = conn.begin()
        self.dbapi.shutdown()

        assert_raises(tsa.exc.DBAPIError, conn.execute, select(1))

        eq_([c.close.mock_calls for c in self.dbapi.connections], [[call()]])
        assert not conn.closed
        assert conn.invalidated
        assert trans.is_active
        assert_raises_message(
            tsa.exc.PendingRollbackError,
            "Can't reconnect until invalid transaction is rolled back",
            conn.execute,
            select(1),
        )
        assert trans.is_active

        assert_raises_message(
            tsa.exc.PendingRollbackError,
            "Can't reconnect until invalid transaction is rolled back",
            trans.commit,
        )

        # now it's inactive...
        assert not trans.is_active

        # but still associated with the connection
        assert_raises_message(
            tsa.exc.PendingRollbackError,
            "Can't reconnect until invalid transaction is rolled back",
            conn.execute,
            select(1),
        )
        assert not trans.is_active

        # still can't commit... error stays the same
        assert_raises_message(
            tsa.exc.PendingRollbackError,
            "Can't reconnect until invalid transaction is rolled back",
            trans.commit,
        )

        trans.rollback()
        assert not trans.is_active
        conn.execute(select(1))
        assert not conn.invalidated
        eq_(
            [c.close.mock_calls for c in self.dbapi.connections],
            [[call()], []],
        )

    def test_invalidate_on_commit_trans(self):
        conn = self.db.connect()
        trans = conn.begin()
        self.dbapi.shutdown("commit")

        assert_raises(tsa.exc.DBAPIError, trans.commit)

        assert not conn.closed
        assert conn.invalidated
        assert not trans.is_active

        # error stays consistent
        assert_raises_message(
            tsa.exc.PendingRollbackError,
            "Can't reconnect until invalid transaction is rolled back",
            conn.execute,
            select(1),
        )
        assert not trans.is_active

        assert_raises_message(
            tsa.exc.PendingRollbackError,
            "Can't reconnect until invalid transaction is rolled back",
            trans.commit,
        )

        assert not trans.is_active

        assert_raises_message(
            tsa.exc.PendingRollbackError,
            "Can't reconnect until invalid transaction is rolled back",
            conn.execute,
            select(1),
        )
        assert not trans.is_active

        trans.rollback()
        assert not trans.is_active
        conn.execute(select(1))
        assert not conn.invalidated

    def test_commit_fails_contextmanager(self):
        # this test is also performed in test/engine/test_transaction.py
        # using real connections
        conn = self.db.connect()

        def go():
            with conn.begin():
                self.dbapi.shutdown("commit_no_disconnect")

        assert_raises(tsa.exc.DBAPIError, go)

        assert not conn.in_transaction()

    def test_commit_fails_trans(self):
        # this test is also performed in test/engine/test_transaction.py
        # using real connections

        conn = self.db.connect()
        trans = conn.begin()
        self.dbapi.shutdown("commit_no_disconnect")

        assert_raises(tsa.exc.DBAPIError, trans.commit)

        assert not conn.closed
        assert not conn.invalidated
        assert not trans.is_active

        # error stays consistent
        assert_raises_message(
            tsa.exc.PendingRollbackError,
            r"Can't reconnect until invalid transaction is rolled back.  "
            r"Please rollback\(\) fully before proceeding",
            conn.execute,
            select(1),
        )
        assert not trans.is_active

        assert_raises_message(
            tsa.exc.PendingRollbackError,
            r"Can't reconnect until invalid transaction is rolled back.  "
            r"Please rollback\(\) fully before proceeding",
            trans.commit,
        )

        assert not trans.is_active

        assert_raises_message(
            tsa.exc.PendingRollbackError,
            r"Can't reconnect until invalid transaction is rolled back.  "
            r"Please rollback\(\) fully before proceeding",
            conn.execute,
            select(1),
        )
        assert not trans.is_active

        trans.rollback()
        assert not trans.is_active
        conn.execute(select(1))
        assert not conn.invalidated

    def test_invalidate_dont_call_finalizer(self):
        conn = self.db.connect()
        finalizer = mock.Mock()
        conn.connection._connection_record.finalize_callback.append(finalizer)
        conn.invalidate()
        assert conn.invalidated
        eq_(finalizer.call_count, 0)

    def test_conn_reusable(self):
        conn = self.db.connect()

        conn.execute(select(1))

        eq_(self.dbapi.connect.mock_calls, [self.mock_connect])

        self.dbapi.shutdown()

        with expect_raises(tsa.exc.DBAPIError):
            conn.execute(select(1))

        assert not conn.closed
        assert conn.invalidated

        eq_([c.close.mock_calls for c in self.dbapi.connections], [[call()]])

        # trans was autobegin.  they have to call rollback
        with expect_raises(tsa.exc.PendingRollbackError):
            conn.execute(select(1))

        # ok
        conn.rollback()

        # now we are good
        # test reconnects
        conn.execute(select(1))
        assert not conn.invalidated

        eq_(
            [c.close.mock_calls for c in self.dbapi.connections],
            [[call()], []],
        )

    def test_invalidated_close(self):
        conn = self.db.connect()

        self.dbapi.shutdown()

        assert_raises(tsa.exc.DBAPIError, conn.execute, select(1))

        conn.close()
        assert conn.closed
        assert not conn.invalidated
        assert_raises_message(
            tsa.exc.ResourceClosedError,
            "This Connection is closed",
            conn.execute,
            select(1),
        )

    def test_noreconnect_execute(self):
        conn = self.db.connect()

        self.dbapi.shutdown("execute_no_disconnect")

        # raises error
        assert_raises_message(
            tsa.exc.DBAPIError,
            "something broke on execute but we didn't lose the connection",
            conn.execute,
            select(1),
        )

        assert not conn.closed
        assert not conn.invalidated

        conn.close()

    def test_noreconnect_rollback(self):
        # this test changes in 2.x due to autobegin.

        conn = self.db.connect()

        conn.execute(select(1))

        self.dbapi.shutdown("rollback_no_disconnect")

        # previously, running a select() here which would fail would then
        # trigger autorollback which would also fail, this is not the
        # case now as autorollback does not normally occur
        with expect_raises_message(
            tsa.exc.DBAPIError,
            r"something broke on rollback but we didn't lose the connection",
        ):
            conn.rollback()

        assert not conn.closed
        assert not conn.invalidated

        conn.close()

        assert_raises_message(
            tsa.exc.ResourceClosedError,
            "This Connection is closed",
            conn.execute,
            select(1),
        )

    def test_reconnect_on_reentrant(self):
        conn = self.db.connect()

        conn.execute(select(1))

        assert len(self.dbapi.connections) == 1

        self.dbapi.shutdown("rollback")

        assert_raises_message(
            tsa.exc.DBAPIError,
            "Lost the DB connection on rollback",
            conn.rollback,
        )

        assert not conn.closed
        assert conn.invalidated

    def test_check_disconnect_no_cursor(self):
        conn = self.db.connect()
        result = conn.execute(select(1))
        result.cursor.close()
        conn.close()

        assert_raises_message(
            tsa.exc.DBAPIError, "cursor closed", list, result
        )

    def test_dialect_initialize_once(self):
        from sqlalchemy.engine.url import URL
        from sqlalchemy.engine.default import DefaultDialect

        dbapi = self.dbapi

        class MyURL(URL):
            def _get_entrypoint(self):
                return Dialect

            def get_dialect(self):
                return Dialect

        class Dialect(DefaultDialect):
            initialize = Mock()

        engine = create_engine(MyURL.create("foo://"), module=dbapi)
        engine.connect()

        # note that the dispose() call replaces the old pool with a new one;
        # this is to test that even though a single pool is using
        # dispatch.exec_once(), by replacing the pool with a new one, the event
        # would normally fire again onless once=True is set on the original
        # listen as well.

        engine.dispose()
        engine.connect()
        eq_(Dialect.initialize.call_count, 1)

    def test_dialect_initialize_retry_if_exception(self):
        from sqlalchemy.engine.url import URL
        from sqlalchemy.engine.default import DefaultDialect

        dbapi = self.dbapi

        class MyURL(URL):
            def _get_entrypoint(self):
                return Dialect

            def get_dialect(self):
                return Dialect

        class Dialect(DefaultDialect):
            initialize = Mock()

        # note that the first_connect hook is only invoked when the pool
        # makes a new DBAPI connection, and not when it checks out an existing
        # connection.  So there is a dependency here that if the initializer
        # raises an exception, the pool-level connection attempt is also
        # failed, meaning no DBAPI connection is pooled.  If the first_connect
        # exception raise did not prevent the connection from being pooled,
        # there could be the case where the pool could return that connection
        # on a subsequent attempt without initialization having proceeded.

        Dialect.initialize.side_effect = TypeError
        engine = create_engine(MyURL.create("foo://"), module=dbapi)

        assert_raises(TypeError, engine.connect)
        eq_(Dialect.initialize.call_count, 1)
        is_true(engine.pool._pool.empty())

        assert_raises(TypeError, engine.connect)
        eq_(Dialect.initialize.call_count, 2)
        is_true(engine.pool._pool.empty())

        engine.dispose()

        assert_raises(TypeError, engine.connect)
        eq_(Dialect.initialize.call_count, 3)
        is_true(engine.pool._pool.empty())

        Dialect.initialize.side_effect = None

        conn = engine.connect()
        eq_(Dialect.initialize.call_count, 4)
        conn.close()
        is_false(engine.pool._pool.empty())

        conn = engine.connect()
        eq_(Dialect.initialize.call_count, 4)
        conn.close()
        is_false(engine.pool._pool.empty())

        engine.dispose()
        conn = engine.connect()

        eq_(Dialect.initialize.call_count, 4)
        conn.close()
        is_false(engine.pool._pool.empty())

    def test_invalidate_conn_w_contextmanager_interrupt(self):
        # test [ticket:3803]
        pool = self.db.pool

        conn = self.db.connect()
        self.dbapi.shutdown("interrupt")

        def go():
            with conn.begin():
                conn.execute(select(1))

        assert_raises(MockExitIsh, go)

        assert conn.invalidated

        eq_(pool._invalidate_time, 0)  # pool not invalidated

        conn.execute(select(1))
        assert not conn.invalidated

    def test_invalidate_conn_interrupt_nodisconnect_workaround(self):
        # test [ticket:3803] workaround for no disconnect on keyboard interrupt

        @event.listens_for(self.db, "handle_error")
        def cancel_disconnect(ctx):
            ctx.is_disconnect = False

        pool = self.db.pool

        conn = self.db.connect()
        self.dbapi.shutdown("interrupt_dont_break")

        def go():
            with conn.begin():
                conn.execute(select(1))

        assert_raises(MockExitIsh, go)

        assert not conn.invalidated

        eq_(pool._invalidate_time, 0)  # pool not invalidated

        conn.execute(select(1))
        assert not conn.invalidated

    def test_invalidate_conn_w_contextmanager_disconnect(self):
        # test [ticket:3803] change maintains old behavior

        pool = self.db.pool

        conn = self.db.connect()
        self.dbapi.shutdown("execute")

        def go():
            with conn.begin():
                conn.execute(select(1))

        assert_raises(exc.DBAPIError, go)  # wraps a MockDisconnect

        assert conn.invalidated

        ne_(pool._invalidate_time, 0)  # pool is invalidated

        conn.execute(select(1))
        assert not conn.invalidated


class CursorErrTest(fixtures.TestBase):
    # this isn't really a "reconnect" test, it's more of
    # a generic "recovery".   maybe this test suite should have been
    # named "test_error_recovery".
    def _fixture(self, explode_on_exec, initialize):
        class DBAPIError(Exception):
            pass

        def MockDBAPI():
            def cursor():
                while True:
                    if explode_on_exec:
                        yield Mock(
                            description=[],
                            close=Mock(side_effect=DBAPIError("explode")),
                            execute=Mock(side_effect=DBAPIError("explode")),
                        )
                    else:
                        yield Mock(
                            description=[],
                            close=Mock(side_effect=Exception("explode")),
                        )

            def connect():
                while True:
                    yield Mock(
                        spec=["cursor", "commit", "rollback", "close"],
                        cursor=Mock(side_effect=cursor()),
                    )

            return Mock(
                Error=DBAPIError,
                paramstyle="qmark",
                connect=Mock(side_effect=connect()),
            )

        dbapi = MockDBAPI()

        from sqlalchemy.engine import default

        url = Mock(
            get_dialect=lambda: default.DefaultDialect,
            _get_entrypoint=lambda: default.DefaultDialect,
            _instantiate_plugins=lambda kwargs: (url, [], kwargs),
            translate_connect_args=lambda: {},
            query={},
        )
        eng = testing_engine(
            url, options=dict(module=dbapi, _initialize=initialize)
        )
        eng.pool.logger = Mock()

        def get_default_schema_name(connection):
            try:
                cursor = connection.connection.cursor()
                connection._cursor_execute(cursor, "statement", {})
                cursor.close()
            except exc.DBAPIError:
                util.warn("Exception attempting to detect")

        eng.dialect._get_default_schema_name = get_default_schema_name
        return eng

    def test_cursor_explode(self):
        db = self._fixture(False, False)
        conn = db.connect()
        result = conn.exec_driver_sql("select foo")
        result.close()
        conn.close()
        eq_(
            db.pool.logger.error.mock_calls,
            [call("Error closing cursor", exc_info=True)],
        )

    def test_cursor_shutdown_in_initialize(self):
        db = self._fixture(True, True)
        assert_warns_message(
            exc.SAWarning, "Exception attempting to detect", db.connect
        )
        eq_(
            db.pool.logger.error.mock_calls,
            [call("Error closing cursor", exc_info=True)],
        )


def _assert_invalidated(fn, *args):
    try:
        fn(*args)
        assert False
    except tsa.exc.DBAPIError as e:
        if not e.connection_invalidated:
            raise


class RealPrePingEventHandlerTest(fixtures.TestBase):
    """real test for issue #5648, which had to be revisited for 2.0 as the
    initial version was not adequately tested and non-implementation for
    mysql, postgresql was not caught

    """

    __backend__ = True
    __requires__ = "graceful_disconnects", "ad_hoc_engines"

    @testing.fixture
    def ping_fixture(self, testing_engine):
        engine = testing_engine(
            options={"pool_pre_ping": True, "_initialize": False}
        )

        existing_connect = engine.dialect.dbapi.connect

        fail = False
        fail_count = itertools.count()
        DBAPIError = engine.dialect.dbapi.Error

        class ExplodeConnection(DBAPIProxyConnection):
            def ping(self, *arg, **kw):
                if fail and next(fail_count) < 1:
                    raise DBAPIError("unhandled disconnect situation")
                else:
                    return True

        class ExplodeCursor(DBAPIProxyCursor):
            def execute(self, stmt, parameters=None, **kw):
                if fail and next(fail_count) < 1:
                    raise DBAPIError("unhandled disconnect situation")
                else:
                    return super().execute(stmt, parameters=parameters, **kw)

        def mock_connect(*arg, **kw):
            real_connection = existing_connect(*arg, **kw)
            return ExplodeConnection(engine, real_connection, ExplodeCursor)

        with mock.patch.object(
            engine.dialect.loaded_dbapi, "connect", mock_connect
        ):
            # set up initial connection.  pre_ping works on subsequent connects
            engine.connect().close()

            # ping / exec will fail
            fail = True

            yield engine

    @testing.fixture
    def ping_fixture_all_errs_disconnect(self, ping_fixture):
        engine = ping_fixture

        with mock.patch.object(
            engine.dialect, "is_disconnect", lambda *arg, **kw: True
        ):
            yield engine

    def test_control(self, ping_fixture):
        """test the fixture raises on connect"""
        engine = ping_fixture

        with expect_raises_message(
            exc.DBAPIError, "unhandled disconnect situation"
        ):
            engine.connect()

    def test_downgrade_control(self, ping_fixture_all_errs_disconnect):
        """test the disconnect fixture doesn't raise, since it considers
        all errors to be disconnect errors.

        """

        engine = ping_fixture_all_errs_disconnect

        conn = engine.connect()
        conn.close()

    def test_event_handler_didnt_upgrade_disconnect(self, ping_fixture):
        """test that having an event handler that doesn't do anything
        keeps the behavior in place for a fatal error.

        """
        engine = ping_fixture

        @event.listens_for(engine, "handle_error")
        def setup_disconnect(ctx):
            assert not ctx.is_disconnect

        with expect_raises_message(
            exc.DBAPIError, "unhandled disconnect situation"
        ):
            engine.connect()

    def test_event_handler_didnt_downgrade_disconnect(
        self, ping_fixture_all_errs_disconnect
    ):
        """test that having an event handler that doesn't do anything
        keeps the behavior in place for a disconnect error.

        """
        engine = ping_fixture_all_errs_disconnect

        @event.listens_for(engine, "handle_error")
        def setup_disconnect(ctx):
            assert ctx.is_pre_ping
            assert ctx.is_disconnect

        conn = engine.connect()
        conn.close()

    def test_event_handler_can_upgrade_disconnect(self, ping_fixture):
        """test that an event hook can receive a fatal error and convert
        it to be a disconnect error during pre-ping"""

        engine = ping_fixture

        @event.listens_for(engine, "handle_error")
        def setup_disconnect(ctx):
            assert ctx.is_pre_ping
            ctx.is_disconnect = True

        conn = engine.connect()
        # no error
        conn.close()

    def test_event_handler_can_downgrade_disconnect(
        self, ping_fixture_all_errs_disconnect
    ):
        """test that an event hook can receive a disconnect error and convert
        it to be a fatal error during pre-ping"""

        engine = ping_fixture_all_errs_disconnect

        @event.listens_for(engine, "handle_error")
        def setup_disconnect(ctx):
            assert ctx.is_disconnect
            if ctx.is_pre_ping:
                ctx.is_disconnect = False

        with expect_raises_message(
            exc.DBAPIError, "unhandled disconnect situation"
        ):
            engine.connect()


class RealReconnectTest(fixtures.TestBase):
    __backend__ = True
    __requires__ = "graceful_disconnects", "ad_hoc_engines"

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

    def teardown_test(self):
        self.engine.dispose()

    def test_reconnect(self):
        with self.engine.connect() as conn:
            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 conn.invalidated

            with expect_raises(tsa.exc.PendingRollbackError):
                conn.execute(select(1))

            conn.rollback()

            eq_(conn.execute(select(1)).scalar(), 1)
            assert not conn.invalidated

            # one more time
            self.engine.test_shutdown()
            _assert_invalidated(conn.execute, select(1))

            assert conn.invalidated
            conn.rollback()

            eq_(conn.execute(select(1)).scalar(), 1)
            assert not conn.invalidated

    def test_detach_invalidated(self):
        with self.engine.connect() as conn:
            conn.invalidate()
            with expect_raises_message(
                exc.InvalidRequestError,
                "Can't detach an invalidated Connection",
            ):
                conn.detach()

    def test_detach_closed(self):
        with self.engine.connect() as conn:
            pass
        with expect_raises_message(
            exc.ResourceClosedError, "This Connection is closed"
        ):
            conn.detach()

    @testing.requires.independent_connections
    def test_multiple_invalidate(self):
        c1 = self.engine.connect()
        c2 = self.engine.connect()

        eq_(c1.execute(select(1)).scalar(), 1)

        self.engine.test_shutdown()

        _assert_invalidated(c1.execute, select(1))

        p2 = self.engine.pool

        _assert_invalidated(c2.execute, select(1))

        # pool isn't replaced
        assert self.engine.pool is p2

    def test_ensure_is_disconnect_gets_connection(self):
        def is_disconnect(e, conn, cursor):
            # connection is still present
            assert conn.dbapi_connection is not None
            # the error usually occurs on connection.cursor(),
            # though MySQLdb we get a non-working cursor.
            # assert cursor is None

        self.engine.dialect.is_disconnect = is_disconnect

        with self.engine.connect() as conn:
            self.engine.test_shutdown()
            assert_raises(tsa.exc.DBAPIError, conn.execute, select(1))

            # aiosqlite is not able to run close() here without an
            # error.
            conn.invalidate()

    def test_rollback_on_invalid_plain(self):
        with self.engine.connect() as conn:
            trans = conn.begin()
            conn.invalidate()
            trans.rollback()

    @testing.requires.two_phase_transactions
    def test_rollback_on_invalid_twophase(self):
        with self.engine.connect() as conn:
            trans = conn.begin_twophase()
            conn.invalidate()
            trans.rollback()

    @testing.requires.savepoints
    def test_rollback_on_invalid_savepoint(self):
        with self.engine.connect() as conn:
            conn.begin()
            trans2 = conn.begin_nested()
            conn.invalidate()

            # this passes silently, as it will often be involved
            # in error catching schemes
            trans2.rollback()

            # still invalid though
            with expect_raises(exc.PendingRollbackError):
                conn.begin_nested()

    def test_no_begin_on_invalid(self):
        with self.engine.connect() as conn:
            conn.begin()
            conn.invalidate()

            with expect_raises(exc.PendingRollbackError):
                conn.commit()

    def test_invalidate_twice(self):
        with self.engine.connect() as conn:
            conn.invalidate()
            conn.invalidate()

    def test_explode_in_initializer(self):
        engine = engines.testing_engine()

        def broken_initialize(connection):
            connection.exec_driver_sql("select fake_stuff from _fake_table")

        engine.dialect.initialize = broken_initialize

        # raises a DBAPIError, not an AttributeError
        assert_raises(exc.DBAPIError, engine.connect)

    def test_explode_in_initializer_disconnect(self):
        engine = engines.testing_engine()

        def broken_initialize(connection):
            connection.exec_driver_sql("select fake_stuff from _fake_table")

        engine.dialect.initialize = broken_initialize

        def is_disconnect(e, conn, cursor):
            return True

        engine.dialect.is_disconnect = is_disconnect

        # invalidate() also doesn't screw up
        assert_raises(exc.DBAPIError, engine.connect)

    def test_null_pool(self):
        engine = engines.reconnecting_engine(
            options=dict(poolclass=pool.NullPool)
        )
        with engine.connect() as conn:
            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
            conn.rollback()
            eq_(conn.execute(select(1)).scalar(), 1)
            assert not conn.invalidated

    def test_close(self):
        with self.engine.connect() as conn:
            eq_(conn.execute(select(1)).scalar(), 1)
            assert not conn.closed

            self.engine.test_shutdown()

            _assert_invalidated(conn.execute, select(1))

        with self.engine.connect() as conn:
            eq_(conn.execute(select(1)).scalar(), 1)

    def test_with_transaction(self):
        with self.engine.connect() as conn:
            trans = conn.begin()
            assert trans.is_valid
            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 not trans.is_valid

            assert_raises_message(
                tsa.exc.PendingRollbackError,
                "Can't reconnect until invalid transaction is rolled back",
                conn.execute,
                select(1),
            )
            assert trans.is_active
            assert not trans.is_valid

            assert_raises_message(
                tsa.exc.PendingRollbackError,
                "Can't reconnect until invalid transaction is rolled back",
                trans.commit,
            )

            # becomes inactive
            assert not trans.is_active
            assert not trans.is_valid

            # still asks us to rollback
            assert_raises_message(
                tsa.exc.PendingRollbackError,
                "Can't reconnect until invalid transaction is rolled back",
                conn.execute,
                select(1),
            )

            # still asks us..
            assert_raises_message(
                tsa.exc.PendingRollbackError,
                "Can't reconnect until invalid transaction is rolled back",
                trans.commit,
            )

            # still...it's being consistent in what it is asking.
            assert_raises_message(
                tsa.exc.PendingRollbackError,
                "Can't reconnect until invalid transaction is rolled back",
                conn.execute,
                select(1),
            )

            #  OK!
            trans.rollback()
            assert not trans.is_active
            assert not trans.is_valid

            # conn still invalid but we can reconnect
            assert conn.invalidated
            eq_(conn.execute(select(1)).scalar(), 1)
            assert not conn.invalidated


class RecycleTest(fixtures.TestBase):
    __backend__ = True

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

        conn = engine.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.connect()
        eq_(conn.execute(select(1)).scalar(), 1)
        conn.close()


class PrePingRealTest(fixtures.TestBase):
    __backend__ = True

    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.dbapi_connection
        conn.close()

        engine.test_shutdown()
        engine.test_restart()

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

        with expect_raises(engine.dialect.dbapi.Error, check_context=False):
            curs = stale_connection.cursor()
            curs.execute("select 1")

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

        if isinstance(engine.pool, pool.QueuePool):
            eq_(engine.pool.checkedin(), 0)
            eq_(engine.pool._overflow, -5)

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

        if isinstance(engine.pool, pool.QueuePool):
            eq_(engine.pool.checkedin(), 1)
            eq_(engine.pool._overflow, -4)

        engine.test_shutdown(stop=True)

        assert_raises(exc.DBAPIError, engine.connect)

        if isinstance(engine.pool, pool.QueuePool):
            eq_(engine.pool.checkedin(), 1)
            eq_(engine.pool._overflow, -4)


class InvalidateDuringResultTest(fixtures.TestBase):
    __backend__ = True

    # test locks SQLite file databases due to unconsumed results
    __requires__ = ("ad_hoc_engines",)

    def setup_test(self):
        self.engine = engines.reconnecting_engine()
        self.meta = MetaData()
        table = Table(
            "sometable",
            self.meta,
            Column("id", Integer, primary_key=True),
            Column("name", String(50)),
        )

        with self.engine.begin() as conn:
            self.meta.create_all(conn)
            conn.execute(
                table.insert(),
                [{"id": i, "name": "row %d" % i} for i in range(1, 100)],
            )

    def teardown_test(self):
        with self.engine.begin() as conn:
            self.meta.drop_all(conn)
        self.engine.dispose()

    def test_invalidate_on_results(self):
        conn = self.engine.connect()
        result = conn.exec_driver_sql(
            "select * from sometable",
        )
        for x in range(20):
            result.fetchone()

        real_cursor = result.cursor
        self.engine.test_shutdown()

        def produce_side_effect():
            # will fail because connection was closed, with an exception
            # that should trigger disconnect routines
            real_cursor.execute("select * from sometable")

        result.cursor = Mock(
            fetchone=mock.Mock(side_effect=produce_side_effect)
        )
        try:
            _assert_invalidated(result.fetchone)
            assert conn.invalidated
        finally:
            conn.invalidate()


class ReconnectRecipeTest(fixtures.TestBase):
    """Test for the reconnect recipe given at doc/build/faq/connections.rst.

    Make sure the above document is updated if changes are made here.

    """

    # this recipe works on PostgreSQL also but only if the connection
    # is cut off from the server side, otherwise the connection.cursor()
    # method rightly fails because we explicitly closed the connection.
    # since we don't have a fixture
    # that can do this we currently rely on the MySQL drivers that allow
    # us to call cursor() even when the connection were closed.   In order
    # to get a real "cut the server off" kind of fixture we'd need to do
    # something in provisioning that seeks out the TCP connection at the
    # OS level and kills it.
    __only_on__ = ("+mysqldb", "+pymysql")

    def make_engine(self, engine):
        num_retries = 3
        retry_interval = 0.5

        def _run_with_retries(fn, context, cursor, statement, *arg, **kw):
            for retry in range(num_retries + 1):
                try:
                    fn(cursor, statement, context=context, *arg)
                except engine.dialect.dbapi.Error as raw_dbapi_err:
                    connection = context.root_connection
                    if engine.dialect.is_disconnect(
                        raw_dbapi_err, connection, cursor
                    ):
                        if retry > num_retries:
                            raise
                        engine.logger.error(
                            "disconnection error, retrying operation",
                            exc_info=True,
                        )
                        connection.invalidate()

                        connection.rollback()

                        time.sleep(retry_interval)
                        context.cursor = cursor = (
                            connection.connection.cursor()
                        )
                    else:
                        raise
                else:
                    return True

        e = engine.execution_options(isolation_level="AUTOCOMMIT")

        @event.listens_for(e, "do_execute_no_params")
        def do_execute_no_params(cursor, statement, context):
            return _run_with_retries(
                context.dialect.do_execute_no_params,
                context,
                cursor,
                statement,
            )

        @event.listens_for(e, "do_execute")
        def do_execute(cursor, statement, parameters, context):
            return _run_with_retries(
                context.dialect.do_execute,
                context,
                cursor,
                statement,
                parameters,
            )

        return e

    __backend__ = True

    def setup_test(self):
        self.engine = engines.reconnecting_engine()
        self.meta = MetaData()
        self.table = Table(
            "sometable",
            self.meta,
            Column("id", Integer, primary_key=True),
            Column("name", String(50)),
        )
        self.meta.create_all(self.engine)

    def teardown_test(self):
        self.meta.drop_all(self.engine)
        self.engine.dispose()

    def test_restart_on_execute_no_txn(self):
        engine = self.make_engine(self.engine)

        with engine.connect() as conn:
            eq_(conn.execute(select(1)).scalar(), 1)

            self.engine.test_shutdown()
            self.engine.test_restart()

            eq_(conn.execute(select(1)).scalar(), 1)

    def test_restart_on_execute_txn(self):
        engine = self.make_engine(self.engine)

        with engine.begin() as conn:
            eq_(conn.execute(select(1)).scalar(), 1)

            self.engine.test_shutdown()
            self.engine.test_restart()

            eq_(conn.execute(select(1)).scalar(), 1)

    def test_autocommits_txn(self):
        engine = self.make_engine(self.engine)

        with engine.begin() as conn:
            conn.execute(
                self.table.insert(),
                [
                    {"id": 1, "name": "some name 1"},
                    {"id": 2, "name": "some name 2"},
                    {"id": 3, "name": "some name 3"},
                ],
            )

            self.engine.test_shutdown()
            self.engine.test_restart()

            eq_(
                conn.execute(
                    select(self.table).order_by(self.table.c.id)
                ).fetchall(),
                [(1, "some name 1"), (2, "some name 2"), (3, "some name 3")],
            )

    def test_fail_on_executemany_txn(self):
        engine = self.make_engine(self.engine)

        with engine.begin() as conn:
            conn.execute(
                self.table.insert(),
                [
                    {"id": 1, "name": "some name 1"},
                    {"id": 2, "name": "some name 2"},
                    {"id": 3, "name": "some name 3"},
                ],
            )

            self.engine.test_shutdown()
            self.engine.test_restart()

            assert_raises(
                exc.DBAPIError,
                conn.execute,
                self.table.insert(),
                [
                    {"id": 4, "name": "some name 4"},
                    {"id": 5, "name": "some name 5"},
                    {"id": 6, "name": "some name 6"},
                ],
            )
            conn.rollback()