mirror of
https://github.com/saltstack/salt.git
synced 2025-04-17 10:10:20 +00:00
Move module tests to pytest
This commit is contained in:
krionbsd
Megan Wilhite
parent
da4bd33571
commit
1bdcbc19fb
8 changed files with 2512 additions and 2495 deletions
311
tests/pytests/unit/modules/test_pf.py
Normal file
311
tests/pytests/unit/modules/test_pf.py
Normal file
|
|
@ -0,0 +1,311 @@
|
|||
import pytest
|
||||
import salt.modules.pf as pf
|
||||
from tests.support.mock import MagicMock, patch
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def configure_loader_modules():
|
||||
return {pf: {}}
|
||||
|
||||
|
||||
def test_enable_when_disabled():
|
||||
"""
|
||||
Tests enabling pf when it's not enabled yet.
|
||||
"""
|
||||
ret = {}
|
||||
ret["stderr"] = "pf enabled"
|
||||
ret["retcode"] = 0
|
||||
mock_cmd = MagicMock(return_value=ret)
|
||||
with patch.dict(pf.__salt__, {"cmd.run_all": mock_cmd}):
|
||||
assert pf.enable()["changes"]
|
||||
|
||||
|
||||
def test_enable_when_enabled():
|
||||
"""
|
||||
Tests enabling pf when it already enabled.
|
||||
"""
|
||||
ret = {}
|
||||
ret["stderr"] = "pfctl: pf already enabled"
|
||||
ret["retcode"] = 1
|
||||
mock_cmd = MagicMock(return_value=ret)
|
||||
with patch.dict(pf.__salt__, {"cmd.run_all": mock_cmd}):
|
||||
assert not pf.enable()["changes"]
|
||||
|
||||
|
||||
def test_disable_when_enabled():
|
||||
"""
|
||||
Tests disabling pf when it's enabled.
|
||||
"""
|
||||
ret = {}
|
||||
ret["stderr"] = "pf disabled"
|
||||
ret["retcode"] = 0
|
||||
mock_cmd = MagicMock(return_value=ret)
|
||||
with patch.dict(pf.__salt__, {"cmd.run_all": mock_cmd}):
|
||||
assert pf.disable()["changes"]
|
||||
|
||||
|
||||
def test_disable_when_disabled():
|
||||
"""
|
||||
Tests disabling pf when it already disabled.
|
||||
"""
|
||||
ret = {}
|
||||
ret["stderr"] = "pfctl: pf not enabled"
|
||||
ret["retcode"] = 1
|
||||
mock_cmd = MagicMock(return_value=ret)
|
||||
with patch.dict(pf.__salt__, {"cmd.run_all": mock_cmd}):
|
||||
assert not pf.disable()["changes"]
|
||||
|
||||
|
||||
def test_loglevel_freebsd():
|
||||
"""
|
||||
Tests setting a loglevel.
|
||||
"""
|
||||
ret = {}
|
||||
ret["retcode"] = 0
|
||||
mock_cmd = MagicMock(return_value=ret)
|
||||
with patch.dict(pf.__salt__, {"cmd.run_all": mock_cmd}), patch.dict(
|
||||
pf.__grains__, {"os": "FreeBSD"}
|
||||
):
|
||||
res = pf.loglevel("urgent")
|
||||
mock_cmd.assert_called_once_with(
|
||||
"pfctl -x urgent", output_loglevel="trace", python_shell=False
|
||||
)
|
||||
assert res["changes"]
|
||||
|
||||
|
||||
def test_loglevel_openbsd():
|
||||
"""
|
||||
Tests setting a loglevel.
|
||||
"""
|
||||
ret = {}
|
||||
ret["retcode"] = 0
|
||||
mock_cmd = MagicMock(return_value=ret)
|
||||
with patch.dict(pf.__salt__, {"cmd.run_all": mock_cmd}), patch.dict(
|
||||
pf.__grains__, {"os": "OpenBSD"}
|
||||
):
|
||||
res = pf.loglevel("crit")
|
||||
mock_cmd.assert_called_once_with(
|
||||
"pfctl -x crit", output_loglevel="trace", python_shell=False
|
||||
)
|
||||
assert res["changes"]
|
||||
|
||||
|
||||
def test_load():
|
||||
"""
|
||||
Tests loading ruleset.
|
||||
"""
|
||||
ret = {}
|
||||
ret["retcode"] = 0
|
||||
mock_cmd = MagicMock(return_value=ret)
|
||||
with patch.dict(pf.__salt__, {"cmd.run_all": mock_cmd}):
|
||||
res = pf.load()
|
||||
mock_cmd.assert_called_once_with(
|
||||
["pfctl", "-f", "/etc/pf.conf"],
|
||||
output_loglevel="trace",
|
||||
python_shell=False,
|
||||
)
|
||||
assert res["changes"]
|
||||
|
||||
|
||||
def test_load_noop():
|
||||
"""
|
||||
Tests evaluating but not actually loading ruleset.
|
||||
"""
|
||||
ret = {}
|
||||
ret["retcode"] = 0
|
||||
mock_cmd = MagicMock(return_value=ret)
|
||||
with patch.dict(pf.__salt__, {"cmd.run_all": mock_cmd}):
|
||||
res = pf.load(noop=True)
|
||||
mock_cmd.assert_called_once_with(
|
||||
["pfctl", "-f", "/etc/pf.conf", "-n"],
|
||||
output_loglevel="trace",
|
||||
python_shell=False,
|
||||
)
|
||||
assert not res["changes"]
|
||||
|
||||
|
||||
def test_flush():
|
||||
"""
|
||||
Tests a regular flush command.
|
||||
"""
|
||||
ret = {}
|
||||
ret["stderr"] = "pf: statistics cleared"
|
||||
ret["retcode"] = 0
|
||||
mock_cmd = MagicMock(return_value=ret)
|
||||
with patch.dict(pf.__salt__, {"cmd.run_all": mock_cmd}):
|
||||
res = pf.flush("info")
|
||||
mock_cmd.assert_called_once_with(
|
||||
"pfctl -v -F info", output_loglevel="trace", python_shell=False
|
||||
)
|
||||
assert res["changes"]
|
||||
|
||||
|
||||
def test_flush_capital():
|
||||
"""
|
||||
Tests a flush command starting with a capital letter.
|
||||
"""
|
||||
ret = {}
|
||||
ret["stderr"] = "2 tables cleared"
|
||||
ret["retcode"] = 0
|
||||
mock_cmd = MagicMock(return_value=ret)
|
||||
with patch.dict(pf.__salt__, {"cmd.run_all": mock_cmd}):
|
||||
res = pf.flush("tables")
|
||||
mock_cmd.assert_called_once_with(
|
||||
"pfctl -v -F Tables", output_loglevel="trace", python_shell=False
|
||||
)
|
||||
assert res["changes"]
|
||||
|
||||
|
||||
def test_flush_without_changes():
|
||||
"""
|
||||
Tests a flush command that has no changes.
|
||||
"""
|
||||
ret = {}
|
||||
ret["stderr"] = "0 tables cleared"
|
||||
ret["retcode"] = 0
|
||||
mock_cmd = MagicMock(return_value=ret)
|
||||
with patch.dict(pf.__salt__, {"cmd.run_all": mock_cmd}):
|
||||
assert not pf.flush("tables")["changes"]
|
||||
|
||||
|
||||
def test_table():
|
||||
"""
|
||||
Tests a regular table command.
|
||||
"""
|
||||
ret = {}
|
||||
ret["stderr"] = "42 addresses deleted"
|
||||
ret["retcode"] = 0
|
||||
mock_cmd = MagicMock(return_value=ret)
|
||||
with patch.dict(pf.__salt__, {"cmd.run_all": mock_cmd}):
|
||||
assert pf.table("flush", table="bad_hosts")["changes"]
|
||||
|
||||
|
||||
def test_table_expire():
|
||||
"""
|
||||
Tests the table expire command.
|
||||
"""
|
||||
ret = {}
|
||||
ret["stderr"] = "1/1 addresses expired."
|
||||
ret["retcode"] = 0
|
||||
mock_cmd = MagicMock(return_value=ret)
|
||||
with patch.dict(pf.__salt__, {"cmd.run_all": mock_cmd}):
|
||||
assert pf.table("expire", table="bad_hosts", number=300)["changes"]
|
||||
|
||||
|
||||
def test_table_add_addresses():
|
||||
"""
|
||||
Tests adding addresses to a table.
|
||||
"""
|
||||
ret = {}
|
||||
ret["stderr"] = "2/2 addressess added."
|
||||
ret["retcode"] = 0
|
||||
mock_cmd = MagicMock(return_value=ret)
|
||||
with patch.dict(pf.__salt__, {"cmd.run_all": mock_cmd}):
|
||||
assert pf.table("add", table="bad_hosts", addresses=["1.2.3.4", "5.6.7.8"])[
|
||||
"changes"
|
||||
]
|
||||
|
||||
|
||||
def test_table_delete_addresses():
|
||||
"""
|
||||
Tests deleting addresses in a table.
|
||||
"""
|
||||
ret = {}
|
||||
ret["stderr"] = "2/2 addressess deleted."
|
||||
ret["retcode"] = 0
|
||||
mock_cmd = MagicMock(return_value=ret)
|
||||
with patch.dict(pf.__salt__, {"cmd.run_all": mock_cmd}):
|
||||
assert pf.table("delete", table="bad_hosts", addresses=["1.2.3.4", "5.6.7.8"])[
|
||||
"changes"
|
||||
]
|
||||
|
||||
|
||||
def test_table_test_address():
|
||||
"""
|
||||
Tests testing addresses in a table.
|
||||
"""
|
||||
ret = {}
|
||||
ret["stderr"] = "1/2 addressess match."
|
||||
ret["retcode"] = 0
|
||||
mock_cmd = MagicMock(return_value=ret)
|
||||
with patch.dict(pf.__salt__, {"cmd.run_all": mock_cmd}):
|
||||
assert pf.table("test", table="bad_hosts", addresses=["1.2.3.4"])["matches"]
|
||||
|
||||
|
||||
def test_table_no_changes():
|
||||
"""
|
||||
Tests a table command that has no changes.
|
||||
"""
|
||||
ret = {}
|
||||
ret["stderr"] = "0/1 addresses expired."
|
||||
ret["retcode"] = 0
|
||||
mock_cmd = MagicMock(return_value=ret)
|
||||
with patch.dict(pf.__salt__, {"cmd.run_all": mock_cmd}):
|
||||
assert not pf.table("expire", table="bad_hosts", number=300)["changes"]
|
||||
|
||||
|
||||
def test_table_show():
|
||||
"""
|
||||
Tests showing table contents.
|
||||
"""
|
||||
ret = {}
|
||||
ret["stdout"] = "1.2.3.4\n5.6.7.8"
|
||||
ret["retcode"] = 0
|
||||
expected = ["1.2.3.4", "5.6.7.8"]
|
||||
mock_cmd = MagicMock(return_value=ret)
|
||||
with patch.dict(pf.__salt__, {"cmd.run_all": mock_cmd}):
|
||||
assert pf.table("show", table="bad_hosts")["comment"] == expected
|
||||
|
||||
|
||||
def test_table_zero():
|
||||
"""
|
||||
Tests clearing all the statistics of a table.
|
||||
"""
|
||||
ret = {}
|
||||
ret["stderr"] = "42 addresses has been cleared"
|
||||
ret["retcode"] = 0
|
||||
mock_cmd = MagicMock(return_value=ret)
|
||||
with patch.dict(pf.__salt__, {"cmd.run_all": mock_cmd}):
|
||||
assert pf.table("zero", table="bad_hosts")["changes"]
|
||||
|
||||
|
||||
def test_show_rules():
|
||||
"""
|
||||
Tests show rules command.
|
||||
"""
|
||||
ret = {}
|
||||
ret["stdout"] = "block return\npass"
|
||||
ret["retcode"] = 0
|
||||
expected = ["block return", "pass"]
|
||||
mock_cmd = MagicMock(return_value=ret)
|
||||
with patch.dict(pf.__salt__, {"cmd.run_all": mock_cmd}):
|
||||
assert pf.show("rules")["comment"] == expected
|
||||
|
||||
|
||||
def test_show_states():
|
||||
"""
|
||||
Tests show states command.
|
||||
"""
|
||||
ret = {}
|
||||
ret["stdout"] = "all udp 192.168.1.1:3478\n"
|
||||
ret["retcode"] = 0
|
||||
expected = ["all udp 192.168.1.1:3478", ""]
|
||||
mock_cmd = MagicMock(return_value=ret)
|
||||
with patch.dict(pf.__salt__, {"cmd.run_all": mock_cmd}):
|
||||
assert pf.show("states")["comment"] == expected
|
||||
|
||||
|
||||
def test_show_tables():
|
||||
"""
|
||||
Tests show tables command.
|
||||
"""
|
||||
ret = {}
|
||||
ret["stdout"] = "bad_hosts"
|
||||
ret["retcode"] = 0
|
||||
mock_cmd = MagicMock(return_value=ret)
|
||||
with patch.dict(pf.__salt__, {"cmd.run_all": mock_cmd}):
|
||||
res = pf.show("tables")
|
||||
mock_cmd.assert_called_once_with(
|
||||
"pfctl -s Tables", output_loglevel="trace", python_shell=False
|
||||
)
|
||||
assert not res["changes"]
|
||||
154
tests/pytests/unit/modules/test_pillar.py
Normal file
154
tests/pytests/unit/modules/test_pillar.py
Normal file
|
|
@ -0,0 +1,154 @@
|
|||
import pytest
|
||||
import salt.modules.pillar as pillarmod
|
||||
from salt.utils.odict import OrderedDict
|
||||
from tests.support.mock import MagicMock, patch
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def pillar_value():
|
||||
pillar_value = dict(a=1, b="very secret")
|
||||
return pillar_value
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def configure_loader_modules():
|
||||
return {pillarmod: {}}
|
||||
|
||||
|
||||
def test_obfuscate_inner_recursion():
|
||||
assert pillarmod._obfuscate_inner(
|
||||
dict(a=[1, 2], b=dict(pwd="secret", deeper=("a", 1)))
|
||||
) == dict(a=["<int>", "<int>"], b=dict(pwd="<str>", deeper=("<str>", "<int>")))
|
||||
|
||||
|
||||
def test_obfuscate_inner_more_types():
|
||||
assert pillarmod._obfuscate_inner(OrderedDict([("key", "value")])) == OrderedDict(
|
||||
[("key", "<str>")]
|
||||
)
|
||||
|
||||
assert pillarmod._obfuscate_inner({1, 2}) == {"<int>"}
|
||||
|
||||
assert pillarmod._obfuscate_inner((1, 2)) == ("<int>", "<int>")
|
||||
|
||||
|
||||
def test_obfuscate(pillar_value):
|
||||
with patch("salt.modules.pillar.items", MagicMock(return_value=pillar_value)):
|
||||
assert pillarmod.obfuscate() == dict(a="<int>", b="<str>")
|
||||
|
||||
|
||||
def test_ls(pillar_value):
|
||||
with patch("salt.modules.pillar.items", MagicMock(return_value=pillar_value)):
|
||||
ls = sorted(pillarmod.ls())
|
||||
assert ls == ["a", "b"]
|
||||
|
||||
|
||||
def test_pillar_get_default_merge():
|
||||
defaults = {
|
||||
"int": 1,
|
||||
"string": "foo",
|
||||
"list": ["foo"],
|
||||
"dict": {"foo": "bar", "subkey": {"foo": "bar"}},
|
||||
}
|
||||
|
||||
pillar_mock = {
|
||||
"int": 2,
|
||||
"string": "bar",
|
||||
"list": ["bar", "baz"],
|
||||
"dict": {"baz": "qux", "subkey": {"baz": "qux"}},
|
||||
}
|
||||
|
||||
# Test that we raise a KeyError when pillar_raise_on_missing is True
|
||||
with patch.dict(pillarmod.__opts__, {"pillar_raise_on_missing": True}):
|
||||
pytest.raises(KeyError, pillarmod.get, "missing")
|
||||
# Test that we return an empty string when it is not
|
||||
with patch.dict(pillarmod.__opts__, {}):
|
||||
assert pillarmod.get("missing") == ""
|
||||
|
||||
with patch.dict(pillarmod.__pillar__, pillar_mock):
|
||||
# Test with no default passed (it should be KeyError) and merge=True.
|
||||
# The merge should be skipped and the value returned from __pillar__
|
||||
# should be returned.
|
||||
for item in pillarmod.__pillar__:
|
||||
assert pillarmod.get(item, merge=True) == pillarmod.__pillar__[item]
|
||||
|
||||
# Test merging when the type of the default value is not the same as
|
||||
# what was returned. Merging should be skipped and the value returned
|
||||
# from __pillar__ should be returned.
|
||||
for default_type in defaults:
|
||||
for data_type in ("dict", "list"):
|
||||
if default_type == data_type:
|
||||
continue
|
||||
assert (
|
||||
pillarmod.get(data_type, default=defaults[default_type], merge=True)
|
||||
== pillarmod.__pillar__[data_type]
|
||||
)
|
||||
|
||||
# Test recursive dict merging
|
||||
assert pillarmod.get("dict", default=defaults["dict"], merge=True) == {
|
||||
"foo": "bar",
|
||||
"baz": "qux",
|
||||
"subkey": {"foo": "bar", "baz": "qux"},
|
||||
}
|
||||
|
||||
# Test list merging
|
||||
assert pillarmod.get("list", default=defaults["list"], merge=True) == [
|
||||
"foo",
|
||||
"bar",
|
||||
"baz",
|
||||
]
|
||||
|
||||
|
||||
def test_pillar_get_default_merge_regression_38558():
|
||||
"""Test for pillar.get(key=..., default=..., merge=True)
|
||||
Do not update the ``default`` value when using ``merge=True``.
|
||||
See: https://github.com/saltstack/salt/issues/38558
|
||||
"""
|
||||
with patch.dict(pillarmod.__pillar__, {"l1": {"l2": {"l3": 42}}}):
|
||||
|
||||
res = pillarmod.get(key="l1")
|
||||
assert {"l2": {"l3": 42}} == res
|
||||
|
||||
default = {"l2": {"l3": 43}}
|
||||
|
||||
res = pillarmod.get(key="l1", default=default)
|
||||
assert {"l2": {"l3": 42}} == res
|
||||
assert {"l2": {"l3": 43}} == default
|
||||
|
||||
res = pillarmod.get(key="l1", default=default, merge=True)
|
||||
assert {"l2": {"l3": 42}} == res
|
||||
assert {"l2": {"l3": 43}} == default
|
||||
|
||||
|
||||
def test_pillar_get_default_merge_regression_39062():
|
||||
"""
|
||||
Confirm that we do not raise an exception if default is None and
|
||||
merge=True.
|
||||
See https://github.com/saltstack/salt/issues/39062 for more info.
|
||||
"""
|
||||
with patch.dict(pillarmod.__pillar__, {"foo": "bar"}):
|
||||
|
||||
assert pillarmod.get(key="foo", default=None, merge=True) == "bar"
|
||||
|
||||
|
||||
def test_pillar_get_int_key():
|
||||
"""
|
||||
Confirm that we can access pillar keys that are integers
|
||||
"""
|
||||
with patch.dict(pillarmod.__pillar__, {12345: "luggage_code"}):
|
||||
|
||||
assert pillarmod.get(key=12345, default=None, merge=True) == "luggage_code"
|
||||
|
||||
with patch.dict(pillarmod.__pillar__, {12345: {"l2": {"l3": "my_luggage_code"}}}):
|
||||
|
||||
res = pillarmod.get(key=12345)
|
||||
assert {"l2": {"l3": "my_luggage_code"}} == res
|
||||
|
||||
default = {"l2": {"l3": "your_luggage_code"}}
|
||||
|
||||
res = pillarmod.get(key=12345, default=default)
|
||||
assert {"l2": {"l3": "my_luggage_code"}} == res
|
||||
assert {"l2": {"l3": "your_luggage_code"}} == default
|
||||
|
||||
res = pillarmod.get(key=12345, default=default, merge=True)
|
||||
assert {"l2": {"l3": "my_luggage_code"}} == res
|
||||
assert {"l2": {"l3": "your_luggage_code"}} == default
|
||||
1775
tests/pytests/unit/modules/test_pip.py
Normal file
1775
tests/pytests/unit/modules/test_pip.py
Normal file
File diff suppressed because it is too large
Load diff
272
tests/pytests/unit/modules/test_pkg_resource.py
Normal file
272
tests/pytests/unit/modules/test_pkg_resource.py
Normal file
|
|
@ -0,0 +1,272 @@
|
|||
"""
|
||||
:codeauthor: Rahul Handay <rahulha@saltstack.com>
|
||||
"""
|
||||
|
||||
import pytest
|
||||
import salt.modules.pkg_resource as pkg_resource
|
||||
import salt.utils.data
|
||||
import salt.utils.yaml
|
||||
import yaml
|
||||
from tests.support.mock import MagicMock, patch
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def configure_loader_modules():
|
||||
return {pkg_resource: {}}
|
||||
|
||||
|
||||
def test_pack_sources():
|
||||
"""
|
||||
Test to accepts list of dicts (or a string representing a
|
||||
list of dicts) and packs the key/value pairs into a single dict.
|
||||
"""
|
||||
with patch.object(
|
||||
salt.utils.yaml,
|
||||
"safe_load",
|
||||
MagicMock(side_effect=yaml.parser.ParserError("f")),
|
||||
):
|
||||
with patch.dict(pkg_resource.__salt__, {"pkg.normalize_name": MagicMock()}):
|
||||
assert pkg_resource.pack_sources("sources") == {}
|
||||
|
||||
assert pkg_resource.pack_sources(["A", "a"]) == {}
|
||||
|
||||
assert pkg_resource.pack_sources([{"A": "a"}])
|
||||
|
||||
|
||||
def test_parse_targets():
|
||||
"""
|
||||
Test to parses the input to pkg.install and
|
||||
returns back the package(s) to be installed. Returns a
|
||||
list of packages, as well as a string noting whether the
|
||||
packages are to come from a repository or a binary package.
|
||||
"""
|
||||
with patch.dict(pkg_resource.__grains__, {"os": "A"}):
|
||||
assert pkg_resource.parse_targets(pkgs="a", sources="a") == (None, None)
|
||||
|
||||
with patch.object(pkg_resource, "_repack_pkgs", return_value=False):
|
||||
assert pkg_resource.parse_targets(pkgs="a") == (None, None)
|
||||
|
||||
with patch.object(pkg_resource, "_repack_pkgs", return_value="A"):
|
||||
assert pkg_resource.parse_targets(pkgs="a") == ("A", "repository")
|
||||
|
||||
with patch.dict(pkg_resource.__grains__, {"os": "MacOS1"}):
|
||||
with patch.object(pkg_resource, "pack_sources", return_value=False):
|
||||
assert pkg_resource.parse_targets(sources="s") == (None, None)
|
||||
|
||||
with patch.object(pkg_resource, "pack_sources", return_value={"A": "/a"}):
|
||||
with patch.dict(
|
||||
pkg_resource.__salt__,
|
||||
{"config.valid_fileproto": MagicMock(return_value=False)},
|
||||
):
|
||||
assert pkg_resource.parse_targets(sources="s") == (["/a"], "file")
|
||||
|
||||
with patch.object(pkg_resource, "pack_sources", return_value={"A": "a"}):
|
||||
with patch.dict(
|
||||
pkg_resource.__salt__,
|
||||
{"config.valid_fileproto": MagicMock(return_value=False)},
|
||||
):
|
||||
assert pkg_resource.parse_targets(name="n") == (
|
||||
{"n": None},
|
||||
"repository",
|
||||
)
|
||||
|
||||
assert pkg_resource.parse_targets() == (None, None)
|
||||
|
||||
|
||||
def test_version():
|
||||
"""
|
||||
Test to Common interface for obtaining the version
|
||||
of installed packages.
|
||||
"""
|
||||
with patch.object(salt.utils.data, "is_true", return_value=True):
|
||||
mock = MagicMock(return_value={"A": "B"})
|
||||
with patch.dict(pkg_resource.__salt__, {"pkg.list_pkgs": mock}):
|
||||
assert pkg_resource.version("A") == "B"
|
||||
|
||||
assert pkg_resource.version() == {}
|
||||
|
||||
mock = MagicMock(return_value={})
|
||||
with patch.dict(pkg_resource.__salt__, {"pkg.list_pkgs": mock}):
|
||||
with patch("builtins.next") as mock_next:
|
||||
mock_next.side_effect = StopIteration()
|
||||
assert pkg_resource.version("A") == ""
|
||||
|
||||
|
||||
def test_add_pkg():
|
||||
"""
|
||||
Test to add a package to a dict of installed packages.
|
||||
"""
|
||||
assert pkg_resource.add_pkg({"pkgs": []}, "name", "version") is None
|
||||
|
||||
|
||||
def test_sort_pkglist():
|
||||
"""
|
||||
Test to accepts a dict obtained from pkg.list_pkgs() and sorts
|
||||
in place the list of versions for any packages that have multiple
|
||||
versions installed, so that two package lists can be compared
|
||||
to one another.
|
||||
"""
|
||||
assert pkg_resource.sort_pkglist({}) is None
|
||||
|
||||
|
||||
def test_format_pkg_list_no_attr():
|
||||
"""
|
||||
Test to output format of the package list with no attr parameter.
|
||||
"""
|
||||
packages = {
|
||||
"glibc": [
|
||||
{
|
||||
"version": "2.12",
|
||||
"epoch": None,
|
||||
"release": "1.212.el6",
|
||||
"arch": "x86_64",
|
||||
}
|
||||
],
|
||||
"glibc.i686": [
|
||||
{
|
||||
"version": "2.12",
|
||||
"epoch": None,
|
||||
"release": "1.212.el6",
|
||||
"arch": "i686",
|
||||
}
|
||||
],
|
||||
"foobar": [
|
||||
{"version": "1.2.0", "epoch": "2", "release": "7", "arch": "x86_64"},
|
||||
{"version": "1.2.3", "epoch": "2", "release": "27", "arch": "x86_64"},
|
||||
],
|
||||
"foobar.something": [
|
||||
{"version": "1.1", "epoch": "3", "release": "23.1", "arch": "i686"}
|
||||
],
|
||||
"foobar.": [
|
||||
{"version": "1.1", "epoch": "3", "release": "23.1", "arch": "i686"}
|
||||
],
|
||||
}
|
||||
expected_pkg_list = {
|
||||
"glibc": "2.12-1.212.el6",
|
||||
"glibc.i686": "2.12-1.212.el6",
|
||||
"foobar": "2:1.2.0-7,2:1.2.3-27",
|
||||
"foobar.something": "3:1.1-23.1",
|
||||
"foobar.": "3:1.1-23.1",
|
||||
}
|
||||
assert pkg_resource.format_pkg_list(packages, False, None) == expected_pkg_list
|
||||
|
||||
|
||||
def test_format_pkg_list_with_attr():
|
||||
"""
|
||||
Test to output format of the package list with attr parameter.
|
||||
In this case, any redundant "arch" reference will be removed
|
||||
from the package name since it's included as part of the requested attr.
|
||||
"""
|
||||
name_arch_mapping = {
|
||||
"glibc": {"name": "glibc", "arch": None},
|
||||
"glibc.i686": {"name": "glibc", "arch": "i686"},
|
||||
"foobar": {"name": "foobar", "arch": None},
|
||||
"foobar.something": {"name": "foobar.something", "arch": None},
|
||||
"foobar.": {"name": "foobar.", "arch": None},
|
||||
}
|
||||
packages = {
|
||||
"glibc": [
|
||||
{
|
||||
"version": "2.12",
|
||||
"epoch": None,
|
||||
"release": "1.212.el6",
|
||||
"arch": "x86_64",
|
||||
}
|
||||
],
|
||||
"glibc.i686": [
|
||||
{
|
||||
"version": "2.12",
|
||||
"epoch": None,
|
||||
"release": "1.212.el6",
|
||||
"arch": "i686",
|
||||
}
|
||||
],
|
||||
"foobar": [
|
||||
{"version": "1.2.0", "epoch": "2", "release": "7", "arch": "x86_64"},
|
||||
{"version": "1.2.3", "epoch": "2", "release": "27", "arch": "x86_64"},
|
||||
],
|
||||
"foobar.something": [
|
||||
{"version": "1.1", "epoch": "3", "release": "23.1", "arch": "i686"}
|
||||
],
|
||||
"foobar.": [
|
||||
{"version": "1.1", "epoch": "3", "release": "23.1", "arch": "i686"}
|
||||
],
|
||||
}
|
||||
expected_pkg_list = {
|
||||
"glibc": [
|
||||
{
|
||||
"arch": "x86_64",
|
||||
"release": "1.212.el6",
|
||||
"epoch": None,
|
||||
"version": "2.12",
|
||||
},
|
||||
{
|
||||
"arch": "i686",
|
||||
"release": "1.212.el6",
|
||||
"epoch": None,
|
||||
"version": "2.12",
|
||||
},
|
||||
],
|
||||
"foobar": [
|
||||
{"arch": "x86_64", "release": "7", "epoch": "2", "version": "1.2.0"},
|
||||
{"arch": "x86_64", "release": "27", "epoch": "2", "version": "1.2.3"},
|
||||
],
|
||||
"foobar.": [
|
||||
{"arch": "i686", "release": "23.1", "epoch": "3", "version": "1.1"}
|
||||
],
|
||||
"foobar.something": [
|
||||
{"arch": "i686", "release": "23.1", "epoch": "3", "version": "1.1"}
|
||||
],
|
||||
}
|
||||
with patch.dict(pkg_resource.__salt__, {"pkg.parse_arch": name_arch_mapping.get}):
|
||||
pkgs = pkg_resource.format_pkg_list(packages, False, attr=["epoch", "release"])
|
||||
assert pkgs == expected_pkg_list
|
||||
|
||||
|
||||
def test_stringify():
|
||||
"""
|
||||
Test to takes a dict of package name/version information
|
||||
and joins each list of
|
||||
installed versions into a string.
|
||||
"""
|
||||
assert pkg_resource.stringify({}) is None
|
||||
|
||||
|
||||
def test_version_clean():
|
||||
"""
|
||||
Test to clean the version string removing extra data.
|
||||
"""
|
||||
with patch.dict(
|
||||
pkg_resource.__salt__, {"pkg.version_clean": MagicMock(return_value="A")}
|
||||
):
|
||||
assert pkg_resource.version_clean("version") == "A"
|
||||
|
||||
assert pkg_resource.version_clean("v") == "v"
|
||||
|
||||
|
||||
def test_check_extra_requirements():
|
||||
"""
|
||||
Test to check if the installed package already
|
||||
has the given requirements.
|
||||
"""
|
||||
with patch.dict(
|
||||
pkg_resource.__salt__,
|
||||
{"pkg.check_extra_requirements": MagicMock(return_value="A")},
|
||||
):
|
||||
assert pkg_resource.check_extra_requirements("a", "b") == "A"
|
||||
|
||||
assert pkg_resource.check_extra_requirements("a", False)
|
||||
|
||||
|
||||
def test_version_compare():
|
||||
"""
|
||||
Test the version_compare function
|
||||
|
||||
TODO: Come up with a good way to test epoch handling across different
|
||||
platforms. This function will look in the ``__salt__`` dunder for a
|
||||
version_cmp function (which not all pkg modules implement) and use that
|
||||
to perform platform-specific handling (including interpretation of
|
||||
epochs), but even an integration test would need to take into account
|
||||
the fact that not all package managers grok epochs.
|
||||
"""
|
||||
assert pkg_resource.version_compare("2.0", "<", "3.0") is True
|
||||
|
|
@ -1,306 +0,0 @@
|
|||
import salt.modules.pf as pf
|
||||
from tests.support.mixins import LoaderModuleMockMixin
|
||||
from tests.support.mock import MagicMock, patch
|
||||
from tests.support.unit import TestCase
|
||||
|
||||
|
||||
class PfTestCase(TestCase, LoaderModuleMockMixin):
|
||||
"""
|
||||
test modules.pf functions
|
||||
"""
|
||||
|
||||
def setup_loader_modules(self):
|
||||
return {pf: {}}
|
||||
|
||||
def test_enable_when_disabled(self):
|
||||
"""
|
||||
Tests enabling pf when it's not enabled yet.
|
||||
"""
|
||||
ret = {}
|
||||
ret["stderr"] = "pf enabled"
|
||||
ret["retcode"] = 0
|
||||
mock_cmd = MagicMock(return_value=ret)
|
||||
with patch.dict(pf.__salt__, {"cmd.run_all": mock_cmd}):
|
||||
self.assertTrue(pf.enable()["changes"])
|
||||
|
||||
def test_enable_when_enabled(self):
|
||||
"""
|
||||
Tests enabling pf when it already enabled.
|
||||
"""
|
||||
ret = {}
|
||||
ret["stderr"] = "pfctl: pf already enabled"
|
||||
ret["retcode"] = 1
|
||||
mock_cmd = MagicMock(return_value=ret)
|
||||
with patch.dict(pf.__salt__, {"cmd.run_all": mock_cmd}):
|
||||
self.assertFalse(pf.enable()["changes"])
|
||||
|
||||
def test_disable_when_enabled(self):
|
||||
"""
|
||||
Tests disabling pf when it's enabled.
|
||||
"""
|
||||
ret = {}
|
||||
ret["stderr"] = "pf disabled"
|
||||
ret["retcode"] = 0
|
||||
mock_cmd = MagicMock(return_value=ret)
|
||||
with patch.dict(pf.__salt__, {"cmd.run_all": mock_cmd}):
|
||||
self.assertTrue(pf.disable()["changes"])
|
||||
|
||||
def test_disable_when_disabled(self):
|
||||
"""
|
||||
Tests disabling pf when it already disabled.
|
||||
"""
|
||||
ret = {}
|
||||
ret["stderr"] = "pfctl: pf not enabled"
|
||||
ret["retcode"] = 1
|
||||
mock_cmd = MagicMock(return_value=ret)
|
||||
with patch.dict(pf.__salt__, {"cmd.run_all": mock_cmd}):
|
||||
self.assertFalse(pf.disable()["changes"])
|
||||
|
||||
def test_loglevel_freebsd(self):
|
||||
"""
|
||||
Tests setting a loglevel.
|
||||
"""
|
||||
ret = {}
|
||||
ret["retcode"] = 0
|
||||
mock_cmd = MagicMock(return_value=ret)
|
||||
with patch.dict(pf.__salt__, {"cmd.run_all": mock_cmd}), patch.dict(
|
||||
pf.__grains__, {"os": "FreeBSD"}
|
||||
):
|
||||
res = pf.loglevel("urgent")
|
||||
mock_cmd.assert_called_once_with(
|
||||
"pfctl -x urgent", output_loglevel="trace", python_shell=False
|
||||
)
|
||||
self.assertTrue(res["changes"])
|
||||
|
||||
def test_loglevel_openbsd(self):
|
||||
"""
|
||||
Tests setting a loglevel.
|
||||
"""
|
||||
ret = {}
|
||||
ret["retcode"] = 0
|
||||
mock_cmd = MagicMock(return_value=ret)
|
||||
with patch.dict(pf.__salt__, {"cmd.run_all": mock_cmd}), patch.dict(
|
||||
pf.__grains__, {"os": "OpenBSD"}
|
||||
):
|
||||
res = pf.loglevel("crit")
|
||||
mock_cmd.assert_called_once_with(
|
||||
"pfctl -x crit", output_loglevel="trace", python_shell=False
|
||||
)
|
||||
self.assertTrue(res["changes"])
|
||||
|
||||
def test_load(self):
|
||||
"""
|
||||
Tests loading ruleset.
|
||||
"""
|
||||
ret = {}
|
||||
ret["retcode"] = 0
|
||||
mock_cmd = MagicMock(return_value=ret)
|
||||
with patch.dict(pf.__salt__, {"cmd.run_all": mock_cmd}):
|
||||
res = pf.load()
|
||||
mock_cmd.assert_called_once_with(
|
||||
["pfctl", "-f", "/etc/pf.conf"],
|
||||
output_loglevel="trace",
|
||||
python_shell=False,
|
||||
)
|
||||
self.assertTrue(res["changes"])
|
||||
|
||||
def test_load_noop(self):
|
||||
"""
|
||||
Tests evaluating but not actually loading ruleset.
|
||||
"""
|
||||
ret = {}
|
||||
ret["retcode"] = 0
|
||||
mock_cmd = MagicMock(return_value=ret)
|
||||
with patch.dict(pf.__salt__, {"cmd.run_all": mock_cmd}):
|
||||
res = pf.load(noop=True)
|
||||
mock_cmd.assert_called_once_with(
|
||||
["pfctl", "-f", "/etc/pf.conf", "-n"],
|
||||
output_loglevel="trace",
|
||||
python_shell=False,
|
||||
)
|
||||
self.assertFalse(res["changes"])
|
||||
|
||||
def test_flush(self):
|
||||
"""
|
||||
Tests a regular flush command.
|
||||
"""
|
||||
ret = {}
|
||||
ret["stderr"] = "pf: statistics cleared"
|
||||
ret["retcode"] = 0
|
||||
mock_cmd = MagicMock(return_value=ret)
|
||||
with patch.dict(pf.__salt__, {"cmd.run_all": mock_cmd}):
|
||||
res = pf.flush("info")
|
||||
mock_cmd.assert_called_once_with(
|
||||
"pfctl -v -F info", output_loglevel="trace", python_shell=False
|
||||
)
|
||||
self.assertTrue(res["changes"])
|
||||
|
||||
def test_flush_capital(self):
|
||||
"""
|
||||
Tests a flush command starting with a capital letter.
|
||||
"""
|
||||
ret = {}
|
||||
ret["stderr"] = "2 tables cleared"
|
||||
ret["retcode"] = 0
|
||||
mock_cmd = MagicMock(return_value=ret)
|
||||
with patch.dict(pf.__salt__, {"cmd.run_all": mock_cmd}):
|
||||
res = pf.flush("tables")
|
||||
mock_cmd.assert_called_once_with(
|
||||
"pfctl -v -F Tables", output_loglevel="trace", python_shell=False
|
||||
)
|
||||
self.assertTrue(res["changes"])
|
||||
|
||||
def test_flush_without_changes(self):
|
||||
"""
|
||||
Tests a flush command that has no changes.
|
||||
"""
|
||||
ret = {}
|
||||
ret["stderr"] = "0 tables cleared"
|
||||
ret["retcode"] = 0
|
||||
mock_cmd = MagicMock(return_value=ret)
|
||||
with patch.dict(pf.__salt__, {"cmd.run_all": mock_cmd}):
|
||||
self.assertFalse(pf.flush("tables")["changes"])
|
||||
|
||||
def test_table(self):
|
||||
"""
|
||||
Tests a regular table command.
|
||||
"""
|
||||
ret = {}
|
||||
ret["stderr"] = "42 addresses deleted"
|
||||
ret["retcode"] = 0
|
||||
mock_cmd = MagicMock(return_value=ret)
|
||||
with patch.dict(pf.__salt__, {"cmd.run_all": mock_cmd}):
|
||||
self.assertTrue(pf.table("flush", table="bad_hosts")["changes"])
|
||||
|
||||
def test_table_expire(self):
|
||||
"""
|
||||
Tests the table expire command.
|
||||
"""
|
||||
ret = {}
|
||||
ret["stderr"] = "1/1 addresses expired."
|
||||
ret["retcode"] = 0
|
||||
mock_cmd = MagicMock(return_value=ret)
|
||||
with patch.dict(pf.__salt__, {"cmd.run_all": mock_cmd}):
|
||||
self.assertTrue(
|
||||
pf.table("expire", table="bad_hosts", number=300)["changes"]
|
||||
)
|
||||
|
||||
def test_table_add_addresses(self):
|
||||
"""
|
||||
Tests adding addresses to a table.
|
||||
"""
|
||||
ret = {}
|
||||
ret["stderr"] = "2/2 addressess added."
|
||||
ret["retcode"] = 0
|
||||
mock_cmd = MagicMock(return_value=ret)
|
||||
with patch.dict(pf.__salt__, {"cmd.run_all": mock_cmd}):
|
||||
self.assertTrue(
|
||||
pf.table("add", table="bad_hosts", addresses=["1.2.3.4", "5.6.7.8"])[
|
||||
"changes"
|
||||
]
|
||||
)
|
||||
|
||||
def test_table_delete_addresses(self):
|
||||
"""
|
||||
Tests deleting addresses in a table.
|
||||
"""
|
||||
ret = {}
|
||||
ret["stderr"] = "2/2 addressess deleted."
|
||||
ret["retcode"] = 0
|
||||
mock_cmd = MagicMock(return_value=ret)
|
||||
with patch.dict(pf.__salt__, {"cmd.run_all": mock_cmd}):
|
||||
self.assertTrue(
|
||||
pf.table("delete", table="bad_hosts", addresses=["1.2.3.4", "5.6.7.8"])[
|
||||
"changes"
|
||||
]
|
||||
)
|
||||
|
||||
def test_table_test_address(self):
|
||||
"""
|
||||
Tests testing addresses in a table.
|
||||
"""
|
||||
ret = {}
|
||||
ret["stderr"] = "1/2 addressess match."
|
||||
ret["retcode"] = 0
|
||||
mock_cmd = MagicMock(return_value=ret)
|
||||
with patch.dict(pf.__salt__, {"cmd.run_all": mock_cmd}):
|
||||
self.assertTrue(
|
||||
pf.table("test", table="bad_hosts", addresses=["1.2.3.4"])["matches"]
|
||||
)
|
||||
|
||||
def test_table_no_changes(self):
|
||||
"""
|
||||
Tests a table command that has no changes.
|
||||
"""
|
||||
ret = {}
|
||||
ret["stderr"] = "0/1 addresses expired."
|
||||
ret["retcode"] = 0
|
||||
mock_cmd = MagicMock(return_value=ret)
|
||||
with patch.dict(pf.__salt__, {"cmd.run_all": mock_cmd}):
|
||||
self.assertFalse(
|
||||
pf.table("expire", table="bad_hosts", number=300)["changes"]
|
||||
)
|
||||
|
||||
def test_table_show(self):
|
||||
"""
|
||||
Tests showing table contents.
|
||||
"""
|
||||
ret = {}
|
||||
ret["stdout"] = "1.2.3.4\n5.6.7.8"
|
||||
ret["retcode"] = 0
|
||||
expected = ["1.2.3.4", "5.6.7.8"]
|
||||
mock_cmd = MagicMock(return_value=ret)
|
||||
with patch.dict(pf.__salt__, {"cmd.run_all": mock_cmd}):
|
||||
self.assertListEqual(
|
||||
pf.table("show", table="bad_hosts")["comment"], expected
|
||||
)
|
||||
|
||||
def test_table_zero(self):
|
||||
"""
|
||||
Tests clearing all the statistics of a table.
|
||||
"""
|
||||
ret = {}
|
||||
ret["stderr"] = "42 addresses has been cleared"
|
||||
ret["retcode"] = 0
|
||||
mock_cmd = MagicMock(return_value=ret)
|
||||
with patch.dict(pf.__salt__, {"cmd.run_all": mock_cmd}):
|
||||
self.assertTrue(pf.table("zero", table="bad_hosts")["changes"])
|
||||
|
||||
def test_show_rules(self):
|
||||
"""
|
||||
Tests show rules command.
|
||||
"""
|
||||
ret = {}
|
||||
ret["stdout"] = "block return\npass"
|
||||
ret["retcode"] = 0
|
||||
expected = ["block return", "pass"]
|
||||
mock_cmd = MagicMock(return_value=ret)
|
||||
with patch.dict(pf.__salt__, {"cmd.run_all": mock_cmd}):
|
||||
self.assertListEqual(pf.show("rules")["comment"], expected)
|
||||
|
||||
def test_show_states(self):
|
||||
"""
|
||||
Tests show states command.
|
||||
"""
|
||||
ret = {}
|
||||
ret["stdout"] = "all udp 192.168.1.1:3478\n"
|
||||
ret["retcode"] = 0
|
||||
expected = ["all udp 192.168.1.1:3478", ""]
|
||||
mock_cmd = MagicMock(return_value=ret)
|
||||
with patch.dict(pf.__salt__, {"cmd.run_all": mock_cmd}):
|
||||
self.assertListEqual(pf.show("states")["comment"], expected)
|
||||
|
||||
def test_show_tables(self):
|
||||
"""
|
||||
Tests show tables command.
|
||||
"""
|
||||
ret = {}
|
||||
ret["stdout"] = "bad_hosts"
|
||||
ret["retcode"] = 0
|
||||
mock_cmd = MagicMock(return_value=ret)
|
||||
with patch.dict(pf.__salt__, {"cmd.run_all": mock_cmd}):
|
||||
res = pf.show("tables")
|
||||
mock_cmd.assert_called_once_with(
|
||||
"pfctl -s Tables", output_loglevel="trace", python_shell=False
|
||||
)
|
||||
self.assertFalse(res["changes"])
|
||||
|
|
@ -1,160 +0,0 @@
|
|||
import salt.modules.pillar as pillarmod
|
||||
from salt.utils.odict import OrderedDict
|
||||
from tests.support.mixins import LoaderModuleMockMixin
|
||||
from tests.support.mock import MagicMock, patch
|
||||
from tests.support.unit import TestCase
|
||||
|
||||
pillar_value_1 = dict(a=1, b="very secret")
|
||||
|
||||
|
||||
class PillarModuleTestCase(TestCase, LoaderModuleMockMixin):
|
||||
def setup_loader_modules(self):
|
||||
return {pillarmod: {}}
|
||||
|
||||
def test_obfuscate_inner_recursion(self):
|
||||
self.assertEqual(
|
||||
pillarmod._obfuscate_inner(
|
||||
dict(a=[1, 2], b=dict(pwd="secret", deeper=("a", 1)))
|
||||
),
|
||||
dict(a=["<int>", "<int>"], b=dict(pwd="<str>", deeper=("<str>", "<int>"))),
|
||||
)
|
||||
|
||||
def test_obfuscate_inner_more_types(self):
|
||||
self.assertEqual(
|
||||
pillarmod._obfuscate_inner(OrderedDict([("key", "value")])),
|
||||
OrderedDict([("key", "<str>")]),
|
||||
)
|
||||
|
||||
self.assertEqual(pillarmod._obfuscate_inner({1, 2}), {"<int>"})
|
||||
|
||||
self.assertEqual(pillarmod._obfuscate_inner((1, 2)), ("<int>", "<int>"))
|
||||
|
||||
def test_obfuscate(self):
|
||||
with patch("salt.modules.pillar.items", MagicMock(return_value=pillar_value_1)):
|
||||
self.assertEqual(pillarmod.obfuscate(), dict(a="<int>", b="<str>"))
|
||||
|
||||
def test_ls(self):
|
||||
with patch("salt.modules.pillar.items", MagicMock(return_value=pillar_value_1)):
|
||||
ls = sorted(pillarmod.ls())
|
||||
self.assertCountEqual(ls, ["a", "b"])
|
||||
|
||||
def test_pillar_get_default_merge(self):
|
||||
defaults = {
|
||||
"int": 1,
|
||||
"string": "foo",
|
||||
"list": ["foo"],
|
||||
"dict": {"foo": "bar", "subkey": {"foo": "bar"}},
|
||||
}
|
||||
|
||||
pillar_mock = {
|
||||
"int": 2,
|
||||
"string": "bar",
|
||||
"list": ["bar", "baz"],
|
||||
"dict": {"baz": "qux", "subkey": {"baz": "qux"}},
|
||||
}
|
||||
|
||||
# Test that we raise a KeyError when pillar_raise_on_missing is True
|
||||
with patch.dict(pillarmod.__opts__, {"pillar_raise_on_missing": True}):
|
||||
self.assertRaises(KeyError, pillarmod.get, "missing")
|
||||
# Test that we return an empty string when it is not
|
||||
with patch.dict(pillarmod.__opts__, {}):
|
||||
self.assertEqual(pillarmod.get("missing"), "")
|
||||
|
||||
with patch.dict(pillarmod.__pillar__, pillar_mock):
|
||||
# Test with no default passed (it should be KeyError) and merge=True.
|
||||
# The merge should be skipped and the value returned from __pillar__
|
||||
# should be returned.
|
||||
for item in pillarmod.__pillar__:
|
||||
self.assertEqual(
|
||||
pillarmod.get(item, merge=True), pillarmod.__pillar__[item]
|
||||
)
|
||||
|
||||
# Test merging when the type of the default value is not the same as
|
||||
# what was returned. Merging should be skipped and the value returned
|
||||
# from __pillar__ should be returned.
|
||||
for default_type in defaults:
|
||||
for data_type in ("dict", "list"):
|
||||
if default_type == data_type:
|
||||
continue
|
||||
self.assertEqual(
|
||||
pillarmod.get(
|
||||
data_type, default=defaults[default_type], merge=True
|
||||
),
|
||||
pillarmod.__pillar__[data_type],
|
||||
)
|
||||
|
||||
# Test recursive dict merging
|
||||
self.assertEqual(
|
||||
pillarmod.get("dict", default=defaults["dict"], merge=True),
|
||||
{"foo": "bar", "baz": "qux", "subkey": {"foo": "bar", "baz": "qux"}},
|
||||
)
|
||||
|
||||
# Test list merging
|
||||
self.assertEqual(
|
||||
pillarmod.get("list", default=defaults["list"], merge=True),
|
||||
["foo", "bar", "baz"],
|
||||
)
|
||||
|
||||
def test_pillar_get_default_merge_regression_38558(self):
|
||||
"""Test for pillar.get(key=..., default=..., merge=True)
|
||||
|
||||
Do not update the ``default`` value when using ``merge=True``.
|
||||
|
||||
See: https://github.com/saltstack/salt/issues/38558
|
||||
"""
|
||||
with patch.dict(pillarmod.__pillar__, {"l1": {"l2": {"l3": 42}}}):
|
||||
|
||||
res = pillarmod.get(key="l1")
|
||||
self.assertEqual({"l2": {"l3": 42}}, res)
|
||||
|
||||
default = {"l2": {"l3": 43}}
|
||||
|
||||
res = pillarmod.get(key="l1", default=default)
|
||||
self.assertEqual({"l2": {"l3": 42}}, res)
|
||||
self.assertEqual({"l2": {"l3": 43}}, default)
|
||||
|
||||
res = pillarmod.get(key="l1", default=default, merge=True)
|
||||
self.assertEqual({"l2": {"l3": 42}}, res)
|
||||
self.assertEqual({"l2": {"l3": 43}}, default)
|
||||
|
||||
def test_pillar_get_default_merge_regression_39062(self):
|
||||
"""
|
||||
Confirm that we do not raise an exception if default is None and
|
||||
merge=True.
|
||||
|
||||
See https://github.com/saltstack/salt/issues/39062 for more info.
|
||||
"""
|
||||
with patch.dict(pillarmod.__pillar__, {"foo": "bar"}):
|
||||
|
||||
self.assertEqual(
|
||||
pillarmod.get(key="foo", default=None, merge=True),
|
||||
"bar",
|
||||
)
|
||||
|
||||
def test_pillar_get_int_key(self):
|
||||
"""
|
||||
Confirm that we can access pillar keys that are integers
|
||||
"""
|
||||
with patch.dict(pillarmod.__pillar__, {12345: "luggage_code"}):
|
||||
|
||||
self.assertEqual(
|
||||
pillarmod.get(key=12345, default=None, merge=True),
|
||||
"luggage_code",
|
||||
)
|
||||
|
||||
with patch.dict(
|
||||
pillarmod.__pillar__, {12345: {"l2": {"l3": "my_luggage_code"}}}
|
||||
):
|
||||
|
||||
res = pillarmod.get(key=12345)
|
||||
self.assertEqual({"l2": {"l3": "my_luggage_code"}}, res)
|
||||
|
||||
default = {"l2": {"l3": "your_luggage_code"}}
|
||||
|
||||
res = pillarmod.get(key=12345, default=default)
|
||||
self.assertEqual({"l2": {"l3": "my_luggage_code"}}, res)
|
||||
self.assertEqual({"l2": {"l3": "your_luggage_code"}}, default)
|
||||
|
||||
res = pillarmod.get(key=12345, default=default, merge=True)
|
||||
self.assertEqual({"l2": {"l3": "my_luggage_code"}}, res)
|
||||
self.assertEqual({"l2": {"l3": "your_luggage_code"}}, default)
|
||||
File diff suppressed because it is too large
Load diff
|
|
@ -1,281 +0,0 @@
|
|||
"""
|
||||
:codeauthor: Rahul Handay <rahulha@saltstack.com>
|
||||
"""
|
||||
|
||||
|
||||
import salt.modules.pkg_resource as pkg_resource
|
||||
import salt.utils.data
|
||||
import salt.utils.yaml
|
||||
import yaml
|
||||
from tests.support.mixins import LoaderModuleMockMixin
|
||||
from tests.support.mock import MagicMock, patch
|
||||
from tests.support.unit import TestCase
|
||||
|
||||
|
||||
class PkgresTestCase(TestCase, LoaderModuleMockMixin):
|
||||
"""
|
||||
Test cases for salt.modules.pkg_resource
|
||||
"""
|
||||
|
||||
def setup_loader_modules(self):
|
||||
return {pkg_resource: {}}
|
||||
|
||||
def test_pack_sources(self):
|
||||
"""
|
||||
Test to accepts list of dicts (or a string representing a
|
||||
list of dicts) and packs the key/value pairs into a single dict.
|
||||
"""
|
||||
with patch.object(
|
||||
salt.utils.yaml,
|
||||
"safe_load",
|
||||
MagicMock(side_effect=yaml.parser.ParserError("f")),
|
||||
):
|
||||
with patch.dict(pkg_resource.__salt__, {"pkg.normalize_name": MagicMock()}):
|
||||
self.assertDictEqual(pkg_resource.pack_sources("sources"), {})
|
||||
|
||||
self.assertDictEqual(pkg_resource.pack_sources(["A", "a"]), {})
|
||||
|
||||
self.assertTrue(pkg_resource.pack_sources([{"A": "a"}]))
|
||||
|
||||
def test_parse_targets(self):
|
||||
"""
|
||||
Test to parses the input to pkg.install and
|
||||
returns back the package(s) to be installed. Returns a
|
||||
list of packages, as well as a string noting whether the
|
||||
packages are to come from a repository or a binary package.
|
||||
"""
|
||||
with patch.dict(pkg_resource.__grains__, {"os": "A"}):
|
||||
self.assertEqual(
|
||||
pkg_resource.parse_targets(pkgs="a", sources="a"), (None, None)
|
||||
)
|
||||
|
||||
with patch.object(pkg_resource, "_repack_pkgs", return_value=False):
|
||||
self.assertEqual(pkg_resource.parse_targets(pkgs="a"), (None, None))
|
||||
|
||||
with patch.object(pkg_resource, "_repack_pkgs", return_value="A"):
|
||||
self.assertEqual(
|
||||
pkg_resource.parse_targets(pkgs="a"), ("A", "repository")
|
||||
)
|
||||
|
||||
with patch.dict(pkg_resource.__grains__, {"os": "MacOS1"}):
|
||||
with patch.object(pkg_resource, "pack_sources", return_value=False):
|
||||
self.assertEqual(pkg_resource.parse_targets(sources="s"), (None, None))
|
||||
|
||||
with patch.object(pkg_resource, "pack_sources", return_value={"A": "/a"}):
|
||||
with patch.dict(
|
||||
pkg_resource.__salt__,
|
||||
{"config.valid_fileproto": MagicMock(return_value=False)},
|
||||
):
|
||||
self.assertEqual(
|
||||
pkg_resource.parse_targets(sources="s"), (["/a"], "file")
|
||||
)
|
||||
|
||||
with patch.object(pkg_resource, "pack_sources", return_value={"A": "a"}):
|
||||
with patch.dict(
|
||||
pkg_resource.__salt__,
|
||||
{"config.valid_fileproto": MagicMock(return_value=False)},
|
||||
):
|
||||
self.assertEqual(
|
||||
pkg_resource.parse_targets(name="n"),
|
||||
({"n": None}, "repository"),
|
||||
)
|
||||
|
||||
self.assertEqual(pkg_resource.parse_targets(), (None, None))
|
||||
|
||||
def test_version(self):
|
||||
"""
|
||||
Test to Common interface for obtaining the version
|
||||
of installed packages.
|
||||
"""
|
||||
with patch.object(salt.utils.data, "is_true", return_value=True):
|
||||
mock = MagicMock(return_value={"A": "B"})
|
||||
with patch.dict(pkg_resource.__salt__, {"pkg.list_pkgs": mock}):
|
||||
self.assertEqual(pkg_resource.version("A"), "B")
|
||||
|
||||
self.assertDictEqual(pkg_resource.version(), {})
|
||||
|
||||
mock = MagicMock(return_value={})
|
||||
with patch.dict(pkg_resource.__salt__, {"pkg.list_pkgs": mock}):
|
||||
with patch("builtins.next") as mock_next:
|
||||
mock_next.side_effect = StopIteration()
|
||||
self.assertEqual(pkg_resource.version("A"), "")
|
||||
|
||||
def test_add_pkg(self):
|
||||
"""
|
||||
Test to add a package to a dict of installed packages.
|
||||
"""
|
||||
self.assertIsNone(pkg_resource.add_pkg({"pkgs": []}, "name", "version"))
|
||||
|
||||
def test_sort_pkglist(self):
|
||||
"""
|
||||
Test to accepts a dict obtained from pkg.list_pkgs() and sorts
|
||||
in place the list of versions for any packages that have multiple
|
||||
versions installed, so that two package lists can be compared
|
||||
to one another.
|
||||
"""
|
||||
self.assertIsNone(pkg_resource.sort_pkglist({}))
|
||||
|
||||
def test_format_pkg_list_no_attr(self):
|
||||
"""
|
||||
Test to output format of the package list with no attr parameter.
|
||||
"""
|
||||
packages = {
|
||||
"glibc": [
|
||||
{
|
||||
"version": "2.12",
|
||||
"epoch": None,
|
||||
"release": "1.212.el6",
|
||||
"arch": "x86_64",
|
||||
}
|
||||
],
|
||||
"glibc.i686": [
|
||||
{
|
||||
"version": "2.12",
|
||||
"epoch": None,
|
||||
"release": "1.212.el6",
|
||||
"arch": "i686",
|
||||
}
|
||||
],
|
||||
"foobar": [
|
||||
{"version": "1.2.0", "epoch": "2", "release": "7", "arch": "x86_64"},
|
||||
{"version": "1.2.3", "epoch": "2", "release": "27", "arch": "x86_64"},
|
||||
],
|
||||
"foobar.something": [
|
||||
{"version": "1.1", "epoch": "3", "release": "23.1", "arch": "i686"}
|
||||
],
|
||||
"foobar.": [
|
||||
{"version": "1.1", "epoch": "3", "release": "23.1", "arch": "i686"}
|
||||
],
|
||||
}
|
||||
expected_pkg_list = {
|
||||
"glibc": "2.12-1.212.el6",
|
||||
"glibc.i686": "2.12-1.212.el6",
|
||||
"foobar": "2:1.2.0-7,2:1.2.3-27",
|
||||
"foobar.something": "3:1.1-23.1",
|
||||
"foobar.": "3:1.1-23.1",
|
||||
}
|
||||
self.assertCountEqual(
|
||||
pkg_resource.format_pkg_list(packages, False, None), expected_pkg_list
|
||||
)
|
||||
|
||||
def test_format_pkg_list_with_attr(self):
|
||||
"""
|
||||
Test to output format of the package list with attr parameter.
|
||||
In this case, any redundant "arch" reference will be removed from the package name since it's
|
||||
include as part of the requested attr.
|
||||
"""
|
||||
NAME_ARCH_MAPPING = {
|
||||
"glibc": {"name": "glibc", "arch": None},
|
||||
"glibc.i686": {"name": "glibc", "arch": "i686"},
|
||||
"foobar": {"name": "foobar", "arch": None},
|
||||
"foobar.something": {"name": "foobar.something", "arch": None},
|
||||
"foobar.": {"name": "foobar.", "arch": None},
|
||||
}
|
||||
packages = {
|
||||
"glibc": [
|
||||
{
|
||||
"version": "2.12",
|
||||
"epoch": None,
|
||||
"release": "1.212.el6",
|
||||
"arch": "x86_64",
|
||||
}
|
||||
],
|
||||
"glibc.i686": [
|
||||
{
|
||||
"version": "2.12",
|
||||
"epoch": None,
|
||||
"release": "1.212.el6",
|
||||
"arch": "i686",
|
||||
}
|
||||
],
|
||||
"foobar": [
|
||||
{"version": "1.2.0", "epoch": "2", "release": "7", "arch": "x86_64"},
|
||||
{"version": "1.2.3", "epoch": "2", "release": "27", "arch": "x86_64"},
|
||||
],
|
||||
"foobar.something": [
|
||||
{"version": "1.1", "epoch": "3", "release": "23.1", "arch": "i686"}
|
||||
],
|
||||
"foobar.": [
|
||||
{"version": "1.1", "epoch": "3", "release": "23.1", "arch": "i686"}
|
||||
],
|
||||
}
|
||||
expected_pkg_list = {
|
||||
"glibc": [
|
||||
{
|
||||
"arch": "x86_64",
|
||||
"release": "1.212.el6",
|
||||
"epoch": None,
|
||||
"version": "2.12",
|
||||
},
|
||||
{
|
||||
"arch": "i686",
|
||||
"release": "1.212.el6",
|
||||
"epoch": None,
|
||||
"version": "2.12",
|
||||
},
|
||||
],
|
||||
"foobar": [
|
||||
{"arch": "x86_64", "release": "7", "epoch": "2", "version": "1.2.0"},
|
||||
{"arch": "x86_64", "release": "27", "epoch": "2", "version": "1.2.3"},
|
||||
],
|
||||
"foobar.": [
|
||||
{"arch": "i686", "release": "23.1", "epoch": "3", "version": "1.1"}
|
||||
],
|
||||
"foobar.something": [
|
||||
{"arch": "i686", "release": "23.1", "epoch": "3", "version": "1.1"}
|
||||
],
|
||||
}
|
||||
with patch.dict(
|
||||
pkg_resource.__salt__, {"pkg.parse_arch": NAME_ARCH_MAPPING.get}
|
||||
):
|
||||
self.assertCountEqual(
|
||||
pkg_resource.format_pkg_list(
|
||||
packages, False, attr=["epoch", "release"]
|
||||
),
|
||||
expected_pkg_list,
|
||||
)
|
||||
|
||||
def test_stringify(self):
|
||||
"""
|
||||
Test to takes a dict of package name/version information
|
||||
and joins each list of
|
||||
installed versions into a string.
|
||||
"""
|
||||
self.assertIsNone(pkg_resource.stringify({}))
|
||||
|
||||
def test_version_clean(self):
|
||||
"""
|
||||
Test to clean the version string removing extra data.
|
||||
"""
|
||||
with patch.dict(
|
||||
pkg_resource.__salt__, {"pkg.version_clean": MagicMock(return_value="A")}
|
||||
):
|
||||
self.assertEqual(pkg_resource.version_clean("version"), "A")
|
||||
|
||||
self.assertEqual(pkg_resource.version_clean("v"), "v")
|
||||
|
||||
def test_check_extra_requirements(self):
|
||||
"""
|
||||
Test to check if the installed package already
|
||||
has the given requirements.
|
||||
"""
|
||||
with patch.dict(
|
||||
pkg_resource.__salt__,
|
||||
{"pkg.check_extra_requirements": MagicMock(return_value="A")},
|
||||
):
|
||||
self.assertEqual(pkg_resource.check_extra_requirements("a", "b"), "A")
|
||||
|
||||
self.assertTrue(pkg_resource.check_extra_requirements("a", False))
|
||||
|
||||
def test_version_compare(self):
|
||||
"""
|
||||
Test the version_compare function
|
||||
|
||||
TODO: Come up with a good way to test epoch handling across different
|
||||
platforms. This function will look in the ``__salt__`` dunder for a
|
||||
version_cmp function (which not all pkg modules implement) and use that
|
||||
to perform platform-specific handling (including interpretation of
|
||||
epochs), but even an integration test would need to take into account
|
||||
the fact that not all package managers grok epochs.
|
||||
"""
|
||||
assert pkg_resource.version_compare("2.0", "<", "3.0") is True
|
||||
Loading…
Add table
Reference in a new issue