community.general/test/units/module_utils/basic/test_run_command.py

# -*- 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)
__metaclass__ = type

import errno
from itertools import product
from io import BytesIO

import pytest

from ansible.module_utils._text import to_native


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_read(fd, nbytes):
        return os._cmd_out[fd].read(nbytes)

    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._cmd_out = {
        # os.read() is returning 'bytes', not strings
        mocker.sentinel.stdout: BytesIO(),
        mocker.sentinel.stderr: BytesIO(),
    }

    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.read.side_effect = mock_os_read
    os.path.abspath.side_effect = mock_os_abspath

    yield os


@pytest.fixture
def mock_subprocess(mocker):
    def mock_select(rlist, wlist, xlist, timeout=1):
        return (rlist, [], [])

    fake_select = mocker.patch('ansible.module_utils.basic.select')
    fake_select.select.side_effect = mock_select

    subprocess = mocker.patch('ansible.module_utils.basic.subprocess')
    cmd = mocker.MagicMock()
    cmd.returncode = 0
    cmd.stdin = OpenBytesIO()
    cmd.stdout.fileno.return_value = mocker.sentinel.stdout
    cmd.stderr.fileno.return_value = 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'], ['/bin/ls', 'a', 'b', 'c'], '/bin/ls a b c'),
                ('/bin/ls a " b" "c "', ['/bin/ls', 'a', ' b', 'c '], '/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._os.getcwd.return_value = '/old'
        rc_am.run_command('/bin/ls', cwd='/new')
        assert rc_am._os.chdir.mock_calls == [mocker.call('/new'), mocker.call('/old'), ]

    @pytest.mark.parametrize('stdin', [{}], indirect=['stdin'])
    def test_cwd_relative_path(self, mocker, rc_am):
        rc_am._os.getcwd.return_value = '/old'
        rc_am.run_command('/bin/ls', cwd='sub-dir')
        assert rc_am._os.chdir.mock_calls == [mocker.call('/old/sub-dir'), mocker.call('/old'), ]

    @pytest.mark.parametrize('stdin', [{}], indirect=['stdin'])
    def test_cwd_not_a_dir(self, mocker, rc_am):
        rc_am._os.getcwd.return_value = '/old'
        rc_am._os.path.isdir.side_effect = lambda d: d != '/not-a-dir'
        rc_am.run_command('/bin/ls', cwd='/not-a-dir')
        assert rc_am._os.chdir.mock_calls == [mocker.call('/old'), ]

    @pytest.mark.parametrize('stdin', [{}], indirect=['stdin'])
    def test_cwd_inaccessible(self, rc_am):
        with pytest.raises(SystemExit):
            rc_am.run_command('/bin/ls', cwd='/inaccessible')
        assert rc_am.fail_json.called
        args, kwargs = rc_am.fail_json.call_args
        assert kwargs['rc'] == errno.EPERM


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._os._cmd_out[mocker.sentinel.stdout] = BytesIO(b'Authentication required!\nEnter password: ')
        (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._os._cmd_out[mocker.sentinel.stdout] = BytesIO(b'hello')
        (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._os._cmd_out[mocker.sentinel.stdout] = BytesIO(u'Žarn§'.encode('utf-8'))
        rc_am._os._cmd_out[mocker.sentinel.stderr] = BytesIO(u'لرئيسية'.encode('utf-8'))
        (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'لرئيسية')