async def METHOD_NAME(
    dask_gateway_stack_deployed_services,
    dask_gateway_entrypoint: str,
    faker: Faker,
    dask_gateway_password: str,
):
    gateway = dask_gateway.Gateway(
        address=dask_gateway_entrypoint,
        auth=dask_gateway.BasicAuth(faker.pystr(), dask_gateway_password),
    )
    with gateway.new_cluster() as cluster:
        _NUM_WORKERS = 2
        cluster.scale(
            _NUM_WORKERS
        )  # when returning we are in the process of creating the workers
        # now wait until we get the workers
        workers = None
        async for attempt in AsyncRetrying(
            reraise=True, wait=wait_fixed(1), stop=stop_after_delay(60)
        ):
            with attempt:
                print(
                    f"--> Waiting to have {_NUM_WORKERS} running,"
                    f" attempt {attempt.retry_state.attempt_number}...",
                )
                assert "workers" in cluster.scheduler_info
                assert len(cluster.scheduler_info["workers"]) == _NUM_WORKERS
                workers = deepcopy(cluster.scheduler_info["workers"])
                print(
                    f"!-- {_NUM_WORKERS} are running,"
                    f" [{json.dumps(attempt.retry_state.retry_object.statistics)}]",
                )
        # now check all this is stable
        _SECONDS_STABLE = 6
        for n in range(_SECONDS_STABLE):
            # NOTE: the scheduler_info gets auto-udpated by the dask-gateway internals
            assert workers == cluster.scheduler_info["workers"]
            await asyncio.sleep(1)
            print(f"!-- {_NUM_WORKERS} stable for {n} seconds")
        # send some work
        def square(x):
            return x**2
        def neg(x):
            return -x
        with cluster.get_client() as client:
            square_of_2 = client.submit(square, 2)
            assert square_of_2.result(timeout=10) == 4
            assert not square_of_2.exception(timeout=10)
            # now send some more stuff just for the fun
            A = client.map(square, range(10))
            B = client.map(neg, A)
            total = client.submit(sum, B)
            print("computation completed", total.result(timeout=120))