Method create_pool

test/asynchronous/test_pooling.py:165–175

(self, pair=None, *args, **kwargs)

        await db.test.insert_many([{} for _ in range(10)])

    async def create_pool(self, pair=None, *args, **kwargs):
        if pair is None:
            pair = (await async_client_context.host, await async_client_context.port)
        # Start the pool with the correct ssl options.
        pool_options = async_client_context.client._topology_settings.pool_options
        kwargs["ssl_context"] = pool_options._ssl_context
        kwargs["tls_allow_invalid_hostnames"] = pool_options.tls_allow_invalid_hostnames
        kwargs["server_api"] = pool_options.server_api
        pool = Pool(pair, PoolOptions(*args, **kwargs))
        await pool.ready()
        return pool


class TestPooling(_TestPoolingBase):