diff --git a/kubernetes/e2e_test/test_utils.py b/kubernetes/e2e_test/test_utils.py
index 646081b8c..da0435638 100644
--- a/kubernetes/e2e_test/test_utils.py
+++ b/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
@@ -51,15 +61,17 @@ def test_create_apps_deployment_from_yaml(self):
Should be able to create an apps/v1 deployment.
"""
k8s_client = client.api_client.ApiClient(configuration=self.config)
- utils.create_from_yaml(
+ utils.process_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)
self.assertEqual("nginx-app", dep.metadata.name)
- self.assertEqual("nginx:1.15.4", dep.spec.template.spec.containers[0].image)
- self.assertEqual(80, dep.spec.template.spec.containers[0].ports[0].container_port)
+ self.assertEqual(
+ "nginx:1.15.4", dep.spec.template.spec.containers[0].image)
+ self.assertEqual(
+ 80, dep.spec.template.spec.containers[0].ports[0].container_port)
self.assertEqual("nginx", dep.spec.template.spec.containers[0].name)
self.assertEqual("nginx", dep.spec.template.metadata.labels["app"])
self.assertEqual(3, dep.spec.replicas)
@@ -79,28 +91,34 @@ def test_create_apps_deployment_from_yaml_with_apply_is_idempotent(self):
"""
k8s_client = client.api_client.ApiClient(configuration=self.config)
try:
- utils.create_from_yaml(
+ utils.process_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")
+ namespace="default")
self.assertIsNotNone(dep)
self.assertEqual("nginx-app", dep.metadata.name)
- self.assertEqual("nginx:1.15.4", dep.spec.template.spec.containers[0].image)
- self.assertEqual(80, dep.spec.template.spec.containers[0].ports[0].container_port)
- self.assertEqual("nginx", dep.spec.template.spec.containers[0].name)
+ self.assertEqual(
+ "nginx:1.15.4", dep.spec.template.spec.containers[0].image)
+ self.assertEqual(
+ 80, dep.spec.template.spec.containers[0].ports[0].container_port)
+ self.assertEqual(
+ "nginx", dep.spec.template.spec.containers[0].name)
self.assertEqual("nginx", dep.spec.template.metadata.labels["app"])
self.assertEqual(3, dep.spec.replicas)
- utils.create_from_yaml(
+ utils.process_from_yaml(
k8s_client, self.path_prefix + "apps-deployment.yaml", apply=True)
dep = app_api.read_namespaced_deployment(name="nginx-app",
- namespace="default")
+ namespace="default")
self.assertIsNotNone(dep)
self.assertEqual("nginx-app", dep.metadata.name)
- self.assertEqual("nginx:1.15.4", dep.spec.template.spec.containers[0].image)
- self.assertEqual(80, dep.spec.template.spec.containers[0].ports[0].container_port)
- self.assertEqual("nginx", dep.spec.template.spec.containers[0].name)
+ self.assertEqual(
+ "nginx:1.15.4", dep.spec.template.spec.containers[0].image)
+ self.assertEqual(
+ 80, dep.spec.template.spec.containers[0].ports[0].container_port)
+ self.assertEqual(
+ "nginx", dep.spec.template.spec.containers[0].name)
self.assertEqual("nginx", dep.spec.template.metadata.labels["app"])
self.assertEqual(3, dep.spec.replicas)
except Exception as e:
@@ -123,7 +141,7 @@ def test_create_apps_deployment_from_yaml_object(self):
_path = self.path_prefix + "apps-deployment.yaml"
with open(path.abspath(_path)) as f:
yaml_objects = yaml.safe_load_all(f)
- utils.create_from_yaml(
+ utils.process_from_yaml(
k8s_client,
yaml_objects=yaml_objects,
)
@@ -132,8 +150,10 @@ def test_create_apps_deployment_from_yaml_object(self):
namespace="default")
self.assertIsNotNone(dep)
self.assertEqual("nginx-app", dep.metadata.name)
- self.assertEqual("nginx:1.15.4", dep.spec.template.spec.containers[0].image)
- self.assertEqual(80, dep.spec.template.spec.containers[0].ports[0].container_port)
+ self.assertEqual(
+ "nginx:1.15.4", dep.spec.template.spec.containers[0].image)
+ self.assertEqual(
+ 80, dep.spec.template.spec.containers[0].ports[0].container_port)
self.assertEqual("nginx", dep.spec.template.spec.containers[0].name)
self.assertEqual("nginx", dep.spec.template.metadata.labels["app"])
self.assertEqual(3, dep.spec.replicas)
@@ -154,15 +174,17 @@ def test_create_apps_deployment_from_yaml_obj(self):
yml_obj["metadata"]["name"] = "nginx-app-3"
- utils.create_from_dict(k8s_client, yml_obj)
+ utils.process_from_dict(k8s_client, yml_obj)
app_api = client.AppsV1Api(k8s_client)
dep = app_api.read_namespaced_deployment(name="nginx-app-3",
namespace="default")
self.assertIsNotNone(dep)
self.assertEqual("nginx-app-3", dep.metadata.name)
- self.assertEqual("nginx:1.15.4", dep.spec.template.spec.containers[0].image)
- self.assertEqual(80, dep.spec.template.spec.containers[0].ports[0].container_port)
+ self.assertEqual(
+ "nginx:1.15.4", dep.spec.template.spec.containers[0].image)
+ self.assertEqual(
+ 80, dep.spec.template.spec.containers[0].ports[0].container_port)
self.assertEqual("nginx", dep.spec.template.spec.containers[0].name)
self.assertEqual("nginx", dep.spec.template.metadata.labels["app"])
self.assertEqual(3, dep.spec.replicas)
@@ -176,7 +198,7 @@ def test_create_pod_from_yaml(self):
Should be able to create a pod.
"""
k8s_client = client.api_client.ApiClient(configuration=self.config)
- utils.create_from_yaml(
+ utils.process_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",
@@ -195,7 +217,7 @@ def test_create_service_from_yaml(self):
Should be able to create a service.
"""
k8s_client = client.api_client.ApiClient(configuration=self.config)
- utils.create_from_yaml(
+ utils.process_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",
@@ -213,7 +235,7 @@ def test_create_namespace_from_yaml(self):
Should be able to create a namespace.
"""
k8s_client = client.api_client.ApiClient(configuration=self.config)
- utils.create_from_yaml(
+ utils.process_from_yaml(
k8s_client, self.path_prefix + "core-namespace.yaml")
core_api = client.CoreV1Api(k8s_client)
nmsp = core_api.read_namespace(name="development")
@@ -228,7 +250,7 @@ def test_create_rbac_role_from_yaml(self):
Should be able to create a rbac role.
"""
k8s_client = client.api_client.ApiClient(configuration=self.config)
- utils.create_from_yaml(
+ utils.process_from_yaml(
k8s_client, self.path_prefix + "rbac-role.yaml")
rbac_api = client.RbacAuthorizationV1Api(k8s_client)
rbac_role = rbac_api.read_namespaced_role(
@@ -245,7 +267,7 @@ def test_create_rbac_role_from_yaml_with_verbose_enabled(self):
Should be able to create a rbac role with verbose enabled.
"""
k8s_client = client.api_client.ApiClient(configuration=self.config)
- utils.create_from_yaml(
+ utils.process_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(
@@ -263,9 +285,9 @@ def test_create_deployment_non_default_namespace_from_yaml(self):
and then create a deployment in the just-created namespace.
"""
k8s_client = client.ApiClient(configuration=self.config)
- utils.create_from_yaml(
+ utils.process_from_yaml(
k8s_client, self.path_prefix + "dep-namespace.yaml")
- utils.create_from_yaml(
+ utils.process_from_yaml(
k8s_client, self.path_prefix + "dep-deployment.yaml")
core_api = client.CoreV1Api(k8s_client)
ext_api = client.AppsV1Api(k8s_client)
@@ -288,7 +310,7 @@ def test_create_apiservice_from_yaml_with_conflict(self):
fail due to conflict.
"""
k8s_client = client.api_client.ApiClient(configuration=self.config)
- utils.create_from_yaml(
+ utils.process_from_yaml(
k8s_client, self.path_prefix + "api-service.yaml")
reg_api = client.ApiregistrationV1Api(k8s_client)
svc = reg_api.read_api_service(
@@ -298,8 +320,8 @@ def test_create_apiservice_from_yaml_with_conflict(self):
self.assertEqual("wardle.k8s.io", svc.spec.group)
self.assertEqual("v1alpha1", svc.spec.version)
- with self.assertRaises(utils.FailToCreateError) as cm:
- utils.create_from_yaml(
+ with self.assertRaises(utils.FailToProcessError) as cm:
+ utils.process_from_yaml(
k8s_client, "kubernetes/e2e_test/test_yaml/api-service.yaml")
exp_error = ('Error from server (Conflict): '
'{"kind":"Status","apiVersion":"v1","metadata":{},'
@@ -323,7 +345,7 @@ def test_create_general_list_from_yaml(self):
from a kind: List yaml file
"""
k8s_client = client.api_client.ApiClient(configuration=self.config)
- utils.create_from_yaml(
+ utils.process_from_yaml(
k8s_client, self.path_prefix + "list.yaml")
core_api = client.CoreV1Api(k8s_client)
ext_api = client.AppsV1Api(k8s_client)
@@ -337,7 +359,8 @@ def test_create_general_list_from_yaml(self):
namespace="default")
self.assertIsNotNone(dep)
self.assertEqual("list-deployment-test", dep.metadata.name)
- self.assertEqual("nginx:1.15.4", dep.spec.template.spec.containers[0].image)
+ self.assertEqual(
+ "nginx:1.15.4", dep.spec.template.spec.containers[0].image)
self.assertEqual(1, dep.spec.replicas)
core_api.delete_namespaced_service(name="list-service-test",
@@ -351,7 +374,7 @@ def test_create_namespace_list_from_yaml(self):
from a kind: NamespaceList yaml file
"""
k8s_client = client.api_client.ApiClient(configuration=self.config)
- utils.create_from_yaml(
+ utils.process_from_yaml(
k8s_client, self.path_prefix + "namespace-list.yaml")
core_api = client.CoreV1Api(k8s_client)
nmsp_1 = core_api.read_namespace(name="mock-1")
@@ -373,8 +396,8 @@ def test_create_implicit_service_list_from_yaml_with_conflict(self):
json file that implicitly indicates the kind of individual objects
"""
k8s_client = client.api_client.ApiClient(configuration=self.config)
- with self.assertRaises(utils.FailToCreateError):
- utils.create_from_yaml(
+ with self.assertRaises(utils.FailToProcessError):
+ utils.process_from_yaml(
k8s_client, self.path_prefix + "implicit-svclist.json")
core_api = client.CoreV1Api(k8s_client)
svc_3 = core_api.read_namespaced_service(name="mock-3",
@@ -402,7 +425,7 @@ def test_create_multi_resource_from_directory(self):
from a directory
"""
k8s_client = client.api_client.ApiClient(configuration=self.config)
- utils.create_from_directory(
+ utils.process_from_directory(
k8s_client, self.path_prefix + "multi-resource/")
core_api = client.CoreV1Api(k8s_client)
svc = core_api.read_namespaced_service(name="mock",
@@ -419,8 +442,10 @@ def test_create_multi_resource_from_directory(self):
self.assertEqual("mock", ctr.spec.template.metadata.labels["app"])
self.assertEqual("mock", ctr.spec.selector["app"])
self.assertEqual(1, ctr.spec.replicas)
- self.assertEqual("k8s.gcr.io/pause:2.0", ctr.spec.template.spec.containers[0].image)
- self.assertEqual("mock-container", ctr.spec.template.spec.containers[0].name)
+ self.assertEqual("k8s.gcr.io/pause:2.0",
+ ctr.spec.template.spec.containers[0].image)
+ self.assertEqual("mock-container",
+ ctr.spec.template.spec.containers[0].name)
core_api.delete_namespaced_replication_controller(
name="mock", namespace="default", propagation_policy="Background")
@@ -435,7 +460,7 @@ def test_create_from_multi_resource_yaml(self):
from a multi-resource yaml file
"""
k8s_client = client.api_client.ApiClient(configuration=self.config)
- utils.create_from_yaml(
+ utils.process_from_yaml(
k8s_client, self.path_prefix + "multi-resource.yaml")
core_api = client.CoreV1Api(k8s_client)
svc = core_api.read_namespaced_service(name="mock",
@@ -452,8 +477,10 @@ def test_create_from_multi_resource_yaml(self):
self.assertEqual("mock", ctr.spec.template.metadata.labels["app"])
self.assertEqual("mock", ctr.spec.selector["app"])
self.assertEqual(1, ctr.spec.replicas)
- self.assertEqual("k8s.gcr.io/pause:2.0", ctr.spec.template.spec.containers[0].image)
- self.assertEqual("mock-container", ctr.spec.template.spec.containers[0].name)
+ self.assertEqual("k8s.gcr.io/pause:2.0",
+ ctr.spec.template.spec.containers[0].image)
+ self.assertEqual("mock-container",
+ ctr.spec.template.spec.containers[0].name)
core_api.delete_namespaced_replication_controller(
name="mock", namespace="default", propagation_policy="Background")
@@ -466,7 +493,7 @@ def test_create_from_list_in_multi_resource_yaml(self):
specified in the multi-resource file
"""
k8s_client = client.api_client.ApiClient(configuration=self.config)
- utils.create_from_yaml(
+ utils.process_from_yaml(
k8s_client, self.path_prefix + "multi-resource-with-list.yaml")
core_api = client.CoreV1Api(k8s_client)
app_api = client.AppsV1Api(k8s_client)
@@ -508,7 +535,7 @@ def test_create_from_multi_resource_yaml_with_conflict(self):
Should raise an exception for failure to create the same service twice.
"""
k8s_client = client.api_client.ApiClient(configuration=self.config)
- utils.create_from_yaml(
+ utils.process_from_yaml(
k8s_client, self.path_prefix + "yaml-conflict-first.yaml")
core_api = client.CoreV1Api(k8s_client)
svc = core_api.read_namespaced_service(name="mock-2",
@@ -519,8 +546,8 @@ def test_create_from_multi_resource_yaml_with_conflict(self):
self.assertEqual("mock-2", svc.spec.selector["app"])
self.assertEqual(99, svc.spec.ports[0].port)
- with self.assertRaises(utils.FailToCreateError) as cm:
- utils.create_from_yaml(
+ with self.assertRaises(utils.FailToProcessError) as cm:
+ utils.process_from_yaml(
k8s_client, self.path_prefix + "yaml-conflict-multi.yaml")
exp_error = ('Error from server (Conflict): {"kind":"Status",'
'"apiVersion":"v1","metadata":{},"status":"Failure",'
@@ -544,8 +571,8 @@ def test_create_from_multi_resource_yaml_with_multi_conflicts(self):
Should raise an exception that contains two error messages.
"""
k8s_client = client.api_client.ApiClient(configuration=self.config)
- with self.assertRaises(utils.FailToCreateError) as cm:
- utils.create_from_yaml(
+ with self.assertRaises(utils.FailToProcessError) as cm:
+ utils.process_from_yaml(
k8s_client, self.path_prefix + "triple-nginx.yaml")
exp_error = ('Error from server (Conflict): {"kind":"Status",'
'"apiVersion":"v1","metadata":{},"status":"Failure",'
@@ -570,7 +597,7 @@ def test_create_namespaced_apps_deployment_from_yaml(self):
in a test namespace.
"""
k8s_client = client.api_client.ApiClient(configuration=self.config)
- utils.create_from_yaml(
+ utils.process_from_yaml(
k8s_client, self.path_prefix + "apps-deployment.yaml",
namespace=self.test_namespace)
app_api = client.AppsV1Api(k8s_client)
@@ -587,7 +614,7 @@ def test_create_from_list_in_multi_resource_yaml_namespaced(self):
specified in the multi-resource file in a test namespace
"""
k8s_client = client.api_client.ApiClient(configuration=self.config)
- utils.create_from_yaml(
+ utils.process_from_yaml(
k8s_client, self.path_prefix + "multi-resource-with-list.yaml",
namespace=self.test_namespace)
core_api = client.CoreV1Api(k8s_client)
@@ -608,6 +635,287 @@ def test_create_from_list_in_multi_resource_yaml_namespaced(self):
app_api.delete_namespaced_deployment(
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.process_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.process_from_yaml(
+ k8s_client, self.path_prefix + "core-namespace.yaml", action="delete")
+ 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.process_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.process_from_yaml(
+ k8s_client, self.path_prefix + "apps-deployment.yaml", action="delete")
+ 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.process_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.process_from_yaml(
+ k8s_client, self.path_prefix + "core-service.yaml", action="delete")
+ 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.process_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.process_from_yaml(
+ k8s_client, self.path_prefix + "core-pod.yaml", action="delete")
+ 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.process_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.process_from_yaml(
+ k8s_client, self.path_prefix + "rbac-role.yaml", action="delete")
+ 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.process_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.process_from_yaml(
+ k8s_client, self.path_prefix + "rbac-role.yaml", verbose=True, action="delete")
+
+ 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.process_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.process_from_yaml(
+ k8s_client, self.path_prefix + "multi-resource.yaml", action="delete")
+ 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.process_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.process_from_yaml(
+ k8s_client, self.path_prefix + "multi-resource-with-list.yaml", action="delete")
+ 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)
+
+ def test_delete_apps_deployment_from_yaml_with_apply(self):
+ """
+ Should be able to create and delete an apps/v1 deployment using apply.
+ """
+ k8s_client = client.api_client.ApiClient(configuration=self.config)
+ # Create the deployment
+ utils.process_from_yaml(
+ k8s_client, self.path_prefix + "apps-deployment.yaml", apply=True)
+ app_api = client.AppsV1Api(k8s_client)
+ dep = app_api.read_namespaced_deployment(name="nginx-app",
+ namespace="default")
+ self.assertIsNotNone(dep)
+ self.assertEqual("nginx-app", dep.metadata.name)
+ self.assertEqual(
+ "nginx:1.15.4", dep.spec.template.spec.containers[0].image)
+ self.assertEqual(
+ 80, dep.spec.template.spec.containers[0].ports[0].container_port)
+ self.assertEqual(
+ "nginx", dep.spec.template.spec.containers[0].name)
+ self.assertEqual("nginx", dep.spec.template.metadata.labels["app"])
+ self.assertEqual(3, dep.spec.replicas)
+
+ # Delete the deployment using apply
+ utils.process_from_yaml(
+ k8s_client, self.path_prefix + "apps-deployment.yaml", apply=True, action="delete")
+ time.sleep(10) # Wait for the deletion to propagate
+
+ # Verify the deployment is deleted
+ with self.assertRaises(ApiException) as cm:
+ app_api.read_namespaced_deployment(name="nginx-app",
+ namespace="default")
+ self.assertEqual(cm.exception.status, 404)
+
class TestUtilsUnitTests(unittest.TestCase):
@@ -624,7 +932,8 @@ def test_parse_quantity(self):
self.assertRaises(
ValueError, lambda: quantity.parse_quantity("1000ki")
)
- self.assertRaises(ValueError, lambda: quantity.parse_quantity("1000foo"))
+ self.assertRaises(
+ ValueError, lambda: quantity.parse_quantity("1000foo"))
self.assertRaises(ValueError, lambda: quantity.parse_quantity("foo"))
# == no suffix ==
@@ -641,14 +950,18 @@ def test_parse_quantity(self):
self.assertEqual(quantity.parse_quantity("0.5Ki"), Decimal(512))
# == base 1000 ==
- self.assertAlmostEqual(quantity.parse_quantity("1n"), Decimal(0.000_000_001))
- self.assertAlmostEqual(quantity.parse_quantity("1u"), Decimal(0.000_001))
+ self.assertAlmostEqual(quantity.parse_quantity(
+ "1n"), Decimal(0.000_000_001))
+ self.assertAlmostEqual(
+ quantity.parse_quantity("1u"), Decimal(0.000_001))
self.assertAlmostEqual(quantity.parse_quantity("1m"), Decimal(0.001))
self.assertEqual(quantity.parse_quantity("1k"), Decimal(1_000))
self.assertEqual(quantity.parse_quantity("1M"), Decimal(1_000_000))
self.assertEqual(quantity.parse_quantity("1G"), Decimal(1_000_000_000))
- self.assertEqual(quantity.parse_quantity("1T"), Decimal(1_000_000_000_000))
- self.assertEqual(quantity.parse_quantity("1P"), Decimal(1_000_000_000_000_000))
+ self.assertEqual(quantity.parse_quantity(
+ "1T"), Decimal(1_000_000_000_000))
+ self.assertEqual(quantity.parse_quantity(
+ "1P"), Decimal(1_000_000_000_000_000))
self.assertEqual(
quantity.parse_quantity("1E"), Decimal(1_000_000_000_000_000_000))
self.assertEqual(quantity.parse_quantity("1000k"), Decimal(1_000_000))
@@ -671,17 +984,25 @@ def test_format_quantity(self):
# == no suffix ==
self.assertEqual(quantity.format_quantity(Decimal(1_000), ""), "1000")
- self.assertEqual(quantity.format_quantity(Decimal(1_000), None), "1000")
+ self.assertEqual(quantity.format_quantity(
+ Decimal(1_000), None), "1000")
# == base 1024 ==
self.assertEqual(quantity.format_quantity(Decimal(1024), "Ki"), "1Ki")
- self.assertEqual(quantity.format_quantity(Decimal(1024**2), "Mi"), "1Mi")
- self.assertEqual(quantity.format_quantity(Decimal(1024**3), "Gi"), "1Gi")
- self.assertEqual(quantity.format_quantity(Decimal(1024**4), "Ti"), "1Ti")
- self.assertEqual(quantity.format_quantity(Decimal(1024**5), "Pi"), "1Pi")
- self.assertEqual(quantity.format_quantity(Decimal(1024**6), "Ei"), "1Ei")
- self.assertEqual(quantity.format_quantity(Decimal(1024**2), "Ki"), "1024Ki")
- self.assertEqual(quantity.format_quantity(Decimal((1024**3) / 2), "Gi"), "0.5Gi")
+ self.assertEqual(quantity.format_quantity(
+ Decimal(1024**2), "Mi"), "1Mi")
+ self.assertEqual(quantity.format_quantity(
+ Decimal(1024**3), "Gi"), "1Gi")
+ self.assertEqual(quantity.format_quantity(
+ Decimal(1024**4), "Ti"), "1Ti")
+ self.assertEqual(quantity.format_quantity(
+ Decimal(1024**5), "Pi"), "1Pi")
+ self.assertEqual(quantity.format_quantity(
+ Decimal(1024**6), "Ei"), "1Ei")
+ self.assertEqual(quantity.format_quantity(
+ Decimal(1024**2), "Ki"), "1024Ki")
+ self.assertEqual(quantity.format_quantity(
+ Decimal((1024**3) / 2), "Gi"), "0.5Gi")
# Decimal((1024**3)/3) are 0.3333333333333333148296162562Gi; expecting to
# be quantized to 0.3Gi
self.assertEqual(
@@ -693,12 +1014,16 @@ def test_format_quantity(self):
"0.3Gi")
# == base 1000 ==
- self.assertEqual(quantity.format_quantity(Decimal(0.000_000_001), "n"), "1n")
- self.assertEqual(quantity.format_quantity(Decimal(0.000_001), "u"), "1u")
+ self.assertEqual(quantity.format_quantity(
+ Decimal(0.000_000_001), "n"), "1n")
+ self.assertEqual(quantity.format_quantity(
+ Decimal(0.000_001), "u"), "1u")
self.assertEqual(quantity.format_quantity(Decimal(0.001), "m"), "1m")
self.assertEqual(quantity.format_quantity(Decimal(1_000), "k"), "1k")
- self.assertEqual(quantity.format_quantity(Decimal(1_000_000), "M"), "1M")
- self.assertEqual(quantity.format_quantity(Decimal(1_000_000_000), "G"), "1G")
+ self.assertEqual(quantity.format_quantity(
+ Decimal(1_000_000), "M"), "1M")
+ self.assertEqual(quantity.format_quantity(
+ Decimal(1_000_000_000), "G"), "1G")
self.assertEqual(
quantity.format_quantity(Decimal(1_000_000_000_000), "T"), "1T"
)
@@ -706,9 +1031,11 @@ def test_format_quantity(self):
quantity.format_quantity(Decimal(1_000_000_000_000_000), "P"), "1P"
)
self.assertEqual(
- quantity.format_quantity(Decimal(1_000_000_000_000_000_000), "E"), "1E"
+ quantity.format_quantity(
+ Decimal(1_000_000_000_000_000_000), "E"), "1E"
)
- self.assertEqual(quantity.format_quantity(Decimal(1_000_000), "k"), "1000k")
+ self.assertEqual(quantity.format_quantity(
+ Decimal(1_000_000), "k"), "1000k")
# Decimal(1_000_000/3) are 333.3333333333333139307796955k; expecting to
# be quantized to 333k
self.assertEqual(
diff --git a/kubernetes/utils/__init__.py b/kubernetes/utils/__init__.py
index 2cd0caa7c..07c356423 100644
--- a/kubernetes/utils/__init__.py
+++ b/kubernetes/utils/__init__.py
@@ -14,7 +14,9 @@
from __future__ import absolute_import
-from .create_from_yaml import (FailToCreateError, create_from_dict,
- create_from_yaml, create_from_directory)
+from .yaml_processor import (FailToCreateError, create_from_dict,
+ create_from_yaml, create_from_directory,
+ FailToProcessError, process_from_dict,
+ process_from_yaml, process_from_directory)
from .quantity import parse_quantity
-from. duration import parse_duration
+from . duration import parse_duration
diff --git a/kubernetes/utils/create_from_yaml.py b/kubernetes/utils/yaml_processor.py
similarity index 55%
rename from kubernetes/utils/create_from_yaml.py
rename to kubernetes/utils/yaml_processor.py
index 562c0edb7..b174ad17a 100644
--- a/kubernetes/utils/create_from_yaml.py
+++ b/kubernetes/utils/yaml_processor.py
@@ -1,4 +1,4 @@
-# Copyright 2019 The Kubernetes Authors.
+# Copyright 2025 The Kubernetes Authors.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
@@ -24,26 +24,32 @@
LOWER_OR_NUM_FOLLOWED_BY_UPPER_RE = re.compile("([a-z0-9])([A-Z])")
-def create_from_directory(
- k8s_client, yaml_dir=None, verbose=False, namespace="default", apply=False, **kwargs
-):
+def process_from_directory(
+ k8s_client,
+ yaml_dir=None,
+ verbose=False,
+ namespace="default",
+ apply=False,
+ action="create",
+ **kwargs):
"""
- Perform an action from files from a directory. Pass True for verbose to
+ Perform an action (create/delete) from files from a directory. Pass True for verbose to
print confirmation information.
Input:
k8s_client: an ApiClient object, initialized with the client args.
yaml_dir: string. Contains the path to directory.
- verbose: If True, print confirmation from the create action.
+ verbose: If True, print confirmation from the action.
Default is False.
- namespace: string. Contains the namespace to create all
+ namespace: string. Contains the namespace to process all
resources inside. The namespace must preexist otherwise
- the resource creation will fail. If the API object in
+ the resource processing will fail. If the API object in
the yaml file already contains a namespace definition
this parameter has no effect.
- apply: bool. If True, use server-side apply for creating resources.
+ apply: bool. If True, use server-side apply for processing resources.
+ action: string. The action to perform, either "create" or "delete".
- Available parameters for creating <kind>:
+ Available parameters for processing <kind>:
:param async_req bool
:param bool include_uninitialized: If true, partially initialized
resources are included in the response.
@@ -55,11 +61,11 @@ def create_from_directory(
Valid values are: - All: all dry run stages will be processed
Returns:
- The list containing the created kubernetes API objects.
+ The list containing the processed kubernetes API objects.
Raises:
- FailToCreateError which holds list of `client.rest.ApiException`
- instances for each object that failed to create.
+ FailToProcessError which holds list of `client.rest.ApiException`
+ instances for each object that failed to process.
"""
if not yaml_dir:
@@ -80,49 +86,52 @@ def create_from_directory(
for file in files:
try:
- k8s_objects = create_from_yaml(
+ k8s_objects = process_from_yaml(
k8s_client,
file,
verbose=verbose,
namespace=namespace,
apply=apply,
+ action=action,
**kwargs,
)
k8s_objects_all.append(k8s_objects)
- except FailToCreateError as failure:
+ except FailToProcessError as failure:
failures.extend(failure.api_exceptions)
if failures:
- raise FailToCreateError(failures)
+ raise FailToProcessError(failures)
return k8s_objects_all
-def create_from_yaml(
+def process_from_yaml(
k8s_client,
yaml_file=None,
yaml_objects=None,
verbose=False,
namespace="default",
apply=False,
+ action="create",
**kwargs,
):
"""
- Perform an action from a yaml file. Pass True for verbose to
+ Perform an action (create/delete) from a yaml file. Pass True for verbose to
print confirmation information.
Input:
yaml_file: string. Contains the path to yaml file.
k8s_client: an ApiClient object, initialized with the client args.
yaml_objects: List[dict]. Optional list of YAML objects; used instead
of reading the `yaml_file`. Default is None.
- verbose: If True, print confirmation from the create action.
+ verbose: If True, print confirmation from the action.
Default is False.
- namespace: string. Contains the namespace to create all
+ namespace: string. Contains the namespace to process all
resources inside. The namespace must preexist otherwise
- the resource creation will fail. If the API object in
+ the resource processing will fail. If the API object in
the yaml file already contains a namespace definition
this parameter has no effect.
- apply: bool. If True, use server-side apply for creating resources.
+ apply: bool. If True, use server-side apply for processing resources.
+ action: string. The action to perform, either "create" or "delete".
- Available parameters for creating <kind>:
+ Available parameters for processing <kind>:
:param async_req bool
:param bool include_uninitialized: If true, partially initialized
resources are included in the response.
@@ -134,33 +143,34 @@ def create_from_yaml(
Valid values are: - All: all dry run stages will be processed
Returns:
- The created kubernetes API objects.
+ The processed kubernetes API objects.
Raises:
- FailToCreateError which holds list of `client.rest.ApiException`
- instances for each object that failed to create.
+ FailToProcessError which holds list of `client.rest.ApiException`
+ instances for each object that failed to process.
"""
- def create_with(objects, apply=apply):
+ def process_with(objects, apply=apply):
failures = []
k8s_objects = []
for yml_document in objects:
if yml_document is None:
continue
try:
- created = create_from_dict(
+ processed = process_from_dict(
k8s_client,
yml_document,
verbose,
namespace=namespace,
apply=apply,
+ action=action,
**kwargs,
)
- k8s_objects.append(created)
- except FailToCreateError as failure:
+ k8s_objects.append(processed)
+ except FailToProcessError as failure:
failures.extend(failure.api_exceptions)
if failures:
- raise FailToCreateError(failures)
+ raise FailToProcessError(failures)
return k8s_objects
class Loader(yaml.loader.SafeLoader):
@@ -170,42 +180,43 @@ class Loader(yaml.loader.SafeLoader):
if yaml_objects:
yml_document_all = yaml_objects
- return create_with(yml_document_all)
+ return process_with(yml_document_all)
elif yaml_file:
with open(os.path.abspath(yaml_file)) as f:
yml_document_all = yaml.load_all(f, Loader=Loader)
- return create_with(yml_document_all, apply)
+ return process_with(yml_document_all, apply)
else:
raise ValueError(
"One of `yaml_file` or `yaml_objects` arguments must be provided"
)
-def create_from_dict(
- k8s_client, data, verbose=False, namespace="default", apply=False, **kwargs
+def process_from_dict(
+ k8s_client, data, verbose=False, namespace="default", apply=False, action="create", **kwargs
):
"""
- Perform an action from a dictionary containing valid kubernetes
+ Perform an action (create/delete) from a dictionary containing valid kubernetes
API object (i.e. List, Service, etc).
Input:
k8s_client: an ApiClient object, initialized with the client args.
data: a dictionary holding valid kubernetes objects
- verbose: If True, print confirmation from the create action.
+ verbose: If True, print confirmation from the action.
Default is False.
- namespace: string. Contains the namespace to create all
+ namespace: string. Contains the namespace to process all
resources inside. The namespace must preexist otherwise
- the resource creation will fail. If the API object in
+ the resource processing will fail. If the API object in
the yaml file already contains a namespace definition
this parameter has no effect.
- apply: bool. If True, use server-side apply for creating resources.
+ apply: bool. If True, use server-side apply for processing resources.
+ action: string. The action to perform, either "create" or "delete".
Returns:
- The created kubernetes API objects.
+ The processed kubernetes API objects.
Raises:
- FailToCreateError which holds list of `client.rest.ApiException`
- instances for each object that failed to create.
+ FailToProcessError which holds list of `client.rest.ApiException`
+ instances for each object that failed to process.
"""
# If it is a list type, will need to iterate its items
api_exceptions = []
@@ -222,48 +233,59 @@ def create_from_dict(
yml_object["apiVersion"] = data["apiVersion"]
yml_object["kind"] = kind
try:
- created = create_from_yaml_single_item(
+ processed = process_from_yaml_single_item(
k8s_client,
yml_object,
verbose,
namespace=namespace,
apply=apply,
+ action=action,
**kwargs,
)
- k8s_objects.append(created)
+ k8s_objects.append(processed)
except client.rest.ApiException as api_exception:
api_exceptions.append(api_exception)
else:
# This is a single object. Call the single item method
try:
- created = create_from_yaml_single_item(
- k8s_client, data, verbose, namespace=namespace, apply=apply, **kwargs
- )
- k8s_objects.append(created)
+ processed = process_from_yaml_single_item(
+ k8s_client,
+ data,
+ verbose,
+ namespace=namespace,
+ apply=apply,
+ action=action,
+ **kwargs)
+ k8s_objects.append(processed)
except client.rest.ApiException as api_exception:
api_exceptions.append(api_exception)
# In case we have exceptions waiting for us, raise them
if api_exceptions:
- raise FailToCreateError(api_exceptions)
+ raise FailToProcessError(api_exceptions)
return k8s_objects
-def create_from_yaml_single_item(
- k8s_client, yml_object, verbose=False, apply=False, **kwargs
+def process_from_yaml_single_item(
+ k8s_client, yml_object, verbose=False, apply=False, action="create", **kwargs
):
-
kind = yml_object["kind"]
if apply is True:
apply_client = DynamicClient(k8s_client).resources.get(
api_version=yml_object["apiVersion"], kind=kind
)
- resp = apply_client.server_side_apply(
- body=yml_object, field_manager="python-client", **kwargs
- )
+ if action == "create":
+ resp = apply_client.server_side_apply(
+ body=yml_object, field_manager="python-client", **kwargs
+ )
+ elif action == "delete":
+ name = yml_object["metadata"]["name"]
+ resp = apply_client.delete(
+ name=name, **kwargs
+ )
if verbose:
- msg = "{0} created.".format(kind)
+ msg = "{0} {1}d.".format(kind, action)
if hasattr(resp, "status"):
msg += " status='{0}'".format(str(resp.status))
print(msg)
@@ -283,30 +305,41 @@ def create_from_yaml_single_item(
# Replace CamelCased action_type into snake_case
kind = UPPER_FOLLOWED_BY_LOWER_RE.sub(r"\1_\2", kind)
kind = LOWER_OR_NUM_FOLLOWED_BY_UPPER_RE.sub(r"\1_\2", kind).lower()
- # Expect the user to create namespaced objects more often
- if hasattr(k8s_api, "create_namespaced_{0}".format(kind)):
+ # Prepare the arguments for the API call
+ api_args = {
+ "body": client.V1DeleteOptions(
+ propagation_policy="Background",
+ grace_period_seconds=5
+ ) if action == "delete" else yml_object,
+ **kwargs
+ }
+
+ if action == "delete":
+ name = yml_object["metadata"]["name"]
+ api_args["name"] = name
+
+ # Expect the user to process namespaced objects more often
+ if hasattr(k8s_api, "{0}_namespaced_{1}".format(action, kind)):
# Decide which namespace we are going to put the object in,
# if any
if "namespace" in yml_object["metadata"]:
namespace = yml_object["metadata"]["namespace"]
- kwargs["namespace"] = namespace
- resp = getattr(k8s_api, "create_namespaced_{0}".format(kind))(
- body=yml_object, **kwargs
- )
+ api_args["namespace"] = namespace
+ resp = getattr(k8s_api, "{0}_namespaced_{1}".format(
+ action, kind))(**api_args)
else:
- kwargs.pop("namespace", None)
- resp = getattr(k8s_api, "create_{0}".format(kind))(
- body=yml_object, **kwargs
- )
+ api_args.pop("namespace", None)
+ resp = getattr(k8s_api, "{0}_{1}".format(action, kind))(**api_args)
+
if verbose:
- msg = "{0} created.".format(kind)
+ msg = "{0} {1}d.".format(kind, action)
if hasattr(resp, "status"):
msg += " status='{0}'".format(str(resp.status))
print(msg)
return resp
-class FailToCreateError(Exception):
+class FailToProcessError(Exception):
"""
An exception class for handling error if an error occurred when
handling a yaml file.
@@ -322,3 +355,138 @@ def __str__(self):
api_exception.reason, api_exception.body
)
return msg
+
+
+def create_from_directory(
+ k8s_client, yaml_dir=None, verbose=False, namespace="default", apply=False, **kwargs
+):
+ """
+ Perform an action from files from a directory. Pass True for verbose to
+ print confirmation information.
+
+ Input:
+ k8s_client: an ApiClient object, initialized with the client args.
+ yaml_dir: string. Contains the path to directory.
+ verbose: If True, print confirmation from the create action.
+ Default is False.
+ namespace: string. Contains the namespace to create all
+ resources inside. The namespace must preexist otherwise
+ the resource creation will fail. If the API object in
+ the yaml file already contains a namespace definition
+ this parameter has no effect.
+ apply: bool. If True, use server-side apply for creating resources.
+
+ Available parameters for creating <kind>:
+ :param async_req bool
+ :param bool include_uninitialized: If true, partially initialized
+ resources are included in the response.
+ :param str pretty: If 'true', then the output is pretty printed.
+ :param str dry_run: When present, indicates that modifications
+ should not be persisted. An invalid or unrecognized dryRun
+ directive will result in an error response and no further
+ processing of the request.
+ Valid values are: - All: all dry run stages will be processed
+
+ Returns:
+ The list containing the created kubernetes API objects.
+
+ Raises:
+ FailToCreateError which holds list of `client.rest.ApiException`
+ instances for each object that failed to create.
+ """
+ return process_from_directory(
+ k8s_client,
+ yaml_dir,
+ verbose=verbose,
+ namespace=namespace,
+ apply=apply,
+ action="create",
+ **kwargs,
+ )
+
+
+def create_from_yaml(
+ k8s_client,
+ yaml_file=None,
+ yaml_objects=None,
+ verbose=False,
+ namespace="default",
+ apply=False,
+ **kwargs):
+ """
+ Perform an action from a yaml file. Pass True for verbose to
+ print confirmation information.
+ Input:
+ yaml_file: string. Contains the path to yaml file.
+ k8s_client: an ApiClient object, initialized with the client args.
+ yaml_objects: List[dict]. Optional list of YAML objects; used instead
+ of reading the `yaml_file`. Default is None.
+ verbose: If True, print confirmation from the create action.
+ Default is False.
+ namespace: string. Contains the namespace to create all
+ resources inside. The namespace must preexist otherwise
+ the resource creation will fail. If the API object in
+ the yaml file already contains a namespace definition
+ this parameter has no effect.
+
+ Available parameters for creating <kind>:
+ :param async_req bool
+ :param bool include_uninitialized: If true, partially initialized
+ resources are included in the response.
+ :param str pretty: If 'true', then the output is pretty printed.
+ :param str dry_run: When present, indicates that modifications
+ should not be persisted. An invalid or unrecognized dryRun
+ directive will result in an error response and no further
+ processing of the request.
+ Valid values are: - All: all dry run stages will be processed
+
+ Returns:
+ The created kubernetes API objects.
+
+ Raises:
+ FailToCreateError which holds list of `client.rest.ApiException`
+ instances for each object that failed to create.
+ """
+ return process_from_yaml(
+ k8s_client,
+ yaml_file,
+ yaml_objects,
+ verbose,
+ namespace,
+ apply,
+ action="create",
+ **kwargs)
+
+
+def create_from_dict(k8s_client, data, verbose=False, namespace='default',
+ **kwargs):
+ """
+ Perform an action from a dictionary containing valid kubernetes
+ API object (i.e. List, Service, etc).
+
+ Input:
+ k8s_client: an ApiClient object, initialized with the client args.
+ data: a dictionary holding valid kubernetes objects
+ verbose: If True, print confirmation from the create action.
+ Default is False.
+ namespace: string. Contains the namespace to create all
+ resources inside. The namespace must preexist otherwise
+ the resource creation will fail. If the API object in
+ the yaml file already contains a namespace definition
+ this parameter has no effect.
+
+ Returns:
+ The created kubernetes API objects.
+
+ Raises:
+ FailToCreateError which holds list of `client.rest.ApiException`
+ instances for each object that failed to create.
+ """
+ return process_from_dict(k8s_client, data, verbose, namespace, action="create", **kwargs)
+
+
+class FailToCreateError(FailToProcessError):
+ """
+ An exception class for handling error if an error occurred when
+ handling a yaml file.
+ """