Set and test the current client

Explicitly set the Dask resource's client as the current client using a
context manager, then test using the current client.

python_modules/libraries/dagster-dask/dagster_dask_tests/test_resources.py

@@ -13,6 +13,7 @@
 )
 import dask.dataframe as dd
 from dask.dataframe.utils import assert_eq
+from dask.distributed import Client
 
 from dagster_dask import DataFrame, dask_resource
 
@@ -25,10 +26,11 @@
     required_resource_keys={"dask"},
 )
 def scheduler_info_solid(context):
-    client = context.resources.dask.client
+    with context.resources.dask.client.as_current():
+        client = Client.current()
 
-    yield Output(client.scheduler_info(), "scheduler_info")
-    yield Output(client.nthreads(), "nthreads")
+        yield Output(client.scheduler_info(), "scheduler_info")
+        yield Output(client.nthreads(), "nthreads")
 
 
 @pipeline(mode_defs=[ModeDefinition(resource_defs={"dask": dask_resource})])