Self-review

Author: crusaderky

distributed/cli/tests/test_dask_scheduler.py

@@ -38,9 +38,6 @@ async def f():
         response = requests.get("http://127.0.0.1:8787/status/")
         response.raise_for_status()
 
-    with pytest.raises(Exception):
-        response = requests.get("http://127.0.0.1:9786/info.json")
-
 
 def test_hostport(loop):
     with popen(["dask-scheduler", "--no-dashboard", "--host", "127.0.0.1:8978"]):
@@ -55,9 +52,8 @@ async def f():
 
 
 def test_no_dashboard(loop):
-    pytest.importorskip("bokeh")
-    with popen(["dask-scheduler", "--no-dashboard"]) as proc:
-        with Client("127.0.0.1:%d" % Scheduler.default_port, loop=loop) as c:
+    with popen(["dask-scheduler", "--no-dashboard"]):
+        with Client(f"127.0.0.1:{Scheduler.default_port}", loop=loop):
             response = requests.get("http://127.0.0.1:8787/status/")
             assert response.status_code == 404
 

distributed/cli/tests/test_dask_worker.py

@@ -256,15 +256,13 @@ def test_nprocs_auto(loop):
 
 def test_nprocs_expands_name(loop):
     with popen(["dask-scheduler", "--no-dashboard"]):
-        with popen(
-            ["dask-worker", "127.0.0.1:8786", "--nprocs", "2", "--name", "0"]
-        ) as worker:
+        with popen(["dask-worker", "127.0.0.1:8786", "--nprocs", "2", "--name", "0"]):
             with popen(["dask-worker", "127.0.0.1:8786", "--nprocs", "2"]):
                 with Client("tcp://127.0.0.1:8786", loop=loop) as c:
                     start = time()
                     while len(c.scheduler_info()["workers"]) < 4:
                         sleep(0.2)
-                        assert time() < start + 10
+                        assert time() < start + 30
 
                     info = c.scheduler_info()
                     names = [d["name"] for d in info["workers"].values()]

distributed/client.py

@@ -1467,10 +1467,10 @@ def close(self, timeout=no_default):
                 pc.stop()
 
         if self.asynchronous:
-            future = self._close()
+            coro = self._close()
             if timeout:
-                future = asyncio.wait_for(future, timeout)
-            return future
+                coro = asyncio.wait_for(coro, timeout)
+            return coro
 
         if self._start_arg is None:
             with suppress(AttributeError):

distributed/comm/tests/test_ws.py

@@ -95,33 +95,26 @@ async def test_expect_scheduler_ssl_when_sharing_server():
                     pass
 
 
-@gen_cluster(nthreads=[], scheduler_kwargs={"protocol": "ws://"})
-async def test_roundtrip(s, a, b):
-    async with Worker(s.address) as w:
-        async with Client(s.address, asynchronous=True) as c:
-            assert c.scheduler.address.startswith("ws://")
-            assert w.address.startswith("ws://")
-            future = c.submit(inc, 1)
-            result = await future
-            assert result == 2
-
-
-@gen_cluster(nthreads=[], scheduler_kwargs={"protocol": "ws://"})
-async def test_collections(s):
+@gen_cluster(client=True, scheduler_kwargs={"protocol": "ws://"})
+async def test_roundtrip(c, s, a, b):
+    assert a.address.startswith("ws://")
+    assert b.address.startswith("ws://")
+    assert c.scheduler.address.startswith("ws://")
+    assert await c.submit(inc, 1) == 2
+
+
+@gen_cluster(client=True, scheduler_kwargs={"protocol": "ws://"})
+async def test_collections(c, s, a, b):
     da = pytest.importorskip("dask.array")
-    async with Worker(s.address), Worker(s.address):
-        async with Client(s.address, asynchronous=True):
-            x = da.random.random((1000, 1000), chunks=(100, 100))
-            x = x + x.T
-            await x.persist()
+    x = da.random.random((1000, 1000), chunks=(100, 100))
+    x = x + x.T
+    await x.persist()
 
 
-@gen_cluster(nthreads=[], scheduler_kwargs={"protocol": "ws://"})
-async def test_large_transfer(s):
+@gen_cluster(client=True, scheduler_kwargs={"protocol": "ws://"})
+async def test_large_transfer(c, s, a, b):
     np = pytest.importorskip("numpy")
-    async with Worker(s.address, protocol="ws://"):
-        async with Client(s.address, asynchronous=True) as c:
-            await c.scatter(np.random.random(1_000_000))
+    await c.scatter(np.random.random(1_000_000))
 
 
 @pytest.mark.asyncio
@@ -181,7 +174,9 @@ async def test_connection_made_with_extra_conn_args(cleanup, protocol, security)
 @gen_test()
 async def test_quiet_close():
     with warnings.catch_warnings(record=True) as record:
-        async with Client(protocol="ws", processes=False, asynchronous=True) as c:
+        async with Client(
+            protocol="ws", processes=False, asynchronous=True, dashboard_address=":0"
+        ):
             pass
 
     # For some reason unrelated @coroutine warnings are showing up

distributed/deploy/tests/test_adaptive.py

@@ -7,7 +7,15 @@
 
 import dask
 
-from distributed import Adaptive, Client, LocalCluster, SpecCluster, Worker, wait
+from distributed import (
+    Adaptive,
+    Client,
+    LocalCluster,
+    Scheduler,
+    SpecCluster,
+    Worker,
+    wait,
+)
 from distributed.compatibility import WINDOWS
 from distributed.metrics import time
 from distributed.utils_test import async_wait_for, clean, gen_test, slowinc
@@ -18,7 +26,7 @@ def test_adaptive_local_cluster(loop):
         n_workers=0,
         scheduler_port=0,
         silence_logs=False,
-        dashboard_address=None,
+        dashboard_address=":0",
         loop=loop,
     ) as cluster:
         alc = cluster.adapt(interval="100 ms")
@@ -48,7 +56,7 @@ async def test_adaptive_local_cluster_multi_workers():
         scheduler_port=0,
         silence_logs=False,
         processes=False,
-        dashboard_address=None,
+        dashboard_address=":0",
         asynchronous=True,
     ) as cluster:
 
@@ -76,8 +84,8 @@ async def test_adaptive_local_cluster_multi_workers():
 
 
 @pytest.mark.xfail(reason="changed API")
-@pytest.mark.asyncio
-async def test_adaptive_scale_down_override(cleanup):
+@gen_test()
+async def test_adaptive_scale_down_override():
     class TestAdaptive(Adaptive):
         def __init__(self, *args, **kwargs):
             self.min_size = kwargs.pop("min_size", 0)
@@ -95,7 +103,9 @@ class TestCluster(LocalCluster):
         def scale_up(self, n, **kwargs):
             assert False
 
-    async with TestCluster(n_workers=10, processes=False, asynchronous=True) as cluster:
+    async with TestCluster(
+        n_workers=10, processes=False, asynchronous=True, dashboard_address=":0"
+    ) as cluster:
         ta = cluster.adapt(
             min_size=2, interval=0.1, scale_factor=2, Adaptive=TestAdaptive
         )
@@ -113,7 +123,7 @@ async def test_min_max():
         scheduler_port=0,
         silence_logs=False,
         processes=False,
-        dashboard_address=None,
+        dashboard_address=":0",
         asynchronous=True,
         threads_per_worker=1,
     )
@@ -169,7 +179,7 @@ async def test_avoid_churn(cleanup):
         processes=False,
         scheduler_port=0,
         silence_logs=False,
-        dashboard_address=None,
+        dashboard_address=":0",
     ) as cluster:
         async with Client(cluster, asynchronous=True) as client:
             adapt = cluster.adapt(interval="20 ms", wait_count=5)
@@ -194,7 +204,7 @@ async def test_adapt_quickly():
         processes=False,
         scheduler_port=0,
         silence_logs=False,
-        dashboard_address=None,
+        dashboard_address=":0",
     )
     client = await Client(cluster, asynchronous=True)
     adapt = cluster.adapt(interval="20 ms", wait_count=5, maximum=10)
@@ -247,7 +257,7 @@ async def test_adapt_down():
         processes=False,
         scheduler_port=0,
         silence_logs=False,
-        dashboard_address=None,
+        dashboard_address=":0",
     ) as cluster:
         async with Client(cluster, asynchronous=True) as client:
             cluster.adapt(interval="20ms", maximum=5)
@@ -274,7 +284,7 @@ async def test_no_more_workers_than_tasks():
             scheduler_port=0,
             silence_logs=False,
             processes=False,
-            dashboard_address=None,
+            dashboard_address=":0",
             asynchronous=True,
         ) as cluster:
             adapt = cluster.adapt(minimum=0, maximum=4, interval="10 ms")
@@ -287,7 +297,7 @@ def test_basic_no_loop(loop):
     with clean(threads=False):
         try:
             with LocalCluster(
-                0, scheduler_port=0, silence_logs=False, dashboard_address=None
+                0, scheduler_port=0, silence_logs=False, dashboard_address=":0"
             ) as cluster:
                 with Client(cluster) as client:
                     cluster.adapt()
@@ -311,7 +321,7 @@ async def test_target_duration():
             processes=False,
             scheduler_port=0,
             silence_logs=False,
-            dashboard_address=None,
+            dashboard_address=":0",
         ) as cluster:
             adapt = cluster.adapt(interval="20ms", minimum=2, target_duration="5s")
             async with Client(cluster, asynchronous=True) as client:
@@ -327,6 +337,7 @@ async def test_target_duration():
 async def test_worker_keys(cleanup):
     """Ensure that redefining adapt with a lower maximum removes workers"""
     async with SpecCluster(
+        scheduler={"cls": Scheduler, "options": {"port": 0, "dashboard_address": ":0"}},
         workers={
             "a-1": {"cls": Worker},
             "a-2": {"cls": Worker},
@@ -360,7 +371,7 @@ async def test_adapt_cores_memory(cleanup):
         scheduler_port=0,
         silence_logs=False,
         processes=False,
-        dashboard_address=None,
+        dashboard_address=":0",
         asynchronous=True,
     ) as cluster:
         adapt = cluster.adapt(minimum_cores=3, maximum_cores=9)
@@ -401,7 +412,7 @@ async def test_update_adaptive(cleanup):
         scheduler_port=0,
         silence_logs=False,
         processes=False,
-        dashboard_address=None,
+        dashboard_address=":0",
         asynchronous=True,
     ) as cluster:
         first = cluster.adapt(maxmimum=1)
@@ -415,7 +426,11 @@ async def test_update_adaptive(cleanup):
 async def test_adaptive_no_memory_limit(cleanup):
     """Make sure that adapt() does not keep creating workers when no memory limit is set."""
     async with LocalCluster(
-        n_workers=0, threads_per_worker=1, memory_limit=0, asynchronous=True
+        n_workers=0,
+        threads_per_worker=1,
+        memory_limit=0,
+        asynchronous=True,
+        dashboard_address=":0",
     ) as cluster:
         cluster.adapt(minimum=1, maximum=10, interval="1 ms")
         async with Client(cluster, asynchronous=True) as client:
@@ -447,7 +462,9 @@ async def _():
 
             return self.sync(_)
 
-    async with RequiresAwaitCluster(n_workers=0, asynchronous=True) as cluster:
+    async with RequiresAwaitCluster(
+        n_workers=0, asynchronous=True, dashboard_address=":0"
+    ) as cluster:
         async with Client(cluster, asynchronous=True) as client:
             futures = client.map(slowinc, range(5), delay=0.05)
             assert len(cluster.workers) == 0
@@ -465,7 +482,9 @@ async def test_adaptive_stopped():
     We should ensure that the adapt PC is actually stopped once the cluster
     stops.
     """
-    async with LocalCluster(n_workers=0, asynchronous=True) as cluster:
+    async with LocalCluster(
+        n_workers=0, asynchronous=True, dashboard_address=":0"
+    ) as cluster:
         instance = cluster.adapt(interval="10ms")
         assert instance.periodic_callback is not None