<commit_before>#!/usr/bin/env python
# -*- coding: utf-8 -*-

#
# This file is subject to the terms and conditions defined in
# file 'LICENSE.md', which is part of this source code package.
#

import unittest
import uuid

from kubernetes import K8sPod, K8sConfig, K8sContainer
from kubernetes.K8sExceptions import *
from kubernetes.models.v1 import Pod, ObjectMeta, PodSpec, PodStatus
from tests import utils


class K8sPodTest(unittest.TestCase):
    def setUp(self):
        pass

    def tearDown(self):
        utils.cleanup_pods()

    # ------------------------------------------------------------------------------------- init

    def test_init_no_args(self):
        try:
            K8sPod()
            self.fail("Should not fail.")
        except SyntaxError:
            pass
        except IOError:
            pass
        except Exception as err:
            self.fail("Unhandled exception: [ {0} ]".format(err.__class__.__name__))

    def test_init_with_invalid_config(self):
        config = object()
        try:
            K8sPod(config=config)
            self.fail("Should not fail.")
        except SyntaxError:
            pass
        except IOError:
            pass
        except Exception as err:
            self.fail("Unhandled exception: [ {0} ]".format(err.__class__.__name__))

    def test_init_with_invalid_name(self):
        name = object()
        try:
            K8sPod(name=name)
            self.fail("Should not fail.")
        except SyntaxError:
            pass
        except IOError:
            pass
        except Exception as err:
            self.fail("Unhandled exception: [ {0} ]".format(err.__class__.__name__))

    def test_init_with_name(self):
        name = "yomama"
        pod = utils.create_pod(name=name)
        self.assertIsNotNone(pod)
        self.assertIsInstance(pod, K8sPod)
        self.assertEqual(pod.name, name)

    def test_init_with_config_and_pull_secrets(self):
        ps = [{'name': 'yomama'}]
        name = "sofat"
        cfg = K8sConfig(kubeconfig=utils.kubeconfig_fallback, pull_secret=ps)
        pod = utils.create_pod(config=cfg, name=name)
        self.assertIsNotNone(pod.config)
        self.assertEqual(ps, pod.config.pull_secret)

    # ------------------------------------------------------------------------------------- struct

    def test_struct_k8spod(self):
        name = "yomama"
        pod = utils.create_pod(name=name)
        self.assertIsNotNone(pod)
        self.assertIsInstance(pod, K8sPod)
        self.assertIsNotNone(pod.model)
        self.assertIsInstance(pod.model, Pod)

    def test_struct_pod(self):
        name = "yomama"
        pod = utils.create_pod(name=name)
        self.assertIsInstance(pod.model, Pod)
        self.assertIsInstance(pod.model.metadata, ObjectMeta)
        self.assertIsInstance(pod.model.spec, PodSpec)
        self.assertIsInstance(pod.model.status, PodStatus)

    # ------------------------------------------------------------------------------------- add annotation

    def test_add_annotation_none_args(self):
        name = "yopod-{0}".format(str(uuid.uuid4()))
        pod = utils.create_pod(name=name)
        try:
            pod.add_annotation()
        except Exception as err:
            self.assertIsInstance(err, SyntaxError)

    def test_add_annotation_invalid_args(self):
        name = "yopod-{0}".format(str(uuid.uuid4()))
        pod = utils.create_pod(name=name)
        k = object()
        v = object()
        try:
            pod.add_annotation(k, v)
        except Exception as err:
            self.assertIsInstance(err, SyntaxError)

    def test_add_annotation(self):
        name = "yopod-{0}".format(str(uuid.uuid4()))
        pod = utils.create_pod(name=name)
        k = "yokey"
        v = "yovalue"
        pod.add_annotation(k, v)
        self.assertEqual(1, len(pod.model.metadata.annotations))
        self.assertIn(k, pod.model.metadata.annotations)
        self.assertEqual(v, pod.model.metadata.annotations[k])

    # --------------------------------------------------------------------------------- add container

    def test_pod_add_container_invalid(self):
        name = "yoname"
        obj = utils.create_pod(name=name)
        c = object()
        with self.assertRaises(SyntaxError):
            obj.add_container(c)

    def test_pod_add_container(self):
        name = "yoname"
        pod = utils.create_pod(name=name)
        self.assertEqual(0, len(pod.model.spec.containers))
        name = "yopod"
        image = "redis"
        c = K8sContainer(name=name, image=image)
        pod.add_container(c)
        self.assertEqual(1, len(pod.model.spec.containers))
        self.assertEqual(c.model, pod.model.spec.containers[0])

    # ------------------------------------------------------------------------------------- add label

    def test_add_label_none_args(self):
        name = "yopod-{0}".format(str(uuid.uuid4()))
        pod = utils.create_pod(name=name)
        with self.assertRaises(SyntaxError):
            pod.add_label()

    def test_add_label_invalid_args(self):
        name = "yopod-{0}".format(str(uuid.uuid4()))
        pod = utils.create_pod(name=name)
        k = object()
        v = object()
        with self.assertRaises(SyntaxError):
            pod.add_label(k, v)

    def test_add_label(self):
        name = "yopod-{0}".format(str(uuid.uuid4()))
        pod = utils.create_pod(name=name)
        k = "yokey"
        v_in = "yovalue"
        pod.add_label(k, v_in)
        v_out = pod.get_label(k)
        self.assertEqual(v_in, v_out)

    # --------------------------------------------------------------------------------- add pull secret

    def test_pod_add_image_pull_secrets_none_arg(self):
        name = "yoname"
        obj = utils.create_pod(name=name)
        secretname = None
        try:
            obj.add_image_pull_secrets(secretname)
            self.fail("Should not fail.")
        except Exception as err:
            self.assertIsInstance(err, SyntaxError)

    def test_pod_add_image_pull_secrets_invalid_arg(self):
        name = "yoname"
        obj = utils.create_pod(name=name)
        secretname = 666
        try:
            obj.add_image_pull_secrets(secretname)
            self.fail("Should not fail.")
        except Exception as err:
            self.assertIsInstance(err, SyntaxError)

    def test_pod_add_image_pull_secrets(self):
        name = "yoname"
        pod = utils.create_pod(name=name)
        secret = [{'name': 'yosecret'}]
        pod.add_image_pull_secrets(secret)
        self.assertEqual(1, len(pod.model.spec.image_pull_secrets))
        self.assertEqual(secret, pod.model.spec.image_pull_secrets[0])

    # --------------------------------------------------------------------------------- add volume

    # def test_pod_add_volume_invalid(self):
    #     name = "yoname"
    #     obj = utils.create_pod(name=name)
    #     vol = object()
    #     with self.assertRaises(SyntaxError):
    #         obj.add_volume(vol)
    #
    # def test_pod_add_volume_emptydir(self):
    #     name = "yoname"
    #     obj = utils.create_pod(name=name)
    #     config = utils.create_config()
    #     vol = K8sVolume(config=config, name=name, mount_path="/var/test", type='emptyDir')
    #     obj.add_volume(vol)
    #     self.assertEqual(1, len(obj.model.model['spec']['volumes']))
    #     self.assertEqual(1, len(obj.model.pod_spec.model['volumes']))
    #     self.assertEqual(name, obj.model.model['spec']['volumes'][0]['name'])<commit_msg>K8sPod: unit tests pass<commit_after>#!/usr/bin/env python
