diff --git a/test/runner/requirements/units.txt b/test/runner/requirements/units.txt index 3af32de945d..a2fb5940268 100644 --- a/test/runner/requirements/units.txt +++ b/test/runner/requirements/units.txt @@ -28,3 +28,6 @@ pyfmg # requirement for aci_rest module xmljson + +# requirement for winrm connection plugin tests +pexpect \ No newline at end of file diff --git a/test/units/plugins/connection/test_winrm.py b/test/units/plugins/connection/test_winrm.py index f9fb403aa29..24f7983124e 100644 --- a/test/units/plugins/connection/test_winrm.py +++ b/test/units/plugins/connection/test_winrm.py @@ -6,8 +6,6 @@ from __future__ import (absolute_import, division, print_function) __metaclass__ = type -import sys - import pytest from io import StringIO @@ -17,6 +15,7 @@ from ansible.errors import AnsibleConnectionFailure from ansible.module_utils._text import to_bytes from ansible.playbook.play_context import PlayContext from ansible.plugins.loader import connection_loader +from ansible.plugins.connection import winrm class TestConnectionWinRM(object): @@ -194,23 +193,13 @@ class TestConnectionWinRM(object): @pytest.mark.parametrize('play, options, direct, expected, kerb', ((p, o, d, e, k) for p, o, d, e, k in OPTIONS_DATA)) def test_set_options(self, play, options, direct, expected, kerb): - if kerb: - kerberos_mock = MagicMock() - modules = {'kerberos': kerberos_mock} - else: - modules = {'kerberos': None} - - module_patcher = patch.dict(sys.modules, modules) - module_patcher.start() + winrm.HAVE_KERBEROS = kerb pc = PlayContext() for attr, value in play.items(): setattr(pc, attr, value) - new_stdin = StringIO() - # ensure we get a fresh connection plugin by clearing the cache - connection_loader._module_cache = {} conn = connection_loader.get('winrm', pc, new_stdin) conn.set_options(var_options=options, direct=direct) @@ -220,149 +209,159 @@ class TestConnectionWinRM(object): "winrm attr '%s', actual '%s' != expected '%s'"\ % (attr, actual, expected) - module_patcher.stop() - class TestWinRMKerbAuth(object): - DATA = ( - # default - ({"_extras": {}}, (["kinit", "user@domain"],), False), - ({"_extras": {}}, ("kinit", ["user@domain"],), True), - - # override kinit path from options - ({"_extras": {}, 'ansible_winrm_kinit_cmd': 'kinit2'}, - (["kinit2", "user@domain"],), False), - ({"_extras": {}, 'ansible_winrm_kinit_cmd': 'kinit2'}, - ("kinit2", ["user@domain"],), True), - - # we expect the -f flag when delegation is set - ({"_extras": {'ansible_winrm_kerberos_delegation': True}}, - (["kinit", "-f", "user@domain"],), False), - ({"_extras": {'ansible_winrm_kerberos_delegation': True}}, - ("kinit", ["-f", "user@domain"],), True), - ) - - # pylint bug: https://github.com/PyCQA/pylint/issues/511 - # pylint: disable=undefined-variable - @pytest.mark.parametrize('options, expected, pexpect', - ((o, e, p) for o, e, p in DATA)) - def test_kinit_success(self, options, expected, pexpect): - def mock_popen_communicate(input=None, timeout=None): + @pytest.mark.parametrize('options, expected', [ + [{"_extras": {}}, + (["kinit", "user@domain"],)], + [{"_extras": {}, 'ansible_winrm_kinit_cmd': 'kinit2'}, + (["kinit2", "user@domain"],)], + [{"_extras": {'ansible_winrm_kerberos_delegation': True}}, + (["kinit", "-f", "user@domain"],)], + ]) + def test_kinit_success_subprocess(self, monkeypatch, options, expected): + def mock_communicate(input=None, timeout=None): return b"", b"" - mock_pexpect = None - if pexpect: - mock_pexpect = MagicMock() - mock_pexpect.spawn.return_value.exitstatus = 0 + mock_popen = MagicMock() + mock_popen.return_value.communicate = mock_communicate + mock_popen.return_value.returncode = 0 + monkeypatch.setattr("subprocess.Popen", mock_popen) - mock_subprocess = MagicMock() - mock_subprocess.Popen.return_value.communicate = mock_popen_communicate - mock_subprocess.Popen.return_value.returncode = 0 - - modules = { - 'pexpect': mock_pexpect, - 'subprocess': mock_subprocess, - } + winrm.HAS_PEXPECT = False + pc = PlayContext() + new_stdin = StringIO() + conn = connection_loader.get('winrm', pc, new_stdin) + conn.set_options(var_options=options) + + conn._kerb_auth("user@domain", "pass") + mock_calls = mock_popen.mock_calls + assert len(mock_calls) == 1 + assert mock_calls[0][1] == expected + actual_env = mock_calls[0][2]['env'] + assert list(actual_env.keys()) == ['KRB5CCNAME'] + assert actual_env['KRB5CCNAME'].startswith("FILE:/") + + @pytest.mark.parametrize('options, expected', [ + [{"_extras": {}}, + ("kinit", ["user@domain"],)], + [{"_extras": {}, 'ansible_winrm_kinit_cmd': 'kinit2'}, + ("kinit2", ["user@domain"],)], + [{"_extras": {'ansible_winrm_kerberos_delegation': True}}, + ("kinit", ["-f", "user@domain"],)], + ]) + def test_kinit_success_pexpect(self, monkeypatch, options, expected): + pytest.importorskip("pexpect") + mock_pexpect = MagicMock() + mock_pexpect.return_value.exitstatus = 0 + monkeypatch.setattr("pexpect.spawn", mock_pexpect) + + winrm.HAS_PEXPECT = True + pc = PlayContext() + new_stdin = StringIO() + conn = connection_loader.get('winrm', pc, new_stdin) + conn.set_options(var_options=options) + + conn._kerb_auth("user@domain", "pass") + mock_calls = mock_pexpect.mock_calls + assert mock_calls[0][1] == expected + actual_env = mock_calls[0][2]['env'] + assert list(actual_env.keys()) == ['KRB5CCNAME'] + assert actual_env['KRB5CCNAME'].startswith("FILE:/") + assert mock_calls[1][0] == "().expect" + assert mock_calls[1][1] == (".*:",) + assert mock_calls[2][0] == "().sendline" + assert mock_calls[2][1] == ("pass",) + assert mock_calls[3][0] == "().read" + assert mock_calls[4][0] == "().wait" + + def test_kinit_with_missing_executable_subprocess(self, monkeypatch): + expected_err = "[Errno 2] No such file or directory: " \ + "'/fake/kinit': '/fake/kinit'" + mock_popen = MagicMock(side_effect=OSError(expected_err)) - with patch.dict(sys.modules, modules): - pc = PlayContext() - new_stdin = StringIO() + monkeypatch.setattr("subprocess.Popen", mock_popen) - connection_loader._module_cache = {} - conn = connection_loader.get('winrm', pc, new_stdin) - conn.set_options(var_options=options) + winrm.HAS_PEXPECT = False + pc = PlayContext() + new_stdin = StringIO() + conn = connection_loader.get('winrm', pc, new_stdin) + options = {"_extras": {}, "ansible_winrm_kinit_cmd": "/fake/kinit"} + conn.set_options(var_options=options) + with pytest.raises(AnsibleConnectionFailure) as err: conn._kerb_auth("user@domain", "pass") - if pexpect: - assert len(mock_pexpect.method_calls) == 1 - assert mock_pexpect.method_calls[0][1] == expected - actual_env = mock_pexpect.method_calls[0][2]['env'] - else: - assert len(mock_subprocess.method_calls) == 1 - assert mock_subprocess.method_calls[0][1] == expected - actual_env = mock_subprocess.method_calls[0][2]['env'] - - assert list(actual_env.keys()) == ['KRB5CCNAME'] - assert actual_env['KRB5CCNAME'].startswith("FILE:/") + assert str(err.value) == "Kerberos auth failure when calling " \ + "kinit cmd '/fake/kinit': %s" % expected_err - # pylint bug: https://github.com/PyCQA/pylint/issues/511 - # pylint: disable=undefined-variable - @pytest.mark.parametrize('use_pexpect', (False, True),) - def test_kinit_with_missing_executable(self, use_pexpect): - expected_err = "[Errno 2] No such file or directory: " \ - "'/fake/kinit': '/fake/kinit'" - mock_subprocess = MagicMock() - mock_subprocess.Popen = MagicMock(side_effect=OSError(expected_err)) - - mock_pexpect = None - if use_pexpect: - expected_err = "The command was not found or was not " \ - "executable: /fake/kinit" + def test_kinit_with_missing_executable_pexpect(self, monkeypatch): + pexpect = pytest.importorskip("pexpect") - mock_pexpect = MagicMock() - mock_pexpect.ExceptionPexpect = Exception - mock_pexpect.spawn = MagicMock(side_effect=Exception(expected_err)) + expected_err = "The command was not found or was not " \ + "executable: /fake/kinit" + mock_pexpect = \ + MagicMock(side_effect=pexpect.ExceptionPexpect(expected_err)) - modules = { - 'pexpect': mock_pexpect, - 'subprocess': mock_subprocess, - } + monkeypatch.setattr("pexpect.spawn", mock_pexpect) - with patch.dict(sys.modules, modules): - pc = PlayContext() - new_stdin = StringIO() - - connection_loader._module_cache = {} - conn = connection_loader.get('winrm', pc, new_stdin) - options = {"_extras": {}, "ansible_winrm_kinit_cmd": "/fake/kinit"} - conn.set_options(var_options=options) + winrm.HAS_PEXPECT = True + pc = PlayContext() + new_stdin = StringIO() + conn = connection_loader.get('winrm', pc, new_stdin) + options = {"_extras": {}, "ansible_winrm_kinit_cmd": "/fake/kinit"} + conn.set_options(var_options=options) - with pytest.raises(AnsibleConnectionFailure) as err: - conn._kerb_auth("user@domain", "pass") - assert str(err.value) == "Kerberos auth failure when calling " \ - "kinit cmd '/fake/kinit': %s" % expected_err + with pytest.raises(AnsibleConnectionFailure) as err: + conn._kerb_auth("user@domain", "pass") + assert str(err.value) == "Kerberos auth failure when calling " \ + "kinit cmd '/fake/kinit': %s" % expected_err - # pylint bug: https://github.com/PyCQA/pylint/issues/511 - # pylint: disable=undefined-variable - @pytest.mark.parametrize('use_pexpect', (False, True),) - def test_kinit_error(self, use_pexpect): - mechanism = "subprocess" + def test_kinit_error_subprocess(self, monkeypatch): expected_err = "kinit: krb5_parse_name: " \ "Configuration file does not specify default realm" - def mock_popen_communicate(input=None, timeout=None): + def mock_communicate(input=None, timeout=None): return b"", to_bytes(expected_err) - mock_subprocess = MagicMock() - mock_subprocess.Popen.return_value.communicate = mock_popen_communicate - mock_subprocess.Popen.return_value.returncode = 1 + mock_popen = MagicMock() + mock_popen.return_value.communicate = mock_communicate + mock_popen.return_value.returncode = 1 + monkeypatch.setattr("subprocess.Popen", mock_popen) + + winrm.HAS_PEXPECT = False + pc = PlayContext() + new_stdin = StringIO() + conn = connection_loader.get('winrm', pc, new_stdin) + conn.set_options(var_options={"_extras": {}}) + + with pytest.raises(AnsibleConnectionFailure) as err: + conn._kerb_auth("invaliduser", "pass") - mock_pexpect = None - if use_pexpect: - mechanism = "pexpect" - expected_err = "Configuration file does not specify default realm" + assert str(err.value) == \ + "Kerberos auth failure for principal invaliduser with " \ + "subprocess: %s" % (expected_err) - mock_pexpect = MagicMock() - mock_pexpect.spawn.return_value.expect = MagicMock(side_effect=OSError) - mock_pexpect.spawn.return_value.read.return_value = to_bytes(expected_err) - mock_pexpect.spawn.return_value.exitstatus = 1 + def test_kinit_error_pexpect(self, monkeypatch): + pytest.importorskip("pexpect") - modules = { - 'pexpect': mock_pexpect, - 'subprocess': mock_subprocess, - } + expected_err = "Configuration file does not specify default realm" + mock_pexpect = MagicMock() + mock_pexpect.return_value.expect = MagicMock(side_effect=OSError) + mock_pexpect.return_value.read.return_value = to_bytes(expected_err) + mock_pexpect.return_value.exitstatus = 1 - with patch.dict(sys.modules, modules): - pc = PlayContext() - new_stdin = StringIO() + monkeypatch.setattr("pexpect.spawn", mock_pexpect) - connection_loader._module_cache = {} - conn = connection_loader.get('winrm', pc, new_stdin) - conn.set_options(var_options={"_extras": {}}) + winrm.HAS_PEXPECT = True + pc = PlayContext() + new_stdin = StringIO() + conn = connection_loader.get('winrm', pc, new_stdin) + conn.set_options(var_options={"_extras": {}}) - with pytest.raises(AnsibleConnectionFailure) as err: - conn._kerb_auth("invaliduser", "pass") + with pytest.raises(AnsibleConnectionFailure) as err: + conn._kerb_auth("invaliduser", "pass") - assert str(err.value) == "Kerberos auth failure for principal " \ - "invaliduser with %s: %s" % (mechanism, expected_err) + assert str(err.value) == \ + "Kerberos auth failure for principal invaliduser with " \ + "pexpect: %s" % (expected_err)