use the loop fixture in even more tests

Refs #6163
follow up to #6642
adjustments for tests added since the first round, and addition
of explicit loop=None for cases where the LoopRunner itself
is being tested

distributed/cli/tests/test_dask_worker.py

@@ -703,7 +703,7 @@ async def test_signal_handling(c, s, nanny, sig):
 
 
 @pytest.mark.parametrize("nanny", ["--nanny", "--no-nanny"])
-def test_error_during_startup(monkeypatch, nanny):
+def test_error_during_startup(monkeypatch, nanny, loop):
     # see https://github.com/dask/distributed/issues/6320
     scheduler_port = open_port()
     scheduler_addr = f"tcp://127.0.0.1:{scheduler_port}"
@@ -716,7 +716,7 @@ def test_error_during_startup(monkeypatch, nanny):
             "--dashboard-address=:0",
         ],
     ):
-        with Client(scheduler_addr) as c:
+        with Client(scheduler_addr, loop=loop) as c:
             with popen(
                 [
                     "dask-worker",

distributed/deploy/tests/test_adaptive.py

@@ -293,7 +293,7 @@ def test_basic_no_loop(cleanup):
     loop = None
     try:
         with LocalCluster(
-            n_workers=0, silence_logs=False, dashboard_address=":0"
+            n_workers=0, silence_logs=False, dashboard_address=":0", loop=None
         ) as cluster:
             with Client(cluster) as client:
                 cluster.adapt()

distributed/tests/test_client.py

@@ -6854,6 +6854,7 @@ def f(x):
             processes=False,
             dashboard_address=":0",
             worker_dashboard_address=":0",
+            loop=None,
         ) as cluster2:
             with Client(cluster2) as c1:
                 c2 = get_client()

distributed/tests/test_scheduler.py

@@ -3600,12 +3600,12 @@ async def test_scheduler_close_fast_deprecated(s, w):
         await s.close(fast=True)
 
 
-def test_runspec_regression_sync():
+def test_runspec_regression_sync(loop):
     # https://github.com/dask/distributed/issues/6624
 
     da = pytest.importorskip("dask.array")
     np = pytest.importorskip("numpy")
-    with Client():
+    with Client(loop=loop):
         v = da.random.random((20, 20), chunks=(5, 5))
 
         overlapped = da.map_overlap(np.sum, v, depth=2, boundary="reflect")