diff --git a/distributed/tests/test_utils_test.py b/distributed/tests/test_utils_test.py index 80632b27c9..67d24ce4db 100755 --- a/distributed/tests/test_utils_test.py +++ b/distributed/tests/test_utils_test.py @@ -10,6 +10,8 @@ import yaml from tornado import gen +import dask.config + from distributed import Client, Nanny, Scheduler, Worker, config, default_client from distributed.core import Server, rpc from distributed.metrics import time @@ -100,11 +102,8 @@ async def test_gen_cluster_parametrized_variadic_workers(c, s, *workers, foo): ) async def test_gen_cluster_set_config_nanny(c, s, a, b): def assert_config(): - import dask - assert dask.config.get("distributed.comm.timeouts.connect") == "1s" assert dask.config.get("new.config.value") == "foo" - return dask.config await c.run(assert_config) await c.run_on_scheduler(assert_config) @@ -535,12 +534,11 @@ async def test_dump_cluster_state_unresponsive_local_worker(s, a, b, tmpdir): @gen_cluster( client=True, Worker=Nanny, - config={"distributed.comm.timeouts.connect": "200ms"}, + config={"distributed.comm.timeouts.connect": "600ms"}, ) async def test_dump_cluster_unresponsive_remote_worker(c, s, a, b, tmpdir): - addr1, addr2 = s.workers clog_fut = asyncio.create_task( - c.run(lambda dask_scheduler: dask_scheduler.stop(), workers=[addr1]) + c.run(lambda dask_scheduler: dask_scheduler.stop(), workers=[a.worker_address]) ) await asyncio.sleep(0.2) @@ -549,7 +547,9 @@ async def test_dump_cluster_unresponsive_remote_worker(c, s, a, b, tmpdir): out = yaml.safe_load(fh) assert out.keys() == {"scheduler", "workers", "versions"} - assert isinstance(out["workers"][addr2], dict) - assert out["workers"][addr1].startswith("OSError('Timed out trying to connect to") + assert isinstance(out["workers"][b.worker_address], dict) + assert out["workers"][a.worker_address].startswith( + "OSError('Timed out trying to connect to" + ) clog_fut.cancel()