test_utils.py

# -*- coding: utf-8 -*-

# Licensed under the Apache License, Version 2.0 (the "License"); you may
# not use this file except in compliance with the License. You may obtain
# a copy of the License at
#
#      http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations
# under the License.

import unittest
from decimal import Decimal
from os import path

import yaml

from kubernetes import client, utils
from kubernetes.client.rest import ApiException
from kubernetes.e2e_test import base
from kubernetes.utils import quantity


class TestUtils(unittest.TestCase):

    @classmethod
    def setUpClass(cls):
        cls.config = base.get_e2e_configuration()
        cls.path_prefix = "kubernetes/e2e_test/test_yaml/"
        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)

    # Tests for creating individual API objects

    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(
            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", dep.spec.template.spec.containers[0].name)
        self.assertEqual("nginx", dep.spec.template.metadata.labels["app"])
        self.assertEqual(3, dep.spec.replicas)

        while True:
            try:
                app_api.delete_namespaced_deployment(
                    name="nginx-app", namespace="default",
                    body={})
                break
            except ApiException:
                continue

    def test_create_apps_deployment_from_yaml_with_apply_is_idempotent(self):
        """
        Should be able to create an apps/v1 deployment.
        """
        k8s_client = client.api_client.ApiClient(configuration=self.config)
        try:
            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)
            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)

            utils.create_from_yaml(
                k8s_client, self.path_prefix + "apps-deployment.yaml", apply=True)
            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)
        except Exception as e:
            self.fail(e)
        finally:
            while True:
                try:
                    app_api.delete_namespaced_deployment(
                        name="nginx-app", namespace="default",
                        body={})
                    break
                except ApiException:
                    continue

    def test_create_apps_deployment_from_yaml_object(self):
        """
        Should be able to pass YAML objects directly to helper function.
        """
        k8s_client = client.api_client.ApiClient(configuration=self.config)
        _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(
                k8s_client,
                yaml_objects=yaml_objects,
            )
        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)

        while True:
            try:
                app_api.delete_namespaced_deployment(
                    name="nginx-app", namespace="default",
                    body={})
                break
            except ApiException:
                continue

    def test_create_apps_deployment_from_yaml_obj(self):
        k8s_client = client.api_client.ApiClient(configuration=self.config)
        with open(self.path_prefix + "apps-deployment.yaml") as f:
            yml_obj = yaml.safe_load(f)

        yml_obj["metadata"]["name"] = "nginx-app-3"

        utils.create_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", dep.spec.template.spec.containers[0].name)
        self.assertEqual("nginx", dep.spec.template.metadata.labels["app"])
        self.assertEqual(3, dep.spec.replicas)

        app_api.delete_namespaced_deployment(
            name="nginx-app-3", namespace="default",
            body={})

    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(
            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)
        self.assertEqual("myapp-pod", pod.metadata.name)
        self.assertEqual("busybox", pod.spec.containers[0].image)
        self.assertEqual("myapp-container", pod.spec.containers[0].name)

        core_api.delete_namespaced_pod(
            name="myapp-pod", namespace="default",
            body={})

    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(
            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)
        self.assertEqual("my-service", svc.metadata.name)
        self.assertEqual("MyApp", svc.spec.selector["app"])

        core_api.delete_namespaced_service(
            name="my-service", namespace="default",
            body={})

    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(
            k8s_client, self.path_prefix + "core-namespace.yaml")
        core_api = client.CoreV1Api(k8s_client)
        nmsp = core_api.read_namespace(name="development")

        self.assertIsNotNone(nmsp)
        self.assertEqual("development", nmsp.metadata.name)

        core_api.delete_namespace(name="development", body={})

    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(
            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)
        self.assertEqual("pod-reader", rbac_role.metadata.name)
        self.assertEqual("pods", rbac_role.rules[0].resources[0])

        rbac_api.delete_namespaced_role(
            name="pod-reader", namespace="default", body={})

    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(
            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)
        self.assertEqual("pod-reader", rbac_role.metadata.name)
        self.assertEqual("pods", rbac_role.rules[0].resources[0])

        rbac_api.delete_namespaced_role(
            name="pod-reader", namespace="default", body={})

    def test_create_deployment_non_default_namespace_from_yaml(self):
        """
        Should be able to create a namespace "dep",
        and then create a deployment in the just-created namespace.
        """
        k8s_client = client.ApiClient(configuration=self.config)
        utils.create_from_yaml(
            k8s_client, self.path_prefix + "dep-namespace.yaml")
        utils.create_from_yaml(
            k8s_client, self.path_prefix + "dep-deployment.yaml")
        core_api = client.CoreV1Api(k8s_client)
        ext_api = client.AppsV1Api(k8s_client)
        nmsp = core_api.read_namespace(name="dep")
        self.assertIsNotNone(nmsp)
        self.assertEqual("dep", nmsp.metadata.name)

        dep = ext_api.read_namespaced_deployment(name="nginx-deployment",
                                                 namespace="dep")
        self.assertIsNotNone(dep)
        ext_api.delete_namespaced_deployment(
            name="nginx-deployment", namespace="dep",
            body={})
        core_api.delete_namespace(name="dep", body={})

    def test_create_apiservice_from_yaml_with_conflict(self):
        """
        Should be able to create an API service.
        Should verify that creating the same API service should
        fail due to conflict.
        """
        k8s_client = client.api_client.ApiClient(configuration=self.config)
        utils.create_from_yaml(
            k8s_client, self.path_prefix + "api-service.yaml")
        reg_api = client.ApiregistrationV1Api(k8s_client)
        svc = reg_api.read_api_service(
            name="v1alpha1.wardle.k8s.io")
        self.assertIsNotNone(svc)
        self.assertEqual("v1alpha1.wardle.k8s.io", svc.metadata.name)
        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(
                k8s_client, "kubernetes/e2e_test/test_yaml/api-service.yaml")
        exp_error = ('Error from server (Conflict): '
                     '{"kind":"Status","apiVersion":"v1","metadata":{},'
                     '"status":"Failure",'
                     '"message":"apiservices.apiregistration.k8s.io '
                     '\\"v1alpha1.wardle.k8s.io\\" already exists",'
                     '"reason":"AlreadyExists",'
                     '"details":{"name":"v1alpha1.wardle.k8s.io",'
                     '"group":"apiregistration.k8s.io","kind":"apiservices"},'
                     '"code":409}\n'
                     )
        self.assertEqual(exp_error, str(cm.exception))
        reg_api.delete_api_service(
            name="v1alpha1.wardle.k8s.io", body={})

    # Tests for creating API objects from lists

    def test_create_general_list_from_yaml(self):
        """
        Should be able to create a service and a deployment
        from a kind: List yaml file
        """
        k8s_client = client.api_client.ApiClient(configuration=self.config)
        utils.create_from_yaml(
            k8s_client, self.path_prefix + "list.yaml")
        core_api = client.CoreV1Api(k8s_client)
        ext_api = client.AppsV1Api(k8s_client)
        svc = core_api.read_namespaced_service(name="list-service-test",
                                               namespace="default")
        self.assertIsNotNone(svc)
        self.assertEqual("list-service-test", svc.metadata.name)
        self.assertEqual("list-deployment-test", svc.spec.selector["app"])

        dep = ext_api.read_namespaced_deployment(name="list-deployment-test",
                                                 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(1, dep.spec.replicas)

        core_api.delete_namespaced_service(name="list-service-test",
                                           namespace="default", body={})
        ext_api.delete_namespaced_deployment(name="list-deployment-test",
                                             namespace="default", body={})

    def test_create_namespace_list_from_yaml(self):
        """
        Should be able to create two namespaces
        from a kind: NamespaceList yaml file
        """
        k8s_client = client.api_client.ApiClient(configuration=self.config)
        utils.create_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")
        self.assertIsNotNone(nmsp_1)
        self.assertEqual("mock-1", nmsp_1.metadata.name)
        self.assertEqual("mock-1", nmsp_1.metadata.labels["name"])

        nmsp_2 = core_api.read_namespace(name="mock-2")
        self.assertIsNotNone(nmsp_2)
        self.assertEqual("mock-2", nmsp_2.metadata.name)
        self.assertEqual("mock-2", nmsp_2.metadata.labels["name"])

        core_api.delete_namespace(name="mock-1", body={})
        core_api.delete_namespace(name="mock-2", body={})

    def test_create_implicit_service_list_from_yaml_with_conflict(self):
        """
        Should be able to create two services from a kind: ServiceList
        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(
                k8s_client, self.path_prefix + "implicit-svclist.json")
        core_api = client.CoreV1Api(k8s_client)
        svc_3 = core_api.read_namespaced_service(name="mock-3",
                                                 namespace="default")
        self.assertIsNotNone(svc_3)
        self.assertEqual("mock-3", svc_3.metadata.name)
        self.assertEqual("mock-3", svc_3.metadata.labels["app"])

        svc_4 = core_api.read_namespaced_service(name="mock-4",
                                                 namespace="default")
        self.assertIsNotNone(svc_4)
        self.assertEqual("mock-4", svc_4.metadata.name)
        self.assertEqual("mock-4", svc_4.metadata.labels["app"])

        core_api.delete_namespaced_service(name="mock-3",
                                           namespace="default", body={})
        core_api.delete_namespaced_service(name="mock-4",
                                           namespace="default", body={})

    # Tests for creating multi-resource from directory

    def test_create_multi_resource_from_directory(self):
        """
        Should be able to create a service and a replication controller
        from a directory
        """
        k8s_client = client.api_client.ApiClient(configuration=self.config)
        utils.create_from_directory(
            k8s_client, self.path_prefix + "multi-resource/")
        core_api = client.CoreV1Api(k8s_client)
        svc = core_api.read_namespaced_service(name="mock",
                                               namespace="default")
        self.assertIsNotNone(svc)
        self.assertEqual("mock", svc.metadata.name)
        self.assertEqual("mock", svc.metadata.labels["app"])
        self.assertEqual("mock", svc.spec.selector["app"])

        ctr = core_api.read_namespaced_replication_controller(
            name="mock", namespace="default")
        self.assertIsNotNone(ctr)
        self.assertEqual("mock", ctr.metadata.name)
        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)

        core_api.delete_namespaced_replication_controller(
            name="mock", namespace="default", propagation_policy="Background")
        core_api.delete_namespaced_service(name="mock",
                                           namespace="default", body={})

    # Tests for multi-resource yaml objects

    def test_create_from_multi_resource_yaml(self):
        """
        Should be able to create a service and a replication controller
        from a multi-resource yaml file
        """
        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)
        self.assertEqual("mock", svc.metadata.name)
        self.assertEqual("mock", svc.metadata.labels["app"])
        self.assertEqual("mock", svc.spec.selector["app"])

        ctr = core_api.read_namespaced_replication_controller(
            name="mock", namespace="default")
        self.assertIsNotNone(ctr)
        self.assertEqual("mock", ctr.metadata.name)
        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)

        core_api.delete_namespaced_replication_controller(
            name="mock", namespace="default", propagation_policy="Background")
        core_api.delete_namespaced_service(name="mock",
                                           namespace="default", body={})

    def test_create_from_list_in_multi_resource_yaml(self):
        """
        Should be able to create the items in the PodList and a deployment
        specified in the multi-resource file
        """
        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)
        self.assertEqual("mock-pod-0", pod_0.metadata.name)
        self.assertEqual("mock-pod-0", pod_0.metadata.labels["app"])
        self.assertEqual("mock-pod-0", pod_0.spec.containers[0].name)
        self.assertEqual("busybox", pod_0.spec.containers[0].image)

        pod_1 = core_api.read_namespaced_pod(
            name="mock-pod-1", namespace="default")
        self.assertIsNotNone(pod_1)
        self.assertEqual("mock-pod-1", pod_1.metadata.name)
        self.assertEqual("mock-pod-1", pod_1.metadata.labels["app"])
        self.assertEqual("mock-pod-1", pod_1.spec.containers[0].name)
        self.assertEqual("busybox", pod_1.spec.containers[0].image)

        dep = app_api.read_namespaced_deployment(
            name="mock", namespace="default")
        self.assertIsNotNone(dep)
        self.assertEqual("mock", dep.metadata.name)
        self.assertEqual("mock", dep.spec.template.metadata.labels["app"])
        self.assertEqual(3, dep.spec.replicas)

        core_api.delete_namespaced_pod(
            name="mock-pod-0", namespace="default", body={})
        core_api.delete_namespaced_pod(
            name="mock-pod-1", namespace="default", body={})
        app_api.delete_namespaced_deployment(
            name="mock", namespace="default", body={})

    def test_create_from_multi_resource_yaml_with_conflict(self):
        """
        Should be able to create a service from the first yaml file.
        Should fail to create the same service from the second yaml file
        and create a replication controller.
        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(
            k8s_client, self.path_prefix + "yaml-conflict-first.yaml")
        core_api = client.CoreV1Api(k8s_client)
        svc = core_api.read_namespaced_service(name="mock-2",
                                               namespace="default")
        self.assertIsNotNone(svc)
        self.assertEqual("mock-2", svc.metadata.name)
        self.assertEqual("mock-2", svc.metadata.labels["app"])
        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(
                k8s_client, self.path_prefix + "yaml-conflict-multi.yaml")
        exp_error = ('Error from server (Conflict): {"kind":"Status",'
                     '"apiVersion":"v1","metadata":{},"status":"Failure",'
                     '"message":"services \\"mock-2\\" already exists",'
                     '"reason":"AlreadyExists","details":{"name":"mock-2",'
                     '"kind":"services"},"code":409}\n'
                     )
        self.assertEqual(exp_error, str(cm.exception))
        ctr = core_api.read_namespaced_replication_controller(
            name="mock-2", namespace="default")
        self.assertIsNotNone(ctr)
        core_api.delete_namespaced_replication_controller(
            name="mock-2", namespace="default", propagation_policy="Background")
        core_api.delete_namespaced_service(name="mock-2",
                                           namespace="default", body={})

    def test_create_from_multi_resource_yaml_with_multi_conflicts(self):
        """
        Should create an apps/v1 deployment
        and fail to create the same deployment twice.
        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(
                k8s_client, self.path_prefix + "triple-nginx.yaml")
        exp_error = ('Error from server (Conflict): {"kind":"Status",'
                     '"apiVersion":"v1","metadata":{},"status":"Failure",'
                     '"message":"deployments.apps \\"triple-nginx\\" '
                     'already exists","reason":"AlreadyExists",'
                     '"details":{"name":"triple-nginx","group":"apps",'
                     '"kind":"deployments"},"code":409}\n'
                     )
        exp_error += exp_error
        self.assertEqual(exp_error, str(cm.exception))
        ext_api = client.AppsV1Api(k8s_client)
        dep = ext_api.read_namespaced_deployment(name="triple-nginx",
                                                 namespace="default")
        self.assertIsNotNone(dep)
        ext_api.delete_namespaced_deployment(
            name="triple-nginx", namespace="default",
            body={})

    def test_create_namespaced_apps_deployment_from_yaml(self):
        """
        Should be able to create an apps/v1beta1 deployment
                in a test namespace.
        """
        k8s_client = client.api_client.ApiClient(configuration=self.config)
        utils.create_from_yaml(
            k8s_client, self.path_prefix + "apps-deployment.yaml",
            namespace=self.test_namespace)
        app_api = client.AppsV1Api(k8s_client)
        dep = app_api.read_namespaced_deployment(name="nginx-app",
                                                 namespace=self.test_namespace)
        self.assertIsNotNone(dep)
        app_api.delete_namespaced_deployment(
            name="nginx-app", namespace=self.test_namespace,
            body={})

    def test_create_from_list_in_multi_resource_yaml_namespaced(self):
        """
        Should be able to create the items in the PodList and a deployment
        specified in the multi-resource file in a test namespace
        """
        k8s_client = client.api_client.ApiClient(configuration=self.config)
        utils.create_from_yaml(
            k8s_client, self.path_prefix + "multi-resource-with-list.yaml",
            namespace=self.test_namespace)
        core_api = client.CoreV1Api(k8s_client)
        app_api = client.AppsV1Api(k8s_client)
        pod_0 = core_api.read_namespaced_pod(
            name="mock-pod-0", namespace=self.test_namespace)
        self.assertIsNotNone(pod_0)
        pod_1 = core_api.read_namespaced_pod(
            name="mock-pod-1", namespace=self.test_namespace)
        self.assertIsNotNone(pod_1)
        dep = app_api.read_namespaced_deployment(
            name="mock", namespace=self.test_namespace)
        self.assertIsNotNone(dep)
        core_api.delete_namespaced_pod(
            name="mock-pod-0", namespace=self.test_namespace, body={})
        core_api.delete_namespaced_pod(
            name="mock-pod-1", namespace=self.test_namespace, body={})
        app_api.delete_namespaced_deployment(
            name="mock", namespace=self.test_namespace, body={})


class TestUtilsUnitTests(unittest.TestCase):

    def test_parse_quantity(self):
        # == trivial returns ==
        self.assertEqual(quantity.parse_quantity(Decimal(1)), Decimal(1))
        self.assertEqual(quantity.parse_quantity(float(1)), Decimal(1))
        self.assertEqual(quantity.parse_quantity(1), Decimal(1))

        # == exceptions ==
        self.assertRaises(
            ValueError, lambda: quantity.parse_quantity("1000kb")
        )
        self.assertRaises(
            ValueError, lambda: quantity.parse_quantity("1000ki")
        )
        self.assertRaises(ValueError, lambda: quantity.parse_quantity("1000foo"))
        self.assertRaises(ValueError, lambda: quantity.parse_quantity("foo"))

        # == no suffix ==
        self.assertEqual(quantity.parse_quantity("1000"), Decimal(1000))

        # == base 1024 ==
        self.assertEqual(quantity.parse_quantity("1Ki"), Decimal(1024))
        self.assertEqual(quantity.parse_quantity("1Mi"), Decimal(1024**2))
        self.assertEqual(quantity.parse_quantity("1Gi"), Decimal(1024**3))
        self.assertEqual(quantity.parse_quantity("1Ti"), Decimal(1024**4))
        self.assertEqual(quantity.parse_quantity("1Pi"), Decimal(1024**5))
        self.assertEqual(quantity.parse_quantity("1Ei"), Decimal(1024**6))
        self.assertEqual(quantity.parse_quantity("1024Ki"), Decimal(1024**2))
        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("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("1E"), Decimal(1_000_000_000_000_000_000))
        self.assertEqual(quantity.parse_quantity("1000k"), Decimal(1_000_000))
        self.assertEqual(quantity.parse_quantity("500k"), Decimal(500_000))

    def test_format_quantity(self):
        """Unit test for quantity.format_quantity. Testing the different SI suffixes and
        function should return the expected string"""

        # == unknown suffixes ==
        self.assertRaises(
            ValueError, lambda: quantity.format_quantity(Decimal(1_000), "kb")
        )
        self.assertRaises(
            ValueError, lambda: quantity.format_quantity(Decimal(1_000), "ki")
        )
        self.assertRaises(
            ValueError, lambda: quantity.format_quantity(Decimal(1_000), "foo")
        )

        # == no suffix ==
        self.assertEqual(quantity.format_quantity(Decimal(1_000), ""), "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")
        # Decimal((1024**3)/3) are 0.3333333333333333148296162562Gi; expecting to
        # be quantized to 0.3Gi
        self.assertEqual(
            quantity.format_quantity(
                Decimal(
                    (1024**3) / 3),
                "Gi",
                quantize=Decimal(.5)),
            "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.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_000_000), "T"), "1T"
        )
        self.assertEqual(
            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"
        )
        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(
            quantity.format_quantity(
                Decimal(1_000_000 / 3), "k", quantize=Decimal(1000)
            ),
            "333k",
        )