Allow passing kwargs and fix broken tests

Author: micw523

examples/create_deployment_from_yaml.py

@@ -24,8 +24,7 @@ def main():
     # utility. If no argument provided, the config will be loaded from
     # default location.
     config.load_kube_config()
-    k8s_client = utils.create_from_yaml("nginx-deployment.yaml")
-    k8s_client
+    k8s_api = utils.create_from_yaml("nginx-deployment.yaml")
 
 if __name__ == '__main__':
     main()

kubernetes/e2e_test/test_utils.py

@@ -24,55 +24,63 @@ def setUpClass(cls):
         cls.config = base.get_e2e_configuration()
 
     def test_app_yaml(self):
-        k8s_client = utils.create_from_yaml("test_yaml/app.yaml")
+        k8s_api = utils.create_from_yaml(
+            "kubernetes/e2e_test/test_yaml/app.yaml", 
+            configuration=self.config)
         self.assertEqual("apps/v1", 
-            k8s_client.get_api_resources().group_version)
-        deployments = k8s_client.get_namespaced_deployment(
+            k8s_api.get_api_resources().group_version)
+        deployments = k8s_api.list_namespaced_deployment(
             namespace="default").items
-        deployment_name = deployments[0].metadata.name
-        self.assertEqual("nginx-deployment-app", deployment_name)
-        resp = k8s_client.delete_namespaced_deployment(
-            name=deployment_name, namespace="default", 
-            body=client.V1DeleteOptions(
-                propagation_policy="Foreground",
-                grace_period_seconds=1))
+        deployment_name = []
+        for item in deployments:
+            deployment_name.append(item.metadata.name)
+        self.assertIn("nginx-deployment-app", deployment_name)
+        resp = k8s_api.delete_namespaced_deployment(
+            name="nginx-deployment-app", namespace="default", 
+            body={})
 
     def test_extension_yaml(self):
-        k8s_client = utils.create_from_yaml("test_yaml/extension.yaml")
+        k8s_api = utils.create_from_yaml(
+            "kubernetes/e2e_test/test_yaml/extension.yaml",
+            configuration=self.config)
         self.assertEqual("extensions/v1beta1", 
-            k8s_client.get_api_resources().group_version)
-        deployments = k8s_client.get_namespaced_deployment(
+            k8s_api.get_api_resources().group_version)
+        deployments = k8s_api.list_namespaced_deployment(
             namespace="default").items
-        deployment_name = deployments[0].metadata.name
-        self.assertEqual("nginx-deployment", deployment_name)
-        resp = k8s_client.delete_namespaced_deployment(
-            name=deployment_name, namespace="default", 
-            body=client.V1DeleteOptions(
-                propagation_policy="Foreground",
-                grace_period_seconds=1))
+        deployment_name = []
+        for item in deployments:
+            deployment_name.append(item.metadata.name)
+        self.assertIn("nginx-deployment", deployment_name)
+        resp = k8s_api.delete_namespaced_deployment(
+            name="nginx-deployment", namespace="default", 
+            body={})
 
     def test_core_pod_yaml(self):
-        k8s_client = utils.create_from_yaml("test_yaml/core-pod.yaml")
+        k8s_api = utils.create_from_yaml(
+            "kubernetes/e2e_test/test_yaml/core-pod.yaml",
+            configuration=self.config)
         self.assertEqual("v1", 
-            k8s_client.get_api_resources().group_version)
-        pods = k8s_client.list_namespaced_pod(namespace="default").items
-        pod_name = pods[0].metadata.name
-        self.assertEqual("myapp-pod", pod_name)
-        resp = k8s_client.delete_namespaced_pod(
-            name=pod_name, namespace="default",
-            body=client.V1DeleteOptions(
-                propagation_policy="Foreground",
-                grace_period_seconds=1))
+            k8s_api.get_api_resources().group_version)
+        pods = k8s_api.list_namespaced_pod(namespace="default").items
+        pod_name = []
+        for item in pods:
+            pod_name.append(item.metadata.name)
+        self.assertIn("myapp-pod", pod_name)
+        resp = k8s_api.delete_namespaced_pod(
+            name="myapp-pod", namespace="default",
+            body={})
 
     def test_core_service_yaml(self):
-        k8s_client = utils.create_from_yaml("test_yaml/core-service.yaml")
+        k8s_api = utils.create_from_yaml(
+            "kubernetes/e2e_test/test_yaml/core-service.yaml",
+            configuration=self.config)
         self.assertEqual("v1", 
-            k8s_client.get_api_resources().group_version)
-        svcs = k8s_client.list_namespaced_service(namespace="default").items
-        svc_name = svcs[1].metadata.name
-        self.assertEqual("my-service", svc_name)
-        resp = k8s_client.delete_namespaced_pod(
-            name=svc_name, namespace="default",
-            body=client.V1DeleteOptions(
-                propagation_policy="Foreground",
-                grace_period_seconds=1))
+            k8s_api.get_api_resources().group_version)
+        svcs = k8s_api.list_namespaced_service(namespace="default").items
+        svc_name = []
+        for item in svcs:
+            svc_name.append(item.metadata.name)
+        self.assertIn("my-service", svc_name)
+        resp = k8s_api.delete_namespaced_service(
+            name="my-service", namespace="default",
+            body={})

kubernetes/utils/create_from_yaml.py

@@ -15,26 +15,59 @@
 from os import path
 import sys
 
+from six import iteritems
+
 import yaml
 
 from kubernetes import client
 
-def create_from_yaml(yaml_file):
+def create_from_yaml(yaml_file, verbose=0, **kwargs):
+    """
+    Perform an action from a yaml file. 
+
+    Available parameters for generating the client:
+    :param configuration: Configuration for the client.
+    :param host: The base path for the server to call.
+    :param header_name: a header to pass when making calls to the API.
+    :param header_value: a header value to pass when making calls to the API.
+
+    Available parameters for performing the subsequent action:
+    :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
+    """
+    
+    client_params = ["configuration", "host", "header_name", "header_value"]
+    action_params = ["async_req", "include_unitialized", "pretty", "dry_run",
+        '_return_http_data_only', '_preload_content', '_request_timeout']
+    client_args = {}
+    action_args = {}
+    params = locals()
+    for key, val in iteritems(params["kwargs"]):
+        if key in client_params:
+            client_args[key] = val
+        elif key in action_params:
+            action_args[key] = val
+
+    k8s_client = client.api_client.ApiClient(**client_args)
     with open(path.abspath(yaml_file)) as f:
         dep = yaml.load(f)
         api_type, _, api_version = dep["apiVersion"].partition("/")
         if api_version == "":
                 api_version = api_type
                 api_type = "core"
-        k8s_client = getattr(client, "%s%sApi" % (
-                api_type.capitalize(), api_version.capitalize()))()
+        k8s_api = getattr(client, "%s%sApi" % (
+                api_type.capitalize(), api_version.capitalize()))(
+                    k8s_client)
         action_type = dep["kind"]
         if "namespace" in dep["metadata"]:
                 dep_namespace = dep["metadata"]["namespace"]
         else:
                 dep_namespace = "default"
-        resp = getattr(k8s_client, "create_namespaced_%s" % action_type.lower())(
-                body=dep, namespace=dep_namespace)
-        print("Deployment created. status='%s'" % str(resp.status))
-        return k8s_client
+        resp = getattr(k8s_api, "create_namespaced_%s" % action_type.lower())(
+                body=dep, namespace=dep_namespace, **action_args)
+        if verbose:
+            print("Deployment created. status='%s'" % str(resp.status))
+        return k8s_api