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__init_one_container(self):
114 pod_name = 'pod_name'
115 _kwargs = {'args': ['arg1', 'arg2'],
116 'image': 'fake_image',
117 'command': 'fake_command'}
118 k8s_obj = kubernetes.KubernetesObject(pod_name, **_kwargs)
119 self.assertEqual(1, len(k8s_obj._containers))
120 container = k8s_obj._containers[0]
121 self.assertEqual(['arg1', 'arg2'], container._args)
122 self.assertEqual('fake_image', container._image)
123 self.assertEqual(['fake_command'], container._command)
124 self.assertEqual([], container._volume_mounts)
126 def test__init_multipe_containers(self):
127 pod_name = 'pod_name'
130 containers.append({'args': ['arg1', 'arg2'],
131 'image': 'fake_image_%s' % i,
132 'command': 'fake_command_%s' % i})
133 _kwargs = {'containers': containers}
134 k8s_obj = kubernetes.KubernetesObject(pod_name, **_kwargs)
135 self.assertEqual(5, len(k8s_obj._containers))
137 container = k8s_obj._containers[i]
138 self.assertEqual(['arg1', 'arg2'], container._args)
139 self.assertEqual('fake_image_%s' % i, container._image)
140 self.assertEqual(['fake_command_%s' % i], container._command)
141 self.assertEqual([], container._volume_mounts)
143 def test__add_volumes(self):
144 volume1 = {'name': 'fake_sshkey',
145 'configMap': {'name': 'fake_sshkey'}}
146 volume2 = {'name': 'volume2',
148 k8s_obj = kubernetes.KubernetesObject('name', ssh_key='fake_sshkey',
150 k8s_obj._add_volumes()
151 volumes = k8s_obj.template['spec']['template']['spec']['volumes']
152 self.assertEqual(sorted([volume1, volume2], key=lambda k: k['name']),
153 sorted(volumes, key=lambda k: k['name']))
155 def test__add_volumes_no_volumes(self):
156 volume1 = {'name': 'fake_sshkey',
157 'configMap': {'name': 'fake_sshkey'}}
158 k8s_obj = kubernetes.KubernetesObject('name', ssh_key='fake_sshkey')
159 k8s_obj._add_volumes()
160 volumes = k8s_obj.template['spec']['template']['spec']['volumes']
161 self.assertEqual([volume1], volumes)
163 def test__create_ssh_key_volume(self):
164 expected = {'name': 'fake_sshkey',
165 'configMap': {'name': 'fake_sshkey'}}
166 k8s_obj = kubernetes.KubernetesObject('name', ssh_key='fake_sshkey')
167 self.assertEqual(expected, k8s_obj._create_ssh_key_volume())
169 def test__create_volume_item(self):
170 for vol_type in kubernetes_utils.get_volume_types():
171 volume = {'name': 'vol_name',
175 kubernetes.KubernetesObject._create_volume_item(volume))
177 def test__create_volume_item_invalid_type(self):
178 volume = {'name': 'vol_name',
179 'invalid_type': 'data'}
180 with self.assertRaises(exceptions.KubernetesTemplateInvalidVolumeType):
181 kubernetes.KubernetesObject._create_volume_item(volume)
183 def test__add_security_context(self):
184 k8s_obj = kubernetes.KubernetesObject('pod_name')
185 self.assertNotIn('securityContext',
186 k8s_obj.template['spec']['template']['spec'])
188 k8s_obj._security_context = {'key_pod': 'value_pod'}
189 k8s_obj._add_security_context()
191 {'key_pod': 'value_pod'},
192 k8s_obj.template['spec']['template']['spec']['securityContext'])
194 def test__add_security_context_by_init(self):
198 {'securityContext': {'key%s' % i: 'value%s' % i}})
199 _kwargs = {'containers': containers,
200 'securityContext': {'key_pod': 'value_pod'}}
201 k8s_obj = kubernetes.KubernetesObject('pod_name', **_kwargs)
203 {'key_pod': 'value_pod'},
204 k8s_obj.template['spec']['template']['spec']['securityContext'])
207 k8s_obj.template['spec']['template']['spec']['containers'][i])
208 self.assertEqual({'key%s' % i: 'value%s' % i},
209 container['securityContext'])
211 def test__add_networks(self):
212 k8s_obj = kubernetes.KubernetesObject(
213 'name', networks=['network1', 'network2', 'network3'])
214 k8s_obj._add_networks()
216 template['spec']['template']['metadata']['annotations']['networks']
217 expected = ('[{"name": "network1"}, {"name": "network2"}, '
218 '{"name": "network3"}]')
219 self.assertEqual(expected, networks)
222 class ContainerObjectTestCase(base.BaseUnitTestCase):
224 def test__create_volume_mounts(self):
225 volume_mount = {'name': 'fake_name',
226 'mountPath': 'fake_path'}
227 ssh_vol = {'name': 'fake_ssh_key',
228 'mountPath': kubernetes.ContainerObject.SSH_MOUNT_PATH,
230 expected = copy.deepcopy(volume_mount)
231 expected['readOnly'] = False
232 expected = [expected, ssh_vol]
233 container_obj = kubernetes.ContainerObject(
234 'cname', 'fake_ssh_key', volumeMounts=[volume_mount])
235 output = container_obj._create_volume_mounts()
236 self.assertEqual(expected, output)
238 def test__create_volume_mounts_no_volume_mounts(self):
239 ssh_vol = {'name': 'fake_ssh_key2',
240 'mountPath': kubernetes.ContainerObject.SSH_MOUNT_PATH,
242 container_obj = kubernetes.ContainerObject('name', 'fake_ssh_key2')
243 output = container_obj._create_volume_mounts()
244 self.assertEqual([ssh_vol], output)
246 def test__create_volume_mounts_item(self):
247 volume_mount = {'name': 'fake_name',
248 'mountPath': 'fake_path'}
249 expected = copy.deepcopy(volume_mount)
250 expected['readOnly'] = False
251 output = kubernetes.ContainerObject._create_volume_mounts_item(
253 self.assertEqual(expected, output)
255 def test_get_container_item(self):
256 volume_mount = {'name': 'fake_name',
257 'mountPath': 'fake_path'}
258 args = ['arg1', 'arg2']
259 container_obj = kubernetes.ContainerObject(
260 'cname', ssh_key='fake_sshkey', volumeMount=[volume_mount],
262 expected = {'args': args,
263 'command': [kubernetes.ContainerObject.COMMAND_DEFAULT],
264 'image': kubernetes.ContainerObject.IMAGE_DEFAULT,
265 'name': 'cname-container',
266 'volumeMounts': container_obj._create_volume_mounts()}
267 self.assertEqual(expected, container_obj.get_container_item())
269 def test_get_container_item_with_security_context(self):
270 volume_mount = {'name': 'fake_name',
271 'mountPath': 'fake_path'}
272 args = ['arg1', 'arg2']
273 container_obj = kubernetes.ContainerObject(
274 'cname', ssh_key='fake_sshkey', volumeMount=[volume_mount],
275 args=args, securityContext={'key': 'value'})
276 expected = {'args': args,
277 'command': [kubernetes.ContainerObject.COMMAND_DEFAULT],
278 'image': kubernetes.ContainerObject.IMAGE_DEFAULT,
279 'name': 'cname-container',
280 'volumeMounts': container_obj._create_volume_mounts(),
281 'securityContext': {'key': 'value'}}
282 self.assertEqual(expected, container_obj.get_container_item())
285 class CustomResourceDefinitionObjectTestCase(base.BaseUnitTestCase):
287 def test__init(self):
290 'name': 'newcrds.ctx_name.com'
293 'group': 'ctx_name.com',
296 'names': {'plural': 'newcrds',
297 'singular': 'newcrd',
301 crd_obj = kubernetes.CustomResourceDefinitionObject(
302 'ctx_name', name='newcrd', version='v2', scope='scope')
303 self.assertEqual('newcrds.ctx_name.com', crd_obj._name)
304 self.assertEqual(template, crd_obj._template)
306 def test__init_missing_parameter(self):
307 with self.assertRaises(exceptions.KubernetesCRDObjectDefinitionError):
308 kubernetes.CustomResourceDefinitionObject('ctx_name',
312 class NetworkObjectTestCase(base.BaseUnitTestCase):
315 self.net_obj = kubernetes.NetworkObject(name='fake_name',
316 plugin='fake_plugin',
319 def test__init_missing_parameter(self):
320 with self.assertRaises(
321 exceptions.KubernetesNetworkObjectDefinitionError):
322 kubernetes.NetworkObject(name='name', plugin='plugin')
323 with self.assertRaises(
324 exceptions.KubernetesNetworkObjectDefinitionError):
325 kubernetes.NetworkObject(name='name', args='args')
326 with self.assertRaises(
327 exceptions.KubernetesNetworkObjectDefinitionError):
328 kubernetes.NetworkObject(args='args', plugin='plugin')
330 @mock.patch.object(kubernetes_utils, 'get_custom_resource_definition')
331 def test_crd(self, mock_get_crd):
332 mock_crd = mock.Mock()
333 mock_get_crd.return_value = mock_crd
334 net_obj = copy.deepcopy(self.net_obj)
335 self.assertEqual(mock_crd, net_obj.crd)
337 def test_template(self):
338 net_obj = copy.deepcopy(self.net_obj)
339 expected = {'apiVersion': 'group.com/v2',
340 'kind': kubernetes.NetworkObject.KIND,
342 'name': 'fake_name'},
343 'plugin': 'fake_plugin',
346 crd.spec.group = 'group.com'
347 crd.spec.version = 'v2'
349 self.assertEqual(expected, net_obj.template)
351 def test_group(self):
352 net_obj = copy.deepcopy(self.net_obj)
353 net_obj._crd = mock.Mock()
354 net_obj._crd.spec.group = 'fake_group'
355 self.assertEqual('fake_group', net_obj.group)
357 def test_version(self):
358 net_obj = copy.deepcopy(self.net_obj)
359 net_obj._crd = mock.Mock()
360 net_obj._crd.spec.version = 'version_4'
361 self.assertEqual('version_4', net_obj.version)
363 def test_plural(self):
364 net_obj = copy.deepcopy(self.net_obj)
365 net_obj._crd = mock.Mock()
366 net_obj._crd.spec.names.plural = 'name_ending_in_s'
367 self.assertEqual('name_ending_in_s', net_obj.plural)
369 def test_scope(self):
370 net_obj = copy.deepcopy(self.net_obj)
371 net_obj._crd = mock.Mock()
372 net_obj._crd.spec.scope = 'Cluster'
373 self.assertEqual('Cluster', net_obj.scope)
375 @mock.patch.object(kubernetes_utils, 'create_network')
376 def test_create(self, mock_create_network):
377 net_obj = copy.deepcopy(self.net_obj)
378 net_obj._scope = 'scope'
379 net_obj._group = 'group'
380 net_obj._version = 'version'
381 net_obj._plural = 'plural'
382 net_obj._template = 'template'
384 mock_create_network.assert_called_once_with(
385 'scope', 'group', 'version', 'plural', 'template')
387 @mock.patch.object(kubernetes_utils, 'delete_network')
388 def test_delete(self, mock_delete_network):
389 net_obj = copy.deepcopy(self.net_obj)
390 net_obj._scope = 'scope'
391 net_obj._group = 'group'
392 net_obj._version = 'version'
393 net_obj._plural = 'plural'
394 net_obj._name = 'name'
396 mock_delete_network.assert_called_once_with(
397 'scope', 'group', 'version', 'plural', 'name')