saltstack/salt
1
0
Fork 0
mirror of https://github.com/saltstack/salt.git synced 2025-04-17 10:10:20 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 2

Move module tests to pytest (#61217)

Browse source
This commit is contained in:
Kirill Ponomarev 2021-11-11 16:26:59 +01:00 • committed by GitHub
parent 4c0968e87f
commit ed9215e24e
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
8 changed files with 830 additions and 866 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

155
tests/pytests/unit/modules/test_pkgin.py Normal file
View file

@ -0,0 +1,155 @@
import os
import pytest
import salt.modules.pkgin as pkgin
from tests.support.mock import MagicMock, patch
@pytest.fixture
def configure_loader_modules(tmp_path):
return {pkgin: {"__opts__": {"cachedir": str(tmp_path)}}}
def test_search():
"""
Test searching for a package
"""
pkgin_out = [
"somepkg-1.0 Some package description here",
"",
"=: package is installed and up-to-date",
"<: package is installed but newer version is available",
">: installed package has a greater version than available package",
]
pkgin__get_version_mock = MagicMock(return_value=["0", "9", "0"])
pkgin__check_pkgin_mock = MagicMock(return_value="/opt/pkg/bin/pkgin")
pkgin_search_cmd = MagicMock(return_value=os.linesep.join(pkgin_out))
with patch("salt.modules.pkgin._get_version", pkgin__get_version_mock), patch(
"salt.modules.pkgin._check_pkgin", pkgin__check_pkgin_mock
), patch.dict(pkgin.__salt__, {"cmd.run": pkgin_search_cmd}):
assert pkgin.search("somepkg") == {"somepkg": "1.0"}
pkgin_out = [
"somepkg-1.0 = Some package description here",
"",
"=: package is installed and up-to-date",
"<: package is installed but newer version is available",
">: installed package has a greater version than available package",
]
pkgin_search_cmd = MagicMock(return_value=os.linesep.join(pkgin_out))
with patch("salt.modules.pkgin._get_version", pkgin__get_version_mock), patch(
"salt.modules.pkgin._check_pkgin", pkgin__check_pkgin_mock
), patch.dict(pkgin.__salt__, {"cmd.run": pkgin_search_cmd}):
assert pkgin.search("somepkg") == {"somepkg": "1.0"}
def test_latest_version():
"""
Test getting the latest version of a package
"""
pkgin_out = [
"somepkg-1.0;;Some package description here",
"",
"=: package is installed and up-to-date",
"<: package is installed but newer version is available",
">: installed package has a greater version than available package",
]
pkgin__get_version_mock = MagicMock(return_value=["0", "9", "0"])
pkgin__check_pkgin_mock = MagicMock(return_value="/opt/pkg/bin/pkgin")
pkgin_refresh_db_mock = MagicMock(return_value=True)
pkgin_search_cmd = MagicMock(return_value=os.linesep.join(pkgin_out))
with patch("salt.modules.pkgin.refresh_db", pkgin_refresh_db_mock), patch(
"salt.modules.pkgin._get_version", pkgin__get_version_mock
), patch("salt.modules.pkgin._check_pkgin", pkgin__check_pkgin_mock), patch.dict(
pkgin.__salt__, {"cmd.run": pkgin_search_cmd}
):
assert pkgin.latest_version("somepkg") == "1.0"
pkgin_out = [
"somepkg-1.1;<;Some package description here",
"",
"=: package is installed and up-to-date",
"<: package is installed but newer version is available",
">: installed package has a greater version than available package",
]
pkgin_refresh_db_mock = MagicMock(return_value=True)
pkgin_search_cmd = MagicMock(return_value=os.linesep.join(pkgin_out))
with patch("salt.modules.pkgin.refresh_db", pkgin_refresh_db_mock), patch(
"salt.modules.pkgin._get_version", pkgin__get_version_mock
), patch("salt.modules.pkgin._check_pkgin", pkgin__check_pkgin_mock), patch.dict(
pkgin.__salt__, {"cmd.run": pkgin_search_cmd}
):
assert pkgin.latest_version("somepkg") == "1.1"
pkgin_out = [
"somepkg-1.2;=;Some package description here",
"",
"=: package is installed and up-to-date",
"<: package is installed but newer version is available",
">: installed package has a greater version than available package",
]
pkgin_refresh_db_mock = MagicMock(return_value=True)
pkgin_search_cmd = MagicMock(return_value=os.linesep.join(pkgin_out))
with patch("salt.modules.pkgin.refresh_db", pkgin_refresh_db_mock), patch(
"salt.modules.pkgin._get_version", pkgin__get_version_mock
), patch("salt.modules.pkgin._check_pkgin", pkgin__check_pkgin_mock), patch.dict(
pkgin.__salt__, {"cmd.run": pkgin_search_cmd}
):
assert pkgin.latest_version("somepkg") == "1.2"
pkgin_out = "No results found for ^boguspkg$"
pkgin_refresh_db_mock = MagicMock(return_value=True)
pkgin_search_cmd = MagicMock(return_value=pkgin_out)
with patch("salt.modules.pkgin.refresh_db", pkgin_refresh_db_mock), patch(
"salt.modules.pkgin._get_version", pkgin__get_version_mock
), patch("salt.modules.pkgin._check_pkgin", pkgin__check_pkgin_mock), patch.dict(
pkgin.__salt__, {"cmd.run": pkgin_search_cmd}
):
assert pkgin.latest_version("boguspkg") == {}
def test_file_dict():
"""
Test that file_dict doesn't crash
"""
pkg_info_stdout = [
"/opt/pkg/bin/pkgin",
"/opt/pkg/man/man1/pkgin.1",
"/opt/pkg/share/examples/pkgin/preferred.conf.example",
"/opt/pkg/share/examples/pkgin/repositories.conf.example",
]
pkg_info_out = {
"pid": 1234,
"retcode": 0,
"stderr": "",
"stdout": os.linesep.join(pkg_info_stdout),
}
pkg_info_cmd = MagicMock(return_value=pkg_info_out)
with patch.dict(pkgin.__salt__, {"cmd.run_all": pkg_info_cmd}):
assert pkgin.file_dict("pkgin") == {
"files": {
"pkgin": [
"/opt/pkg/bin/pkgin",
"/opt/pkg/man/man1/pkgin.1",
"/opt/pkg/share/examples/pkgin/preferred.conf.example",
"/opt/pkg/share/examples/pkgin/repositories.conf.example",
]
}
}

74
tests/pytests/unit/modules/test_powerpath.py Normal file
View file

@ -0,0 +1,74 @@
"""
:codeauthor: Jayesh Kariya <jayeshk@saltstack.com>
"""
import pytest
import salt.modules.powerpath as powerpath
from tests.support.mock import MagicMock, patch
@pytest.fixture
def configure_loader_modules():
return {powerpath: {}}
def test_has_powerpath():
"""
Test for powerpath
"""
with patch("os.path.exists") as mock_exists:
mock_exists.return_value = True
assert powerpath.has_powerpath()
mock_exists.return_value = False
assert not powerpath.has_powerpath()
def test_list_licenses():
"""
Test to returns a list of applied powerpath license keys
"""
with patch.dict(powerpath.__salt__, {"cmd.run": MagicMock(return_value="A\nB")}):
assert powerpath.list_licenses() == []
def test_add_license():
"""
Test to add a license
"""
with patch.object(powerpath, "has_powerpath", return_value=False):
assert powerpath.add_license("key") == {
"output": "PowerPath is not installed",
"result": False,
"retcode": -1,
}
mock = MagicMock(return_value={"retcode": 1, "stderr": "stderr"})
with patch.object(powerpath, "has_powerpath", return_value=True):
with patch.dict(powerpath.__salt__, {"cmd.run_all": mock}):
assert powerpath.add_license("key") == {
"output": "stderr",
"result": False,
"retcode": 1,
}
def test_remove_license():
"""
Test to remove a license
"""
with patch.object(powerpath, "has_powerpath", return_value=False):
assert powerpath.remove_license("key") == {
"output": "PowerPath is not installed",
"result": False,
"retcode": -1,
}
mock = MagicMock(return_value={"retcode": 1, "stderr": "stderr"})
with patch.object(powerpath, "has_powerpath", return_value=True):
with patch.dict(powerpath.__salt__, {"cmd.run_all": mock}):
assert powerpath.remove_license("key") == {
"output": "stderr",
"result": False,
"retcode": 1,
}

499
tests/pytests/unit/modules/test_proxy.py Normal file
View file

@ -0,0 +1,499 @@
import pytest
import salt.modules.proxy as proxy
from tests.support.mock import MagicMock, call, patch
@pytest.fixture
def configure_loader_modules():
return {proxy: {"__grains__": {"os": "Darwin"}}}
def test_get_http_proxy_macos():
"""
Test to make sure that we correctly get the current proxy info
on macOS
"""
mock = MagicMock(
return_value=(
"Enabled: Yes\nServer: 192.168.0.1\nPort: 3128\nAuthenticated Proxy"
" Enabled: 0"
)
)
expected = {"enabled": True, "server": "192.168.0.1", "port": "3128"}
with patch.dict(proxy.__salt__, {"cmd.run": mock}):
out = proxy.get_http_proxy()
mock.assert_called_once_with("networksetup -getwebproxy Ethernet")
assert expected == out
def test_get_https_proxy_macos():
"""
Test to make sure that we correctly get the current proxy info
on macOS
"""
mock = MagicMock(
return_value=(
"Enabled: Yes\nServer: 192.168.0.1\nPort: 3128\nAuthenticated Proxy"
" Enabled: 0"
)
)
expected = {"enabled": True, "server": "192.168.0.1", "port": "3128"}
with patch.dict(proxy.__salt__, {"cmd.run": mock}):
out = proxy.get_https_proxy()
mock.assert_called_once_with("networksetup -getsecurewebproxy Ethernet")
assert expected == out
def test_get_ftp_proxy_macos():
"""
Test to make sure that we correctly get the current proxy info
on macOS
"""
mock = MagicMock(
return_value=(
"Enabled: Yes\nServer: 192.168.0.1\nPort: 3128\nAuthenticated Proxy"
" Enabled: 0"
)
)
expected = {"enabled": True, "server": "192.168.0.1", "port": "3128"}
with patch.dict(proxy.__salt__, {"cmd.run": mock}):
out = proxy.get_ftp_proxy()
mock.assert_called_once_with("networksetup -getftpproxy Ethernet")
assert expected == out
def test_get_http_proxy_macos_none():
"""
Test to make sure that we correctly return when there's no proxy set
"""
mock = MagicMock(
return_value="Enabled: No\nServer:\nPort: 0\nAuthenticated Proxy Enabled: 0"
)
with patch.dict(proxy.__salt__, {"cmd.run": mock}):
out = proxy.get_http_proxy()
mock.assert_called_once_with("networksetup -getwebproxy Ethernet")
assert {} == out
def test_set_http_proxy_macos():
"""
Test to make sure that we correctly set the proxy info
on macOS
"""
mock = MagicMock()
with patch.dict(proxy.__salt__, {"cmd.run": mock}):
out = proxy.set_http_proxy(
"192.168.0.1",
3128,
"frank",
"badpassw0rd",
bypass_hosts=".moo.com,.salt.com",
)
mock.assert_called_once_with(
"networksetup -setwebproxy Ethernet 192.168.0.1 3128 On frank"
" badpassw0rd"
)
assert out
def test_set_https_proxy_macos():
"""
Test to make sure that we correctly set the proxy info
on macOS
"""
mock = MagicMock()
with patch.dict(proxy.__salt__, {"cmd.run": mock}):
out = proxy.set_https_proxy(
"192.168.0.1",
3128,
"frank",
"passw0rd",
bypass_hosts=".moo.com,.salt.com",
)
mock.assert_called_once_with(
"networksetup -setsecurewebproxy Ethernet 192.168.0.1 3128 On frank"
" passw0rd"
)
assert out
def test_set_ftp_proxy_macos():
"""
Test to make sure that we correctly set the proxy info
on macOS
"""
mock = MagicMock()
with patch.dict(proxy.__salt__, {"cmd.run": mock}):
out = proxy.set_ftp_proxy(
"192.168.0.1",
3128,
"frank",
"badpassw0rd",
bypass_hosts=".moo.com,.salt.com",
)
mock.assert_called_once_with(
"networksetup -setftpproxy Ethernet 192.168.0.1 3128 On frank"
" badpassw0rd"
)
assert out
def test_get_proxy_win():
"""
Test to make sure that we correctly get the current proxy info on
Windows
"""
result = [
{
"vdata": (
"http=192.168.0.1:3128;https=192.168.0.2:3128;ftp=192.168.0.3:3128"
)
},
{"vdata": 1},
]
mock_reg_read = MagicMock(side_effect=result)
expected = {
"enabled": True,
"http": {"server": "192.168.0.1", "port": "3128"},
"https": {"server": "192.168.0.2", "port": "3128"},
"ftp": {"server": "192.168.0.3", "port": "3128"},
}
with patch.dict(proxy.__grains__, {"os": "Windows"}), patch.dict(
proxy.__utils__, {"reg.read_value": mock_reg_read}
):
out = proxy.get_proxy_win()
assert out == expected
mock_reg_read.assert_any_call(
hive="HKEY_CURRENT_USER",
key="SOFTWARE\\Microsoft\\Windows\\CurrentVersion\\Internet Settings",
vname="ProxyServer",
)
mock_reg_read.assert_any_call(
hive="HKEY_CURRENT_USER",
key="SOFTWARE\\Microsoft\\Windows\\CurrentVersion\\Internet Settings",
vname="ProxyEnable",
)
def test_get_http_proxy_windows():
"""
Test to make sure that we correctly get the current proxy info on
Windows
"""
result = {
"vdata": "http=192.168.0.1:3128;https=192.168.0.2:3128;ftp=192.168.0.3:3128"
}
mock = MagicMock(return_value=result)
expected = {"server": "192.168.0.1", "port": "3128"}
with patch.dict(proxy.__grains__, {"os": "Windows"}), patch.dict(
proxy.__utils__, {"reg.read_value": mock}
):
out = proxy.get_http_proxy()
mock.assert_called_once_with(
hive="HKEY_CURRENT_USER",
key="SOFTWARE\\Microsoft\\Windows\\CurrentVersion\\Internet Settings",
vname="ProxyServer",
)
assert expected == out
def test_get_https_proxy_windows():
"""
Test to make sure that we correctly get the current proxy info on
Windows
"""
result = {
"vdata": "http=192.168.0.1:3128;https=192.168.0.2:3128;ftp=192.168.0.3:3128"
}
mock = MagicMock(return_value=result)
expected = {"server": "192.168.0.2", "port": "3128"}
with patch.dict(proxy.__grains__, {"os": "Windows"}), patch.dict(
proxy.__utils__, {"reg.read_value": mock}
):
out = proxy.get_https_proxy()
mock.assert_called_once_with(
hive="HKEY_CURRENT_USER",
key="SOFTWARE\\Microsoft\\Windows\\CurrentVersion\\Internet Settings",
vname="ProxyServer",
)
assert expected == out
def test_get_ftp_proxy_windows():
"""
Test to make sure that we correctly get the current proxy info on
Windows
"""
result = {
"vdata": "http=192.168.0.1:3128;https=192.168.0.2:3128;ftp=192.168.0.3:3128"
}
mock = MagicMock(return_value=result)
expected = {"server": "192.168.0.3", "port": "3128"}
with patch.dict(proxy.__grains__, {"os": "Windows"}), patch.dict(
proxy.__utils__, {"reg.read_value": mock}
):
out = proxy.get_ftp_proxy()
mock.assert_called_once_with(
hive="HKEY_CURRENT_USER",
key="SOFTWARE\\Microsoft\\Windows\\CurrentVersion\\Internet Settings",
vname="ProxyServer",
)
assert expected == out
def test_get_all_proxies_macos_fails():
mock = MagicMock()
with patch.dict(proxy.__utils__, {"reg.read_value": mock}):
out = proxy.get_proxy_win()
assert not mock.called
assert out is None
def test_get_all_proxies_windows():
"""
Test to make sure that we correctly get the current proxy info on
Windows
"""
results = [
{
"vdata": (
"http=192.168.0.1:3128;https=192.168.0.2:3128;ftp=192.168.0.3:3128"
)
},
{"vdata": 1},
]
mock = MagicMock(side_effect=results)
expected = {
"enabled": True,
"http": {"server": "192.168.0.1", "port": "3128"},
"https": {"server": "192.168.0.2", "port": "3128"},
"ftp": {"server": "192.168.0.3", "port": "3128"},
}
calls = [
call(
hive="HKEY_CURRENT_USER",
key="SOFTWARE\\Microsoft\\Windows\\CurrentVersion\\Internet Settings",
vname="ProxyServer",
),
call(
hive="HKEY_CURRENT_USER",
key="SOFTWARE\\Microsoft\\Windows\\CurrentVersion\\Internet Settings",
vname="ProxyEnable",
),
]
with patch.dict(proxy.__grains__, {"os": "Windows"}), patch.dict(
proxy.__utils__, {"reg.read_value": mock}
):
out = proxy.get_proxy_win()
mock.assert_has_calls(calls)
assert expected == out
def test_set_http_proxy_windows():
"""
Test to make sure that we correctly set the proxy info on Windows
"""
calls = [
call(
hive="HKEY_CURRENT_USER",
key="SOFTWARE\\Microsoft\\Windows\\CurrentVersion\\Internet Settings",
vname="ProxyServer",
vdata="http=192.168.0.1:3128;",
),
call(
hive="HKEY_CURRENT_USER",
key="SOFTWARE\\Microsoft\\Windows\\CurrentVersion\\Internet Settings",
vname="ProxyEnable",
vdata=1,
vtype="REG_DWORD",
),
call(
hive="HKEY_CURRENT_USER",
key="SOFTWARE\\Microsoft\\Windows\\CurrentVersion\\Internet Settings",
vname="ProxyOverride",
vdata="<local>;.moo.com;.salt.com",
),
]
mock_reg = MagicMock()
mock_cmd = MagicMock()
with patch.dict(proxy.__grains__, {"os": "Windows"}), patch.dict(
proxy.__utils__, {"reg.set_value": mock_reg}
), patch.dict(proxy.__salt__, {"cmd.run": mock_cmd}):
out = proxy.set_http_proxy(
server="192.168.0.1",
port=3128,
bypass_hosts=[".moo.com", ".salt.com"],
)
mock_reg.assert_has_calls(calls)
mock_cmd.assert_called_once_with("netsh winhttp import proxy source=ie")
assert out
def test_set_https_proxy_windows():
"""
Test to make sure that we correctly set the proxy info on Windows
"""
calls = [
call(
hive="HKEY_CURRENT_USER",
key="SOFTWARE\\Microsoft\\Windows\\CurrentVersion\\Internet Settings",
vname="ProxyServer",
vdata="https=192.168.0.1:3128;",
),
call(
hive="HKEY_CURRENT_USER",
key="SOFTWARE\\Microsoft\\Windows\\CurrentVersion\\Internet Settings",
vname="ProxyEnable",
vdata=1,
vtype="REG_DWORD",
),
call(
hive="HKEY_CURRENT_USER",
key="SOFTWARE\\Microsoft\\Windows\\CurrentVersion\\Internet Settings",
vname="ProxyOverride",
vdata="<local>;.moo.com;.salt.com",
),
]
mock_reg = MagicMock()
mock_cmd = MagicMock()
with patch.dict(proxy.__grains__, {"os": "Windows"}), patch.dict(
proxy.__utils__, {"reg.set_value": mock_reg}
), patch.dict(proxy.__salt__, {"cmd.run": mock_cmd}):
out = proxy.set_https_proxy(
server="192.168.0.1",
port=3128,
bypass_hosts=[".moo.com", ".salt.com"],
)
mock_reg.assert_has_calls(calls)
mock_cmd.assert_called_once_with("netsh winhttp import proxy source=ie")
assert out
def test_set_ftp_proxy_windows():
"""
Test to make sure that we correctly set the proxy info on Windows
"""
calls = [
call(
hive="HKEY_CURRENT_USER",
key="SOFTWARE\\Microsoft\\Windows\\CurrentVersion\\Internet Settings",
vname="ProxyServer",
vdata="ftp=192.168.0.1:3128;",
),
call(
hive="HKEY_CURRENT_USER",
key="SOFTWARE\\Microsoft\\Windows\\CurrentVersion\\Internet Settings",
vname="ProxyEnable",
vdata=1,
vtype="REG_DWORD",
),
call(
hive="HKEY_CURRENT_USER",
key="SOFTWARE\\Microsoft\\Windows\\CurrentVersion\\Internet Settings",
vname="ProxyOverride",
vdata="<local>;.moo.com;.salt.com",
),
]
mock_reg = MagicMock()
mock_cmd = MagicMock()
with patch.dict(proxy.__grains__, {"os": "Windows"}), patch.dict(
proxy.__utils__, {"reg.set_value": mock_reg}
), patch.dict(proxy.__salt__, {"cmd.run": mock_cmd}):
out = proxy.set_ftp_proxy(
server="192.168.0.1",
port=3128,
bypass_hosts=[".moo.com", ".salt.com"],
)
mock_reg.assert_has_calls(calls)
mock_cmd.assert_called_once_with("netsh winhttp import proxy source=ie")
assert out
def test_set_proxy_windows():
"""
Test to make sure that we correctly set the proxy info on Windows
"""
calls = [
call(
hive="HKEY_CURRENT_USER",
key="SOFTWARE\\Microsoft\\Windows\\CurrentVersion\\Internet Settings",
vname="ProxyServer",
vdata=(
"http=192.168.0.1:3128;https=192.168.0.1:3128;ftp=192.168.0.1:3128;"
),
),
call(
hive="HKEY_CURRENT_USER",
key="SOFTWARE\\Microsoft\\Windows\\CurrentVersion\\Internet Settings",
vname="ProxyEnable",
vdata=1,
vtype="REG_DWORD",
),
call(
hive="HKEY_CURRENT_USER",
key="SOFTWARE\\Microsoft\\Windows\\CurrentVersion\\Internet Settings",
vname="ProxyOverride",
vdata="<local>;.moo.com;.salt.com",
),
]
mock_reg = MagicMock()
mock_cmd = MagicMock()
with patch.dict(proxy.__grains__, {"os": "Windows"}), patch.dict(
proxy.__utils__, {"reg.set_value": mock_reg}
), patch.dict(proxy.__salt__, {"cmd.run": mock_cmd}):
out = proxy.set_proxy_win(
server="192.168.0.1",
port=3128,
bypass_hosts=[".moo.com", ".salt.com"],
)
mock_reg.assert_has_calls(calls)
mock_cmd.assert_called_once_with("netsh winhttp import proxy source=ie")
assert out
def test_set_proxy_windows_no_ftp():
"""
Test to make sure that we correctly set the proxy info on Windows
"""
calls = [
call(
hive="HKEY_CURRENT_USER",
key="SOFTWARE\\Microsoft\\Windows\\CurrentVersion\\Internet Settings",
vname="ProxyServer",
vdata="http=192.168.0.1:3128;https=192.168.0.1:3128;",
),
call(
hive="HKEY_CURRENT_USER",
key="SOFTWARE\\Microsoft\\Windows\\CurrentVersion\\Internet Settings",
vname="ProxyEnable",
vdata=1,
vtype="REG_DWORD",
),
call(
hive="HKEY_CURRENT_USER",
key="SOFTWARE\\Microsoft\\Windows\\CurrentVersion\\Internet Settings",
vname="ProxyOverride",
vdata="<local>;.moo.com;.salt.com",
),
]
mock_reg = MagicMock()
mock_cmd = MagicMock()
with patch.dict(proxy.__grains__, {"os": "Windows"}), patch.dict(
proxy.__utils__, {"reg.set_value": mock_reg}
), patch.dict(proxy.__salt__, {"cmd.run": mock_cmd}):
out = proxy.set_proxy_win(
server="192.168.0.1",
port=3128,
types=["http", "https"],
bypass_hosts=[".moo.com", ".salt.com"],
)
mock_reg.assert_has_calls(calls)
mock_cmd.assert_called_once_with("netsh winhttp import proxy source=ie")
assert out

102
tests/pytests/unit/modules/test_publish.py Normal file
View file

@ -0,0 +1,102 @@
"""
:codeauthor: Jayesh Kariya <jayeshk@saltstack.com>
"""
import pytest
import salt.modules.publish as publish
from salt.exceptions import SaltReqTimeoutError
from tests.support.mock import MagicMock, patch
class SAuth:
"""
Mock SAuth class
"""
def __init__(self, __opts__):
self.tok = None
def gen_token(self, tok):
"""
Mock gen_token method
"""
self.tok = tok
return "salt_tok"
class Channel:
"""
Mock Channel class
"""
flag = None
def __init__(self):
self.tok = None
self.load = None
def factory(self, tok):
"""
Mock factory method
"""
self.tok = tok
return Channel()
def send(self, load):
"""
Mock send method
"""
self.load = load
if self.flag == 1:
raise SaltReqTimeoutError
return True
def close(self):
pass
def __enter__(self):
return self
def __exit__(self, *args):
pass
@pytest.fixture
def configure_loader_modules():
return {publish: {}}
def test_publish():
"""
Test if it publish a command from the minion out to other minions.
"""
assert publish.publish("os:Fedora", "publish.salt") == {}
def test_full_data():
"""
Test if it return the full data about the publication
"""
assert publish.publish("*", "publish.salt") == {}
def test_runner():
"""
Test if it execute a runner on the master and return the data
from the runner function
"""
ret = "No access to master. If using salt-call with --local, please remove."
assert publish.runner("manage.down") == ret
mock = MagicMock(return_value=True)
mock_id = MagicMock(return_value="salt_id")
with patch("salt.crypt.SAuth", return_value=SAuth(publish.__opts__)):
with patch("salt.transport.client.ReqChannel", Channel()):
with patch.dict(publish.__opts__, {"master_uri": mock, "id": mock_id}):
Channel.flag = 0
assert publish.runner("manage.down")
Channel.flag = 1
assert (
publish.runner("manage.down")
== "'manage.down' runner publish timed out"
)

175
tests/unit/modules/test_pkgin.py
View file

@ -1,175 +0,0 @@
import os
import salt.modules.pkgin as pkgin
from tests.support.mixins import LoaderModuleMockMixin
from tests.support.mock import MagicMock, patch
from tests.support.unit import TestCase
class PkginTestCase(TestCase, LoaderModuleMockMixin):
"""
Test cases for salt.modules.pkgin
"""
def setup_loader_modules(self):
return {pkgin: {"__opts__": {"cachedir": "/tmp"}}}
def test_search(self):
"""
Test searching for a package
"""
# Test searching for an available and uninstalled package
pkgin_out = [
"somepkg-1.0 Some package description here",
"",
"=: package is installed and up-to-date",
"<: package is installed but newer version is available",
">: installed package has a greater version than available package",
]
pkgin__get_version_mock = MagicMock(return_value=["0", "9", "0"])
pkgin__check_pkgin_mock = MagicMock(return_value="/opt/pkg/bin/pkgin")
pkgin_search_cmd = MagicMock(return_value=os.linesep.join(pkgin_out))
with patch("salt.modules.pkgin._get_version", pkgin__get_version_mock), patch(
"salt.modules.pkgin._check_pkgin", pkgin__check_pkgin_mock
), patch.dict(pkgin.__salt__, {"cmd.run": pkgin_search_cmd}):
self.assertDictEqual(pkgin.search("somepkg"), {"somepkg": "1.0"})
# Test searching for an available and installed package
pkgin_out = [
"somepkg-1.0 = Some package description here",
"",
"=: package is installed and up-to-date",
"<: package is installed but newer version is available",
">: installed package has a greater version than available package",
]
pkgin_search_cmd = MagicMock(return_value=os.linesep.join(pkgin_out))
with patch("salt.modules.pkgin._get_version", pkgin__get_version_mock), patch(
"salt.modules.pkgin._check_pkgin", pkgin__check_pkgin_mock
), patch.dict(pkgin.__salt__, {"cmd.run": pkgin_search_cmd}):
self.assertDictEqual(pkgin.search("somepkg"), {"somepkg": "1.0"})
def test_latest_version(self):
"""
Test getting the latest version of a package
"""
# Test getting the latest version of an uninstalled package
pkgin_out = [
"somepkg-1.0;;Some package description here",
"",
"=: package is installed and up-to-date",
"<: package is installed but newer version is available",
">: installed package has a greater version than available package",
]
pkgin__get_version_mock = MagicMock(return_value=["0", "9", "0"])
pkgin__check_pkgin_mock = MagicMock(return_value="/opt/pkg/bin/pkgin")
pkgin_refresh_db_mock = MagicMock(return_value=True)
pkgin_search_cmd = MagicMock(return_value=os.linesep.join(pkgin_out))
with patch("salt.modules.pkgin.refresh_db", pkgin_refresh_db_mock), patch(
"salt.modules.pkgin._get_version", pkgin__get_version_mock
), patch(
"salt.modules.pkgin._check_pkgin", pkgin__check_pkgin_mock
), patch.dict(
pkgin.__salt__, {"cmd.run": pkgin_search_cmd}
):
self.assertEqual(pkgin.latest_version("somepkg"), "1.0")
# Test getting the latest version of an installed package
pkgin_out = [
"somepkg-1.1;<;Some package description here",
"",
"=: package is installed and up-to-date",
"<: package is installed but newer version is available",
">: installed package has a greater version than available package",
]
pkgin_refresh_db_mock = MagicMock(return_value=True)
pkgin_search_cmd = MagicMock(return_value=os.linesep.join(pkgin_out))
with patch("salt.modules.pkgin.refresh_db", pkgin_refresh_db_mock), patch(
"salt.modules.pkgin._get_version", pkgin__get_version_mock
), patch(
"salt.modules.pkgin._check_pkgin", pkgin__check_pkgin_mock
), patch.dict(
pkgin.__salt__, {"cmd.run": pkgin_search_cmd}
):
self.assertEqual(pkgin.latest_version("somepkg"), "1.1")
# Test getting the latest version of a package that is already installed
# and is already at the latest version
pkgin_out = [
"somepkg-1.2;=;Some package description here",
"",
"=: package is installed and up-to-date",
"<: package is installed but newer version is available",
">: installed package has a greater version than available package",
]
pkgin_refresh_db_mock = MagicMock(return_value=True)
pkgin_search_cmd = MagicMock(return_value=os.linesep.join(pkgin_out))
with patch("salt.modules.pkgin.refresh_db", pkgin_refresh_db_mock), patch(
"salt.modules.pkgin._get_version", pkgin__get_version_mock
), patch(
"salt.modules.pkgin._check_pkgin", pkgin__check_pkgin_mock
), patch.dict(
pkgin.__salt__, {"cmd.run": pkgin_search_cmd}
):
self.assertEqual(pkgin.latest_version("somepkg"), "1.2")
# Test getting the latest version of a bogus package
pkgin_out = "No results found for ^boguspkg$"
pkgin_refresh_db_mock = MagicMock(return_value=True)
pkgin_search_cmd = MagicMock(return_value=pkgin_out)
with patch("salt.modules.pkgin.refresh_db", pkgin_refresh_db_mock), patch(
"salt.modules.pkgin._get_version", pkgin__get_version_mock
), patch(
"salt.modules.pkgin._check_pkgin", pkgin__check_pkgin_mock
), patch.dict(
pkgin.__salt__, {"cmd.run": pkgin_search_cmd}
):
self.assertEqual(pkgin.latest_version("boguspkg"), {})
def test_file_dict(self):
"""
Test that file_dict doesn't crash
"""
pkg_info_stdout = [
"/opt/pkg/bin/pkgin",
"/opt/pkg/man/man1/pkgin.1",
"/opt/pkg/share/examples/pkgin/preferred.conf.example",
"/opt/pkg/share/examples/pkgin/repositories.conf.example",
]
pkg_info_out = {
"pid": 1234,
"retcode": 0,
"stderr": "",
"stdout": os.linesep.join(pkg_info_stdout),
}
pkg_info_cmd = MagicMock(return_value=pkg_info_out)
with patch.dict(pkgin.__salt__, {"cmd.run_all": pkg_info_cmd}):
self.assertDictEqual(
pkgin.file_dict("pkgin"),
{
"files": {
"pkgin": [
"/opt/pkg/bin/pkgin",
"/opt/pkg/man/man1/pkgin.1",
"/opt/pkg/share/examples/pkgin/preferred.conf.example",
"/opt/pkg/share/examples/pkgin/repositories.conf.example",
]
}
},
)

81
tests/unit/modules/test_powerpath.py
View file

@ -1,81 +0,0 @@
"""
:codeauthor: Jayesh Kariya <jayeshk@saltstack.com>
"""
import salt.modules.powerpath as powerpath
from tests.support.mixins import LoaderModuleMockMixin
from tests.support.mock import MagicMock, patch
from tests.support.unit import TestCase
class PowerpathTestCase(TestCase, LoaderModuleMockMixin):
"""
Test cases for salt.modules.powerpath
"""
def setup_loader_modules(self):
return {powerpath: {}}
def test_has_powerpath(self):
"""
Test for powerpath
"""
with patch("os.path.exists") as mock_exists:
mock_exists.return_value = True
self.assertTrue(powerpath.has_powerpath())
mock_exists.return_value = False
self.assertFalse(powerpath.has_powerpath())
def test_list_licenses(self):
"""
Test to returns a list of applied powerpath license keys
"""
with patch.dict(
powerpath.__salt__, {"cmd.run": MagicMock(return_value="A\nB")}
):
self.assertListEqual(powerpath.list_licenses(), [])
def test_add_license(self):
"""
Test to add a license
"""
with patch.object(powerpath, "has_powerpath", return_value=False):
self.assertDictEqual(
powerpath.add_license("key"),
{
"output": "PowerPath is not installed",
"result": False,
"retcode": -1,
},
)
mock = MagicMock(return_value={"retcode": 1, "stderr": "stderr"})
with patch.object(powerpath, "has_powerpath", return_value=True):
with patch.dict(powerpath.__salt__, {"cmd.run_all": mock}):
self.assertDictEqual(
powerpath.add_license("key"),
{"output": "stderr", "result": False, "retcode": 1},
)
def test_remove_license(self):
"""
Test to remove a license
"""
with patch.object(powerpath, "has_powerpath", return_value=False):
self.assertDictEqual(
powerpath.remove_license("key"),
{
"output": "PowerPath is not installed",
"result": False,
"retcode": -1,
},
)
mock = MagicMock(return_value={"retcode": 1, "stderr": "stderr"})
with patch.object(powerpath, "has_powerpath", return_value=True):
with patch.dict(powerpath.__salt__, {"cmd.run_all": mock}):
self.assertDictEqual(
powerpath.remove_license("key"),
{"output": "stderr", "result": False, "retcode": 1},
)

486
tests/unit/modules/test_proxy.py
View file

@ -1,486 +0,0 @@
import salt.modules.proxy as proxy
from tests.support.mixins import LoaderModuleMockMixin
from tests.support.mock import MagicMock, call, patch
from tests.support.unit import TestCase
class ProxyTestCase(TestCase, LoaderModuleMockMixin):
"""
Test cases for salt.modules.proxy
"""
def setup_loader_modules(self):
return {proxy: {"__grains__": {"os": "Darwin"}}}
def test_get_http_proxy_macos(self):
"""
Test to make sure that we correctly get the current proxy info
on macOS
"""
mock = MagicMock(
return_value=(
"Enabled: Yes\nServer: 192.168.0.1\nPort: 3128\nAuthenticated Proxy"
" Enabled: 0"
)
)
expected = {"enabled": True, "server": "192.168.0.1", "port": "3128"}
with patch.dict(proxy.__salt__, {"cmd.run": mock}):
out = proxy.get_http_proxy()
mock.assert_called_once_with("networksetup -getwebproxy Ethernet")
self.assertEqual(expected, out)
def test_get_https_proxy_macos(self):
"""
Test to make sure that we correctly get the current proxy info
on macOS
"""
mock = MagicMock(
return_value=(
"Enabled: Yes\nServer: 192.168.0.1\nPort: 3128\nAuthenticated Proxy"
" Enabled: 0"
)
)
expected = {"enabled": True, "server": "192.168.0.1", "port": "3128"}
with patch.dict(proxy.__salt__, {"cmd.run": mock}):
out = proxy.get_https_proxy()
mock.assert_called_once_with("networksetup -getsecurewebproxy Ethernet")
self.assertEqual(expected, out)
def test_get_ftp_proxy_macos(self):
"""
Test to make sure that we correctly get the current proxy info
on macOS
"""
mock = MagicMock(
return_value=(
"Enabled: Yes\nServer: 192.168.0.1\nPort: 3128\nAuthenticated Proxy"
" Enabled: 0"
)
)
expected = {"enabled": True, "server": "192.168.0.1", "port": "3128"}
with patch.dict(proxy.__salt__, {"cmd.run": mock}):
out = proxy.get_ftp_proxy()
mock.assert_called_once_with("networksetup -getftpproxy Ethernet")
self.assertEqual(expected, out)
def test_get_http_proxy_macos_none(self):
"""
Test to make sure that we correctly return when there's no proxy set
"""
mock = MagicMock(
return_value="Enabled: No\nServer:\nPort: 0\nAuthenticated Proxy Enabled: 0"
)
with patch.dict(proxy.__salt__, {"cmd.run": mock}):
out = proxy.get_http_proxy()
mock.assert_called_once_with("networksetup -getwebproxy Ethernet")
self.assertEqual({}, out)
def test_set_http_proxy_macos(self):
"""
Test to make sure that we correctly set the proxy info
on macOS
"""
mock = MagicMock()
with patch.dict(proxy.__salt__, {"cmd.run": mock}):
out = proxy.set_http_proxy(
"192.168.0.1",
3128,
"frank",
"badpassw0rd",
bypass_hosts=".moo.com,.salt.com",
)
mock.assert_called_once_with(
"networksetup -setwebproxy Ethernet 192.168.0.1 3128 On frank"
" badpassw0rd"
)
self.assertTrue(out)
def test_set_https_proxy_macos(self):
"""
Test to make sure that we correctly set the proxy info
on macOS
"""
mock = MagicMock()
with patch.dict(proxy.__salt__, {"cmd.run": mock}):
out = proxy.set_https_proxy(
"192.168.0.1",
3128,
"frank",
"passw0rd",
bypass_hosts=".moo.com,.salt.com",
)
mock.assert_called_once_with(
"networksetup -setsecurewebproxy Ethernet 192.168.0.1 3128 On frank"
" passw0rd"
)
self.assertTrue(out)
def test_set_ftp_proxy_macos(self):
"""
Test to make sure that we correctly set the proxy info
on macOS
"""
mock = MagicMock()
with patch.dict(proxy.__salt__, {"cmd.run": mock}):
out = proxy.set_ftp_proxy(
"192.168.0.1",
3128,
"frank",
"badpassw0rd",
bypass_hosts=".moo.com,.salt.com",
)
mock.assert_called_once_with(
"networksetup -setftpproxy Ethernet 192.168.0.1 3128 On frank"
" badpassw0rd"
)
self.assertTrue(out)
def test_get_proxy_win(self):
"""
Test to make sure that we correctly get the current proxy info on
Windows
"""
result = [
{
"vdata": (
"http=192.168.0.1:3128;https=192.168.0.2:3128;ftp=192.168.0.3:3128"
)
},
{"vdata": 1},
]
mock_reg_read = MagicMock(side_effect=result)
expected = {
"enabled": True,
"http": {"server": "192.168.0.1", "port": "3128"},
"https": {"server": "192.168.0.2", "port": "3128"},
"ftp": {"server": "192.168.0.3", "port": "3128"},
}
with patch.dict(proxy.__grains__, {"os": "Windows"}), patch.dict(
proxy.__utils__, {"reg.read_value": mock_reg_read}
):
out = proxy.get_proxy_win()
self.assertDictEqual(out, expected)
mock_reg_read.assert_any_call(
hive="HKEY_CURRENT_USER",
key="SOFTWARE\\Microsoft\\Windows\\CurrentVersion\\Internet Settings",
vname="ProxyServer",
)
mock_reg_read.assert_any_call(
hive="HKEY_CURRENT_USER",
key="SOFTWARE\\Microsoft\\Windows\\CurrentVersion\\Internet Settings",
vname="ProxyEnable",
)
def test_get_http_proxy_windows(self):
"""
Test to make sure that we correctly get the current proxy info on
Windows
"""
result = {
"vdata": "http=192.168.0.1:3128;https=192.168.0.2:3128;ftp=192.168.0.3:3128"
}
mock = MagicMock(return_value=result)
expected = {"server": "192.168.0.1", "port": "3128"}
with patch.dict(proxy.__grains__, {"os": "Windows"}), patch.dict(
proxy.__utils__, {"reg.read_value": mock}
):
out = proxy.get_http_proxy()
mock.assert_called_once_with(
hive="HKEY_CURRENT_USER",
key="SOFTWARE\\Microsoft\\Windows\\CurrentVersion\\Internet Settings",
vname="ProxyServer",
)
self.assertEqual(expected, out)
def test_get_https_proxy_windows(self):
"""
Test to make sure that we correctly get the current proxy info on
Windows
"""
result = {
"vdata": "http=192.168.0.1:3128;https=192.168.0.2:3128;ftp=192.168.0.3:3128"
}
mock = MagicMock(return_value=result)
expected = {"server": "192.168.0.2", "port": "3128"}
with patch.dict(proxy.__grains__, {"os": "Windows"}), patch.dict(
proxy.__utils__, {"reg.read_value": mock}
):
out = proxy.get_https_proxy()
mock.assert_called_once_with(
hive="HKEY_CURRENT_USER",
key="SOFTWARE\\Microsoft\\Windows\\CurrentVersion\\Internet Settings",
vname="ProxyServer",
)
self.assertEqual(expected, out)
def test_get_ftp_proxy_windows(self):
"""
Test to make sure that we correctly get the current proxy info on
Windows
"""
result = {
"vdata": "http=192.168.0.1:3128;https=192.168.0.2:3128;ftp=192.168.0.3:3128"
}
mock = MagicMock(return_value=result)
expected = {"server": "192.168.0.3", "port": "3128"}
with patch.dict(proxy.__grains__, {"os": "Windows"}), patch.dict(
proxy.__utils__, {"reg.read_value": mock}
):
out = proxy.get_ftp_proxy()
mock.assert_called_once_with(
hive="HKEY_CURRENT_USER",
key="SOFTWARE\\Microsoft\\Windows\\CurrentVersion\\Internet Settings",
vname="ProxyServer",
)
self.assertEqual(expected, out)
def test_get_all_proxies_macos_fails(self):
mock = MagicMock()
with patch.dict(proxy.__utils__, {"reg.read_value": mock}):
out = proxy.get_proxy_win()
assert not mock.called
self.assertEqual(out, None)
def test_get_all_proxies_windows(self):
"""
Test to make sure that we correctly get the current proxy info on
Windows
"""
results = [
{
"vdata": (
"http=192.168.0.1:3128;https=192.168.0.2:3128;ftp=192.168.0.3:3128"
)
},
{"vdata": 1},
]
mock = MagicMock(side_effect=results)
expected = {
"enabled": True,
"http": {"server": "192.168.0.1", "port": "3128"},
"https": {"server": "192.168.0.2", "port": "3128"},
"ftp": {"server": "192.168.0.3", "port": "3128"},
}
calls = [
call(
hive="HKEY_CURRENT_USER",
key="SOFTWARE\\Microsoft\\Windows\\CurrentVersion\\Internet Settings",
vname="ProxyServer",
),
call(
hive="HKEY_CURRENT_USER",
key="SOFTWARE\\Microsoft\\Windows\\CurrentVersion\\Internet Settings",
vname="ProxyEnable",
),
]
with patch.dict(proxy.__grains__, {"os": "Windows"}), patch.dict(
proxy.__utils__, {"reg.read_value": mock}
):
out = proxy.get_proxy_win()
mock.assert_has_calls(calls)
self.assertEqual(expected, out)
def test_set_http_proxy_windows(self):
"""
Test to make sure that we correctly set the proxy info on Windows
"""
calls = [
call(
hive="HKEY_CURRENT_USER",
key="SOFTWARE\\Microsoft\\Windows\\CurrentVersion\\Internet Settings",
vname="ProxyServer",
vdata="http=192.168.0.1:3128;",
),
call(
hive="HKEY_CURRENT_USER",
key="SOFTWARE\\Microsoft\\Windows\\CurrentVersion\\Internet Settings",
vname="ProxyEnable",
vdata=1,
vtype="REG_DWORD",
),
call(
hive="HKEY_CURRENT_USER",
key="SOFTWARE\\Microsoft\\Windows\\CurrentVersion\\Internet Settings",
vname="ProxyOverride",
vdata="<local>;.moo.com;.salt.com",
),
]
mock_reg = MagicMock()
mock_cmd = MagicMock()
with patch.dict(proxy.__grains__, {"os": "Windows"}), patch.dict(
proxy.__utils__, {"reg.set_value": mock_reg}
), patch.dict(proxy.__salt__, {"cmd.run": mock_cmd}):
out = proxy.set_http_proxy(
server="192.168.0.1",
port=3128,
bypass_hosts=[".moo.com", ".salt.com"],
)
mock_reg.assert_has_calls(calls)
mock_cmd.assert_called_once_with("netsh winhttp import proxy source=ie")
self.assertTrue(out)
def test_set_https_proxy_windows(self):
"""
Test to make sure that we correctly set the proxy info on Windows
"""
calls = [
call(
hive="HKEY_CURRENT_USER",
key="SOFTWARE\\Microsoft\\Windows\\CurrentVersion\\Internet Settings",
vname="ProxyServer",
vdata="https=192.168.0.1:3128;",
),
call(
hive="HKEY_CURRENT_USER",
key="SOFTWARE\\Microsoft\\Windows\\CurrentVersion\\Internet Settings",
vname="ProxyEnable",
vdata=1,
vtype="REG_DWORD",
),
call(
hive="HKEY_CURRENT_USER",
key="SOFTWARE\\Microsoft\\Windows\\CurrentVersion\\Internet Settings",
vname="ProxyOverride",
vdata="<local>;.moo.com;.salt.com",
),
]
mock_reg = MagicMock()
mock_cmd = MagicMock()
with patch.dict(proxy.__grains__, {"os": "Windows"}), patch.dict(
proxy.__utils__, {"reg.set_value": mock_reg}
), patch.dict(proxy.__salt__, {"cmd.run": mock_cmd}):
out = proxy.set_https_proxy(
server="192.168.0.1",
port=3128,
bypass_hosts=[".moo.com", ".salt.com"],
)
mock_reg.assert_has_calls(calls)
mock_cmd.assert_called_once_with("netsh winhttp import proxy source=ie")
self.assertTrue(out)
def test_set_ftp_proxy_windows(self):
"""
Test to make sure that we correctly set the proxy info on Windows
"""
calls = [
call(
hive="HKEY_CURRENT_USER",
key="SOFTWARE\\Microsoft\\Windows\\CurrentVersion\\Internet Settings",
vname="ProxyServer",
vdata="ftp=192.168.0.1:3128;",
),
call(
hive="HKEY_CURRENT_USER",
key="SOFTWARE\\Microsoft\\Windows\\CurrentVersion\\Internet Settings",
vname="ProxyEnable",
vdata=1,
vtype="REG_DWORD",
),
call(
hive="HKEY_CURRENT_USER",
key="SOFTWARE\\Microsoft\\Windows\\CurrentVersion\\Internet Settings",
vname="ProxyOverride",
vdata="<local>;.moo.com;.salt.com",
),
]
mock_reg = MagicMock()
mock_cmd = MagicMock()
with patch.dict(proxy.__grains__, {"os": "Windows"}), patch.dict(
proxy.__utils__, {"reg.set_value": mock_reg}
), patch.dict(proxy.__salt__, {"cmd.run": mock_cmd}):
out = proxy.set_ftp_proxy(
server="192.168.0.1",
port=3128,
bypass_hosts=[".moo.com", ".salt.com"],
)
mock_reg.assert_has_calls(calls)
mock_cmd.assert_called_once_with("netsh winhttp import proxy source=ie")
self.assertTrue(out)
def test_set_proxy_windows(self):
"""
Test to make sure that we correctly set the proxy info on Windows
"""
calls = [
call(
hive="HKEY_CURRENT_USER",
key="SOFTWARE\\Microsoft\\Windows\\CurrentVersion\\Internet Settings",
vname="ProxyServer",
vdata=(
"http=192.168.0.1:3128;https=192.168.0.1:3128;ftp=192.168.0.1:3128;"
),
),
call(
hive="HKEY_CURRENT_USER",
key="SOFTWARE\\Microsoft\\Windows\\CurrentVersion\\Internet Settings",
vname="ProxyEnable",
vdata=1,
vtype="REG_DWORD",
),
call(
hive="HKEY_CURRENT_USER",
key="SOFTWARE\\Microsoft\\Windows\\CurrentVersion\\Internet Settings",
vname="ProxyOverride",
vdata="<local>;.moo.com;.salt.com",
),
]
mock_reg = MagicMock()
mock_cmd = MagicMock()
with patch.dict(proxy.__grains__, {"os": "Windows"}), patch.dict(
proxy.__utils__, {"reg.set_value": mock_reg}
), patch.dict(proxy.__salt__, {"cmd.run": mock_cmd}):
out = proxy.set_proxy_win(
server="192.168.0.1",
port=3128,
bypass_hosts=[".moo.com", ".salt.com"],
)
mock_reg.assert_has_calls(calls)
mock_cmd.assert_called_once_with("netsh winhttp import proxy source=ie")
self.assertTrue(out)
def test_set_proxy_windows_no_ftp(self):
"""
Test to make sure that we correctly set the proxy info on Windows
"""
calls = [
call(
hive="HKEY_CURRENT_USER",
key="SOFTWARE\\Microsoft\\Windows\\CurrentVersion\\Internet Settings",
vname="ProxyServer",
vdata="http=192.168.0.1:3128;https=192.168.0.1:3128;",
),
call(
hive="HKEY_CURRENT_USER",
key="SOFTWARE\\Microsoft\\Windows\\CurrentVersion\\Internet Settings",
vname="ProxyEnable",
vdata=1,
vtype="REG_DWORD",
),
call(
hive="HKEY_CURRENT_USER",
key="SOFTWARE\\Microsoft\\Windows\\CurrentVersion\\Internet Settings",
vname="ProxyOverride",
vdata="<local>;.moo.com;.salt.com",
),
]
mock_reg = MagicMock()
mock_cmd = MagicMock()
with patch.dict(proxy.__grains__, {"os": "Windows"}), patch.dict(
proxy.__utils__, {"reg.set_value": mock_reg}
), patch.dict(proxy.__salt__, {"cmd.run": mock_cmd}):
out = proxy.set_proxy_win(
server="192.168.0.1",
port=3128,
types=["http", "https"],
bypass_hosts=[".moo.com", ".salt.com"],
)
mock_reg.assert_has_calls(calls)
mock_cmd.assert_called_once_with("netsh winhttp import proxy source=ie")
self.assertTrue(out)

124
tests/unit/modules/test_publish.py
View file

@ -1,124 +0,0 @@
"""
:codeauthor: Jayesh Kariya <jayeshk@saltstack.com>
"""
import salt.modules.publish as publish
from salt.exceptions import SaltReqTimeoutError
from tests.support.mixins import LoaderModuleMockMixin
from tests.support.mock import MagicMock, patch
from tests.support.unit import TestCase
class SAuth:
"""
Mock SAuth class
"""
def __init__(self, __opts__):
self.tok = None
def gen_token(self, tok):
"""
Mock gen_token method
"""
self.tok = tok
return "salt_tok"
class Channel:
"""
Mock Channel class
"""
flag = None
def __init__(self):
self.tok = None
self.load = None
def factory(self, tok):
"""
Mock factory method
"""
self.tok = tok
return Channel()
def send(self, load):
"""
Mock send method
"""
self.load = load
if self.flag == 1:
raise SaltReqTimeoutError
return True
def close(self):
pass
def __enter__(self):
return self
def __exit__(self, *args):
pass
class PublishTestCase(TestCase, LoaderModuleMockMixin):
"""
Test cases for salt.modules.publish
"""
def setup_loader_modules(self):
return {publish: {}}
@classmethod
def setUpClass(cls):
cls.channel_patcher = patch("salt.transport.client.ReqChannel", Channel())
cls.channel_patcher.start()
@classmethod
def tearDownClass(cls):
cls.channel_patcher.stop()
del cls.channel_patcher
def setUp(self):
patcher = patch("salt.crypt.SAuth", return_value=SAuth(publish.__opts__))
patcher.start()
self.addCleanup(patcher.stop)
# 'publish' function tests: 1
def test_publish(self):
"""
Test if it publish a command from the minion out to other minions.
"""
self.assertDictEqual(publish.publish("os:Fedora", "publish.salt"), {})
# 'full_data' function tests: 1
def test_full_data(self):
"""
Test if it return the full data about the publication
"""
self.assertDictEqual(publish.publish("*", "publish.salt"), {})
# 'runner' function tests: 1
def test_runner(self):
"""
Test if it execute a runner on the master and return the data
from the runner function
"""
ret = "No access to master. If using salt-call with --local, please remove."
self.assertEqual(publish.runner("manage.down"), ret)
mock = MagicMock(return_value=True)
mock_id = MagicMock(return_value="salt_id")
with patch.dict(publish.__opts__, {"master_uri": mock, "id": mock_id}):
Channel.flag = 0
self.assertTrue(publish.runner("manage.down"))
Channel.flag = 1
self.assertEqual(
publish.runner("manage.down"), "'manage.down' runner publish timed out"
)