Add ability to remove kubeconfig

Signed-off-by: michal.gubricky <[email protected]>

Author: michal-gubricky

Tests/kaas/plugin/interface.py

@@ -55,6 +55,13 @@ def create(self, name="scs-cluster", version=None, kubeconfig_filepath=None):
         """
         self.cluster_name = name
         self.cluster_version = version
+<<<<<<< HEAD
+=======
+        self.kubeconfig_filepath = kubeconfig_filepath
+
+        try:
+            self._create_cluster()
+>>>>>>> Add ability to remove kubeconfig
 
         self._create_cluster()  # TODO: maybe we do not need to use try exept here?
         # try:

Tests/kaas/plugin/plugin_cluster_stacks.py

@@ -193,10 +193,11 @@ def _create_cluster(self):
             raise RuntimeError(f"Error waiting for kubeadmcontrolplane to be available: {error}")
 
         # Step 11: Get kubeconfig of the workload k8s cluster
+        self.kubeconfig_cs_cluster_path = f"{self.kubeconfig_filepath}/{self.kubeconfig_cs_cluster_filename}"
         try:
-            kubeconfig_command = f"clusterctl get kubeconfig {self.cs_cluster_name} > ./{self.kubeconfig_cs_cluster_filename}"
+            kubeconfig_command = f"clusterctl get kubeconfig {self.cs_cluster_name} > {self.kubeconfig_cs_cluster_path}"
             subprocess.run(kubeconfig_command, shell=True, check=True)
-            print(f"Kubeconfig of the workload k8s cluster has been saved to {self.kubeconfig_cs_cluster_filename}.")
+            print(f"Kubeconfig of the workload k8s cluster has been saved to {self.kubeconfig_cs_cluster_path}.")
         except subprocess.CalledProcessError as error:
             raise RuntimeError(f"Error getting kubeconfig of the workload k8s cluster: {error}")
 
@@ -210,7 +211,7 @@ def _create_cluster(self):
         # Step 13: Wait for all system pods in the workload k8s cluster to become ready
         try:
             wait_pods_command = (
-                f"kubectl wait -n kube-system --for=condition=Ready --timeout=300s pod --all --kubeconfig {self.kubeconfig_cs_cluster_filename}"
+                f"kubectl wait -n kube-system --for=condition=Ready --timeout=300s pod --all --kubeconfig {self.kubeconfig_cs_cluster_path}"
             )
             subprocess.run(wait_pods_command, shell=True, check=True)
             print("All system pods in the workload Kubernetes cluster are ready.")
@@ -220,29 +221,41 @@ def _create_cluster(self):
     def _delete_cluster(self):
         # Step 1: Check if the cluster exists and if so delete it
         try:
-            check_cluster_command = f"kubectl get cluster {self.cs_cluster_name} --kubeconfig {self.kubeconfig}"
+            check_cluster_command = f"kubectl get cluster {self.cs_cluster_name} --kubeconfig kubeconfig"
             result = subprocess.run(check_cluster_command, shell=True, check=True, capture_output=True, text=True)
 
             if result.returncode == 0:
                 print(f"Cluster {self.cs_cluster_name} exists. Proceeding with deletion.")
                 # Step 2: Delete the cluster with a timeout
                 delete_cluster_command = (
-                    f"kubectl delete cluster {self.cs_cluster_name} --kubeconfig {self.kubeconfig}"
-                    f"--timeout=300s"
+                    f"kubectl delete cluster {self.cs_cluster_name} --kubeconfig kubeconfig --timeout=300s"
                 )
-                subprocess.run(delete_cluster_command, shell=True, check=True)
-                print(f"Cluster {self.cs_cluster_name} deleted successfully.")
+                try:
+                    subprocess.run(delete_cluster_command, shell=True, check=True, timeout=300)
+                    print(f"Cluster {self.cs_cluster_name} deleted successfully.")
+                except subprocess.TimeoutExpired:
+                    raise TimeoutError(f"Timed out while deleting the cluster {self.cs_cluster_name}.")
             else:
                 print(f"No cluster named {self.cs_cluster_name} found. Nothing to delete.")
 
         except subprocess.CalledProcessError as error:
             if "NotFound" in error.stderr:
                 print(f"Cluster {self.cs_cluster_name} not found. Skipping deletion.")
             else:
-                print(f"Error checking for cluster existence: {error}")
-                raise
-        # Todo: Maybe it is worth to remove the kubeconfig file
-        # Step 2: Delete the Kind cluster
+                raise RuntimeError(f"Error checking for cluster existence: {error}")
+
+        # Step 3: Remove the kubeconfig file if it exists
+        kubeconfig_path = self.kubeconfig_cs_cluster_path
+        if os.path.exists(kubeconfig_path):
+            try:
+                os.remove(kubeconfig_path)
+                print(f"Kubeconfig file at {kubeconfig_path} has been successfully removed.")
+            except OSError as e:
+                print(f"Error while trying to remove kubeconfig file: {e}")
+        else:
+            print(f"Kubeconfig file at {kubeconfig_path} does not exist or has already been removed.")
+
+        # Step 4: Delete the Kind cluster
         try:
             self.cluster = KindCluster(self.cluster_name)
             self.cluster.delete()