controller/rootfs/scheduler/tests/test_services.py at 0998f3f7514e4f24ee96ed1e8a63cd7de50984aa · 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
129
130
131
132
133
134
135
136
137
138
139
"""
Unit tests for the Drycc 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 ServicesTest(TestCase):
    """Tests scheduler service calls"""

    def create(self, port=5000, protocol="TCP", target_port=5000):
        """
        Helper function to create and verify a service on the namespace
        """
        name = generate_random_name()
        service = self.scheduler.svc.create(self.namespace, name, ports=[{
            "port": port,
            "protocol": protocol,
            "targetPort": target_port,
        }])
        data = service.json()
        self.assertEqual(service.status_code, 201, data)
        self.assertEqual(data['metadata']['name'], name)
        return name

    def test_create_failure(self):
        # Kubernetes does not throw a 404 if queried on a non-existant Namespace
        with self.assertRaises(
            KubeHTTPException,
            msg='failed to create Service doesnotexist in Namespace {}: 404 Not Found'.format(self.namespace)  # noqa
        ):
            self.scheduler.svc.create('doesnotexist', 'doesnotexist')

    def test_create(self):
        # helper method takes care of the verification
        name = self.create()

        service = self.scheduler.svc.get(self.namespace, name).json()
        self.assertEqual(service['spec']['ports'][0]['targetPort'], 5000, service)

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

    def test_patch(self):
        name = self.create()
        expect = {
            "ports": [{
                "port": 6000,
                "protocol": "UDP",
                "targetPort": "6000",
            }],
            "version": 1,
        }
        self.scheduler.svc.patch(self.namespace, name, **expect)
        service = self.scheduler.svc.get(self.namespace, name).json()
        self.assertEqual(expect, {
            "ports": [{
                "port": service['spec']['ports'][0]['port'],
                "protocol": service['spec']['ports'][0]['protocol'],
                "targetPort": service['spec']['ports'][0]['targetPort'],
            }],
            "version": service['metadata']['resourceVersion']
        })

    def test_update(self):
        # test success
        name = self.create()
        service = self.scheduler.svc.get(self.namespace, name).json()
        self.assertEqual(service['spec']['ports'][0]['targetPort'], 5000, service)

        response = self.scheduler.svc.patch(self.namespace, name, ports=[{
            "port": service['spec']['ports'][0]['port'],
            "protocol": service['spec']['ports'][0]['protocol'],
            "targetPort": 5001,
        }])
        self.assertEqual(response.status_code, 200, response.json())

        service = self.scheduler.svc.get(self.namespace, name).json()
        self.assertEqual(service['spec']['ports'][0]['targetPort'], 5001, service)

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

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

    def test_get_services(self):
        # test success
        name = self.create()
        response = self.scheduler.svc.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_service_failure(self):
        # test failure
        with self.assertRaises(
            KubeHTTPException,
            msg='failed to get Service doesnotexist in Namespace {}: 404 Not Found'.format(self.namespace)  # noqa
        ):
            self.scheduler.svc.get(self.namespace, 'doesnotexist')

    def test_get_service(self):
        # test success
        name = self.create()
        response = self.scheduler.svc.get(self.namespace, name)
        data = response.json()
        self.assertEqual(response.status_code, 200, data)
        self.assertEqual(data['apiVersion'], 'v1')
        self.assertEqual(data['kind'], 'Service')
        metadata = {
            'name': name,
            'labels': {
                'app': self.namespace,
                'heritage': 'drycc'
            }
        }
        self.assertEqual(data['metadata'], data['metadata'] | metadata)
        self.assertEqual(data['spec']['ports'][0]['targetPort'], 5000)