KubeCluster.shutdown_on_close should be False when created from_name (#727)

* KubeCluster.shutdown_on_close should be False when created from_name

* Allow users to specify crete_mode and shutdown_on_close options

* Update cluster from_name test

* Forgot to add async arg

* async cm

* Update for kr8s PR

Author: Matt711

dask_kubernetes/operator/kubecluster/kubecluster.py

@@ -878,7 +878,12 @@ def from_name(cls, name, **kwargs):
         --------
         >>> cluster = KubeCluster.from_name(name="simple-cluster")
         """
-        return cls(name=name, create_mode=CreateMode.CONNECT_ONLY, **kwargs)
+        defaults = {"create_mode": CreateMode.CONNECT_ONLY, "shutdown_on_close": False}
+        kwargs = defaults | kwargs
+        return cls(
+            name=name,
+            **kwargs,
+        )
 
 
 def make_cluster_spec(

dask_kubernetes/operator/kubecluster/tests/test_kubecluster.py

@@ -4,6 +4,7 @@
 from dask.distributed import Client
 from distributed.utils import TimeoutError
 
+from dask_kubernetes.operator.objects import DaskCluster
 from dask_kubernetes.operator import KubeCluster, make_cluster_spec
 from dask_kubernetes.exceptions import SchedulerStartupError
 
@@ -95,13 +96,22 @@ def test_multiple_clusters_simultaneously_same_loop(kopf_runner, docker_image):
                 assert client2.submit(lambda x: x + 1, 10).result() == 11
 
 
-def test_cluster_from_name(kopf_runner, docker_image, ns):
+@pytest.mark.asyncio
+async def test_cluster_from_name(kopf_runner, docker_image, ns):
     with kopf_runner:
-        with KubeCluster(
-            name="abc", namespace=ns, image=docker_image, n_workers=1
+        async with KubeCluster(
+            name="abc",
+            namespace=ns,
+            image=docker_image,
+            n_workers=1,
+            asynchronous=True,
         ) as firstcluster:
-            with KubeCluster.from_name("abc", namespace=ns) as secondcluster:
+            async with KubeCluster.from_name(
+                "abc", namespace=ns, asynchronous=True
+            ) as secondcluster:
                 assert firstcluster == secondcluster
+            cluster = await DaskCluster.get("abc", namespace=ns)
+            assert cluster.status["phase"] == "Running"
 
 
 def test_cluster_scheduler_info_updated(kopf_runner, docker_image, ns):