1 ##############################################################################
2 # Copyright (c) 2017 Intel Corporation
4 # All rights reserved. This program and the accompanying materials
5 # are made available under the terms of the Apache License, Version 2.0
6 # which accompanies this distribution, and is available at
7 # http://www.apache.org/licenses/LICENSE-2.0
8 ##############################################################################
14 from yardstick.common import exceptions
15 from yardstick.common import kubernetes_utils
16 from yardstick.orchestrator import kubernetes
17 from yardstick.tests.unit import base
20 class GetTemplateTestCase(base.BaseUnitTestCase):
22 def test_get_template(self):
25 "kind": "ReplicationController",
27 "name": "host-k8s-86096c30"
34 "app": "host-k8s-86096c30"
42 "chmod 700 ~/.ssh; chmod 600 ~/.ssh/*; \
43 service ssh restart;while true ; do sleep 10000; done"
48 "image": "openretriever/yardstick",
49 "name": "host-k8s-86096c30-container",
52 "mountPath": "/tmp/.ssh/",
53 "name": "k8s-86096c30-key",
62 "name": "k8s-86096c30-key"
64 "name": "k8s-86096c30-key"
68 "kubernetes.io/hostname": "node-01"
75 'command': '/bin/bash',
76 'args': ['-c', 'chmod 700 ~/.ssh; chmod 600 ~/.ssh/*; \
77 service ssh restart;while true ; do sleep 10000; done'],
78 'ssh_key': 'k8s-86096c30-key',
79 'nodeSelector': {'kubernetes.io/hostname': 'node-01'},
82 name = 'host-k8s-86096c30'
83 output_r = kubernetes.KubernetesObject(name, **input_s).get_template()
84 self.assertEqual(output_r, output_t)
87 class GetRcPodsTestCase(base.BaseUnitTestCase):
89 @mock.patch('yardstick.orchestrator.kubernetes.k8s_utils.get_pod_list')
90 def test_get_rc_pods(self, mock_get_pod_list):
93 'image': 'openretriever/yardstick',
94 'command': '/bin/bash',
95 'args': ['-c', 'chmod 700 ~/.ssh; chmod 600 ~/.ssh/*; \
96 service ssh restart;while true ; do sleep 10000; done']
99 'image': 'openretriever/yardstick',
100 'command': '/bin/bash',
101 'args': ['-c', 'chmod 700 ~/.ssh; chmod 600 ~/.ssh/*; \
102 service ssh restart;while true ; do sleep 10000; done']
105 k8s_template = kubernetes.KubernetesTemplate('k8s-86096c30', servers)
106 mock_get_pod_list.return_value.items = []
107 pods = k8s_template.get_rc_pods()
108 self.assertEqual(pods, [])
111 class KubernetesObjectTestCase(base.BaseUnitTestCase):
113 def test__add_volumes(self):
114 volume1 = {'name': 'fake_sshkey',
115 'configMap': {'name': 'fake_sshkey'}}
116 volume2 = {'name': 'volume2',
118 k8s_obj = kubernetes.KubernetesObject('name', ssh_key='fake_sshkey',
120 k8s_obj._add_volumes()
121 volumes = k8s_obj.template['spec']['template']['spec']['volumes']
122 self.assertEqual(sorted([volume1, volume2], key=lambda k: k['name']),
123 sorted(volumes, key=lambda k: k['name']))
125 def test__add_volumes_no_volumes(self):
126 volume1 = {'name': 'fake_sshkey',
127 'configMap': {'name': 'fake_sshkey'}}
128 k8s_obj = kubernetes.KubernetesObject('name', ssh_key='fake_sshkey')
129 k8s_obj._add_volumes()
130 volumes = k8s_obj.template['spec']['template']['spec']['volumes']
131 self.assertEqual([volume1], volumes)
133 def test__create_ssh_key_volume(self):
134 expected = {'name': 'fake_sshkey',
135 'configMap': {'name': 'fake_sshkey'}}
136 k8s_obj = kubernetes.KubernetesObject('name', ssh_key='fake_sshkey')
137 self.assertEqual(expected, k8s_obj._create_ssh_key_volume())
139 def test__create_volume_item(self):
140 for vol_type in kubernetes_utils.get_volume_types():
141 volume = {'name': 'vol_name',
145 kubernetes.KubernetesObject._create_volume_item(volume))
147 def test__create_volume_item_invalid_type(self):
148 volume = {'name': 'vol_name',
149 'invalid_type': 'data'}
150 with self.assertRaises(exceptions.KubernetesTemplateInvalidVolumeType):
151 kubernetes.KubernetesObject._create_volume_item(volume)
153 def test__create_volume_mounts(self):
154 volume_mount = {'name': 'fake_name',
155 'mountPath': 'fake_path'}
156 ssh_vol = {'name': kubernetes.KubernetesObject.SSHKEY_DEFAULT,
157 'mountPath': kubernetes.KubernetesObject.SSH_MOUNT_PATH,
159 expected = copy.deepcopy(volume_mount)
160 expected['readOnly'] = False
161 expected = [expected, ssh_vol]
162 k8s_obj = kubernetes.KubernetesObject('name',
163 volumeMounts=[volume_mount])
164 output = k8s_obj._create_volume_mounts()
165 self.assertEqual(expected, output)
167 def test__create_volume_mounts_no_volume_mounts(self):
168 ssh_vol = {'name': kubernetes.KubernetesObject.SSHKEY_DEFAULT,
169 'mountPath': kubernetes.KubernetesObject.SSH_MOUNT_PATH,
171 k8s_obj = kubernetes.KubernetesObject('name')
172 output = k8s_obj._create_volume_mounts()
173 self.assertEqual([ssh_vol], output)
175 def test__create_volume_mounts_item(self):
176 volume_mount = {'name': 'fake_name',
177 'mountPath': 'fake_path'}
178 expected = copy.deepcopy(volume_mount)
179 expected['readOnly'] = False
180 output = kubernetes.KubernetesObject._create_volume_mounts_item(
182 self.assertEqual(expected, output)
184 def test__create_container_item(self):
185 volume_mount = {'name': 'fake_name',
186 'mountPath': 'fake_path'}
187 args = ['arg1', 'arg2']
188 k8s_obj = kubernetes.KubernetesObject(
189 'cname', ssh_key='fake_sshkey', volumeMount=[volume_mount],
191 expected = {'args': args,
192 'command': [kubernetes.KubernetesObject.COMMAND_DEFAULT],
193 'image': kubernetes.KubernetesObject.IMAGE_DEFAULT,
194 'name': 'cname-container',
195 'volumeMounts': k8s_obj._create_volume_mounts()}
196 self.assertEqual(expected, k8s_obj._create_container_item())