diff options
Diffstat (limited to 'test/units/module_utils/basic/test_run_command.py')
-rw-r--r-- | test/units/module_utils/basic/test_run_command.py | 278 |
1 files changed, 278 insertions, 0 deletions
diff --git a/test/units/module_utils/basic/test_run_command.py b/test/units/module_utils/basic/test_run_command.py new file mode 100644 index 0000000..04211e2 --- /dev/null +++ b/test/units/module_utils/basic/test_run_command.py @@ -0,0 +1,278 @@ +# -*- coding: utf-8 -*- +# Copyright (c) 2017 Ansible Project +# GNU General Public License v3.0+ (see COPYING or https://www.gnu.org/licenses/gpl-3.0.txt) + +# Make coding more python3-ish +from __future__ import (absolute_import, division, print_function) +__metaclass__ = type + +import errno +from itertools import product +from io import BytesIO + +import pytest + +from ansible.module_utils._text import to_native +from ansible.module_utils.six import PY2 +from ansible.module_utils.compat import selectors + + +class OpenBytesIO(BytesIO): + """BytesIO with dummy close() method + + So that you can inspect the content after close() was called. + """ + + def close(self): + pass + + +@pytest.fixture +def mock_os(mocker): + def mock_os_chdir(path): + if path == '/inaccessible': + raise OSError(errno.EPERM, "Permission denied: '/inaccessible'") + + def mock_os_abspath(path): + if path.startswith('/'): + return path + else: + return os.getcwd.return_value + '/' + path + + os = mocker.patch('ansible.module_utils.basic.os') + + os.path.expandvars.side_effect = lambda x: x + os.path.expanduser.side_effect = lambda x: x + os.environ = {'PATH': '/bin'} + os.getcwd.return_value = '/home/foo' + os.path.isdir.return_value = True + os.chdir.side_effect = mock_os_chdir + os.path.abspath.side_effect = mock_os_abspath + + yield os + + +class DummyFileObj(): + def __init__(self, fileobj): + self.fileobj = fileobj + + +class SpecialBytesIO(BytesIO): + def __init__(self, *args, **kwargs): + fh = kwargs.pop('fh', None) + super(SpecialBytesIO, self).__init__(*args, **kwargs) + self.fh = fh + + def fileno(self): + return self.fh + + # We need to do this because some of our tests create a new value for stdout and stderr + # The new value is able to affect the string that is returned by the subprocess stdout and + # stderr but by the time the test gets it, it is too late to change the SpecialBytesIO that + # subprocess.Popen returns for stdout and stderr. If we could figure out how to change those as + # well, then we wouldn't need this. + def __eq__(self, other): + if id(self) == id(other) or self.fh == other.fileno(): + return True + return False + + +class DummyKey: + def __init__(self, fileobj): + self.fileobj = fileobj + + +@pytest.fixture +def mock_subprocess(mocker): + + class MockSelector(selectors.BaseSelector): + def __init__(self): + super(MockSelector, self).__init__() + self._file_objs = [] + + def register(self, fileobj, events, data=None): + self._file_objs.append(fileobj) + + def unregister(self, fileobj): + self._file_objs.remove(fileobj) + + def select(self, timeout=None): + ready = [] + for file_obj in self._file_objs: + ready.append((DummyKey(subprocess._output[file_obj.fileno()]), selectors.EVENT_READ)) + return ready + + def get_map(self): + return self._file_objs + + def close(self): + super(MockSelector, self).close() + self._file_objs = [] + + selectors.DefaultSelector = MockSelector + + subprocess = mocker.patch('ansible.module_utils.basic.subprocess') + subprocess._output = {mocker.sentinel.stdout: SpecialBytesIO(b'', fh=mocker.sentinel.stdout), + mocker.sentinel.stderr: SpecialBytesIO(b'', fh=mocker.sentinel.stderr)} + + cmd = mocker.MagicMock() + cmd.returncode = 0 + cmd.stdin = OpenBytesIO() + cmd.stdout = subprocess._output[mocker.sentinel.stdout] + cmd.stderr = subprocess._output[mocker.sentinel.stderr] + subprocess.Popen.return_value = cmd + + yield subprocess + + +@pytest.fixture() +def rc_am(mocker, am, mock_os, mock_subprocess): + am.fail_json = mocker.MagicMock(side_effect=SystemExit) + am._os = mock_os + am._subprocess = mock_subprocess + yield am + + +class TestRunCommandArgs: + # Format is command as passed to run_command, command to Popen as list, command to Popen as string + ARGS_DATA = ( + (['/bin/ls', 'a', 'b', 'c'], [b'/bin/ls', b'a', b'b', b'c'], b'/bin/ls a b c'), + ('/bin/ls a " b" "c "', [b'/bin/ls', b'a', b' b', b'c '], b'/bin/ls a " b" "c "'), + ) + + # pylint bug: https://github.com/PyCQA/pylint/issues/511 + # pylint: disable=undefined-variable + @pytest.mark.parametrize('cmd, expected, shell, stdin', + ((arg, cmd_str if sh else cmd_lst, sh, {}) + for (arg, cmd_lst, cmd_str), sh in product(ARGS_DATA, (True, False))), + indirect=['stdin']) + def test_args(self, cmd, expected, shell, rc_am): + rc_am.run_command(cmd, use_unsafe_shell=shell) + assert rc_am._subprocess.Popen.called + args, kwargs = rc_am._subprocess.Popen.call_args + assert args == (expected, ) + assert kwargs['shell'] == shell + + @pytest.mark.parametrize('stdin', [{}], indirect=['stdin']) + def test_tuple_as_args(self, rc_am): + with pytest.raises(SystemExit): + rc_am.run_command(('ls', '/')) + assert rc_am.fail_json.called + + +class TestRunCommandCwd: + @pytest.mark.parametrize('stdin', [{}], indirect=['stdin']) + def test_cwd(self, mocker, rc_am): + rc_am.run_command('/bin/ls', cwd='/new') + assert rc_am._subprocess.Popen.mock_calls[0][2]['cwd'] == b'/new' + + @pytest.mark.parametrize('stdin', [{}], indirect=['stdin']) + def test_cwd_relative_path(self, mocker, rc_am): + rc_am.run_command('/bin/ls', cwd='sub-dir') + assert rc_am._subprocess.Popen.mock_calls[0][2]['cwd'] == b'/home/foo/sub-dir' + + @pytest.mark.parametrize('stdin', [{}], indirect=['stdin']) + def test_cwd_not_a_dir(self, mocker, rc_am): + rc_am.run_command('/bin/ls', cwd='/not-a-dir') + assert rc_am._subprocess.Popen.mock_calls[0][2]['cwd'] == b'/not-a-dir' + + @pytest.mark.parametrize('stdin', [{}], indirect=['stdin']) + def test_cwd_not_a_dir_noignore(self, rc_am): + rc_am._os.path.isdir.side_effect = lambda d: d != b'/not-a-dir' + with pytest.raises(SystemExit): + rc_am.run_command('/bin/ls', cwd='/not-a-dir', ignore_invalid_cwd=False) + assert rc_am.fail_json.called + + +class TestRunCommandPrompt: + @pytest.mark.parametrize('stdin', [{}], indirect=['stdin']) + def test_prompt_bad_regex(self, rc_am): + with pytest.raises(SystemExit): + rc_am.run_command('foo', prompt_regex='[pP)assword:') + assert rc_am.fail_json.called + + @pytest.mark.parametrize('stdin', [{}], indirect=['stdin']) + def test_prompt_no_match(self, mocker, rc_am): + rc_am._os._cmd_out[mocker.sentinel.stdout] = BytesIO(b'hello') + (rc, _, _) = rc_am.run_command('foo', prompt_regex='[pP]assword:') + assert rc == 0 + + @pytest.mark.parametrize('stdin', [{}], indirect=['stdin']) + def test_prompt_match_wo_data(self, mocker, rc_am): + rc_am._subprocess._output = {mocker.sentinel.stdout: + SpecialBytesIO(b'Authentication required!\nEnter password: ', + fh=mocker.sentinel.stdout), + mocker.sentinel.stderr: + SpecialBytesIO(b'', fh=mocker.sentinel.stderr)} + (rc, _, _) = rc_am.run_command('foo', prompt_regex=r'[pP]assword:', data=None) + assert rc == 257 + + +class TestRunCommandRc: + @pytest.mark.parametrize('stdin', [{}], indirect=['stdin']) + def test_check_rc_false(self, rc_am): + rc_am._subprocess.Popen.return_value.returncode = 1 + (rc, _, _) = rc_am.run_command('/bin/false', check_rc=False) + assert rc == 1 + + @pytest.mark.parametrize('stdin', [{}], indirect=['stdin']) + def test_check_rc_true(self, rc_am): + rc_am._subprocess.Popen.return_value.returncode = 1 + with pytest.raises(SystemExit): + rc_am.run_command('/bin/false', check_rc=True) + assert rc_am.fail_json.called + args, kwargs = rc_am.fail_json.call_args + assert kwargs['rc'] == 1 + + +class TestRunCommandOutput: + @pytest.mark.parametrize('stdin', [{}], indirect=['stdin']) + def test_text_stdin(self, rc_am): + (rc, stdout, stderr) = rc_am.run_command('/bin/foo', data='hello world') + assert rc_am._subprocess.Popen.return_value.stdin.getvalue() == b'hello world\n' + + @pytest.mark.parametrize('stdin', [{}], indirect=['stdin']) + def test_ascii_stdout(self, mocker, rc_am): + rc_am._subprocess._output = {mocker.sentinel.stdout: + SpecialBytesIO(b'hello', fh=mocker.sentinel.stdout), + mocker.sentinel.stderr: + SpecialBytesIO(b'', fh=mocker.sentinel.stderr)} + (rc, stdout, stderr) = rc_am.run_command('/bin/cat hello.txt') + assert rc == 0 + # module_utils function. On py3 it returns text and py2 it returns + # bytes because it's returning native strings + assert stdout == 'hello' + + @pytest.mark.parametrize('stdin', [{}], indirect=['stdin']) + def test_utf8_output(self, mocker, rc_am): + rc_am._subprocess._output = {mocker.sentinel.stdout: + SpecialBytesIO(u'Žarn§'.encode('utf-8'), + fh=mocker.sentinel.stdout), + mocker.sentinel.stderr: + SpecialBytesIO(u'لرئيسية'.encode('utf-8'), + fh=mocker.sentinel.stderr)} + (rc, stdout, stderr) = rc_am.run_command('/bin/something_ugly') + assert rc == 0 + # module_utils function. On py3 it returns text and py2 it returns + # bytes because it's returning native strings + assert stdout == to_native(u'Žarn§') + assert stderr == to_native(u'لرئيسية') + + +@pytest.mark.parametrize('stdin', [{}], indirect=['stdin']) +def test_run_command_fds(mocker, rc_am): + subprocess_mock = mocker.patch('ansible.module_utils.basic.subprocess') + subprocess_mock.Popen.side_effect = AssertionError + + try: + rc_am.run_command('synchronize', pass_fds=(101, 42)) + except SystemExit: + pass + + if PY2: + assert subprocess_mock.Popen.call_args[1]['close_fds'] is False + assert 'pass_fds' not in subprocess_mock.Popen.call_args[1] + + else: + assert subprocess_mock.Popen.call_args[1]['pass_fds'] == (101, 42) + assert subprocess_mock.Popen.call_args[1]['close_fds'] is True |