self._resources = kwargs.get('resources', {})
self._ports = kwargs.get('ports', [])
self._image_pull_policy = kwargs.get('imagePullPolicy')
+ self._tty = kwargs.get('tty')
+ self._stdin = kwargs.get('stdin')
@staticmethod
def _parse_commands(command):
if self._image_pull_policy not in self.IMAGE_PULL_POLICY:
raise exceptions.KubernetesContainerWrongImagePullPolicy()
container['imagePullPolicy'] = self._image_pull_policy
+ if self._stdin is not None:
+ container['stdin'] = self._stdin
+ if self._tty is not None:
+ container['tty'] = self._tty
return container
'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'],