def _test_overflow(self, thread_count, max_overflow):
        gc_collect()

        dbapi = MockDBAPI()
        mutex = threading.Lock()

        def creator():
            time.sleep(.05)
            with mutex:
                return dbapi.connect()

        p = pool.QueuePool(creator=creator,
                           pool_size=3, timeout=2,
                           max_overflow=max_overflow)
        peaks = []

        def whammy():
            for i in range(10):
                try:
                    con = p.connect()
                    time.sleep(.005)
                    peaks.append(p.overflow())
                    con.close()
                    del con
                except tsa.exc.TimeoutError:
                    pass
        threads = []
        for i in range(thread_count):
            th = threading.Thread(target=whammy)
            th.start()
            threads.append(th)
        for th in threads:
            th.join(join_timeout)

        self.assert_(max_overflow <= max(peaks))

        lazy_gc()
        assert not pool._refs
dataset/ETHPy150Open zzzeek/sqlalchemy/test/engine/test_pool.py/QueuePoolTest._test_overflow/swapped operands of `<=`