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.ReplicationControllerObject(
84 name, **input_s).get_template()
85 self.assertEqual(output_r, output_t)
88 class GetRcPodsTestCase(base.BaseUnitTestCase):
90 @mock.patch('yardstick.orchestrator.kubernetes.k8s_utils.get_pod_list')
91 def test_get_rc_pods(self, mock_get_pod_list):
94 'image': 'openretriever/yardstick',
95 'command': '/bin/bash',
96 'args': ['-c', 'chmod 700 ~/.ssh; chmod 600 ~/.ssh/*; \
97 service ssh restart;while true ; do sleep 10000; done']
100 'image': 'openretriever/yardstick',
101 'command': '/bin/bash',
102 'args': ['-c', 'chmod 700 ~/.ssh; chmod 600 ~/.ssh/*; \
103 service ssh restart;while true ; do sleep 10000; done']
106 k8s_template = kubernetes.KubernetesTemplate('k8s-86096c30', servers)
107 mock_get_pod_list.return_value.items = []
108 pods = k8s_template.get_rc_pods()
109 self.assertEqual(pods, [])
112 class ReplicationControllerObjectTestCase(base.BaseUnitTestCase):
114 def test__init_one_container(self):
115 pod_name = 'pod_name'
116 _kwargs = {'args': ['arg1', 'arg2'],
117 'image': 'fake_image',
118 'command': 'fake_command'}
119 k8s_obj = kubernetes.ReplicationControllerObject(pod_name, **_kwargs)
120 self.assertEqual(1, len(k8s_obj._containers))
121 container = k8s_obj._containers[0]
122 self.assertEqual(['arg1', 'arg2'], container._args)
123 self.assertEqual('fake_image', container._image)
124 self.assertEqual(['fake_command'], container._command)
125 self.assertEqual([], container._volume_mounts)
127 def test__init_multipe_containers(self):
128 pod_name = 'pod_name'
131 containers.append({'args': ['arg1', 'arg2'],
132 'image': 'fake_image_%s' % i,
133 'command': 'fake_command_%s' % i})
134 _kwargs = {'containers': containers}
135 k8s_obj = kubernetes.ReplicationControllerObject(pod_name, **_kwargs)
136 self.assertEqual(5, len(k8s_obj._containers))
138 container = k8s_obj._containers[i]
139 self.assertEqual(['arg1', 'arg2'], container._args)
140 self.assertEqual('fake_image_%s' % i, container._image)
141 self.assertEqual(['fake_command_%s' % i], container._command)
142 self.assertEqual([], container._volume_mounts)
144 def test__add_volumes(self):
145 volume1 = {'name': 'fake_sshkey',
146 'configMap': {'name': 'fake_sshkey'}}
147 volume2 = {'name': 'volume2',
149 k8s_obj = kubernetes.ReplicationControllerObject(
150 'name', ssh_key='fake_sshkey', volumes=[volume2])
151 k8s_obj._add_volumes()
152 volumes = k8s_obj.template['spec']['template']['spec']['volumes']
153 self.assertEqual(sorted([volume1, volume2], key=lambda k: k['name']),
154 sorted(volumes, key=lambda k: k['name']))
156 def test__add_volumes_no_volumes(self):
157 volume1 = {'name': 'fake_sshkey',
158 'configMap': {'name': 'fake_sshkey'}}
159 k8s_obj = kubernetes.ReplicationControllerObject(
160 'name', ssh_key='fake_sshkey')
161 k8s_obj._add_volumes()
162 volumes = k8s_obj.template['spec']['template']['spec']['volumes']
163 self.assertEqual([volume1], volumes)
165 def test__create_ssh_key_volume(self):
166 expected = {'name': 'fake_sshkey',
167 'configMap': {'name': 'fake_sshkey'}}
168 k8s_obj = kubernetes.ReplicationControllerObject(
169 'name', ssh_key='fake_sshkey')
170 self.assertEqual(expected, k8s_obj._create_ssh_key_volume())
172 def test__create_volume_item(self):
173 for vol_type in kubernetes_utils.get_volume_types():
174 volume = {'name': 'vol_name',
178 kubernetes.ReplicationControllerObject.
179 _create_volume_item(volume))
181 def test__create_volume_item_invalid_type(self):
182 volume = {'name': 'vol_name',
183 'invalid_type': 'data'}
184 with self.assertRaises(exceptions.KubernetesTemplateInvalidVolumeType):
185 kubernetes.ReplicationControllerObject._create_volume_item(volume)
187 def test__add_security_context(self):
188 k8s_obj = kubernetes.ReplicationControllerObject('pod_name')
189 self.assertNotIn('securityContext',
190 k8s_obj.template['spec']['template']['spec'])
192 k8s_obj._security_context = {'key_pod': 'value_pod'}
193 k8s_obj._add_security_context()
195 {'key_pod': 'value_pod'},
196 k8s_obj.template['spec']['template']['spec']['securityContext'])
198 def test__add_security_context_by_init(self):
202 {'securityContext': {'key%s' % i: 'value%s' % i}})
203 _kwargs = {'containers': containers,
204 'securityContext': {'key_pod': 'value_pod'}}
205 k8s_obj = kubernetes.ReplicationControllerObject('pod_name', **_kwargs)
207 {'key_pod': 'value_pod'},
208 k8s_obj.template['spec']['template']['spec']['securityContext'])
211 k8s_obj.template['spec']['template']['spec']['containers'][i])
212 self.assertEqual({'key%s' % i: 'value%s' % i},
213 container['securityContext'])
215 def test__add_networks(self):
216 k8s_obj = kubernetes.ReplicationControllerObject(
217 'name', networks=['network1', 'network2', 'network3'])
218 k8s_obj._add_networks()
220 template['spec']['template']['metadata']['annotations']['networks']
221 expected = ('[{"name": "network1"}, {"name": "network2"}, '
222 '{"name": "network3"}]')
223 self.assertEqual(expected, networks)
226 class ContainerObjectTestCase(base.BaseUnitTestCase):
228 def test__create_volume_mounts(self):
229 volume_mount = {'name': 'fake_name',
230 'mountPath': 'fake_path'}
231 ssh_vol = {'name': 'fake_ssh_key',
232 'mountPath': kubernetes.ContainerObject.SSH_MOUNT_PATH,
234 expected = copy.deepcopy(volume_mount)
235 expected['readOnly'] = False
236 expected = [expected, ssh_vol]
237 container_obj = kubernetes.ContainerObject(
238 'cname', 'fake_ssh_key', volumeMounts=[volume_mount])
239 output = container_obj._create_volume_mounts()
240 self.assertEqual(expected, output)
242 def test__create_volume_mounts_no_volume_mounts(self):
243 ssh_vol = {'name': 'fake_ssh_key2',
244 'mountPath': kubernetes.ContainerObject.SSH_MOUNT_PATH,
246 container_obj = kubernetes.ContainerObject('name', 'fake_ssh_key2')
247 output = container_obj._create_volume_mounts()
248 self.assertEqual([ssh_vol], output)
250 def test__create_volume_mounts_item(self):
251 volume_mount = {'name': 'fake_name',
252 'mountPath': 'fake_path'}
253 expected = copy.deepcopy(volume_mount)
254 expected['readOnly'] = False
255 output = kubernetes.ContainerObject._create_volume_mounts_item(
257 self.assertEqual(expected, output)
259 def test_get_container_item(self):
260 volume_mount = {'name': 'fake_name',
261 'mountPath': 'fake_path'}
262 args = ['arg1', 'arg2']
263 container_obj = kubernetes.ContainerObject(
264 'cname', ssh_key='fake_sshkey', volumeMount=[volume_mount],
266 expected = {'args': args,
267 'command': [kubernetes.ContainerObject.COMMAND_DEFAULT],
268 'image': kubernetes.ContainerObject.IMAGE_DEFAULT,
269 'name': 'cname-container',
270 'volumeMounts': container_obj._create_volume_mounts()}
271 self.assertEqual(expected, container_obj.get_container_item())
273 def test_get_container_item_with_security_context(self):
274 volume_mount = {'name': 'fake_name',
275 'mountPath': 'fake_path'}
276 args = ['arg1', 'arg2']
277 container_obj = kubernetes.ContainerObject(
278 'cname', ssh_key='fake_sshkey', volumeMount=[volume_mount],
279 args=args, securityContext={'key': 'value'})
280 expected = {'args': args,
281 'command': [kubernetes.ContainerObject.COMMAND_DEFAULT],
282 'image': kubernetes.ContainerObject.IMAGE_DEFAULT,
283 'name': 'cname-container',
284 'volumeMounts': container_obj._create_volume_mounts(),
285 'securityContext': {'key': 'value'}}
286 self.assertEqual(expected, container_obj.get_container_item())
288 def test_get_container_item_with_env(self):
289 volume_mount = {'name': 'fake_name',
290 'mountPath': 'fake_path'}
291 args = ['arg1', 'arg2']
292 container_obj = kubernetes.ContainerObject(
293 'cname', ssh_key='fake_sshkey', volumeMount=[volume_mount],
294 args=args, env=[{'name': 'fake_var_name',
295 'value': 'fake_var_value'}])
296 expected = {'args': args,
297 'command': [kubernetes.ContainerObject.COMMAND_DEFAULT],
298 'image': kubernetes.ContainerObject.IMAGE_DEFAULT,
299 'name': 'cname-container',
300 'volumeMounts': container_obj._create_volume_mounts(),
301 'env': [{'name': 'fake_var_name',
302 'value': 'fake_var_value'}]}
303 self.assertEqual(expected, container_obj.get_container_item())
305 def test_get_container_item_with_resources(self):
306 volume_mount = {'name': 'fake_name',
307 'mountPath': 'fake_path'}
308 args = ['arg1', 'arg2']
309 resources = {'requests': {'key1': 'val1'},
310 'limits': {'key2': 'val2'},
311 'other_key': {'key3': 'val3'}}
312 container_obj = kubernetes.ContainerObject(
313 'cname', ssh_key='fake_sshkey', volumeMount=[volume_mount],
314 args=args, resources=resources)
315 expected = {'args': args,
316 'command': [kubernetes.ContainerObject.COMMAND_DEFAULT],
317 'image': kubernetes.ContainerObject.IMAGE_DEFAULT,
318 'name': 'cname-container',
319 'volumeMounts': container_obj._create_volume_mounts(),
320 'resources': {'requests': {'key1': 'val1'},
321 'limits': {'key2': 'val2'}}}
322 self.assertEqual(expected, container_obj.get_container_item())
324 class CustomResourceDefinitionObjectTestCase(base.BaseUnitTestCase):
326 def test__init(self):
329 'name': 'newcrds.ctx_name.com'
332 'group': 'ctx_name.com',
335 'names': {'plural': 'newcrds',
336 'singular': 'newcrd',
340 crd_obj = kubernetes.CustomResourceDefinitionObject(
341 'ctx_name', name='newcrd', version='v2', scope='scope')
342 self.assertEqual('newcrds.ctx_name.com', crd_obj._name)
343 self.assertEqual(template, crd_obj._template)
345 def test__init_missing_parameter(self):
346 with self.assertRaises(exceptions.KubernetesCRDObjectDefinitionError):
347 kubernetes.CustomResourceDefinitionObject('ctx_name',
351 class NetworkObjectTestCase(base.BaseUnitTestCase):
354 self.net_obj = kubernetes.NetworkObject(name='fake_name',
355 plugin='fake_plugin',
358 def test__init_missing_parameter(self):
359 with self.assertRaises(
360 exceptions.KubernetesNetworkObjectDefinitionError):
361 kubernetes.NetworkObject(name='name', plugin='plugin')
362 with self.assertRaises(
363 exceptions.KubernetesNetworkObjectDefinitionError):
364 kubernetes.NetworkObject(name='name', args='args')
365 with self.assertRaises(
366 exceptions.KubernetesNetworkObjectDefinitionError):
367 kubernetes.NetworkObject(args='args', plugin='plugin')
369 @mock.patch.object(kubernetes_utils, 'get_custom_resource_definition')
370 def test_crd(self, mock_get_crd):
371 mock_crd = mock.Mock()
372 mock_get_crd.return_value = mock_crd
373 net_obj = copy.deepcopy(self.net_obj)
374 self.assertEqual(mock_crd, net_obj.crd)
376 def test_template(self):
377 net_obj = copy.deepcopy(self.net_obj)
378 expected = {'apiVersion': 'group.com/v2',
379 'kind': kubernetes.NetworkObject.KIND,
381 'name': 'fake_name'},
382 'plugin': 'fake_plugin',
385 crd.spec.group = 'group.com'
386 crd.spec.version = 'v2'
388 self.assertEqual(expected, net_obj.template)
390 def test_group(self):
391 net_obj = copy.deepcopy(self.net_obj)
392 net_obj._crd = mock.Mock()
393 net_obj._crd.spec.group = 'fake_group'
394 self.assertEqual('fake_group', net_obj.group)
396 def test_version(self):
397 net_obj = copy.deepcopy(self.net_obj)
398 net_obj._crd = mock.Mock()
399 net_obj._crd.spec.version = 'version_4'
400 self.assertEqual('version_4', net_obj.version)
402 def test_plural(self):
403 net_obj = copy.deepcopy(self.net_obj)
404 net_obj._crd = mock.Mock()
405 net_obj._crd.spec.names.plural = 'name_ending_in_s'
406 self.assertEqual('name_ending_in_s', net_obj.plural)
408 def test_scope(self):
409 net_obj = copy.deepcopy(self.net_obj)
410 net_obj._crd = mock.Mock()
411 net_obj._crd.spec.scope = 'Cluster'
412 self.assertEqual('Cluster', net_obj.scope)
414 @mock.patch.object(kubernetes_utils, 'create_network')
415 def test_create(self, mock_create_network):
416 net_obj = copy.deepcopy(self.net_obj)
417 net_obj._scope = 'scope'
418 net_obj._group = 'group'
419 net_obj._version = 'version'
420 net_obj._plural = 'plural'
421 net_obj._template = 'template'
423 mock_create_network.assert_called_once_with(
424 'scope', 'group', 'version', 'plural', 'template')
426 @mock.patch.object(kubernetes_utils, 'delete_network')
427 def test_delete(self, mock_delete_network):
428 net_obj = copy.deepcopy(self.net_obj)
429 net_obj._scope = 'scope'
430 net_obj._group = 'group'
431 net_obj._version = 'version'
432 net_obj._plural = 'plural'
433 net_obj._name = 'name'
435 mock_delete_network.assert_called_once_with(
436 'scope', 'group', 'version', 'plural', 'name')
439 class ServiceNodePortObjectTestCase(base.BaseUnitTestCase):
441 def test__init(self):
442 with mock.patch.object(kubernetes.ServiceNodePortObject, '_add_port') \
444 kubernetes.ServiceNodePortObject('fake_name',
445 node_ports=[{'port': 80}])
447 mock_add_port.assert_has_calls([mock.call(22, protocol='TCP'),
450 def test__add_port(self):
451 nodeport_object = kubernetes.ServiceNodePortObject('fake_name')
452 port_ssh = {'port': 22,
454 port_definition = {'port': 80,
459 port = copy.deepcopy(port_definition)
461 nodeport_object._add_port(80, **port)
462 self.assertEqual([port_ssh, port_definition],
463 nodeport_object.template['spec']['ports'])
465 @mock.patch.object(kubernetes_utils, 'create_service')
466 def test_create(self, mock_create_service):
467 nodeport_object = kubernetes.ServiceNodePortObject('fake_name')
468 nodeport_object.template = 'fake_template'
469 nodeport_object.create()
470 mock_create_service.assert_called_once_with('fake_template')
472 @mock.patch.object(kubernetes_utils, 'delete_service')
473 def test_delete(self, mock_delete_service):
474 nodeport_object = kubernetes.ServiceNodePortObject('fake_name')
475 nodeport_object.delete()
476 mock_delete_service.assert_called_once_with('fake_name-service')