+class ReplicationControllerObjectTestCase(base.BaseUnitTestCase):
+
+ def test__init_one_container(self):
+ pod_name = 'pod_name'
+ _kwargs = {'args': ['arg1', 'arg2'],
+ 'image': 'fake_image',
+ 'command': 'fake_command'}
+ k8s_obj = kubernetes.ReplicationControllerObject(pod_name, **_kwargs)
+ self.assertEqual(1, len(k8s_obj._containers))
+ container = k8s_obj._containers[0]
+ self.assertEqual(['arg1', 'arg2'], container._args)
+ self.assertEqual('fake_image', container._image)
+ self.assertEqual(['fake_command'], container._command)
+ self.assertEqual([], container._volume_mounts)
+
+ def test__init_multipe_containers(self):
+ pod_name = 'pod_name'
+ containers = []
+ for i in range(5):
+ containers.append({'args': ['arg1', 'arg2'],
+ 'image': 'fake_image_%s' % i,
+ 'command': 'fake_command_%s' % i})
+ _kwargs = {'containers': containers}
+ k8s_obj = kubernetes.ReplicationControllerObject(pod_name, **_kwargs)
+ self.assertEqual(5, len(k8s_obj._containers))
+ for i in range(5):
+ container = k8s_obj._containers[i]
+ self.assertEqual(['arg1', 'arg2'], container._args)
+ self.assertEqual('fake_image_%s' % i, container._image)
+ self.assertEqual(['fake_command_%s' % i], container._command)
+ self.assertEqual([], container._volume_mounts)
+
+ def test__add_volumes(self):
+ volume1 = {'name': 'fake_sshkey',
+ 'configMap': {'name': 'fake_sshkey'}}
+ volume2 = {'name': 'volume2',
+ 'configMap': 'data'}
+ k8s_obj = kubernetes.ReplicationControllerObject(
+ 'name', ssh_key='fake_sshkey', volumes=[volume2])
+ k8s_obj._add_volumes()
+ volumes = k8s_obj.template['spec']['template']['spec']['volumes']
+ self.assertEqual(sorted([volume1, volume2], key=lambda k: k['name']),
+ sorted(volumes, key=lambda k: k['name']))
+
+ def test__add_volumes_no_volumes(self):
+ volume1 = {'name': 'fake_sshkey',
+ 'configMap': {'name': 'fake_sshkey'}}
+ k8s_obj = kubernetes.ReplicationControllerObject(
+ 'name', ssh_key='fake_sshkey')
+ k8s_obj._add_volumes()
+ volumes = k8s_obj.template['spec']['template']['spec']['volumes']
+ self.assertEqual([volume1], volumes)
+
+ def test__create_ssh_key_volume(self):
+ expected = {'name': 'fake_sshkey',
+ 'configMap': {'name': 'fake_sshkey'}}
+ k8s_obj = kubernetes.ReplicationControllerObject(
+ 'name', ssh_key='fake_sshkey')
+ self.assertEqual(expected, k8s_obj._create_ssh_key_volume())
+
+ def test__create_volume_item(self):
+ for vol_type in kubernetes_utils.get_volume_types():
+ volume = {'name': 'vol_name',
+ vol_type: 'data'}
+ self.assertEqual(
+ volume,
+ kubernetes.ReplicationControllerObject.
+ _create_volume_item(volume))
+
+ def test__create_volume_item_invalid_type(self):
+ volume = {'name': 'vol_name',
+ 'invalid_type': 'data'}
+ with self.assertRaises(exceptions.KubernetesTemplateInvalidVolumeType):
+ kubernetes.ReplicationControllerObject._create_volume_item(volume)
+
+ def test__add_security_context(self):
+ k8s_obj = kubernetes.ReplicationControllerObject('pod_name')
+ self.assertNotIn('securityContext',
+ k8s_obj.template['spec']['template']['spec'])
+
+ k8s_obj._security_context = {'key_pod': 'value_pod'}
+ k8s_obj._add_security_context()
+ self.assertEqual(
+ {'key_pod': 'value_pod'},
+ k8s_obj.template['spec']['template']['spec']['securityContext'])
+
+ def test__add_security_context_by_init(self):
+ containers = []
+ for i in range(5):
+ containers.append(
+ {'securityContext': {'key%s' % i: 'value%s' % i}})
+ _kwargs = {'containers': containers,
+ 'securityContext': {'key_pod': 'value_pod'}}
+ k8s_obj = kubernetes.ReplicationControllerObject('pod_name', **_kwargs)
+ self.assertEqual(
+ {'key_pod': 'value_pod'},
+ k8s_obj.template['spec']['template']['spec']['securityContext'])
+ for i in range(5):
+ container = (
+ k8s_obj.template['spec']['template']['spec']['containers'][i])
+ self.assertEqual({'key%s' % i: 'value%s' % i},
+ container['securityContext'])
+
+ def test__add_networks(self):
+ k8s_obj = kubernetes.ReplicationControllerObject(
+ 'name', networks=['network1', 'network2', 'network3'])
+ k8s_obj._add_networks()
+ networks = k8s_obj.\
+ template['spec']['template']['metadata']['annotations']['networks']
+ expected = ('[{"name": "network1"}, {"name": "network2"}, '
+ '{"name": "network3"}]')
+ self.assertEqual(expected, networks)
+
+ def test__add_tolerations(self):
+ _kwargs = {'tolerations': [{'key': 'key1',
+ 'value': 'value2',
+ 'effect': 'effect3',
+ 'operator': 'operator4',
+ 'wrong_key': 'error_key'}]
+ }
+ k8s_obj = kubernetes.ReplicationControllerObject('pod_name', **_kwargs)
+ k8s_obj._add_tolerations()
+ _tol = k8s_obj.template['spec']['template']['spec']['tolerations']
+ self.assertEqual(1, len(_tol))
+ self.assertEqual({'key': 'key1',
+ 'value': 'value2',
+ 'effect': 'effect3',
+ 'operator': 'operator4'},
+ _tol[0])
+
+ def test__add_tolerations_default(self):
+ k8s_obj = kubernetes.ReplicationControllerObject('pod_name')
+ k8s_obj._add_tolerations()
+ _tol = k8s_obj.template['spec']['template']['spec']['tolerations']
+ self.assertEqual(1, len(_tol))
+ self.assertEqual({'operator': 'Exists'}, _tol[0])
+
+
+class ContainerObjectTestCase(base.BaseUnitTestCase):
+
+ def test__create_volume_mounts(self):
+ volume_mount = {'name': 'fake_name',
+ 'mountPath': 'fake_path'}
+ ssh_vol = {'name': 'fake_ssh_key',
+ 'mountPath': kubernetes.ContainerObject.SSH_MOUNT_PATH,
+ 'readOnly': False}
+ expected = copy.deepcopy(volume_mount)
+ expected['readOnly'] = False
+ expected = [expected, ssh_vol]
+ container_obj = kubernetes.ContainerObject(
+ 'cname', 'fake_ssh_key', volumeMounts=[volume_mount])
+ output = container_obj._create_volume_mounts()
+ self.assertEqual(expected, output)
+
+ def test__create_volume_mounts_no_volume_mounts(self):
+ ssh_vol = {'name': 'fake_ssh_key2',
+ 'mountPath': kubernetes.ContainerObject.SSH_MOUNT_PATH,
+ 'readOnly': False}
+ container_obj = kubernetes.ContainerObject('name', 'fake_ssh_key2')
+ output = container_obj._create_volume_mounts()
+ self.assertEqual([ssh_vol], output)
+
+ def test__create_volume_mounts_item(self):
+ volume_mount = {'name': 'fake_name',
+ 'mountPath': 'fake_path'}
+ expected = copy.deepcopy(volume_mount)
+ expected['readOnly'] = False
+ output = kubernetes.ContainerObject._create_volume_mounts_item(
+ volume_mount)
+ self.assertEqual(expected, output)
+
+ def test_get_container_item(self):
+ volume_mount = {'name': 'fake_name',
+ 'mountPath': 'fake_path'}
+ args = ['arg1', 'arg2']
+ container_obj = kubernetes.ContainerObject(
+ 'cname', ssh_key='fake_sshkey', volumeMount=[volume_mount],
+ args=args)
+ expected = {'args': args,
+ 'command': kubernetes.ContainerObject.COMMAND_DEFAULT,
+ 'image': kubernetes.ContainerObject.IMAGE_DEFAULT,
+ 'name': 'cname-container',
+ 'volumeMounts': container_obj._create_volume_mounts()}
+ self.assertEqual(expected, container_obj.get_container_item())
+
+ def test_get_container_item_with_security_context(self):
+ volume_mount = {'name': 'fake_name',
+ 'mountPath': 'fake_path'}
+ args = ['arg1', 'arg2']
+ container_obj = kubernetes.ContainerObject(
+ 'cname', ssh_key='fake_sshkey', volumeMount=[volume_mount],
+ args=args, securityContext={'key': 'value'})
+ expected = {'args': args,
+ 'command': kubernetes.ContainerObject.COMMAND_DEFAULT,
+ 'image': kubernetes.ContainerObject.IMAGE_DEFAULT,
+ 'name': 'cname-container',
+ 'volumeMounts': container_obj._create_volume_mounts(),
+ 'securityContext': {'key': 'value'}}
+ self.assertEqual(expected, container_obj.get_container_item())
+
+ def test_get_container_item_with_env(self):
+ volume_mount = {'name': 'fake_name',
+ 'mountPath': 'fake_path'}
+ args = ['arg1', 'arg2']
+ container_obj = kubernetes.ContainerObject(
+ 'cname', ssh_key='fake_sshkey', volumeMount=[volume_mount],
+ args=args, env=[{'name': 'fake_var_name',
+ 'value': 'fake_var_value'}])
+ expected = {'args': args,
+ 'command': kubernetes.ContainerObject.COMMAND_DEFAULT,
+ 'image': kubernetes.ContainerObject.IMAGE_DEFAULT,
+ 'name': 'cname-container',
+ 'volumeMounts': container_obj._create_volume_mounts(),
+ 'env': [{'name': 'fake_var_name',
+ 'value': 'fake_var_value'}]}
+ self.assertEqual(expected, container_obj.get_container_item())
+
+ def test_get_container_item_with_ports_multi_parameter(self):
+ volume_mount = {'name': 'fake_name',
+ 'mountPath': 'fake_path'}
+ args = ['arg1', 'arg2']
+ container_obj = kubernetes.ContainerObject(
+ 'cname', ssh_key='fake_sshkey', volumeMount=[volume_mount],
+ args=args, ports=[{'containerPort': 'fake_port_name',
+ 'hostPort': 'fake_host_port',
+ 'name': 'fake_name',
+ 'protocol': 'fake_protocol',
+ 'invalid_varible': 'fakeinvalid_varible',
+ 'hostIP': 'fake_port_number'}])
+ expected = {'args': args,
+ 'command': kubernetes.ContainerObject.COMMAND_DEFAULT,
+ 'image': kubernetes.ContainerObject.IMAGE_DEFAULT,
+ 'name': 'cname-container',
+ 'volumeMounts': container_obj._create_volume_mounts(),
+ 'ports': [{'containerPort': 'fake_port_name',
+ 'hostPort': 'fake_host_port',
+ 'name': 'fake_name',
+ 'protocol': 'fake_protocol',
+ 'hostIP': 'fake_port_number'}]}
+ self.assertEqual(expected, container_obj.get_container_item())
+
+ def test_get_container_item_with_ports_no_container_port(self):
+ with self.assertRaises(exceptions.KubernetesContainerPortNotDefined):
+ volume_mount = {'name': 'fake_name',
+ 'mountPath': 'fake_path'}
+ args = ['arg1', 'arg2']
+ container_obj = kubernetes.ContainerObject(
+ 'cname', ssh_key='fake_sshkey', volumeMount=[volume_mount],
+ args=args, ports=[{'hostPort': 'fake_host_port',
+ 'name': 'fake_name',
+ 'protocol': 'fake_protocol',
+ 'hostIP': 'fake_port_number'}])
+ container_obj.get_container_item()
+
+ def test_get_container_item_with_resources(self):
+ volume_mount = {'name': 'fake_name',
+ 'mountPath': 'fake_path'}
+ args = ['arg1', 'arg2']
+ resources = {'requests': {'key1': 'val1'},
+ 'limits': {'key2': 'val2'},
+ 'other_key': {'key3': 'val3'}}
+ container_obj = kubernetes.ContainerObject(
+ 'cname', ssh_key='fake_sshkey', volumeMount=[volume_mount],
+ args=args, resources=resources)
+ expected = {'args': args,
+ 'command': kubernetes.ContainerObject.COMMAND_DEFAULT,
+ 'image': kubernetes.ContainerObject.IMAGE_DEFAULT,
+ 'name': 'cname-container',
+ 'volumeMounts': container_obj._create_volume_mounts(),
+ 'resources': {'requests': {'key1': 'val1'},
+ 'limits': {'key2': 'val2'}}}
+ self.assertEqual(expected, container_obj.get_container_item())
+
+ def test_get_container_item_image_pull_policy(self):
+ container_obj = kubernetes.ContainerObject(
+ 'cname', ssh_key='fake_sshkey', imagePullPolicy='Always')
+ expected = {'args': [],
+ 'command': kubernetes.ContainerObject.COMMAND_DEFAULT,
+ 'image': kubernetes.ContainerObject.IMAGE_DEFAULT,
+ 'name': 'cname-container',
+ 'volumeMounts': container_obj._create_volume_mounts(),
+ 'imagePullPolicy':'Always'}
+ self.assertEqual(expected, container_obj.get_container_item())
+
+ def test_get_container_item_with_tty_stdin(self):
+ args = ['arg1', 'arg2']
+ container_obj = kubernetes.ContainerObject(
+ 'cname', 'fake_sshkey', args=args, tty=False, stdin=True)
+ expected = {'args': args,
+ 'command': kubernetes.ContainerObject.COMMAND_DEFAULT,
+ 'image': kubernetes.ContainerObject.IMAGE_DEFAULT,
+ 'name': 'cname-container',
+ 'volumeMounts': container_obj._create_volume_mounts(),
+ 'tty': False,
+ 'stdin': True}
+ self.assertEqual(expected, container_obj.get_container_item())
+
+ def test__parse_commands_string(self):
+ container_obj = kubernetes.ContainerObject('cname', 'fake_sshkey')
+ self.assertEqual(['fake command'],
+ container_obj._parse_commands('fake command'))
+
+ def test__parse_commands_list(self):
+ container_obj = kubernetes.ContainerObject('cname', 'fake_sshkey')
+ self.assertEqual(['cmd1', 'cmd2'],
+ container_obj._parse_commands(['cmd1', 'cmd2']))
+
+ def test__parse_commands_exception(self):
+ container_obj = kubernetes.ContainerObject('cname', 'fake_sshkey')
+ with self.assertRaises(exceptions.KubernetesContainerCommandType):
+ container_obj._parse_commands({})
+