10
10
import yaml
11
11
from tornado import gen
12
12
13
+ import dask .config
14
+
13
15
from distributed import Client , Nanny , Scheduler , Worker , config , default_client
14
16
from distributed .core import Server , rpc
15
17
from distributed .metrics import time
@@ -100,11 +102,8 @@ async def test_gen_cluster_parametrized_variadic_workers(c, s, *workers, foo):
100
102
)
101
103
async def test_gen_cluster_set_config_nanny (c , s , a , b ):
102
104
def assert_config ():
103
- import dask
104
-
105
105
assert dask .config .get ("distributed.comm.timeouts.connect" ) == "1s"
106
106
assert dask .config .get ("new.config.value" ) == "foo"
107
- return dask .config
108
107
109
108
await c .run (assert_config )
110
109
await c .run_on_scheduler (assert_config )
@@ -535,12 +534,11 @@ async def test_dump_cluster_state_unresponsive_local_worker(s, a, b, tmpdir):
535
534
@gen_cluster (
536
535
client = True ,
537
536
Worker = Nanny ,
538
- config = {"distributed.comm.timeouts.connect" : "200ms " },
537
+ config = {"distributed.comm.timeouts.connect" : "600ms " },
539
538
)
540
539
async def test_dump_cluster_unresponsive_remote_worker (c , s , a , b , tmpdir ):
541
- addr1 , addr2 = s .workers
542
540
clog_fut = asyncio .create_task (
543
- c .run (lambda dask_scheduler : dask_scheduler .stop (), workers = [addr1 ])
541
+ c .run (lambda dask_scheduler : dask_scheduler .stop (), workers = [a . worker_address ])
544
542
)
545
543
await asyncio .sleep (0.2 )
546
544
@@ -549,7 +547,9 @@ async def test_dump_cluster_unresponsive_remote_worker(c, s, a, b, tmpdir):
549
547
out = yaml .safe_load (fh )
550
548
551
549
assert out .keys () == {"scheduler" , "workers" , "versions" }
552
- assert isinstance (out ["workers" ][addr2 ], dict )
553
- assert out ["workers" ][addr1 ].startswith ("OSError('Timed out trying to connect to" )
550
+ assert isinstance (out ["workers" ][b .worker_address ], dict )
551
+ assert out ["workers" ][a .worker_address ].startswith (
552
+ "OSError('Timed out trying to connect to"
553
+ )
554
554
555
555
clog_fut .cancel ()
0 commit comments