ansible/community.general
1
0
Fork 0
mirror of https://github.com/ansible-collections/community.general.git synced 2024-09-14 20:13:21 +02:00
Code Releases Activity

winrm: fix up unit tests (#41112)

Browse source
This commit is contained in:
Jordan Borean 2018-06-07 06:09:10 +08:00 committed by GitHub
parent 453a6f4047
commit ad8e13e9f8
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
2 changed files with 129 additions and 127 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

3
test/runner/requirements/units.txt
View file

@ -28,3 +28,6 @@ pyfmg
# requirement for aci_rest module # requirement for aci_rest module
xmljson xmljson
# requirement for winrm connection plugin tests
pexpect

253
test/units/plugins/connection/test_winrm.py
View file

@ -6,8 +6,6 @@
from __future__ import (absolute_import, division, print_function) from __future__ import (absolute_import, division, print_function)
__metaclass__ = type __metaclass__ = type
import sys
import pytest import pytest
from io import StringIO from io import StringIO
@ -17,6 +15,7 @@ from ansible.errors import AnsibleConnectionFailure
from ansible.module_utils._text import to_bytes from ansible.module_utils._text import to_bytes
from ansible.playbook.play_context import PlayContext from ansible.playbook.play_context import PlayContext
from ansible.plugins.loader import connection_loader from ansible.plugins.loader import connection_loader
from ansible.plugins.connection import winrm
class TestConnectionWinRM(object): class TestConnectionWinRM(object):
@ -194,23 +193,13 @@ class TestConnectionWinRM(object):
@pytest.mark.parametrize('play, options, direct, expected, kerb', @pytest.mark.parametrize('play, options, direct, expected, kerb',
((p, o, d, e, k) for p, o, d, e, k in OPTIONS_DATA)) ((p, o, d, e, k) for p, o, d, e, k in OPTIONS_DATA))
def test_set_options(self, play, options, direct, expected, kerb): def test_set_options(self, play, options, direct, expected, kerb):
if kerb: winrm.HAVE_KERBEROS = kerb
kerberos_mock = MagicMock()
modules = {'kerberos': kerberos_mock}
else:
modules = {'kerberos': None}
module_patcher = patch.dict(sys.modules, modules)
module_patcher.start()
pc = PlayContext() pc = PlayContext()
for attr, value in play.items(): for attr, value in play.items():
setattr(pc, attr, value) setattr(pc, attr, value)
new_stdin = StringIO() 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 = connection_loader.get('winrm', pc, new_stdin)
conn.set_options(var_options=options, direct=direct) conn.set_options(var_options=options, direct=direct)
@ -220,149 +209,159 @@ class TestConnectionWinRM(object):
"winrm attr '%s', actual '%s' != expected '%s'"\ "winrm attr '%s', actual '%s' != expected '%s'"\
% (attr, actual, expected) % (attr, actual, expected)
module_patcher.stop()
class TestWinRMKerbAuth(object): class TestWinRMKerbAuth(object):
DATA = ( @pytest.mark.parametrize('options, expected', [
# default [{"_extras": {}},
({"_extras": {}}, (["kinit", "user@domain"],), False), (["kinit", "user@domain"],)],
({"_extras": {}}, ("kinit", ["user@domain"],), True), [{"_extras": {}, 'ansible_winrm_kinit_cmd': 'kinit2'},
(["kinit2", "user@domain"],)],
# override kinit path from options [{"_extras": {'ansible_winrm_kerberos_delegation': True}},
({"_extras": {}, 'ansible_winrm_kinit_cmd': 'kinit2'}, (["kinit", "-f", "user@domain"],)],
(["kinit2", "user@domain"],), False), ])
({"_extras": {}, 'ansible_winrm_kinit_cmd': 'kinit2'}, def test_kinit_success_subprocess(self, monkeypatch, options, expected):
("kinit2", ["user@domain"],), True), def mock_communicate(input=None, timeout=None):
# 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):
return b"", b"" return b"", b""
mock_pexpect = None mock_popen = MagicMock()
if pexpect: mock_popen.return_value.communicate = mock_communicate
mock_pexpect = MagicMock() mock_popen.return_value.returncode = 0
mock_pexpect.spawn.return_value.exitstatus = 0 monkeypatch.setattr("subprocess.Popen", mock_popen)
mock_subprocess = MagicMock() winrm.HAS_PEXPECT = False
mock_subprocess.Popen.return_value.communicate = mock_popen_communicate pc = PlayContext()
mock_subprocess.Popen.return_value.returncode = 0 new_stdin = StringIO()
conn = connection_loader.get('winrm', pc, new_stdin)
conn.set_options(var_options=options)
modules = { conn._kerb_auth("user@domain", "pass")
'pexpect': mock_pexpect, mock_calls = mock_popen.mock_calls
'subprocess': mock_subprocess, 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:/")
with patch.dict(sys.modules, modules): @pytest.mark.parametrize('options, expected', [
pc = PlayContext() [{"_extras": {}},
new_stdin = StringIO() ("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)
connection_loader._module_cache = {} winrm.HAS_PEXPECT = True
conn = connection_loader.get('winrm', pc, new_stdin) pc = PlayContext()
conn.set_options(var_options=options) new_stdin = StringIO()
conn = connection_loader.get('winrm', pc, new_stdin)
conn.set_options(var_options=options)
conn._kerb_auth("user@domain", "pass") conn._kerb_auth("user@domain", "pass")
if pexpect: mock_calls = mock_pexpect.mock_calls
assert len(mock_pexpect.method_calls) == 1 assert mock_calls[0][1] == expected
assert mock_pexpect.method_calls[0][1] == expected actual_env = mock_calls[0][2]['env']
actual_env = mock_pexpect.method_calls[0][2]['env'] assert list(actual_env.keys()) == ['KRB5CCNAME']
else: assert actual_env['KRB5CCNAME'].startswith("FILE:/")
assert len(mock_subprocess.method_calls) == 1 assert mock_calls[1][0] == "().expect"
assert mock_subprocess.method_calls[0][1] == expected assert mock_calls[1][1] == (".*:",)
actual_env = mock_subprocess.method_calls[0][2]['env'] assert mock_calls[2][0] == "().sendline"
assert mock_calls[2][1] == ("pass",)
assert mock_calls[3][0] == "().read"
assert mock_calls[4][0] == "().wait"
assert list(actual_env.keys()) == ['KRB5CCNAME'] def test_kinit_with_missing_executable_subprocess(self, monkeypatch):
assert actual_env['KRB5CCNAME'].startswith("FILE:/")
# 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: " \ expected_err = "[Errno 2] No such file or directory: " \
"'/fake/kinit': '/fake/kinit'" "'/fake/kinit': '/fake/kinit'"
mock_subprocess = MagicMock() mock_popen = MagicMock(side_effect=OSError(expected_err))
mock_subprocess.Popen = MagicMock(side_effect=OSError(expected_err))
mock_pexpect = None monkeypatch.setattr("subprocess.Popen", mock_popen)
if use_pexpect:
expected_err = "The command was not found or was not " \
"executable: /fake/kinit"
mock_pexpect = MagicMock() winrm.HAS_PEXPECT = False
mock_pexpect.ExceptionPexpect = Exception pc = PlayContext()
mock_pexpect.spawn = MagicMock(side_effect=Exception(expected_err)) 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)
modules = { with pytest.raises(AnsibleConnectionFailure) as err:
'pexpect': mock_pexpect, conn._kerb_auth("user@domain", "pass")
'subprocess': mock_subprocess, assert str(err.value) == "Kerberos auth failure when calling " \
} "kinit cmd '/fake/kinit': %s" % expected_err
with patch.dict(sys.modules, modules): def test_kinit_with_missing_executable_pexpect(self, monkeypatch):
pc = PlayContext() pexpect = pytest.importorskip("pexpect")
new_stdin = StringIO()
connection_loader._module_cache = {} expected_err = "The command was not found or was not " \
conn = connection_loader.get('winrm', pc, new_stdin) "executable: /fake/kinit"
options = {"_extras": {}, "ansible_winrm_kinit_cmd": "/fake/kinit"} mock_pexpect = \
conn.set_options(var_options=options) MagicMock(side_effect=pexpect.ExceptionPexpect(expected_err))
with pytest.raises(AnsibleConnectionFailure) as err: monkeypatch.setattr("pexpect.spawn", mock_pexpect)
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 winrm.HAS_PEXPECT = True
# pylint: disable=undefined-variable pc = PlayContext()
@pytest.mark.parametrize('use_pexpect', (False, True),) new_stdin = StringIO()
def test_kinit_error(self, use_pexpect): conn = connection_loader.get('winrm', pc, new_stdin)
mechanism = "subprocess" 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
def test_kinit_error_subprocess(self, monkeypatch):
expected_err = "kinit: krb5_parse_name: " \ expected_err = "kinit: krb5_parse_name: " \
"Configuration file does not specify default realm" "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) return b"", to_bytes(expected_err)
mock_subprocess = MagicMock() mock_popen = MagicMock()
mock_subprocess.Popen.return_value.communicate = mock_popen_communicate mock_popen.return_value.communicate = mock_communicate
mock_subprocess.Popen.return_value.returncode = 1 mock_popen.return_value.returncode = 1
monkeypatch.setattr("subprocess.Popen", mock_popen)
mock_pexpect = None winrm.HAS_PEXPECT = False
if use_pexpect: pc = PlayContext()
mechanism = "pexpect" new_stdin = StringIO()
expected_err = "Configuration file does not specify default realm" conn = connection_loader.get('winrm', pc, new_stdin)
conn.set_options(var_options={"_extras": {}})
mock_pexpect = MagicMock() with pytest.raises(AnsibleConnectionFailure) as err:
mock_pexpect.spawn.return_value.expect = MagicMock(side_effect=OSError) conn._kerb_auth("invaliduser", "pass")
mock_pexpect.spawn.return_value.read.return_value = to_bytes(expected_err)
mock_pexpect.spawn.return_value.exitstatus = 1
modules = { assert str(err.value) == \
'pexpect': mock_pexpect, "Kerberos auth failure for principal invaliduser with " \
'subprocess': mock_subprocess, "subprocess: %s" % (expected_err)
}
with patch.dict(sys.modules, modules): def test_kinit_error_pexpect(self, monkeypatch):
pc = PlayContext() pytest.importorskip("pexpect")
new_stdin = StringIO()
connection_loader._module_cache = {} expected_err = "Configuration file does not specify default realm"
conn = connection_loader.get('winrm', pc, new_stdin) mock_pexpect = MagicMock()
conn.set_options(var_options={"_extras": {}}) 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 pytest.raises(AnsibleConnectionFailure) as err: monkeypatch.setattr("pexpect.spawn", mock_pexpect)
conn._kerb_auth("invaliduser", "pass")
assert str(err.value) == "Kerberos auth failure for principal " \ winrm.HAS_PEXPECT = True
"invaliduser with %s: %s" % (mechanism, expected_err) 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")
assert str(err.value) == \
"Kerberos auth failure for principal invaliduser with " \
"pexpect: %s" % (expected_err)