Method test_reset_during_update_pool

test/test_client.py:1772–1812

(self)

    @client_context.require_sync
    def test_reset_during_update_pool(self):
        client = self.rs_or_single_client(minPoolSize=10)
        client.admin.command("ping")
        pool = get_pool(client)
        generation = pool.gen.get_overall()

        # Continuously reset the pool.
        class ResetPoolThread(threading.Thread):
            def __init__(self, pool):
                super().__init__()
                self.running = True
                self.pool = pool

            def stop(self):
                self.running = False

            def _run(self):
                while self.running:
                    exc = AutoReconnect("mock pool error")
                    ctx = _ErrorContext(exc, 0, pool.gen.get_overall(), False, None)
                    client._topology.handle_error(pool.address, ctx)
                    time.sleep(0.001)

            def run(self):
                self._run()

        t = ResetPoolThread(pool)
        t.start()

        # Ensure that update_pool completes without error even when the pool
        # is reset concurrently.
        try:
            while True:
                for _ in range(10):
                    client._topology.update_pool()
                if generation != pool.gen.get_overall():
                    break
        finally:
            t.stop()
            t.join()
        client.admin.command("ping")

    def test_background_connections_do_not_hold_locks(self):
        min_pool_size = 10