test_pods.py

"""
Unit tests for the Deis scheduler module.

Run the tests with './manage.py test scheduler'
"""
from unittest import mock
from datetime import datetime, timedelta
from scheduler import KubeHTTPException
from scheduler.tests import TestCase
from scheduler.utils import generate_random_name


class PodsTest(TestCase):
    """Tests scheduler pod calls"""

    def create(self, namespace=None, name=generate_random_name(), **kwargs):
        """
        Helper function to create and verify a pod on the namespace
        """
        namespace = self.namespace if namespace is None else namespace
        # these are all required even if it is kwargs...
        kwargs = {
            'app_type': kwargs.get('app_type', 'web'),
            'version': kwargs.get('version', 'v99'),
            'replicas': kwargs.get('replicas', 4),
            'pod_termination_grace_period_seconds': 2,
            'image': 'quay.io/fake/image',
            'entrypoint': 'sh',
            'command': 'start',
            'deploy_timeout': 10,
        }

        pod = self.scheduler.pod.create(namespace, name, **kwargs)
        self.assertEqual(pod.status_code, 201, pod.json())
        return name

    def test_create_failure(self):
        with self.assertRaises(
            KubeHTTPException,
            msg='failed to create Pod doesnotexist in Namespace {}: 404 Not Found'.format(self.namespace)  # noqa
        ):
            self.create('doesnotexist', 'doesnotexist')

    def test_create(self):
        self.create()

    def test_delete_failure(self):
        # test failure
        with self.assertRaises(
            KubeHTTPException,
            msg='failed to delete Pod foo in Namespace {}: 404 Not Found'.format(self.namespace)  # noqa
        ):
            self.scheduler.pod.delete(self.namespace, 'foo')

    def test_delete(self):
        # test success
        name = self.create()
        response = self.scheduler.pod.delete(self.namespace, name)
        data = response.json()
        self.assertEqual(response.status_code, 200, data)

    def test_get_pods(self):
        # test success
        name = self.create()
        response = self.scheduler.pod.get(self.namespace)
        data = response.json()
        self.assertEqual(response.status_code, 200, data)
        self.assertIn('items', data)
        self.assertEqual(1, len(data['items']), data['items'])
        # simple verify of data
        self.assertEqual(data['items'][0]['metadata']['name'], name, data)

    def test_get_pod_failure(self):
        # test failure
        with self.assertRaises(
            KubeHTTPException,
            msg='failed to get Pod doesnotexist in Namespace {}: 404 Not Found'.format(self.namespace)  # noqa
        ):
            self.scheduler.pod.get(self.namespace, 'doesnotexist')

    def test_get_pod(self):
        # test success
        name = self.create()
        response = self.scheduler.pod.get(self.namespace, name)
        data = response.json()
        self.assertEqual(response.status_code, 200, data)
        self.assertEqual(data['kind'], 'Pod')
        self.assertEqual(data['metadata']['name'], name)
        self.assertDictContainsSubset(
            {
                'app': self.namespace,
                'heritage': 'deis'
            },
            data['metadata']['labels']
        )

    def test_liveness_status(self):
        # Missing Ready type means pod has passed liveness check
        pod = {'status': {'conditions': []}}
        self.assertTrue(self.scheduler.pod.liveness_status(pod))

        # fake out a minimal pod structure for success
        pod['status']['conditions'].append({
            'type': 'Ready',
            'status': 'True'
        })
        self.assertTrue(self.scheduler.pod.liveness_status(pod))

        # fake out a minimal pod structure for failure
        pod['status']['conditions'][0]['status'] = 'False'
        self.assertFalse(self.scheduler.pod.liveness_status(pod))

    def test_readiness_status(self):
        # create a pod to then manipulate
        name = self.create()
        pod = self.scheduler.pod.get(self.namespace, name).json()

        # on a newly created pod has an overall "Running" status with ready state on a container
        self.assertEqual(self.scheduler.pod.readiness_status(pod), 'Running')

        # on a newly created pod has been deleted with ready state on a container
        pod['metadata']['deletionTimestamp'] = 'fake'
        self.assertEqual(self.scheduler.pod.readiness_status(pod), 'Terminating')
        del pod['metadata']['deletionTimestamp']

        # now say the app container is not Ready
        container = pod['status']['containerStatuses'][0]
        container['ready'] = False

        # state of the container is Starting when container is not ready but says Running
        self.assertTrue('running' in container['state'].keys())
        self.assertEqual(self.scheduler.pod.readiness_status(pod), 'Starting')

        # verify Terminating status
        del container['state']['running']
        container['state']['terminated'] = 'fake'
        self.assertEqual(self.scheduler.pod.readiness_status(pod), 'Terminating')

        # test metdata terminating
        del container['state']['terminated']
        pod['metadata']['deletionTimestamp'] = 'fake'
        self.assertEqual(self.scheduler.pod.readiness_status(pod), 'Terminating')
        del pod['metadata']['deletionTimestamp']

        # inject fake state
        container['state']['random'] = 'fake'
        self.assertEqual(self.scheduler.pod.readiness_status(pod), 'Unknown')

        # no containers around means Unknown
        pod['status']['containerStatuses'] = []
        self.assertEqual(self.scheduler.pod.readiness_status(pod), 'Unknown')

    def test_ready(self):
        # create a pod to then manipulate
        name = self.create()
        pod = self.scheduler.pod.get(self.namespace, name).json()

        # pod itself shouldn't be ready yet
        self.assertFalse(self.scheduler.pod.ready(pod))

        # only pod but no other probe
        pod['status']['phase'] = 'Running'
        # fake out functions for failure
        with mock.patch('scheduler.resources.pod.Pod.readiness_status') as ready:
            ready.return_value = 'Starting'

            # only readiness is ready so overall ready status is False
            self.assertFalse(self.scheduler.pod.ready(pod))

            with mock.patch('scheduler.resources.pod.Pod.liveness_status') as liveness:
                liveness.return_value = False

                # all things have lined up, go time
                self.assertFalse(self.scheduler.pod.ready(pod))

        # fake out other functions since they are tested by themselves
        with mock.patch('scheduler.resources.pod.Pod.readiness_status') as ready:
            ready.return_value = 'Running'
            with mock.patch('scheduler.resources.pod.Pod.liveness_status') as liveness:
                # keep liveness as failing for now
                liveness.return_value = False

                # only readiness is ready so overall ready status is False
                self.assertFalse(self.scheduler.pod.ready(pod))

                # all things have lined up, go time
                liveness.return_value = True
                self.assertTrue(self.scheduler.pod.ready(pod))

    def test_deleted(self):
        # create a pod to then manipulate
        name = self.create()
        pod = self.scheduler.pod.get(self.namespace, name).json()

        # pod should no be deleted yet
        self.assertFalse(self.scheduler.pod.deleted(pod))

        # set deleted 10 minutes in the past
        ts_deleted = datetime.utcnow() - timedelta(minutes=10)
        pod['metadata']['deletionTimestamp'] = ts_deleted.strftime(self.scheduler.DATETIME_FORMAT)
        self.assertTrue(self.scheduler.pod.deleted(pod))