Rollup kubectl commands to a common library

Author: addyess

src/kube_galaxy/cmd/status.py

@@ -4,6 +4,8 @@
 import subprocess
 from pathlib import Path
 
+from kube_galaxy.pkg.utils.client import get_context, get_nodes
+from kube_galaxy.pkg.utils.errors import ClusterError
 from kube_galaxy.pkg.utils.logging import info, print_dict, section
 
 
@@ -49,37 +51,23 @@ def status() -> None:
     if shutil.which("kubectl"):
         info("")
         try:
-            result = subprocess.run(
-                ["kubectl", "get", "nodes"],
-                capture_output=True,
-                text=True,
-                check=False,
-            )
-            if result.returncode == 0 and result.stdout:
-                lines = result.stdout.strip().split("\n")
+            nodes_output = get_nodes()
+            if nodes_output:
+                lines = nodes_output.strip().split("\n")
                 info(f"Cluster Nodes: {len(lines) - 1}")  # Subtract header
                 for line in lines[1:]:  # Skip header
                     if line:
                         info(f"    {line}")
-        except Exception:
+        except (ClusterError, Exception):
             pass
 
     # Show active cluster
     if shutil.which("kubectl"):
         info("")
         try:
-            result = subprocess.run(
-                ["kubectl", "config", "current-context"],
-                capture_output=True,
-                text=True,
-                check=False,
-            )
-            if result.returncode == 0:
-                context = result.stdout.strip()
-                info(f"Active Cluster: {context}")
-            else:
-                info("Active Cluster: none")
-        except Exception:
+            context = get_context()
+            info(f"Active Cluster: {context}")
+        except (ClusterError, Exception):
             info("Active Cluster: error checking")
 
 

src/kube_galaxy/cmd/test.py

@@ -1,6 +1,5 @@
 """Test command handler."""
 
-import subprocess
 from pathlib import Path
 
 import typer
@@ -9,6 +8,8 @@
 from kube_galaxy.pkg.manifest.loader import load_manifest
 from kube_galaxy.pkg.manifest.validator import validate_manifest
 from kube_galaxy.pkg.testing.spread import collect_test_results, run_spread_tests
+from kube_galaxy.pkg.utils.client import get_context, verify_connectivity
+from kube_galaxy.pkg.utils.errors import ClusterError
 from kube_galaxy.pkg.utils.logging import error, exception, info, section, success, warning
 
 
@@ -22,25 +23,10 @@ def spread(manifest_path: str) -> None:
 
     try:
         # Check if kubectl can connect
-        result = subprocess.run(
-            ["kubectl", "cluster-info"],
-            capture_output=True,
-            text=True,
-            check=False,
-        )
-        if result.returncode != 0:
-            error("No Kubernetes cluster available. Please set up a cluster first.")
-            info("You can create a test cluster with: kube-galaxy setup")
-            raise typer.Exit(code=1)
+        verify_connectivity()
 
         # Get cluster context
-        result = subprocess.run(
-            ["kubectl", "config", "current-context"],
-            capture_output=True,
-            text=True,
-            check=True,
-        )
-        cluster_context = result.stdout.strip()
+        cluster_context = get_context()
         success(f"Connected to cluster: {cluster_context}")
 
         # Run spread tests from manifest
@@ -57,7 +43,7 @@ def spread(manifest_path: str) -> None:
 
         success("Spread tests completed")
 
-    except Exception as e:
+    except ClusterError as e:
         exception("Spread tests failed", e)
         raise typer.Exit(code=1) from e
 

src/kube_galaxy/pkg/components/_base.py

@@ -18,6 +18,7 @@
 from kube_galaxy.pkg.arch.detector import ArchInfo
 from kube_galaxy.pkg.literals import Commands, Permissions, SystemPaths, Timeouts
 from kube_galaxy.pkg.manifest.models import ComponentConfig, InstallMethod, Manifest
