Add input validations for resources and n_workers (#702)

* create new pr, initial changes

* switch tests order

* add checks for both  and

* check if field are dict types

* edit the loop that checks dictionary fields

* fix failing tests

* explicitly set error messaege from  exception

dask_kubernetes/operator/kubecluster/kubecluster.py

@@ -230,6 +230,22 @@ def __init__(
         if isinstance(self.worker_command, str):
             self.worker_command = self.worker_command.split(" ")
 
+        if self.n_workers is not None and not isinstance(self.n_workers, int):
+            raise TypeError(f"n_workers must be an integer, got {type(self.n_workers)}")
+
+        try:
+            # Validate `resources` param is a dictionary whose
+            # keys must either be 'limits' or 'requests'
+            assert isinstance(self.resources, dict)
+
+            for field in self.resources:
+                if field in ("limits", "requests"):
+                    assert isinstance(self.resources[field], dict)
+                else:
+                    raise ValueError(f"Unknown field '{field}' in resources")
+        except TypeError as e:
+            raise TypeError(f"invalid '{type(resources)}' for resources type") from e
+
         name = name.format(
             user=getpass.getuser(), uuid=str(uuid.uuid4())[:10], **os.environ
         )

dask_kubernetes/operator/kubecluster/tests/test_kubecluster.py

@@ -153,3 +153,22 @@ def test_custom_spec(kopf_runner, docker_image):
         with KubeCluster(custom_cluster_spec=spec, n_workers=1) as cluster:
             with Client(cluster) as client:
                 assert client.submit(lambda x: x + 1, 10).result() == 11
+
+
+def test_for_noninteger_n_workers(kopf_runner):
+    with kopf_runner:
+        with pytest.raises(TypeError, match="n_workers must be an integer"):
+            KubeCluster(name="foo", n_workers="1")
+
+
+def test_typo_resource_limits(kopf_runner):
+    with kopf_runner:
+        with pytest.raises(ValueError):
+            KubeCluster(
+                name="foo",
+                resources={
+                    "limit": {  # <-- Typo, should be `limits`
+                        "CPU": "1",
+                    },
+                },
+            )