Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

utils and utils.create_from_yaml improvements #783

Closed
wants to merge 3 commits into from
Closed

utils and utils.create_from_yaml improvements #783

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
bddaca1
Add initial work on `utils` improvements
Mar 18, 2019
93e8902
Make all e2e_test use create_from_file
Mar 18, 2019
2fed56a
Change create_from_yaml to accept string or file
Mar 19, 2019
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
94 changes: 54 additions & 40 deletions kubernetes/e2e_test/test_utils.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -17,6 +17,7 @@
from kubernetes import utils, client
from kubernetes.e2e_test import base


class TestUtils(unittest.TestCase):

@classmethod
Expand All @@ -25,92 +26,105 @@ def setUpClass(cls):

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",
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):
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/extensions-deployment.yaml")
self.assertEqual("extensions/v1beta1",
self.assertEqual(
"extensions/v1beta1",
k8s_api.get_api_resources().group_version)
dep = k8s_api.read_namespaced_deployment(name="nginx-deployment",
dep = k8s_api.read_namespaced_deployment(
name="nginx-deployment",
namespace="default")
self.assertIsNotNone(dep)
resp = k8s_api.delete_namespaced_deployment(
name="nginx-deployment", namespace="default",
k8s_api.delete_namespaced_deployment(
name="nginx-deployment", namespace="default",
body={})

def test_core_pod_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/core-pod.yaml")
self.assertEqual("v1",
self.assertEqual(
"v1",
k8s_api.get_api_resources().group_version)
pod = k8s_api.read_namespaced_pod(name="myapp-pod",
namespace="default")
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={})

def test_core_service_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/core-service.yaml")
self.assertEqual("v1",
k8s_api.get_api_resources().group_version)
svc = k8s_api.read_namespaced_service(name="my-service",
self.assertEqual("v1", k8s_api.get_api_resources().group_version)
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={})

def test_core_namespace_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/core-namespace.yaml")
self.assertEqual("v1",
k8s_api.get_api_resources().group_version)
self.assertEqual("v1", 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)
core_api = utils.create_from_yaml(k8s_client,
core_api = utils.create_from_yaml(
k8s_client,
"kubernetes/e2e_test/test_yaml/core-namespace-dep.yaml")
self.assertEqual("v1",
core_api.get_api_resources().group_version)
self.assertEqual("v1", core_api.get_api_resources().group_version)
nmsp = core_api.read_namespace(name="dep")
self.assertIsNotNone(nmsp)
dep_api = utils.create_from_yaml(k8s_client,
dep_api = utils.create_from_yaml(
k8s_client,
"kubernetes/e2e_test/test_yaml/extensions-deployment-dep.yaml")
dep = dep_api.read_namespaced_deployment(name="nginx-deployment",
dep = dep_api.read_namespaced_deployment(
name="nginx-deployment",
namespace="dep")
self.assertIsNotNone(dep)
resp = dep_api.delete_namespaced_deployment(
name="nginx-deployment", namespace="dep",
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)
k8s_api = utils.create_from_yaml(k8s_client,
k8s_api = utils.create_from_yaml(
k8s_client,
"kubernetes/e2e_test/test_yaml/api-service.yaml")
self.assertEqual("apiregistration.k8s.io/v1beta1",
self.assertEqual(
"apiregistration.k8s.io/v1beta1",
k8s_api.get_api_resources().group_version)
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={})
2 changes: 1 addition & 1 deletion kubernetes/utils/__init__.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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_map, create_from_yaml
124 changes: 79 additions & 45 deletions kubernetes/utils/create_from_yaml.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -14,60 +14,94 @@


import re
import sys
from os import path

import yaml
from six import iteritems

from kubernetes import client


def create_from_yaml(k8s_client, yaml_file, verbose=False, **kwargs):
def create_from_yaml(k8s_client, yml, verbose=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_cline: an ApiClient object, initialized with the client args.
Perform an action from a yaml string or file.

:param k8s_client: an ApiClient object, initialized with the client args
:yml: a string containing valid yaml or a file name.
:type yml: str
:param verbose: print confimation information
:type verbose: bool

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
:param async_req:
:type async_req: bool
:param include_uninitialized: include partially initialized resources in
the response.
:type include_uninitialized: bool
:param str pretty: pretty print the output
:param str dry_run: When present, indicates that modifications should not
be persisted. An invalid or unrecognized dry_run 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
"""
if path.exists(yml):
with open(path.abspath(yml)) as f:
stream = yaml.safe_load_all(f.read())
else:
stream = yaml.safe_load_all(yml)
for obj in stream:
k8s_api = create_from_map(k8s_client, obj, verbose, **kwargs)

return k8s_api


def create_from_map(k8s_client, yml_object, verbose, **kwargs):
"""
performe an action from a valid parsed yaml object (as dict).

:param k8s_client: an ApiClient objcet initialized with the client args
:yml_object dict: a parsed yaml object
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I don't think one needs to explicitly reference yaml here as what the function wants is a kubernetes manifest in python dictionary form
though it probably also does not harm as yaml is the most common representation form of that data.

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

this is why I added create_from_map

:param verbose: print confimation information
:type verbose: bool

with open(path.abspath(yaml_file)) as f:
yml_object = yaml.load(f)
# TODO: case of yaml file containing multiple objects
group, _, version = yml_object["apiVersion"].partition("/")
if version == "":
version = group
group = "core"
# Take care for the case e.g. api_type is "apiextensions.k8s.io"
# Only replace the last instance
group = "".join(group.rsplit(".k8s.io", 1))
fcn_to_call = "{0}{1}Api".format(group.capitalize(),
version.capitalize())
k8s_api = getattr(client, fcn_to_call)(k8s_client)
# Replace CamelCased action_type into snake_case
kind = yml_object["kind"]
kind = re.sub('(.)([A-Z][a-z]+)', r'\1_\2', kind)
kind = re.sub('([a-z0-9])([A-Z])', r'\1_\2', kind).lower()
# Decide which namespace we are going to put the object in,
# if any
if "namespace" in yml_object["metadata"]:
namespace = yml_object["metadata"]["namespace"]
else:
namespace = "default"
# Expect the user to create namespaced objects more often
if hasattr(k8s_api, "create_namespaced_{0}".format(kind)):
resp = getattr(k8s_api, "create_namespaced_{0}".format(kind))(
body=yml_object, namespace=namespace, **kwargs)
else:
resp = getattr(k8s_api, "create_{0}".format(kind))(
body=yml_object, **kwargs)
if verbose:
print("{0} created. status='{1}'".format(kind, str(resp.status)))
return k8s_api
Available parameters for performing the subsequent action:
:param async_req:
:type async_req: bool
:param include_uninitialized: include partially initialized resources in
the response.
:type include_uninitialized: bool
:param str pretty: pretty print the output
:param str dry_run: When present, indicates that modifications should not
be persisted. An invalid or unrecognized dry_run 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
"""
group, _, version = yml_object["apiVersion"].partition("/")
if version == "":
version = group
group = "core"
# Take care for the case e.g. api_type is "apiextensions.k8s.io"
# Only replace the last instance
group = "".join(group.rsplit(".k8s.io", 1))
fcn_to_call = "{0}{1}Api".format(group.capitalize(),
version.capitalize())
k8s_api = getattr(client, fcn_to_call)(k8s_client)
# Replace CamelCased action_type into snake_case
kind = yml_object["kind"]
kind = re.sub('(.)([A-Z][a-z]+)', r'\1_\2', kind)
kind = re.sub('([a-z0-9])([A-Z])', r'\1_\2', kind).lower()
# Decide which namespace we are going to put the object in,
# if any
if "namespace" in yml_object["metadata"]:
namespace = yml_object["metadata"]["namespace"]
else:
namespace = "default"
# Expect the user to create namespaced objects more often
if hasattr(k8s_api, "create_namespaced_{0}".format(kind)):
resp = getattr(k8s_api, "create_namespaced_{0}".format(kind))(
body=yml_object, namespace=namespace, **kwargs)
else:
resp = getattr(k8s_api, "create_{0}".format(kind))(
body=yml_object, **kwargs)
if verbose:
print("{0} created. status='{1}'".format(kind, str(resp.status)))
return k8s_api