controller/rootfs/scheduler/tests/test_secrets.py at 23e70e8a91645c6276ceedce7043489088baec63 · 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
"""
Unit tests for the Drycc scheduler module.

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


class SecretsTest(TestCase):
    """Tests scheduler secret calls"""

    def create(self):
        """
        Helper function to create and verify a secret on the namespace
        """
        name = generate_random_name()
        data = {
            'foo': 'bar',
            'this': 'that',
            'empty': None,
        }
        secret = self.scheduler.secret.create(self.namespace, name, data)
        data = secret.json()
        self.assertEqual(secret.status_code, 201, data)
        self.assertEqual(data['metadata']['name'], name)
        self.assertIn('foo', data['data'])
        self.assertIn('this', data['data'])
        return name

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

        with self.assertRaises(
            KubeException,
            msg='invlaid is not a supported secret type. Use one of the following: Opaque, kubernetes.io/dockerconfigjson'  # noqa
        ):
            self.scheduler.secret.create(self.namespace, 'foo', {}, secret_type='invalid')

    def test_create(self):
        name = self.create()
        secret = self.scheduler.secret.get(self.namespace, name).json()
        self.assertEqual(secret['data']['foo'], 'bar', secret)
        self.assertEqual(secret['data']['this'], 'that', secret)
        self.assertEqual(secret['type'], 'Opaque')

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

    def test_update(self):
        # test success
        name = self.create()
        secret = self.scheduler.secret.get(self.namespace, name).json()
        self.assertEqual(secret['data']['foo'], 'bar', secret)
        self.assertEqual(secret['data']['this'], 'that', secret)
        self.assertEqual(secret['type'], 'Opaque')

        secret['data']['foo'] = 5001
        response = self.scheduler.secret.update(self.namespace, name, secret['data'])
        self.assertEqual(response.status_code, 200, response.json())

        secret = self.scheduler.secret.get(self.namespace, name).json()
        self.assertEqual(secret['data']['foo'], '5001', secret)

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

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

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

    def test_get_secret(self):
        # test success
        name = self.create()
        response = self.scheduler.secret.get(self.namespace, name)
        data = response.json()
        self.assertEqual(response.status_code, 200, data)
        self.assertEqual(data['apiVersion'], 'v1')
        self.assertEqual(data['kind'], 'Secret')
        metadata = {
            'name': name,
            'labels': {
                'app': self.namespace,
                'heritage': 'drycc'
            }
        }
        self.assertEqual(data['metadata'], data['metadata'] | metadata)
        self.assertEqual(data['data']['foo'], 'bar', data)
        self.assertEqual(data['data']['this'], 'that', data)
        self.assertEqual(data['type'], 'Opaque')