+from kube_galaxy.pkg.utils.client import apply_manifest
 from kube_galaxy.pkg.utils.components import (
     download_file,
     extract_archive,
@@ -243,8 +244,7 @@ def bootstrap_hook(self) -> None:
                     raise ComponentError(
                         f"{comp_name} manifest not downloaded. Run download hook first."
                     )
-                run(["kubectl", "apply", "-f", str(self.manifest_path)], check=True)
-                info(f"Applied manifest for {comp_name}")
+                apply_manifest(self.manifest_path)
 
         pass
 

src/kube_galaxy/pkg/components/kubeadm.py

@@ -14,6 +14,11 @@
 
 from kube_galaxy.pkg.components import ClusterComponentBase, register_component
 from kube_galaxy.pkg.literals import Commands, SystemPaths, URLs
+from kube_galaxy.pkg.utils.client import (
+    get_api_server_status,
+    verify_connectivity,
+    wait_for_nodes,
+)
 from kube_galaxy.pkg.utils.errors import ComponentError
 from kube_galaxy.pkg.utils.logging import info
 from kube_galaxy.pkg.utils.shell import run
@@ -184,26 +189,9 @@ def verify_hook(self) -> None:
 
         Checks cluster connectivity and waits for nodes/pods to be ready.
         """
-
-        # Check cluster info
-        run(["kubectl", "cluster-info"], check=True)
-
-        # Wait for nodes to be ready
-        run(
-            ["kubectl", "wait", "--for=condition=Ready", "nodes", "--all", "--timeout=300s"],
-            check=True,
-        )
-
-        # Wait for api-server to be ready
-        run(
-            [
-                "kubectl",
-                "get",
-                "--raw=/readyz",
-                "--request-timeout=300s",
-            ],
-            check=True,
-        )
+        verify_connectivity()
+        wait_for_nodes(timeout=300)
+        get_api_server_status(timeout=300)
 
     def stop_hook(self) -> None:
         """

src/kube_galaxy/pkg/testing/spread.py

@@ -18,6 +18,7 @@
     task_path_for_component,
     validate_component_test_structure,
 )
+from kube_galaxy.pkg.utils.client import create_namespace, delete_namespace, verify_connectivity
 from kube_galaxy.pkg.utils.errors import ClusterError
 from kube_galaxy.pkg.utils.logging import error, info, section, success, warning
 from kube_galaxy.pkg.utils.shell import ShellError, run
@@ -111,9 +112,7 @@ def run_spread_tests(
 def _verify_test_prerequisites() -> None:
     """Verify kubectl and spread are available."""
     try:
-        info("Verifying cluster connectivity...")
-        run(["kubectl", "cluster-info"], check=True, capture_output=True)
-        success("Connected to Kubernetes cluster")
+        verify_connectivity()
 
         # Check for spread
         info("Verifying spread test framework...")
@@ -148,23 +147,15 @@ def _create_test_namespace(component_name: str) -> str:
     namespace = f"kube-galaxy-test-{component_name.lower().replace('_', '-')}"
 
     try:
-        info(f"  Creating test namespace: {namespace}")
-
-        # Apply with labels
-        run(["kubectl", "create", "namespace", namespace], check=True)
-
-        # Label namespace
-        label = "app.kubernetes.io/managed-by=kube-galaxy"
-        run(
-            ["kubectl", "label", "namespace", namespace, label, f"component={component_name}"],
-            check=True,
-        )
-
-        success(f"Namespace created: {namespace}")
+        labels = {
+            "app.kubernetes.io/managed-by": "kube-galaxy",
+            "component": component_name,
+        }
+        create_namespace(namespace, labels)
         return namespace
 
-    except ShellError as exc:
-        raise ClusterError(f"Failed to create namespace {namespace}: {exc}") from exc
+    except ClusterError:
+        raise
 
 
 def _cleanup_test_namespace(namespace: str, timeout: int = 60) -> None:
@@ -179,22 +170,9 @@ def _cleanup_test_namespace(namespace: str, timeout: int = 60) -> None:
         ClusterError: If namespace deletion fails
     """
     try:
-        info(f"  Cleaning up namespace: {namespace}")
-
-        # Delete namespace
-        run(
-            ["kubectl", "delete", "namespace", namespace, "--timeout", f"{timeout}s"],
-            check=True,
-        )
-
-        success(f"Namespace deleted: {namespace}")
-
-    except ShellError as exc:
-        # Don't fail if namespace doesn't exist
-        if "not found" in str(exc):
-            warning(f"  Namespace {namespace} not found (may already be deleted)")
-        else:
-            raise ClusterError(f"Failed to delete namespace {namespace}: {exc}") from exc
+        delete_namespace(namespace, timeout)
+    except ClusterError:
+        raise
 
 
 def _generate_orchestration_spread_yaml(