Revive PR kubernetes-client#1392: Implement delete_from_yaml (original by DiptoChakrabarty) and update to latest

Author: kocarba

kubernetes/e2e_test/test_utils.py

@@ -17,6 +17,7 @@
 from os import path
 
 import yaml
+import time
 
 from kubernetes import client, utils
 from kubernetes.client.rest import ApiException
@@ -36,6 +37,15 @@ def setUpClass(cls):
         body = client.V1Namespace(
             metadata=client.V1ObjectMeta(
                 name=cls.test_namespace))
+        
+        # Delete the namespace if it already exists
+        try:
+            core_v1.delete_namespace(name=cls.test_namespace)
+            time.sleep(10)  # Wait for the namespace to be deleted
+        except ApiException as e:
+            if e.status != 404:
+                raise
+
         core_v1.create_namespace(body=body)
 
     @classmethod
@@ -609,6 +619,253 @@ def test_create_from_list_in_multi_resource_yaml_namespaced(self):
             name="mock", namespace=self.test_namespace, body={})
 
 
+    def test_delete_namespace_from_yaml(self):
+        """
+        Should be able to delete a namespace
+        Create namespace from file first and ensure it is created
+        """
+        k8s_client = client.api_client.ApiClient(configuration=self.config)
+        utils.create_from_yaml(
+            k8s_client, self.path_prefix + "core-namespace.yaml")
+        core_api = client.CoreV1Api(k8s_client)
+        nmsp = core_api.read_namespace(name="development")
+        self.assertIsNotNone(nmsp)
+        """
+        Delete namespace from yaml
+        """
+        utils.delete_from_yaml(k8s_client, self.path_prefix + "core-namespace.yaml")
+        time.sleep(10)
+        namespace_status=False
+        try:
+            nmsp=core_api.read_namespace(name="development")
+            namespace_status=True
+        except Exception as e:
+            self.assertFalse(namespace_status)
+        self.assertFalse(namespace_status)
+
+    def test_delete_apps_deployment_from_yaml(self):
+        """
+        Should delete a deployment
+        First create deployment from file and ensure it is created
+        """
+        k8s_client = client.api_client.ApiClient(configuration=self.config)
+        utils.create_from_yaml(
+            k8s_client, self.path_prefix + "apps-deployment.yaml")
+        app_api = client.AppsV1Api(k8s_client)
+        dep = app_api.read_namespaced_deployment(name="nginx-app",
+                                                 namespace="default")
+        self.assertIsNotNone(dep)
+        """
+        Deployment should be created 
+        Now delete deployment using delete_from_yaml method
+        """
+        utils.delete_from_yaml(k8s_client, self.path_prefix + "apps-deployment.yaml")
+        deployment_status=False
+        time.sleep(10)
+        try:
+            response=app_api.read_namespaced_deployment(name="nginx-app",namespace="default")
+            deployment_status=True
+        except Exception as e:
+            self.assertFalse(deployment_status)
+
+        self.assertFalse(deployment_status)
+
+    def test_delete_service_from_yaml(self):
+        """
+        Should be able to delete a service
+        Create service from yaml first and ensure it is created
+        """
+        k8s_client = client.api_client.ApiClient(configuration=self.config)
+        utils.create_from_yaml(
+            k8s_client, self.path_prefix + "core-service.yaml")
+        core_api = client.CoreV1Api(k8s_client)
+        svc = core_api.read_namespaced_service(name="my-service",
+                                               namespace="default")
+        self.assertIsNotNone(svc)
+        """
+        Delete service from yaml
+        """
+        utils.delete_from_yaml(
+            k8s_client, self.path_prefix + "core-service.yaml")
+        service_status=False
+        time.sleep(10)
+        try:
+            response = core_api.read_namespaced_service(name="my-service",namespace="default")
+            service_status=True
+        except Exception as e:
+            self.assertFalse(service_status)
+        self.assertFalse(service_status)
+
+    def test_delete_pod_from_yaml(self):
+        """
+        Should be able to delete pod
+        Create pod from file first and ensure it is created
+        """
+        k8s_client = client.api_client.ApiClient(configuration=self.config)
+        utils.create_from_yaml(
+            k8s_client, self.path_prefix + "core-pod.yaml")
+        core_api = client.CoreV1Api(k8s_client)
+        pod = core_api.read_namespaced_pod(name="myapp-pod",
+                                           namespace="default")
+        self.assertIsNotNone(pod)
+        """
+        Delete pod using delete_from_yaml
+        """
+        utils.delete_from_yaml(
+            k8s_client, self.path_prefix + "core-pod.yaml")
+        time.sleep(10)
+        pod_status=False
+        try:
+            response = core_api.read_namespaced_pod(name="myapp-pod",
+                                                    namespace="default")
+            pod_status=True
+        except Exception as e:
+            self.assertFalse(pod_status)
+        self.assertFalse(pod_status)
+
+
+    def test_delete_rbac_role_from_yaml(self):
+        """
+        Should be able to delete rbac role
+        Create rbac role from file first and ensure it is created
+        """
+        k8s_client = client.api_client.ApiClient(configuration=self.config)
+        utils.create_from_yaml(
+            k8s_client, self.path_prefix + "rbac-role.yaml")
+        rbac_api = client.RbacAuthorizationV1Api(k8s_client)
+        rbac_role = rbac_api.read_namespaced_role(
+            name="pod-reader", namespace="default")
+        self.assertIsNotNone(rbac_role)
+        """
+        Delete rbac role from yaml
+        """
+        utils.delete_from_yaml(
+            k8s_client, self.path_prefix + "rbac-role.yaml")
+        rbac_role_status=False
+        time.sleep(10)
+        try:
+            response = rbac_api.read_namespaced_role(
+                name="pod-reader", namespace="default")
+            rbac_role_status=True
+        except Exception as e:
+            self.assertFalse(rbac_role_status)
+        self.assertFalse(rbac_role_status)
+
+    def test_delete_rbac_role_from_yaml_with_verbose_enabled(self):
+        """
+        Should delete a rbac role with verbose enabled
+        Create rbac role with verbose enabled and ensure it is created
+        """
+        k8s_client = client.api_client.ApiClient(configuration=self.config)
+        utils.create_from_yaml(
+            k8s_client, self.path_prefix + "rbac-role.yaml", verbose=True)
+        rbac_api = client.RbacAuthorizationV1Api(k8s_client)
+        rbac_role = rbac_api.read_namespaced_role(
+            name="pod-reader", namespace="default")
+        self.assertIsNotNone(rbac_role)
+        """
+        Delete the rbac role from yaml
+        """
+        utils.delete_from_yaml(
+            k8s_client, self.path_prefix + "rbac-role.yaml", verbose=True)
+
+        rbac_role_status=False
+        time.sleep(10)
+        try:
+            response=rbac_api.read_namespaced_role(
+                name="pod-reader", namespace="default")
+            rbac_role_status=True
+        except Exception as e:
+            self.assertFalse(rbac_role_status)
+        self.assertFalse(rbac_role_status)
+
+
+    # Deletion Tests for multi resource objects in yaml files
+
+    def test_delete_from_multi_resource_yaml(self):
+        """
+        Should be able to delete service and replication controller
+        from the multi resource yaml files
+        Create the resources first and ensure they exist
+        """
+        k8s_client = client.api_client.ApiClient(configuration=self.config)
+        utils.create_from_yaml(
+            k8s_client, self.path_prefix + "multi-resource.yaml")
+        core_api = client.CoreV1Api(k8s_client)
+        svc = core_api.read_namespaced_service(name="mock",
+                                               namespace="default")
+        self.assertIsNotNone(svc)
+        ctr = core_api.read_namespaced_replication_controller(
+            name="mock", namespace="default")
+        self.assertIsNotNone(ctr)
+        """
+        Delete service and replication controller using yaml file
+        """
+        utils.delete_from_yaml(
+            k8s_client, self.path_prefix + "multi-resource.yaml")
+        svc_status=False
+        replication_status=False
+        time.sleep(10)
+        try:
+            resp_svc= core_api.read_namespaced_service(name="mock",
+                                                       namespace="default")
+            svc_status=True
+            resp_repl= core_api.read_namespaced_replication_controller(
+                name="mock", namespace="default")
+            repl_status = True
+        except Exception as e:
+            self.assertFalse(svc_status)
+            self.assertFalse(replication_status)
+        self.assertFalse(svc_status)
+        self.assertFalse(replication_status)
+
+    def test_delete_from_list_in_multi_resource_yaml(self):
+        """
+        Should delete the items in PodList and the deployment in the yaml file
+        Create the items first and ensure they exist
+        """
+        k8s_client = client.api_client.ApiClient(configuration=self.config)
+        utils.create_from_yaml(
+            k8s_client, self.path_prefix + "multi-resource-with-list.yaml")
+        core_api = client.CoreV1Api(k8s_client)
+        app_api = client.AppsV1Api(k8s_client)
+        pod_0 = core_api.read_namespaced_pod(
+            name="mock-pod-0", namespace="default")
+        self.assertIsNotNone(pod_0)
+        pod_1 = core_api.read_namespaced_pod(
+            name="mock-pod-1", namespace="default")
+        self.assertIsNotNone(pod_1)
+        dep = app_api.read_namespaced_deployment(
+            name="mock", namespace="default")
+        self.assertIsNotNone(dep)
+        """
+        Delete the PodList and Deployment using the yaml file
+        """
+        utils.delete_from_yaml(
+            k8s_client, self.path_prefix + "multi-resource-with-list.yaml")
+        time.sleep(10)
+        pod0_status=False
+        pod1_status=False
+        deploy_status=False
+        try:
+            core_api.read_namespaced_pod(
+                name="mock-pod-0", namespace="default")
+            core_api.read_namespaced_pod(
+                name="mock-pod-1", namespace="default")
+            app_api.read_namespaced_deployment(
+                name="mock", namespace="default")
+            pod0_status=True
+            pod1_status=True
+            deploy_status=True
+        except Exception as e:
+            self.assertFalse(pod0_status)
+            self.assertFalse(pod1_status)
+            self.assertFalse(deploy_status)
+        self.assertFalse(pod0_status)
+        self.assertFalse(pod1_status)
+        self.assertFalse(deploy_status)
+
+
 class TestUtilsUnitTests(unittest.TestCase):
 
     def test_parse_quantity(self):

kubernetes/utils/__init__.py

@@ -16,5 +16,7 @@
 
 from .create_from_yaml import (FailToCreateError, create_from_dict,
                                create_from_yaml, create_from_directory)
+from .delete_from_yaml import (FailToDeleteError, delete_from_dict,
+                               delete_from_yaml, delete_from_directory)
 from .quantity import parse_quantity
 from. duration import parse_duration