Add e2e test for create_namespaced_from_yaml

Author: johscheuer

kubernetes/e2e_test/test_utils.py

@@ -17,23 +17,35 @@
 from kubernetes import utils, client
 from kubernetes.e2e_test import base
 
+
 class TestUtils(unittest.TestCase):
 
     @classmethod
     def setUpClass(cls):
         cls.config = base.get_e2e_configuration()
+        cls.test_namespace = "e2e-test-utils"
+        k8s_client = client.api_client.ApiClient(configuration=cls.config)
+        core_v1 = client.CoreV1Api(api_client=k8s_client)
+        body = client.V1Namespace(metadata=client.V1ObjectMeta(name=cls.test_namespace))
+        core_v1.create_namespace(body=body)
+
+    @classmethod
+    def tearDownClass(cls):
+        k8s_client = client.api_client.ApiClient(configuration=cls.config)
+        core_v1 = client.CoreV1Api(api_client=k8s_client)
+        core_v1.delete_namespace(name=cls.test_namespace)
 
     def test_app_yaml(self):
         k8s_client = client.api_client.ApiClient(configuration=self.config)
-        k8s_api = utils.create_from_yaml(k8s_client, 
+        k8s_api = utils.create_from_yaml(k8s_client,
             "kubernetes/e2e_test/test_yaml/apps-deployment.yaml")
-        self.assertEqual("apps/v1beta1", 
+        self.assertEqual("apps/v1beta1",
             k8s_api.get_api_resources().group_version)
         dep = k8s_api.read_namespaced_deployment(name="nginx-app",
             namespace="default")
         self.assertIsNotNone(dep)
-        resp = k8s_api.delete_namespaced_deployment(
-            name="nginx-app", namespace="default", 
+        k8s_api.delete_namespaced_deployment(
+            name="nginx-app", namespace="default",
             body={})
 
     def test_extension_yaml(self):
@@ -45,7 +57,7 @@ def test_extension_yaml(self):
         dep = k8s_api.read_namespaced_deployment(name="nginx-deployment", 
             namespace="default")
         self.assertIsNotNone(dep)
-        resp = k8s_api.delete_namespaced_deployment(
+        k8s_api.delete_namespaced_deployment(
             name="nginx-deployment", namespace="default", 
             body={})
 
@@ -58,7 +70,7 @@ def test_core_pod_yaml(self):
         pod = k8s_api.read_namespaced_pod(name="myapp-pod", 
             namespace="default")
         self.assertIsNotNone(pod)
-        resp = k8s_api.delete_namespaced_pod(
+        k8s_api.delete_namespaced_pod(
             name="myapp-pod", namespace="default",
             body={})
 
@@ -71,7 +83,7 @@ def test_core_service_yaml(self):
         svc = k8s_api.read_namespaced_service(name="my-service",
             namespace="default")
         self.assertIsNotNone(svc)
-        resp = k8s_api.delete_namespaced_service(
+        k8s_api.delete_namespaced_service(
             name="my-service", namespace="default",
             body={})
 
@@ -83,7 +95,7 @@ def test_core_namespace_yaml(self):
             k8s_api.get_api_resources().group_version)
         nmsp = k8s_api.read_namespace(name="development")
         self.assertIsNotNone(nmsp)
-        resp = k8s_api.delete_namespace(name="development", body={})
+        k8s_api.delete_namespace(name="development", body={})
 
     def test_deployment_in_namespace(self):
         k8s_client = client.ApiClient(configuration=self.config)
@@ -98,10 +110,10 @@ def test_deployment_in_namespace(self):
         dep = dep_api.read_namespaced_deployment(name="nginx-deployment", 
             namespace="dep")
         self.assertIsNotNone(dep)
-        resp = dep_api.delete_namespaced_deployment(
+        dep_api.delete_namespaced_deployment(
             name="nginx-deployment", namespace="dep", 
             body={})
-        resp = core_api.delete_namespace(name="dep", body={})
+        core_api.delete_namespace(name="dep", body={})
 
     def test_api_service(self):
         k8s_client = client.api_client.ApiClient(configuration=self.config)
@@ -112,5 +124,61 @@ def test_api_service(self):
         svc = k8s_api.read_api_service(
             name="v1alpha1.wardle.k8s.io")
         self.assertIsNotNone(svc)
-        resp = k8s_api.delete_api_service(
-            name="v1alpha1.wardle.k8s.io", body={})
+        k8s_api.delete_api_service(
+            name="v1alpha1.wardle.k8s.io", body={})
+
+    def test_namespaced_app_yaml(self):
+        k8s_client = client.api_client.ApiClient(configuration=self.config)
+        k8s_api = utils.create_namespaced_from_yaml(k8s_client,
+            "kubernetes/e2e_test/test_yaml/apps-deployment.yaml",
+            namespace=self.test_namespace)
+        self.assertEqual("apps/v1beta1",
+            k8s_api.get_api_resources().group_version)
+        dep = k8s_api.read_namespaced_deployment(name="nginx-app",
+            namespace=self.test_namespace,)
+        self.assertIsNotNone(dep)
+        k8s_api.delete_namespaced_deployment(
+            name="nginx-app", namespace=self.test_namespace,
+            body={})
+
+    def test_namespaced_extension_yaml(self):
+        k8s_client = client.api_client.ApiClient(configuration=self.config)
+        k8s_api = utils.create_namespaced_from_yaml(k8s_client,
+            "kubernetes/e2e_test/test_yaml/extensions-deployment.yaml",
+            namespace=self.test_namespace)
+        self.assertEqual("extensions/v1beta1",
+            k8s_api.get_api_resources().group_version)
+        dep = k8s_api.read_namespaced_deployment(name="nginx-deployment",
+            namespace=self.test_namespace)
+        self.assertIsNotNone(dep)
+        k8s_api.delete_namespaced_deployment(
+            name="nginx-deployment", namespace=self.test_namespace,
+            body={})
+
+    def test_namespaced_core_pod_yaml(self):
+        k8s_client = client.api_client.ApiClient(configuration=self.config)
+        k8s_api = utils.create_namespaced_from_yaml(k8s_client,
+            "kubernetes/e2e_test/test_yaml/core-pod.yaml",
+            namespace=self.test_namespace)
+        self.assertEqual("v1",
+            k8s_api.get_api_resources().group_version)
+        pod = k8s_api.read_namespaced_pod(name="myapp-pod",
+            namespace=self.test_namespace)
+        self.assertIsNotNone(pod)
+        k8s_api.delete_namespaced_pod(
+            name="myapp-pod", namespace=self.test_namespace,
+            body={})
+
+    def test_namespaced_core_service_yaml(self):
+        k8s_client = client.api_client.ApiClient(configuration=self.config)
+        k8s_api = utils.create_namespaced_from_yaml(k8s_client,
+            "kubernetes/e2e_test/test_yaml/core-service.yaml",
+            namespace=self.test_namespace)
+        self.assertEqual("v1",
+            k8s_api.get_api_resources().group_version)
+        svc = k8s_api.read_namespaced_service(name="my-service",
+            namespace=self.test_namespace)
+        self.assertIsNotNone(svc)
+        k8s_api.delete_namespaced_service(
+            name="my-service", namespace=self.test_namespace,
+            body={})

kubernetes/utils/__init__.py

@@ -12,4 +12,4 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-from .create_from_yaml import create_from_yaml
+from .create_from_yaml import create_from_yaml, create_namespaced_from_yaml

kubernetes/utils/create_from_yaml.py

@@ -18,12 +18,16 @@
 from os import path
 
 import yaml
-from six import iteritems
-
 from kubernetes import client
+from six import iteritems
 
 
-def create_namespaced_from_yaml(k8s_client, yaml_file, verbose=False, namespace=None, **kwargs):
+def create_namespaced_from_yaml(
+        k8s_client,
+        yaml_file,
+        verbose=False,
+        namespace=None,
+        **kwargs):
     """
     Perform an action from a yaml file. Pass True for verbose to
     print confirmation information.
@@ -94,4 +98,5 @@ def create_from_yaml(k8s_client, yaml_file, verbose=False, **kwargs):
     :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
     """
-    return create_namespaced_from_yaml(k8s_client, yaml_file, verbose, None, **kwargs)
+    return create_namespaced_from_yaml(
+        k8s_client, yaml_file, verbose, None, **kwargs)