Method test_recycle_pool_no_race

test/engine/test_pool.py:1816–1888

(self)

    @testing.requires.timing_intensive
    def test_recycle_pool_no_race(self):
        def slow_close():
            slow_closing_connection._slow_close()
            time.sleep(0.5)

        slow_closing_connection = Mock()
        slow_closing_connection.connect.return_value.close = slow_close

        class Error(Exception):
            pass

        dialect = Mock()
        dialect.is_disconnect = lambda *arg, **kw: True
        dialect.dispatch = Mock(handle_error=[])
        dialect.dbapi.Error = dialect.loaded_dbapi.Error = Error

        pools = []

        class TrackQueuePool(pool.QueuePool):
            def __init__(self, *arg, **kw):
                pools.append(self)
                super().__init__(*arg, **kw)

        def creator():
            return slow_closing_connection.connect()

        p1 = TrackQueuePool(creator=creator, pool_size=20)

        from sqlalchemy import create_engine

        eng = create_engine(testing.db.url, pool=p1, _initialize=False)
        eng.dialect = dialect

        # 15 total connections
        conns = [eng.connect() for i in range(15)]

        # return 8 back to the pool
        for conn in conns[3:10]:
            conn.close()

        def attempt(conn):
            time.sleep(random.random())
            try:
                conn._handle_dbapi_exception(
                    Error(), "statement", {}, Mock(), Mock()
                )
            except tsa.exc.DBAPIError:
                pass

        # run an error + invalidate operation on the remaining 7 open
        # connections
        threads = []
        for conn in conns:
            t = threading.Thread(target=attempt, args=(conn,))
            t.start()
            threads.append(t)

        for t in threads: