test_nanny_worker_port_range hangs on Windows

distributed/cli/tests/test_dask_scheduler.py

@@ -63,13 +63,13 @@ def test_no_dashboard(loop):
 def test_dashboard(loop):
     pytest.importorskip("bokeh")
 
-    with popen(["dask-scheduler"]) as proc:
-        for line in proc.stderr:
+    with popen(["dask-scheduler"], flush_output=False) as proc:
+        for line in proc.stdout:
             if b"dashboard at" in line:
                 dashboard_port = int(line.decode().split(":")[-1].strip())
                 break
         else:
-            raise Exception("dashboard not found")
+            assert False  # pragma: nocover
 
         with Client(f"127.0.0.1:{Scheduler.default_port}", loop=loop):
             pass
@@ -217,12 +217,17 @@ def test_scheduler_port_zero(loop):
 
 def test_dashboard_port_zero(loop):
     pytest.importorskip("bokeh")
-    with popen(["dask-scheduler", "--dashboard-address", ":0"]) as proc:
-        for line in proc.stderr:
+    with popen(
+        ["dask-scheduler", "--dashboard-address", ":0"],
+        flush_output=False,
+    ) as proc:
+        for line in proc.stdout:
             if b"dashboard at" in line:
                 dashboard_port = int(line.decode().split(":")[-1].strip())
                 assert dashboard_port != 0
                 break
+        else:
+            assert False  # pragma: nocover
 
 
 PRELOAD_TEXT = """

distributed/cli/tests/test_dask_spec.py

@@ -80,13 +80,17 @@ def test_errors():
             '{"foo": "bar"}',
             "--spec-file",
             "foo.yaml",
-        ]
+        ],
+        flush_output=False,
     ) as proc:
         line = proc.stdout.readline().decode()
         assert "exactly one" in line
         assert "--spec" in line and "--spec-file" in line
 
-    with popen([sys.executable, "-m", "distributed.cli.dask_spec"]) as proc:
+    with popen(
+        [sys.executable, "-m", "distributed.cli.dask_spec"],
+        flush_output=False,
+    ) as proc:
         line = proc.stdout.readline().decode()
         assert "exactly one" in line
         assert "--spec" in line and "--spec-file" in line

distributed/cli/tests/test_dask_ssh.py

@@ -18,23 +18,27 @@ def test_version_option():
     assert result.exit_code == 0
 
 
+@pytest.mark.slow
 def test_ssh_cli_nprocs_renamed_to_nworkers(loop):
-    n_workers = 2
     with popen(
-        ["dask-ssh", f"--nprocs={n_workers}", "--nohost", "localhost"]
-    ) as cluster:
+        ["dask-ssh", "--nprocs=2", "--nohost", "localhost"],
+        flush_output=False,
+    ) as proc:
         with Client("tcp://127.0.0.1:8786", timeout="15 seconds", loop=loop) as c:
-            c.wait_for_workers(n_workers, timeout="15 seconds")
+            c.wait_for_workers(2, timeout="15 seconds")
         # This interrupt is necessary for the cluster to place output into the stdout
         # and stderr pipes
-        cluster.send_signal(2)
-        _, stderr = cluster.communicate()
-
-    assert any(b"renamed to --nworkers" in l for l in stderr.splitlines())
+        proc.send_signal(2)
+        assert any(
+            b"renamed to --nworkers" in proc.stdout.readline() for _ in range(15)
+        )
 
 
 def test_ssh_cli_nworkers_with_nprocs_is_an_error():
-    with popen(["dask-ssh", "localhost", "--nprocs=2", "--nworkers=2"]) as c:
+    with popen(
+        ["dask-ssh", "localhost", "--nprocs=2", "--nworkers=2"],
+        flush_output=False,
+    ) as proc:
         assert any(
-            b"Both --nprocs and --nworkers" in c.stderr.readline() for i in range(15)
+            b"Both --nprocs and --nworkers" in proc.stdout.readline() for _ in range(15)
         )