# -*- coding: utf-8 -*-

#
# This file is subject to the terms and conditions defined in
# file 'LICENSE.md', which is part of this source code package.
#

import unittest
import uuid

from kubernetes import K8sPod, K8sConfig, K8sContainer
from kubernetes.K8sExceptions import *
from kubernetes.models.v1 import Pod, ObjectMeta, PodSpec, PodStatus
from tests import utils


class K8sPodTest(unittest.TestCase):
    def setUp(self):
        pass

    def tearDown(self):
        utils.cleanup_pods()

    # ------------------------------------------------------------------------------------- init

    def test_init_no_args(self):
        try:
            K8sPod()
            self.fail("Should not fail.")
        except SyntaxError:
            pass
        except IOError:
            pass
        except Exception as err:
            self.fail("Unhandled exception: [ {0} ]".format(err.__class__.__name__))

    def test_init_with_invalid_config(self):
        config = object()
        try:
            K8sPod(config=config)
            self.fail("Should not fail.")
        except SyntaxError:
            pass
        except IOError:
            pass
        except Exception as err:
            self.fail("Unhandled exception: [ {0} ]".format(err.__class__.__name__))

    def test_init_with_invalid_name(self):
        name = object()
        try:
            K8sPod(name=name)
            self.fail("Should not fail.")
        except SyntaxError:
            pass
        except IOError:
            pass
        except Exception as err:
            self.fail("Unhandled exception: [ {0} ]".format(err.__class__.__name__))

    def test_init_with_name(self):
        name = "yomama"
        pod = utils.create_pod(name=name)
        self.assertIsNotNone(pod)
        self.assertIsInstance(pod, K8sPod)
        self.assertEqual(pod.name, name)

    def test_init_with_config_and_pull_secrets(self):
        ps = [{'name': 'yomama'}]
        name = "sofat"
        cfg = K8sConfig(kubeconfig=utils.kubeconfig_fallback, pull_secret=ps)
        pod = utils.create_pod(config=cfg, name=name)
        self.assertIsNotNone(pod.config)
        self.assertEqual(ps, pod.config.pull_secret)

    # ------------------------------------------------------------------------------------- struct

    def test_struct_k8spod(self):
        name = "yomama"
        pod = utils.create_pod(name=name)
        self.assertIsNotNone(pod)
        self.assertIsInstance(pod, K8sPod)
        self.assertIsNotNone(pod.model)
        self.assertIsInstance(pod.model, Pod)

    def test_struct_pod(self):
        name = "yomama"
        pod = utils.create_pod(name=name)
        self.assertIsInstance(pod.model, Pod)
        self.assertIsInstance(pod.model.metadata, ObjectMeta)
        self.assertIsInstance(pod.model.spec, PodSpec)
        self.assertIsInstance(pod.model.status, PodStatus)

    # ------------------------------------------------------------------------------------- add annotation

    def test_add_annotation_none_args(self):
        name = "yopod-{0}".format(str(uuid.uuid4()))
        pod = utils.create_pod(name=name)
        try:
            pod.add_annotation()
        except Exception as err:
            self.assertIsInstance(err, SyntaxError)

    def test_add_annotation_invalid_args(self):
        name = "yopod-{0}".format(str(uuid.uuid4()))
        pod = utils.create_pod(name=name)
        k = object()
        v = object()
        try:
            pod.add_annotation(k, v)
        except Exception as err:
            self.assertIsInstance(err, SyntaxError)

    def test_add_annotation(self):
        name = "yopod-{0}".format(str(uuid.uuid4()))
        pod = utils.create_pod(name=name)
        k = "yokey"
        v = "yovalue"
        pod.add_annotation(k, v)
        self.assertEqual(1, len(pod.model.metadata.annotations))
        self.assertIn(k, pod.model.metadata.annotations)
        self.assertEqual(v, pod.model.metadata.annotations[k])

    # --------------------------------------------------------------------------------- add container

    def test_pod_add_container_invalid(self):
        name = "yoname"
        obj = utils.create_pod(name=name)
        c = object()
        with self.assertRaises(SyntaxError):
            obj.add_container(c)

    def test_pod_add_container(self):
        name = "yoname"
        pod = utils.create_pod(name=name)
        self.assertEqual(0, len(pod.model.spec.containers))
        name = "yopod"
        image = "redis"
        c = K8sContainer(name=name, image=image)
        pod.add_container(c)
        self.assertEqual(1, len(pod.model.spec.containers))
        self.assertEqual(c.model, pod.model.spec.containers[0])

    # ------------------------------------------------------------------------------------- add label

    def test_add_label_none_args(self):
        name = "yopod-{0}".format(str(uuid.uuid4()))
        pod = utils.create_pod(name=name)
        with self.assertRaises(SyntaxError):
            pod.add_label()

    def test_add_label_invalid_args(self):
        name = "yopod-{0}".format(str(uuid.uuid4()))
        pod = utils.create_pod(name=name)
        k = object()
        v = object()
        with self.assertRaises(SyntaxError):
            pod.add_label(k, v)

    def test_add_label(self):
        name = "yopod-{0}".format(str(uuid.uuid4()))
        pod = utils.create_pod(name=name)
        k = "yokey"
        v_in = "yovalue"
        pod.add_label(k, v_in)
        v_out = pod.get_label(k)
        self.assertEqual(v_in, v_out)

    # --------------------------------------------------------------------------------- add pull secret

    def test_pod_add_image_pull_secrets_none_arg(self):
        name = "yoname"
        obj = utils.create_pod(name=name)
        secretname = None
        try:
            obj.add_image_pull_secrets(secretname)
            self.fail("Should not fail.")
        except Exception as err:
            self.assertIsInstance(err, SyntaxError)

    def test_pod_add_image_pull_secrets_invalid_arg(self):
        name = "yoname"
        obj = utils.create_pod(name=name)
        secretname = 666
        try:
            obj.add_image_pull_secrets(secretname)
            self.fail("Should not fail.")
        except Exception as err:
            self.assertIsInstance(err, SyntaxError)

    def test_pod_add_image_pull_secrets(self):
        name = "yoname"
        pod = utils.create_pod(name=name)
        secret = [{'name': 'yosecret'}]
        pod.add_image_pull_secrets(secret)
        self.assertEqual(1, len(pod.model.spec.image_pull_secrets))
        self.assertEqual(secret, pod.model.spec.image_pull_secrets)

    # --------------------------------------------------------------------------------- add volume

    # def test_pod_add_volume_invalid(self):
    #     name = "yoname"
    #     obj = utils.create_pod(name=name)
    #     vol = object()
    #     with self.assertRaises(SyntaxError):
    #         obj.add_volume(vol)
    #
    # def test_pod_add_volume_emptydir(self):
    #     name = "yoname"
    #     obj = utils.create_pod(name=name)
    #     config = utils.create_config()
    #     vol = K8sVolume(config=config, name=name, mount_path="/var/test", type='emptyDir')
    #     obj.add_volume(vol)
    #     self.assertEqual(1, len(obj.model.model['spec']['volumes']))
    #     self.assertEqual(1, len(obj.model.pod_spec.model['volumes']))
    #     self.assertEqual(name, obj.model.model['spec']['volumes'][0]['name'])
{'fraction_non_alphanumeric': 0.18311352640996753, 'fraction_numerical': 0.005699516203857114, 'mean_word_length': 1.8493202416918428, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '6849075', 'n_tokens_mistral': 4552, 'n_tokens_neox': 4481, 'n_words': 790}