Disable multi-config support on Windows (#367)

kr8s/_auth.py

@@ -95,14 +95,16 @@ async def _load_kubeconfig(self) -> None:
         if isinstance(self._kubeconfig_path_or_dict, str) or isinstance(
             self._kubeconfig_path_or_dict, pathlib.Path
         ):
-            self._kubeconfig_path_or_dict = os.path.expanduser(
-                self._kubeconfig_path_or_dict
-            )
-            if not os.path.exists(self._kubeconfig_path_or_dict):
+            try:
+                if os.name != "nt":
+                    self.kubeconfig = await KubeConfigSet(
+                        *str(self._kubeconfig_path_or_dict).split(":")
+                    )
+                else:
+                    # Windows doesn't support multiple configs in a path
+                    self.kubeconfig = await KubeConfigSet(self._kubeconfig_path_or_dict)
+            except ValueError:
                 return
-            self.kubeconfig = await KubeConfigSet(
-                *self._kubeconfig_path_or_dict.split(":")
-            )
         else:
             self.kubeconfig = await KubeConfigSet(self._kubeconfig_path_or_dict)
         if self._use_context:

kr8s/_config.py

@@ -20,12 +20,14 @@
 
 class KubeConfigSet(object):
     def __init__(self, *paths_or_dicts: Union[List[str], List[Dict]]):
-        if isinstance(paths_or_dicts[0], str) or isinstance(
-            paths_or_dicts[0], pathlib.Path
-        ):
-            self._configs = [KubeConfig(path) for path in paths_or_dicts]
-        else:
-            self._configs = [KubeConfig(config) for config in paths_or_dicts]
+        self._configs = []
+        for path_or_dict in paths_or_dicts:
+            try:
+                self._configs.append(KubeConfig(path_or_dict))
+            except ValueError:
+                pass
+        if not self._configs:
+            raise ValueError("No valid kubeconfig provided")
 
     def __await__(self):
         async def f():
@@ -181,7 +183,9 @@ def __init__(self, path_or_config: Union[str, Dict]):
         self.path = None
         self._raw = None
         if isinstance(path_or_config, str) or isinstance(path_or_config, pathlib.Path):
-            self.path = path_or_config
+            self.path = pathlib.Path(path_or_config).expanduser()
+            if not self.path.exists():
+                raise ValueError(f"File {self.path} does not exist")
         else:
             self._raw = path_or_config
 

kr8s/tests/test_auth.py

@@ -100,6 +100,14 @@ async def test_kubeconfig(k8s_cluster):
     assert await api.whoami() == "kubernetes-admin"
 
 
+async def test_kubeconfig_multi(k8s_cluster):
+    api = await kr8s.asyncio.api(
+        kubeconfig=f"{k8s_cluster.kubeconfig_path}:{k8s_cluster.kubeconfig_path}"
+    )
+    assert await api.get("pods", namespace=kr8s.ALL)
+    assert await api.whoami() == "kubernetes-admin"
+
+
 async def test_kubeconfig_dict(k8s_cluster):
     config = yaml.safe_load(k8s_cluster.kubeconfig_path.read_text())
     assert isinstance(config, dict)

kr8s/tests/test_config.py

@@ -21,7 +21,7 @@ def temp_kubeconfig(k8s_cluster):
 
 async def test_load_kubeconfig(temp_kubeconfig):
     config = await KubeConfig(temp_kubeconfig)
-    assert config.path == temp_kubeconfig
+    assert str(config.path) == temp_kubeconfig
     assert config._raw
     assert "current-context" in config._raw
     assert config.current_context
@@ -34,7 +34,7 @@ async def test_load_kubeconfig(temp_kubeconfig):
 async def test_load_kubeconfig_set(temp_kubeconfig):
     configs = await KubeConfigSet(temp_kubeconfig)
     assert len(configs._configs) == 1
-    assert configs._configs[0].path == temp_kubeconfig
+    assert str(configs._configs[0].path) == temp_kubeconfig
     assert configs._configs[0]._raw
     assert "current-context" in configs._configs[0]._raw
     assert configs.current_context