Use config_for_cluster_tests in sync fixture

distributed/tests/test_utils_test.py

@@ -1091,3 +1091,19 @@ async def test_ensure_no_new_clients():
         async with Client(s.address, asynchronous=True):
             with ensure_no_new_clients():
                 pass
+
+
+@pytest.mark.parametrize("nanny", [False, True])
+def test_cluster_uses_config_for_test(nanny):
+    key = "distributed.admin.tick.interval"
+    local = dask.config.get(key)
+
+    with cluster(nanny=nanny, nworkers=1) as (scheduler, workers):
+        with Client(scheduler["address"]) as client:
+            s_remote = client.run_on_scheduler(dask.config.get, key)
+            assert s_remote != local
+
+            w_remote = client.run(dask.config.get, key)
+            w_remote = next(iter(w_remote.values()))
+            assert w_remote != local
+            assert w_remote == s_remote

distributed/utils_test.py

@@ -398,7 +398,7 @@ async def _():
 
 
 def run_worker(q, scheduler_q, config, **kwargs):
-    with dask.config.set(config):
+    with config_for_cluster_tests(**config):
         from distributed import Worker
 
         reset_logger_locks()
@@ -422,7 +422,7 @@ async def _():
 
 @log_errors
 def run_nanny(q, scheduler_q, config, **kwargs):
-    with dask.config.set(config):
+    with config_for_cluster_tests(**config):
         scheduler_addr = scheduler_q.get()
 
         async def _():
@@ -617,7 +617,7 @@ def cluster(
 
     enable_proctitle_on_children()
 
-    with check_process_leak(check=True), check_instances(), config_for_cluster_tests():
+    with check_process_leak(check=True), check_instances():
         if nanny:
             _run_worker = run_nanny
         else: