controller/rootfs/scheduler/tests/test_replicationcontrollers.py at 7306202c34d67218c9080f8545de660da32723ff · drycc/controller · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
"""
Unit tests for the Deis scheduler module.

Run the tests with './manage.py test scheduler'
"""
from scheduler import KubeHTTPException
from scheduler.tests import TestCase
from scheduler.utils import generate_random_name


class ReplicationControllersTest(TestCase):
    """Tests scheduler rc calls"""

    def create(self, namespace=None, name=generate_random_name(), **kwargs):
        """
        Helper function to create and verify a rc 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),
        }

        rc = self.scheduler.rc.create(namespace, name, 'quay.io/fake/image',
                                      'sh', 'start', **kwargs)
        data = rc.json()
        self.assertEqual(rc.status_code, 201, data)
        return name

    def scale_rc(self, namespace=None, name=generate_random_name(), **kwargs):
        """
        Helper function to scale and verify a deployment 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),
            'deploy_timeout': 120
        }

        self.scheduler.scale_rc(namespace, name, 'quay.io/fake/image', 'sh', 'start', **kwargs)
        return name

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

    def test_create(self):
        self.create()

    def test_update_rc_failure(self):
        # test failure
        with self.assertRaises(
            KubeHTTPException,
            msg='failed to update ReplicationController foo in Namespace {}: 404 Not Found'.format(self.namespace)  # noqa
        ):
            self.scheduler.rc.update(self.namespace, 'foo', {})

    def test_update(self):
        # test success
        name = self.create()
        rc = self.scheduler.rc.get(self.namespace, name).json()
        self.assertEqual(rc['spec']['replicas'], 4, rc)

        rc['spec']['replicas'] = 2
        response = self.scheduler.rc.update(self.namespace, name, rc)
        self.assertEqual(response.status_code, 200, response.json())

        rc = self.scheduler.rc.get(self.namespace, name).json()
        self.assertEqual(rc['spec']['replicas'], 2, rc)

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

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

    def test_get_rcs(self):
        # test success
        name = self.create()
        response = self.scheduler.rc.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)

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

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