fix(providers): support per-run isolated KUBECONFIG in vcluster credential resolution

Author: isadominguez314

devops_bench/providers/vcluster.py

@@ -134,6 +134,15 @@ def _is_allowlisted_context(context_name: str, kubeconfig_path: str) -> bool:
 
 def _default_kubeconfig_path(cluster_name: str) -> str:
     """Determine default target kubeconfig path for a virtual cluster."""
+    env_kube = get_env("KUBECONFIG")
+    if env_kube:
+        try:
+            resolved_env = str(Path(env_kube).expanduser().resolve())
+            default_kube = str(Path("~/.kube/config").expanduser().resolve())
+            if resolved_env != default_kube:
+                return resolved_env
+        except Exception:
+            pass
     return str(Path(tempfile.gettempdir()) / f"vcluster-{cluster_name}-kubeconfig.yaml")
 
 
@@ -223,6 +232,7 @@ def ensure_cluster_credentials(
                 ) from exc
             f.write(kubeconfig_yaml)
         _log.info("Wrote virtual cluster kubeconfig to %s (mode 0600)", resolved_target)
+        os.environ["KUBECONFIG"] = str(resolved_target)
 
         return ClusterInfo.from_dict(
             {

tests/unit/providers/test_vcluster_provider.py

@@ -17,6 +17,7 @@
 from __future__ import annotations
 
 import json
+import os
 import re
 import tempfile
 from pathlib import Path
@@ -143,6 +144,18 @@ def test_vcluster_resolve_variables_parallel_kubeconfig(
     run_kubeconfig = str(tmp_path / "run-kubeconfig")
     monkeypatch.setenv("KUBECONFIG", run_kubeconfig)
 
+    variables = VClusterProvider().resolve_variables(ctx, {})
+    assert variables["kubeconfig_path"] == str(Path(run_kubeconfig).resolve())
+
+
+def test_vcluster_resolve_variables_default_kubeconfig_falls_back_to_temp(
+    ctx: ResolveContext,
+    fake_kubeconfig: str,
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    monkeypatch.setenv("HOST_KUBECONFIG", fake_kubeconfig)
+    monkeypatch.setenv("KUBECONFIG", "~/.kube/config")
+
     variables = VClusterProvider().resolve_variables(ctx, {})
     assert variables["kubeconfig_path"] == str(
         Path(tempfile.gettempdir()) / "vcluster-test-cluster-kubeconfig.yaml"
@@ -169,6 +182,7 @@ def test_vcluster_ensure_cluster_credentials(tmp_path: Path) -> None:
     assert info.location == "local"
     assert info.project == "local-vcluster"
     assert info.kubeconfig_path == str(target_path.resolve())
+    assert os.environ.get("KUBECONFIG") == str(target_path.resolve())
 
     assert target_path.exists()
     assert "https://127.0.0.1:8443" in target_path.read_text(encoding="utf-8")