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

Use the minion_opts and master_opts fixtures instead.

Browse source 
Signed-off-by: Pedro Algarvio <palgarvio@vmware.com>
This commit is contained in:
Pedro Algarvio 2022-11-13 11:07:48 +00:00 • committed by Megan Wilhite
parent 673143e702
commit 1b18b4e5ab
44 changed files with 279 additions and 472 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

17
tests/pytests/unit/cache/test_mysql_cache.py vendored
View file

@ -8,8 +8,6 @@ import logging
import pytest
import salt.cache.mysql_cache as mysql_cache
import salt.payload
import salt.utils.files
from salt.exceptions import SaltCacheError
from tests.support.mock import MagicMock, call, patch
@ -27,13 +25,6 @@ def configure_loader_modules():
return {mysql_cache: {}}
@pytest.fixture
def master_config():
opts = salt.config.DEFAULT_MASTER_OPTS.copy()
opts["__role"] = "master"
return opts
def test_run_query():
"""
Tests that a SaltCacheError is raised when there is a problem writing to the
@ -45,7 +36,7 @@ def test_run_query():
mock_connect.assert_has_calls((expected_calls,), True)
def test_store(master_config):
def test_store():
"""
Tests that the store function writes the data to the serializer for storage.
"""
@ -101,7 +92,7 @@ def test_store(master_config):
assert expected in str(exc_info.value)
def test_fetch(master_config):
def test_fetch():
"""
Tests that the fetch function reads the data from the serializer for storage.
"""
@ -157,7 +148,7 @@ def test_flush():
mock_run_query.assert_has_calls(expected_calls, True)
def test_init_client(master_config):
def test_init_client():
"""
Tests that the _init_client places the correct information in __context__
"""
@ -204,7 +195,7 @@ def test_init_client(master_config):
)
def test_create_table(master_config):
def test_create_table():
"""
Tests that the _create_table
"""

8
tests/pytests/unit/client/test_netapi.py
View file

@ -1,22 +1,20 @@
import logging
import salt.client.netapi
import salt.config
from tests.support.mock import Mock, patch
def test_run_log(caplog):
def test_run_log(caplog, master_opts):
"""
test salt.client.netapi logs correct message
"""
opts = salt.config.DEFAULT_MASTER_OPTS.copy()
opts["rest_cherrypy"] = {"port": 8000}
master_opts["rest_cherrypy"] = {"port": 8000}
mock_process = Mock()
mock_process.add_process.return_value = True
patch_process = patch.object(salt.utils.process, "ProcessManager", mock_process)
with caplog.at_level(logging.INFO):
with patch_process:
netapi = salt.client.netapi.NetapiClient(opts)
netapi = salt.client.netapi.NetapiClient(master_opts)
netapi.run()
assert "Starting RunNetapi(salt.loaded.int.netapi.rest_cherrypy)" in caplog.text

14
tests/pytests/unit/cloud/test_cloud.py
View file

@ -1,20 +1,16 @@
import copy
import pytest
import salt.config
from salt.cloud import Cloud
from tests.support.mock import MagicMock, patch
@pytest.fixture
def master_config():
opts = copy.deepcopy(salt.config.DEFAULT_MASTER_OPTS)
opts["parallel"] = False
opts["providers"] = {
def master_config(master_opts):
master_opts["parallel"] = False
master_opts["providers"] = {
"test": {},
}
return opts
return master_opts
@pytest.fixture
@ -125,5 +121,5 @@ def test_vm_config_merger():
"size": "t2.micro",
"name": "test_vm",
}
vm = salt.cloud.Cloud.vm_config("test_vm", main, provider, profile, {})
vm = Cloud.vm_config("test_vm", main, provider, profile, {})
assert expected == vm

6
tests/pytests/unit/engines/test_script.py
View file

@ -4,16 +4,14 @@ unit tests for the script engine
import pytest
import salt.config
import salt.engines.script as script
from salt.exceptions import CommandExecutionError
from tests.support.mock import patch
@pytest.fixture
def configure_loader_modules():
opts = salt.config.DEFAULT_MASTER_OPTS
return {script: {"__opts__": opts}}
def configure_loader_modules(master_opts):
return {script: {"__opts__": master_opts}}
def test__get_serializer():

4
tests/pytests/unit/engines/test_slack.py
View file

@ -92,13 +92,13 @@ def configure_loader_modules():
@pytest.fixture
def slack_client():
def slack_client(minion_opts):
mock_opts = salt.config.DEFAULT_MINION_OPTS.copy()
app_token = "xapp-x-xxxxxxxxxxx-xxxxxxxxxxxxx-xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx"
bot_token = "xoxb-xxxxxxxxxx-xxxxxxxxxxxxxxxxxxxxxxxx"
trigger = "!"
with patch.dict(slack_engine.__opts__, mock_opts):
with patch.dict(slack_engine.__opts__, minion_opts):
with patch(
"slack_bolt.App", MagicMock(autospec=True, return_value=MockSlackBoltApp())
):

5
tests/pytests/unit/modules/dockermod/test_module.py
View file

@ -6,7 +6,6 @@ import logging
import pytest
import salt.config
import salt.loader
import salt.modules.dockermod as docker_mod
import salt.utils.platform
@ -21,9 +20,9 @@ pytest.importorskip(
@pytest.fixture
def configure_loader_modules():
def configure_loader_modules(minion_opts):
utils = salt.loader.utils(
salt.config.DEFAULT_MINION_OPTS.copy(),
minion_opts,
whitelist=[
"args",
"docker",

6
tests/pytests/unit/modules/test_ansiblegate.py
View file

@ -4,7 +4,6 @@ import os
import pytest
import salt.config
import salt.loader
import salt.modules.ansiblegate as ansiblegate
import salt.utils.json
@ -142,7 +141,7 @@ def test_ansible_playbooks_return_retcode():
assert "retcode" in ret
def test_ansible_targets():
def test_ansible_targets(minion_opts):
"""
Test ansible.targets execution module function.
:return:
@ -174,8 +173,7 @@ def test_ansible_targets():
"""
ansible_inventory_mock = MagicMock(return_value=ansible_inventory_ret)
with patch("salt.utils.path.which", MagicMock(return_value=True)):
opts = salt.config.DEFAULT_MINION_OPTS.copy()
utils = salt.loader.utils(opts, whitelist=["ansible"])
utils = salt.loader.utils(minion_opts, whitelist=["ansible"])
with patch("salt.modules.cmdmod.run", ansible_inventory_mock), patch.dict(
ansiblegate.__utils__, utils
), patch("os.path.isfile", MagicMock(return_value=True)):

12
tests/pytests/unit/modules/test_beacons.py
View file

@ -7,7 +7,6 @@ import os
import pytest
import salt.config
import salt.modules.beacons as beacons
from salt.utils.event import SaltEvent
from tests.support.mock import MagicMock, call, mock_open, patch
@ -16,15 +15,8 @@ log = logging.getLogger(__name__)
@pytest.fixture
def sock_dir(tmp_path):
return str(tmp_path / "test-socks")
@pytest.fixture
def configure_loader_modules(sock_dir):
opts = salt.config.DEFAULT_MINION_OPTS.copy()
opts["sock_dir"] = sock_dir
return {beacons: {"__opts__": opts}}
def configure_loader_modules(minion_opts):
return {beacons: {"__opts__": minion_opts}}
@pytest.mark.slow_test

5
tests/pytests/unit/modules/test_cmdmod.py
View file

@ -29,9 +29,8 @@ MOCK_SHELL_FILE = "# List of acceptable shells\n\n/bin/bash\n"
@pytest.fixture
def configure_loader_modules():
opts = salt.config.DEFAULT_MINION_OPTS.copy()
return {cmdmod: {"__opts__": opts}}
def configure_loader_modules(minion_opts):
return {cmdmod: {"__opts__": minion_opts}}
@pytest.fixture(scope="module")

117
tests/pytests/unit/modules/test_schedule.py
View file

@ -28,26 +28,20 @@ def job1():
@pytest.fixture
def sock_dir(tmp_path):
return str(tmp_path / "test-socks")
@pytest.fixture
def configure_loader_modules():
return {schedule: {}}
def configure_loader_modules(minion_opts):
minion_opts["schedule"] = {}
return {schedule: {"__opts__": minion_opts}}
# 'purge' function tests: 1
@pytest.mark.slow_test
def test_purge(sock_dir, job1):
def test_purge(job1):
"""
Test if it purge all the jobs currently scheduled on the minion.
"""
_schedule_data = {"job1": job1}
patch_makedirs = patch("os.makedirs", MagicMock(return_value=True))
patch_schedule_opts = patch.dict(
schedule.__opts__, {"schedule": {}, "sock_dir": sock_dir}
)
patch_schedule_opts = patch.dict(schedule.__opts__, {"schedule": {}})
patch_schedule_event_fire = patch.dict(
schedule.__salt__, {"event.fire": MagicMock(return_value=True)}
)
@ -74,9 +68,7 @@ def test_purge(sock_dir, job1):
changes = {"job1": "removed", "job2": "removed", "job3": "removed"}
patch_schedule_opts = patch.dict(
schedule.__opts__, {"schedule": {"job1": "salt"}, "sock_dir": sock_dir}
)
patch_schedule_opts = patch.dict(schedule.__opts__, {"schedule": {"job1": "salt"}})
patch_schedule_list = patch.object(
schedule, "list_", MagicMock(return_value=_schedule_data)
)
@ -98,14 +90,12 @@ def test_purge(sock_dir, job1):
# 'delete' function tests: 1
@pytest.mark.slow_test
def test_delete(sock_dir, job1):
def test_delete(job1):
"""
Test if it delete a job from the minion's schedule.
"""
patch_makedirs = patch("os.makedirs", MagicMock(return_value=True))
patch_schedule_opts = patch.dict(
schedule.__opts__, {"schedule": {}, "sock_dir": sock_dir}
)
patch_schedule_opts = patch.dict(schedule.__opts__, {"schedule": {}})
patch_schedule_event_fire = patch.dict(
schedule.__salt__, {"event.fire": MagicMock(return_value=True)}
)
@ -125,9 +115,7 @@ def test_delete(sock_dir, job1):
schedule, "list_", MagicMock(return_value=_schedule_data)
)
patch_schedule_opts = patch.dict(
schedule.__opts__, {"schedule": {"job1": "salt"}, "sock_dir": sock_dir}
)
patch_schedule_opts = patch.dict(schedule.__opts__, {"schedule": {"job1": "salt"}})
comm = "Deleted Job job1 from schedule."
changes = {"job1": "removed"}
@ -148,7 +136,7 @@ def test_delete(sock_dir, job1):
# 'build_schedule_item' function tests: 1
def test_build_schedule_item(sock_dir):
def test_build_schedule_item():
"""
Test if it build a schedule job.
"""
@ -183,7 +171,7 @@ def test_build_schedule_item(sock_dir):
# 'build_schedule_item_invalid_when' function tests: 1
def test_build_schedule_item_invalid_when(sock_dir):
def test_build_schedule_item_invalid_when():
"""
Test if it build a schedule job.
"""
@ -194,7 +182,7 @@ def test_build_schedule_item_invalid_when(sock_dir):
) == {"comment": comment, "result": False}
def test_build_schedule_item_invalid_jobs_args(sock_dir):
def test_build_schedule_item_invalid_jobs_args():
"""
Test failure if job_arg and job_kwargs are passed correctly
"""
@ -214,7 +202,7 @@ def test_build_schedule_item_invalid_jobs_args(sock_dir):
@pytest.mark.slow_test
def test_add(sock_dir):
def test_add():
"""
Test if it add a job to the schedule.
"""
@ -227,9 +215,7 @@ def test_add(sock_dir):
comm4 = "Job: job2 would be added to schedule."
patch_makedirs = patch("os.makedirs", MagicMock(return_value=True))
patch_schedule_opts = patch.dict(
schedule.__opts__, {"schedule": {"job1": "salt"}, "sock_dir": sock_dir}
)
patch_schedule_opts = patch.dict(schedule.__opts__, {"schedule": {"job1": "salt"}})
patch_schedule_event_fire = patch.dict(
schedule.__salt__, {"event.fire": MagicMock(return_value=True)}
)
@ -304,15 +290,13 @@ def test_add(sock_dir):
@pytest.mark.slow_test
def test_run_job(sock_dir, job1):
def test_run_job(job1):
"""
Test if it run a scheduled job on the minion immediately.
"""
patch_makedirs = patch("os.makedirs", MagicMock(return_value=True))
patch_schedule_opts = patch.dict(
schedule.__opts__, {"schedule": {"job1": job1}, "sock_dir": sock_dir}
)
patch_schedule_opts = patch.dict(schedule.__opts__, {"schedule": {"job1": job1}})
patch_schedule_event_fire = patch.dict(
schedule.__salt__, {"event.fire": MagicMock(return_value=True)}
@ -334,15 +318,13 @@ def test_run_job(sock_dir, job1):
@pytest.mark.slow_test
def test_enable_job(sock_dir):
def test_enable_job():
"""
Test if it enable a job in the minion's schedule.
"""
patch_makedirs = patch("os.makedirs", MagicMock(return_value=True))
patch_schedule_opts = patch.dict(
schedule.__opts__, {"schedule": {"job1": job1}, "sock_dir": sock_dir}
)
patch_schedule_opts = patch.dict(schedule.__opts__, {"schedule": {"job1": job1}})
patch_schedule_event_fire = patch.dict(
schedule.__salt__, {"event.fire": MagicMock(return_value=True)}
@ -363,15 +345,13 @@ def test_enable_job(sock_dir):
@pytest.mark.slow_test
def test_disable_job(sock_dir):
def test_disable_job():
"""
Test if it disable a job in the minion's schedule.
"""
patch_makedirs = patch("os.makedirs", MagicMock(return_value=True))
patch_schedule_opts = patch.dict(
schedule.__opts__, {"schedule": {"job1": job1}, "sock_dir": sock_dir}
)
patch_schedule_opts = patch.dict(schedule.__opts__, {"schedule": {"job1": job1}})
patch_schedule_event_fire = patch.dict(
schedule.__salt__, {"event.fire": MagicMock(return_value=True)}
@ -392,14 +372,14 @@ def test_disable_job(sock_dir):
@pytest.mark.slow_test
def test_save(sock_dir):
def test_save():
"""
Test if it save all scheduled jobs on the minion.
"""
comm1 = "Schedule (non-pillar items) saved."
with patch.dict(
schedule.__opts__,
{"schedule": {}, "default_include": "/tmp", "sock_dir": sock_dir},
{"default_include": "/tmp"},
):
with patch("os.makedirs", MagicMock(return_value=True)):
mock = MagicMock(return_value=True)
@ -412,7 +392,7 @@ def test_save(sock_dir):
# 'enable' function tests: 1
def test_enable(sock_dir):
def test_enable():
"""
Test if it enable all scheduled jobs on the minion.
"""
@ -426,7 +406,7 @@ def test_enable(sock_dir):
# 'disable' function tests: 1
def test_disable(sock_dir):
def test_disable():
"""
Test if it disable all scheduled jobs on the minion.
"""
@ -441,7 +421,7 @@ def test_disable(sock_dir):
@pytest.mark.slow_test
def test_move(sock_dir, job1):
def test_move(job1):
"""
Test if it move scheduled job to another minion or minions.
"""
@ -451,9 +431,7 @@ def test_move(sock_dir, job1):
patch_makedirs = patch("os.makedirs", MagicMock(return_value=True))
patch_schedule_opts = patch.dict(
schedule.__opts__, {"schedule": {"job1": job1}, "sock_dir": sock_dir}
)
patch_schedule_opts = patch.dict(schedule.__opts__, {"schedule": {"job1": job1}})
patch_schedule_event_fire = patch.dict(
schedule.__salt__, {"event.fire": MagicMock(return_value=True)}
@ -545,7 +523,7 @@ def test_move(sock_dir, job1):
@pytest.mark.slow_test
def test_copy(sock_dir, job1):
def test_copy(job1):
"""
Test if it copy scheduled job to another minion or minions.
"""
@ -555,9 +533,7 @@ def test_copy(sock_dir, job1):
patch_makedirs = patch("os.makedirs", MagicMock(return_value=True))
patch_schedule_opts = patch.dict(
schedule.__opts__, {"schedule": {"job1": job1}, "sock_dir": sock_dir}
)
patch_schedule_opts = patch.dict(schedule.__opts__, {"schedule": {"job1": job1}})
patch_schedule_event_fire = patch.dict(
schedule.__salt__, {"event.fire": MagicMock(return_value=True)}
@ -648,7 +624,7 @@ def test_copy(sock_dir, job1):
@pytest.mark.slow_test
def test_modify(sock_dir, job1):
def test_modify(job1):
"""
Test if modifying job to the schedule.
"""
@ -713,9 +689,7 @@ def test_modify(sock_dir, job1):
patch_makedirs = patch("os.makedirs", MagicMock(return_value=True))
patch_schedule_opts = patch.dict(
schedule.__opts__, {"schedule": {"job1": job1}, "sock_dir": sock_dir}
)
patch_schedule_opts = patch.dict(schedule.__opts__, {"schedule": {"job1": job1}})
patch_schedule_event_fire = patch.dict(
schedule.__salt__, {"event.fire": MagicMock(return_value=True)}
@ -857,7 +831,7 @@ def test_modify(sock_dir, job1):
# 'is_enabled' function tests: 1
def test_is_enabled(sock_dir):
def test_is_enabled():
"""
Test is_enabled
"""
@ -870,9 +844,7 @@ def test_is_enabled(sock_dir):
mock_lst = MagicMock(return_value=mock_schedule)
patch_makedirs = patch("os.makedirs", MagicMock(return_value=True))
patch_schedule_opts = patch.dict(
schedule.__opts__, {"schedule": {}, "sock_dir": sock_dir}
)
patch_schedule_opts = patch.dict(schedule.__opts__, {"schedule": {}})
patch_schedule_salt = patch.dict(
schedule.__salt__,
{"event.fire": MagicMock(return_value=True), "schedule.list": mock_lst},
@ -894,7 +866,7 @@ def test_is_enabled(sock_dir):
# 'job_status' function tests: 1
def test_job_status(sock_dir):
def test_job_status():
"""
Test is_enabled
"""
@ -912,9 +884,7 @@ def test_job_status(sock_dir):
mock_lst = MagicMock(return_value=mock_schedule)
patch_makedirs = patch("os.makedirs", MagicMock(return_value=True))
patch_schedule_opts = patch.dict(
schedule.__opts__, {"schedule": {"job1": job1}, "sock_dir": sock_dir}
)
patch_schedule_opts = patch.dict(schedule.__opts__, {"schedule": {"job1": job1}})
patch_schedule_salt = patch.dict(
schedule.__salt__,
{"event.fire": MagicMock(return_value=True), "schedule.list": mock_lst},
@ -935,7 +905,7 @@ def test_job_status(sock_dir):
# 'purge' function tests: 1
@pytest.mark.slow_test
def test_list(sock_dir, job1):
def test_list(job1):
"""
Test schedule.list
"""
@ -965,9 +935,7 @@ def test_list(sock_dir, job1):
"""
patch_makedirs = patch("os.makedirs", MagicMock(return_value=True))
patch_schedule_opts = patch.dict(
schedule.__opts__, {"schedule": {}, "sock_dir": sock_dir}
)
patch_schedule_opts = patch.dict(schedule.__opts__, {"schedule": {}})
patch_schedule_event_fire = patch.dict(
schedule.__salt__, {"event.fire": MagicMock(return_value=True)}
)
@ -1011,7 +979,6 @@ def test_list(sock_dir, job1):
ret = schedule.list_()
assert ret == expected
_schedule_data = {"job1": job1}
_ret_schedule_data = {
"function": "test.ping",
"seconds": 10,
@ -1048,7 +1015,7 @@ def test_list(sock_dir, job1):
@pytest.mark.slow_test
def test_list_global_enabled(sock_dir, job1):
def test_list_global_enabled(job1):
"""
Test schedule.list when enabled globally
"""
@ -1084,9 +1051,7 @@ def test_list_global_enabled(sock_dir, job1):
seconds: 10
"""
patch_makedirs = patch("os.makedirs", MagicMock(return_value=True))
patch_schedule_opts = patch.dict(
schedule.__opts__, {"schedule": {}, "sock_dir": sock_dir}
)
patch_schedule_opts = patch.dict(schedule.__opts__, {"schedule": {}})
patch_schedule_event_fire = patch.dict(
schedule.__salt__, {"event.fire": MagicMock(return_value=True)}
)
@ -1106,7 +1071,7 @@ def test_list_global_enabled(sock_dir, job1):
@pytest.mark.slow_test
def test_list_global_disabled(sock_dir, job1):
def test_list_global_disabled(job1):
"""
Test schedule.list when disabled globally
"""
@ -1143,9 +1108,7 @@ def test_list_global_disabled(sock_dir, job1):
seconds: 10
"""
patch_makedirs = patch("os.makedirs", MagicMock(return_value=True))
patch_schedule_opts = patch.dict(
schedule.__opts__, {"schedule": {}, "sock_dir": sock_dir}
)
patch_schedule_opts = patch.dict(schedule.__opts__, {"schedule": {}})
patch_schedule_event_fire = patch.dict(
schedule.__salt__, {"event.fire": MagicMock(return_value=True)}
)

7
tests/pytests/unit/modules/test_zfs.py
View file

@ -25,12 +25,11 @@ def utils_patch():
@pytest.fixture
def configure_loader_modules():
opts = salt.config.DEFAULT_MINION_OPTS.copy()
def configure_loader_modules(minion_opts):
utils = salt.loader.utils(
opts, whitelist=["zfs", "args", "systemd", "path", "platform"]
minion_opts, whitelist=["zfs", "args", "systemd", "path", "platform"]
)
zfs_obj = {zfs: {"__opts__": opts, "__utils__": utils}}
zfs_obj = {zfs: {"__opts__": minion_opts, "__utils__": utils}}
return zfs_obj

10
tests/pytests/unit/modules/test_zfs_solaris10.py
View file

@ -4,7 +4,6 @@ Tests for salt.modules.zfs on Solaris
import pytest
import salt.config
import salt.loader
import salt.modules.zfs as zfs
import salt.utils.zfs
@ -18,12 +17,11 @@ def utils_patch():
@pytest.fixture
def configure_loader_modules():
opts = salt.config.DEFAULT_MINION_OPTS.copy()
utils = salt.loader.utils(opts, whitelist=["zfs"])
def configure_loader_modules(minion_opts):
utils = salt.loader.utils(minion_opts, whitelist=["zfs"])
zfs_obj = {
zfs: {
"__opts__": opts,
"__opts__": minion_opts,
"__grains__": {
"osarch": "sparcv9",
"os_family": "Solaris",
@ -38,7 +36,7 @@ def configure_loader_modules():
return zfs_obj
@pytest.mark.skip_unless_on_sunos(reason="test to ensure no -t only applies to Solaris")
@pytest.mark.skip_unless_on_sunos
def test_get_success_solaris():
"""
Tests zfs get success

10
tests/pytests/unit/modules/test_zfs_solaris11.py
View file

@ -4,7 +4,6 @@ Tests for salt.modules.zfs on Solaris
import pytest
import salt.config
import salt.loader
import salt.modules.zfs as zfs
import salt.utils.zfs
@ -18,12 +17,11 @@ def utils_patch():
@pytest.fixture
def configure_loader_modules():
opts = salt.config.DEFAULT_MINION_OPTS.copy()
utils = salt.loader.utils(opts, whitelist=["zfs"])
def configure_loader_modules(minion_opts):
utils = salt.loader.utils(minion_opts, whitelist=["zfs"])
zfs_obj = {
zfs: {
"__opts__": opts,
"__opts__": minion_opts,
"__grains__": {
"osarch": "sparcv9",
"os_family": "Solaris",
@ -38,7 +36,7 @@ def configure_loader_modules():
return zfs_obj
@pytest.mark.skip_unless_on_sunos(reason="test to ensure no -t only applies to Solaris")
@pytest.mark.skip_unless_on_sunos
def test_get_success_solaris():
"""
Tests zfs get success

12
tests/pytests/unit/modules/test_zpool.py
View file

@ -26,12 +26,16 @@ def utils_patch():
@pytest.fixture
def configure_loader_modules():
opts = salt.config.DEFAULT_MINION_OPTS.copy()
def configure_loader_modules(minion_opts):
utils = salt.loader.utils(
opts, whitelist=["zfs", "args", "systemd", "path", "platform"]
minion_opts, whitelist=["zfs", "args", "systemd", "path", "platform"]
)
zpool_obj = {zpool: {"__opts__": opts, "__utils__": utils}}
zpool_obj = {
zpool: {
"__opts__": minion_opts,
"__utils__": utils,
},
}
return zpool_obj

13
tests/pytests/unit/output/test_highstate.py
View file

@ -4,8 +4,8 @@ import sys
import pytest
import salt.config
import salt.output.highstate as highstate
import salt.utils.stringutils
from tests.support.mock import patch
from tests.support.unit import skipIf
@ -13,15 +13,8 @@ log = logging.getLogger(__name__)
@pytest.fixture
def configure_loader_modules():
minion_opts = salt.config.DEFAULT_MINION_OPTS.copy()
overrides = {
"extension_modules": "",
"optimization_order": [0, 1, 2],
"color": False,
"state_output_pct": True,
}
minion_opts.update(overrides)
def configure_loader_modules(minion_opts):
minion_opts.update({"color": False, "state_output_pct": True})
return {highstate: {"__opts__": minion_opts}}

9
tests/pytests/unit/renderers/test_gpg.py
View file

@ -4,18 +4,18 @@ from textwrap import dedent
import pytest
import salt.config
import salt.renderers.gpg as gpg
from salt.exceptions import SaltRenderError
from tests.support.mock import MagicMock, Mock, call, patch
@pytest.fixture
def configure_loader_modules():
def configure_loader_modules(minion_opts):
"""
GPG renderer configuration
"""
return {gpg: {"__opts__": {"gpg_decrypt_must_succeed": True}}}
minion_opts["gpg_decrypt_must_succeed"] = True
return {gpg: {"__opts__": minion_opts}}
def test__get_gpg_exec():
@ -256,7 +256,7 @@ def test_render_without_cache():
popen_mock.assert_has_calls([gpg_call] * 3)
def test_render_with_cache():
def test_render_with_cache(minion_opts):
key_dir = "/etc/salt/gpgkeys"
secret = "Use more salt."
expected = "\n".join([secret] * 3)
@ -274,7 +274,6 @@ def test_render_with_cache():
"""
)
minion_opts = salt.config.DEFAULT_MINION_OPTS.copy()
minion_opts["gpg_cache"] = True
with patch.dict(gpg.__opts__, minion_opts):
with patch("salt.renderers.gpg.Popen") as popen_mock:

6
tests/pytests/unit/spm/test_spm.py
View file

@ -3,7 +3,6 @@ import shutil
import pytest
import salt.config
import salt.spm
import salt.utils.files
from tests.support.mock import MagicMock, patch
@ -61,8 +60,8 @@ class SPMTestUserInterface(salt.spm.SPMUserInterface):
@pytest.fixture()
def setup_spm(tmp_path):
minion_config = salt.config.DEFAULT_MINION_OPTS.copy()
def setup_spm(tmp_path, minion_opts):
minion_config = minion_opts.copy()
minion_config.update(
{
"spm_logfile": str(tmp_path / "log"),
@ -90,7 +89,6 @@ def setup_spm(tmp_path):
)
ui = SPMTestUserInterface()
client = salt.spm.SPMClient(ui, minion_config)
minion_opts = salt.config.DEFAULT_MINION_OPTS.copy()
return tmp_path, ui, client, minion_config, minion_opts

80
tests/pytests/unit/state/test_state_compiler.py
View file

@ -4,7 +4,7 @@
import logging
import pytest # pylint: disable=unused-import
import pytest
import salt.exceptions
import salt.state
@ -34,7 +34,7 @@ def test_format_log_non_ascii_character():
@pytest.mark.slow_test
def test_render_error_on_invalid_requisite():
def test_render_error_on_invalid_requisite(minion_opts):
"""
Test that the state compiler correctly deliver a rendering
exception when a requisite cannot be resolved
@ -74,14 +74,13 @@ def test_render_error_on_invalid_requisite():
]
)
}
minion_opts = salt.config.DEFAULT_MINION_OPTS.copy()
minion_opts["pillar"] = {"git": OrderedDict([("test1", "test")])}
state_obj = salt.state.State(minion_opts)
with pytest.raises(salt.exceptions.SaltRenderError):
state_obj.call_high(high_data)
def test_verify_onlyif_parse():
def test_verify_onlyif_parse(minion_opts):
low_data = {
"onlyif": [{"fun": "test.arg", "args": ["arg1", "arg2"]}],
"name": "mysql-server-5.7",
@ -100,13 +99,12 @@ def test_verify_onlyif_parse():
expected_result[key] = low_data.get(key)
with patch("salt.state.State._gather_pillar"):
minion_opts = salt.config.DEFAULT_MINION_OPTS.copy()
state_obj = salt.state.State(minion_opts)
return_result = state_obj._run_check_onlyif(low_data, "")
assert expected_result == return_result
def test_verify_onlyif_parse_deep_return():
def test_verify_onlyif_parse_deep_return(minion_opts):
low_data = {
"state": "test",
"name": "foo",
@ -128,13 +126,12 @@ def test_verify_onlyif_parse_deep_return():
expected_result[key] = low_data.get(key)
with patch("salt.state.State._gather_pillar"):
minion_opts = salt.config.DEFAULT_MINION_OPTS.copy()
state_obj = salt.state.State(minion_opts)
return_result = state_obj._run_check_onlyif(low_data, "")
assert expected_result == return_result
def test_verify_onlyif_cmd_error():
def test_verify_onlyif_cmd_error(minion_opts):
"""
Simulates a failure in cmd.retcode from onlyif
This could occur if runas is specified with a user that does not exist
@ -159,7 +156,6 @@ def test_verify_onlyif_cmd_error():
expected_result[key] = low_data.get(key)
with patch("salt.state.State._gather_pillar"):
minion_opts = salt.config.DEFAULT_MINION_OPTS.copy()
state_obj = salt.state.State(minion_opts)
mock = MagicMock(side_effect=CommandExecutionError("Boom!"))
with patch.dict(state_obj.functions, {"cmd.retcode": mock}):
@ -170,7 +166,7 @@ def test_verify_onlyif_cmd_error():
assert expected_result == return_result
def test_verify_unless_cmd_error():
def test_verify_unless_cmd_error(minion_opts):
"""
Simulates a failure in cmd.retcode from unless
This could occur if runas is specified with a user that does not exist
@ -195,7 +191,6 @@ def test_verify_unless_cmd_error():
expected_result[key] = low_data.get(key)
with patch("salt.state.State._gather_pillar"):
minion_opts = salt.config.DEFAULT_MINION_OPTS.copy()
state_obj = salt.state.State(minion_opts)
mock = MagicMock(side_effect=CommandExecutionError("Boom!"))
with patch.dict(state_obj.functions, {"cmd.retcode": mock}):
@ -206,7 +201,7 @@ def test_verify_unless_cmd_error():
assert expected_result == return_result
def test_verify_unless_list_cmd():
def test_verify_unless_list_cmd(minion_opts):
"""
If any of the unless commands return False (non 0) then the state should
run (no skip_watch).
@ -228,13 +223,12 @@ def test_verify_unless_list_cmd():
for key in ("__sls__", "__id__", "name"):
expected_result[key] = low_data.get(key)
with patch("salt.state.State._gather_pillar"):
minion_opts = salt.config.DEFAULT_MINION_OPTS.copy()
state_obj = salt.state.State(minion_opts)
return_result = state_obj._run_check_unless(low_data, {})
assert expected_result == return_result
def test_verify_unless_list_cmd_different_order():
def test_verify_unless_list_cmd_different_order(minion_opts):
"""
If any of the unless commands return False (non 0) then the state should
run (no skip_watch). The order shouldn't matter.
@ -256,13 +250,12 @@ def test_verify_unless_list_cmd_different_order():
for key in ("__sls__", "__id__", "name"):
expected_result[key] = low_data.get(key)
with patch("salt.state.State._gather_pillar"):
minion_opts = salt.config.DEFAULT_MINION_OPTS.copy()
state_obj = salt.state.State(minion_opts)
return_result = state_obj._run_check_unless(low_data, {})
assert expected_result == return_result
def test_verify_onlyif_list_cmd_different_order():
def test_verify_onlyif_list_cmd_different_order(minion_opts):
low_data = {
"state": "cmd",
"name": 'echo "something"',
@ -281,13 +274,12 @@ def test_verify_onlyif_list_cmd_different_order():
for key in ("__sls__", "__id__", "name"):
expected_result[key] = low_data.get(key)
with patch("salt.state.State._gather_pillar"):
minion_opts = salt.config.DEFAULT_MINION_OPTS.copy()
state_obj = salt.state.State(minion_opts)
return_result = state_obj._run_check_onlyif(low_data, {})
assert expected_result == return_result
def test_verify_unless_list_cmd_valid():
def test_verify_unless_list_cmd_valid(minion_opts):
"""
If any of the unless commands return False (non 0) then the state should
run (no skip_watch). This tests all commands return False.
@ -306,13 +298,12 @@ def test_verify_unless_list_cmd_valid():
for key in ("__sls__", "__id__", "name"):
expected_result[key] = low_data.get(key)
with patch("salt.state.State._gather_pillar"):
minion_opts = salt.config.DEFAULT_MINION_OPTS.copy()
state_obj = salt.state.State(minion_opts)
return_result = state_obj._run_check_unless(low_data, {})
assert expected_result == return_result
def test_verify_onlyif_list_cmd_valid():
def test_verify_onlyif_list_cmd_valid(minion_opts):
low_data = {
"state": "cmd",
"name": 'echo "something"',
@ -327,13 +318,12 @@ def test_verify_onlyif_list_cmd_valid():
for key in ("__sls__", "__id__", "name"):
expected_result[key] = low_data.get(key)
with patch("salt.state.State._gather_pillar"):
minion_opts = salt.config.DEFAULT_MINION_OPTS.copy()
state_obj = salt.state.State(minion_opts)
return_result = state_obj._run_check_onlyif(low_data, {})
assert expected_result == return_result
def test_verify_unless_list_cmd_invalid():
def test_verify_unless_list_cmd_invalid(minion_opts):
"""
If any of the unless commands return False (non 0) then the state should
run (no skip_watch). This tests all commands return True
@ -356,13 +346,12 @@ def test_verify_unless_list_cmd_invalid():
for key in ("__sls__", "__id__", "name"):
expected_result[key] = low_data.get(key)
with patch("salt.state.State._gather_pillar"):
minion_opts = salt.config.DEFAULT_MINION_OPTS.copy()
state_obj = salt.state.State(minion_opts)
return_result = state_obj._run_check_unless(low_data, {})
assert expected_result == return_result
def test_verify_onlyif_list_cmd_invalid():
def test_verify_onlyif_list_cmd_invalid(minion_opts):
low_data = {
"state": "cmd",
"name": 'echo "something"',
@ -381,13 +370,12 @@ def test_verify_onlyif_list_cmd_invalid():
for key in ("__sls__", "__id__", "name"):
expected_result[key] = low_data.get(key)
with patch("salt.state.State._gather_pillar"):
minion_opts = salt.config.DEFAULT_MINION_OPTS.copy()
state_obj = salt.state.State(minion_opts)
return_result = state_obj._run_check_onlyif(low_data, {})
assert expected_result == return_result
def test_verify_unless_parse():
def test_verify_unless_parse(minion_opts):
low_data = {
"unless": [{"fun": "test.arg", "args": ["arg1", "arg2"]}],
"name": "mysql-server-5.7",
@ -410,13 +398,12 @@ def test_verify_unless_parse():
expected_result[key] = low_data.get(key)
with patch("salt.state.State._gather_pillar"):
minion_opts = salt.config.DEFAULT_MINION_OPTS.copy()
state_obj = salt.state.State(minion_opts)
return_result = state_obj._run_check_unless(low_data, "")
assert expected_result == return_result
def test_verify_unless_parse_deep_return():
def test_verify_unless_parse_deep_return(minion_opts):
low_data = {
"state": "test",
"name": "foo",
@ -438,13 +425,12 @@ def test_verify_unless_parse_deep_return():
expected_result[key] = low_data.get(key)
with patch("salt.state.State._gather_pillar"):
minion_opts = salt.config.DEFAULT_MINION_OPTS.copy()
state_obj = salt.state.State(minion_opts)
return_result = state_obj._run_check_unless(low_data, "")
assert expected_result == return_result
def test_verify_creates():
def test_verify_creates(minion_opts):
low_data = {
"state": "cmd",
"name": 'echo "something"',
@ -457,7 +443,6 @@ def test_verify_creates():
}
with patch("salt.state.State._gather_pillar"):
minion_opts = salt.config.DEFAULT_MINION_OPTS.copy()
state_obj = salt.state.State(minion_opts)
with patch("os.path.exists") as path_mock:
path_mock.return_value = True
@ -482,7 +467,7 @@ def test_verify_creates():
assert expected_result == return_result
def test_verify_creates_list():
def test_verify_creates_list(minion_opts):
low_data = {
"state": "cmd",
"name": 'echo "something"',
@ -495,7 +480,6 @@ def test_verify_creates_list():
}
with patch("salt.state.State._gather_pillar"):
minion_opts = salt.config.DEFAULT_MINION_OPTS.copy()
state_obj = salt.state.State(minion_opts)
with patch("os.path.exists") as path_mock:
path_mock.return_value = True
@ -534,7 +518,7 @@ def _expand_win_path(path):
return path
def test_verify_onlyif_parse_slots(tmp_path):
def test_verify_onlyif_parse_slots(tmp_path, minion_opts):
name = str(tmp_path / "testfile.txt")
with salt.utils.files.fopen(name, "w") as fp:
fp.write("file-contents")
@ -561,13 +545,12 @@ def test_verify_onlyif_parse_slots(tmp_path):
for key in ("__sls__", "__id__", "name"):
expected_result[key] = low_data.get(key)
with patch("salt.state.State._gather_pillar"):
minion_opts = salt.config.DEFAULT_MINION_OPTS.copy()
state_obj = salt.state.State(minion_opts)
return_result = state_obj._run_check_onlyif(low_data, "")
assert expected_result == return_result
def test_verify_onlyif_list_cmd():
def test_verify_onlyif_list_cmd(minion_opts):
low_data = {
"state": "cmd",
"name": 'echo "something"',
@ -586,13 +569,12 @@ def test_verify_onlyif_list_cmd():
for key in ("__sls__", "__id__", "name"):
expected_result[key] = low_data.get(key)
with patch("salt.state.State._gather_pillar"):
minion_opts = salt.config.DEFAULT_MINION_OPTS.copy()
state_obj = salt.state.State(minion_opts)
return_result = state_obj._run_check_onlyif(low_data, {})
assert expected_result == return_result
def test_verify_onlyif_cmd_args():
def test_verify_onlyif_cmd_args(minion_opts):
"""
Verify cmd.run state arguments are properly passed to cmd.retcode in onlyif
"""
@ -617,7 +599,6 @@ def test_verify_onlyif_cmd_args():
}
with patch("salt.state.State._gather_pillar"):
minion_opts = salt.config.DEFAULT_MINION_OPTS.copy()
state_obj = salt.state.State(minion_opts)
mock = MagicMock()
with patch.dict(state_obj.functions, {"cmd.retcode": mock}):
@ -639,7 +620,7 @@ def test_verify_onlyif_cmd_args():
)
def test_verify_unless_parse_slots(tmp_path):
def test_verify_unless_parse_slots(tmp_path, minion_opts):
name = str(tmp_path / "testfile.txt")
with salt.utils.files.fopen(name, "w") as fp:
fp.write("file-contents")
@ -671,13 +652,12 @@ def test_verify_unless_parse_slots(tmp_path):
expected_result[key] = low_data.get(key)
with patch("salt.state.State._gather_pillar"):
minion_opts = salt.config.DEFAULT_MINION_OPTS.copy()
state_obj = salt.state.State(minion_opts)
return_result = state_obj._run_check_unless(low_data, "")
assert expected_result == return_result
def test_verify_retry_parsing():
def test_verify_retry_parsing(minion_opts):
low_data = {
"state": "file",
"name": "/tmp/saltstack.README.rst",
@ -706,7 +686,6 @@ def test_verify_retry_parsing():
expected_result[key] = low_data.get(key)
with patch("salt.state.State._gather_pillar"):
minion_opts = salt.config.DEFAULT_MINION_OPTS.copy()
minion_opts["test"] = True
minion_opts["file_client"] = "local"
state_obj = salt.state.State(minion_opts)
@ -719,7 +698,7 @@ def test_verify_retry_parsing():
assert set(expected_result).issubset(set(state_obj.call(low_data)))
def test_render_requisite_require_disabled(tmp_path):
def test_render_requisite_require_disabled(tmp_path, minion_opts):
"""
Test that the state compiler correctly deliver a rendering
exception when a requisite cannot be resolved
@ -749,7 +728,6 @@ def test_render_requisite_require_disabled(tmp_path):
},
}
minion_opts = salt.config.DEFAULT_MINION_OPTS.copy()
minion_opts["cachedir"] = str(tmp_path)
minion_opts["disabled_requisites"] = ["require"]
state_obj = salt.state.State(minion_opts)
@ -760,7 +738,7 @@ def test_render_requisite_require_disabled(tmp_path):
assert run_num == 0
def test_render_requisite_require_in_disabled(tmp_path):
def test_render_requisite_require_in_disabled(tmp_path, minion_opts):
"""
Test that the state compiler correctly deliver a rendering
exception when a requisite cannot be resolved
@ -795,7 +773,6 @@ def test_render_requisite_require_in_disabled(tmp_path):
),
}
minion_opts = salt.config.DEFAULT_MINION_OPTS.copy()
minion_opts["cachedir"] = str(tmp_path)
minion_opts["disabled_requisites"] = ["require_in"]
state_obj = salt.state.State(minion_opts)
@ -806,7 +783,7 @@ def test_render_requisite_require_in_disabled(tmp_path):
assert run_num == 0
def test_call_chunk_sub_state_run():
def test_call_chunk_sub_state_run(minion_opts):
"""
Test running a batch of states with an external runner
that returns sub_state_run
@ -840,7 +817,6 @@ def test_call_chunk_sub_state_run():
)
with patch("salt.state.State._gather_pillar"):
with patch("salt.state.State.call", return_value=mock_call_return):
minion_opts = salt.config.DEFAULT_MINION_OPTS.copy()
minion_opts["disabled_requisites"] = ["require"]
state_obj = salt.state.State(minion_opts)
ret = state_obj.call_chunk(low_data, {}, {})
@ -852,7 +828,7 @@ def test_call_chunk_sub_state_run():
assert sub_state["__sls__"] == "external"
def test_aggregate_requisites():
def test_aggregate_requisites(minion_opts):
"""
Test to ensure that the requisites are included in the aggregated low state.
"""
@ -960,7 +936,6 @@ def test_aggregate_requisites():
]
with patch("salt.state.State._gather_pillar"):
minion_opts = salt.config.DEFAULT_MINION_OPTS.copy()
state_obj = salt.state.State(minion_opts)
low_ret = state_obj._aggregate_requisites(low, chunks)
@ -975,7 +950,7 @@ def test_aggregate_requisites():
]
def test_mod_aggregate():
def test_mod_aggregate(minion_opts):
"""
Test to ensure that the requisites are included in the aggregated low state.
"""
@ -1046,7 +1021,6 @@ def test_mod_aggregate():
}
with patch("salt.state.State._gather_pillar"):
minion_opts = salt.config.DEFAULT_MINION_OPTS.copy()
state_obj = salt.state.State(minion_opts)
with patch.dict(
state_obj.states,

4
tests/pytests/unit/state/test_state_format_slots.py
View file

@ -16,10 +16,8 @@ log = logging.getLogger(__name__)
@pytest.fixture
def state_obj(tmp_path):
def state_obj(minion_opts):
with patch("salt.state.State._gather_pillar"):
minion_opts = salt.config.DEFAULT_MINION_OPTS.copy()
minion_opts["cachedir"] = str(tmp_path)
yield salt.state.State(minion_opts)

8
tests/pytests/unit/state/test_state_options.py
View file

@ -2,16 +2,15 @@ import itertools
import pytest
import salt.config
import salt.state
@pytest.fixture
def master_opts():
def master_opts(master_opts):
"""
Return a subset of master options to the minion
"""
opts = salt.config.DEFAULT_MASTER_OPTS.copy()
opts = master_opts.copy()
mopts = {}
mopts["file_roots"] = opts["file_roots"]
mopts["top_file_merging_strategy"] = opts["top_file_merging_strategy"]
@ -40,11 +39,10 @@ class MockBaseHighStateClient:
return self.opts
def test_state_aggregate_option_behavior(master_opts):
def test_state_aggregate_option_behavior(master_opts, minion_opts):
"""
Ensure state_aggregate can be overridden on the minion
"""
minion_opts = salt.config.DEFAULT_MINION_OPTS.copy()
possible = [None, True, False, ["pkg"]]
expected_result = [
True,

17
tests/pytests/unit/states/test_boto_cloudfront.py
View file

@ -6,28 +6,15 @@ import textwrap
import pytest
import salt.config
import salt.loader
import salt.states.boto_cloudfront as boto_cloudfront
from tests.support.mock import MagicMock, patch
@pytest.fixture
def setup_vars():
opts = salt.config.DEFAULT_MINION_OPTS.copy()
name = "my_distribution"
base_ret = {"name": name, "changes": {}}
# Most attributes elided since there are so many required ones
config = {"Enabled": True, "HttpVersion": "http2"}
tags = {"test_tag1": "value1"}
@pytest.fixture
def configure_loader_modules(setup_vars):
def configure_loader_modules(minion_opts):
utils = salt.loader.utils(
salt.config.DEFAULT_MINION_OPTS.copy(),
minion_opts,
whitelist=[
"boto3",
"dictdiffer",

16
tests/pytests/unit/states/test_boto_cloudtrail.py
View file

@ -4,7 +4,6 @@ import string
import pytest
import salt.config
import salt.loader
import salt.states.boto_cloudtrail as boto_cloudtrail
from tests.support.mock import MagicMock, patch
@ -64,21 +63,20 @@ def global_config():
@pytest.fixture
def configure_loader_modules():
opts = salt.config.DEFAULT_MINION_OPTS.copy()
opts["grains"] = salt.loader.grains(opts)
def configure_loader_modules(minion_opts):
minion_opts["grains"] = salt.loader.grains(minion_opts)
ctx = {}
utils = salt.loader.utils(
opts,
minion_opts,
whitelist=["boto", "boto3", "args", "systemd", "path", "platform", "reg"],
context=ctx,
)
serializers = salt.loader.serializers(opts)
serializers = salt.loader.serializers(minion_opts)
funcs = salt.loader.minion_mods(
opts, context=ctx, utils=utils, whitelist=["boto_cloudtrail"]
minion_opts, context=ctx, utils=utils, whitelist=["boto_cloudtrail"]
)
salt_states = salt.loader.states(
opts=opts,
opts=minion_opts,
functions=funcs,
utils=utils,
whitelist=["boto_cloudtrail"],
@ -86,7 +84,7 @@ def configure_loader_modules():
)
return {
boto_cloudtrail: {
"__opts__": opts,
"__opts__": minion_opts,
"__salt__": funcs,
"__utils__": utils,
"__states__": salt_states,

16
tests/pytests/unit/states/test_boto_cloudwatch_event.py
View file

@ -4,7 +4,6 @@ import string
import pytest
import salt.config
import salt.loader
import salt.states.boto_cloudwatch_event as boto_cloudwatch_event
from tests.support.mock import MagicMock, patch
@ -52,21 +51,20 @@ def global_config():
@pytest.fixture
def configure_loader_modules():
opts = salt.config.DEFAULT_MINION_OPTS.copy()
opts["grains"] = salt.loader.grains(opts)
def configure_loader_modules(minion_opts):
minion_opts["grains"] = salt.loader.grains(minion_opts)
ctx = {}
utils = salt.loader.utils(
opts,
minion_opts,
whitelist=["boto3", "args", "systemd", "path", "platform"],
context=ctx,
)
serializers = salt.loader.serializers(opts)
serializers = salt.loader.serializers(minion_opts)
funcs = funcs = salt.loader.minion_mods(
opts, context=ctx, utils=utils, whitelist=["boto_cloudwatch_event"]
minion_opts, context=ctx, utils=utils, whitelist=["boto_cloudwatch_event"]
)
salt_states = salt.loader.states(
opts=opts,
opts=minion_opts,
functions=funcs,
utils=utils,
whitelist=["boto_cloudwatch_event"],
@ -74,7 +72,7 @@ def configure_loader_modules():
)
return {
boto_cloudwatch_event: {
"__opts__": opts,
"__opts__": minion_opts,
"__salt__": funcs,
"__utils__": utils,
"__states__": salt_states,

15
tests/pytests/unit/states/test_boto_elasticsearch_domain.py
View file

@ -51,21 +51,20 @@ class GlobalConfig:
@pytest.fixture
def configure_loader_modules():
opts = salt.config.DEFAULT_MINION_OPTS.copy()
opts["grains"] = salt.loader.grains(opts)
def configure_loader_modules(minion_opts):
minion_opts["grains"] = salt.loader.grains(minion_opts)
ctx = {}
utils = salt.loader.utils(
opts,
minion_opts,
whitelist=["boto3", "args", "systemd", "path", "platform"],
context=ctx,
)
serializers = salt.loader.serializers(opts)
serializers = salt.loader.serializers(minion_opts)
funcs = funcs = salt.loader.minion_mods(
opts, context=ctx, utils=utils, whitelist=["boto_elasticsearch_domain"]
minion_opts, context=ctx, utils=utils, whitelist=["boto_elasticsearch_domain"]
)
salt_states = salt.loader.states(
opts=opts,
opts=minion_opts,
functions=funcs,
utils=utils,
whitelist=["boto_elasticsearch_domain"],
@ -73,7 +72,7 @@ def configure_loader_modules():
)
return {
boto_elasticsearch_domain: {
"__opts__": opts,
"__opts__": minion_opts,
"__salt__": funcs,
"__utils__": utils,
"__states__": salt_states,

15
tests/pytests/unit/states/test_boto_iot.py
View file

@ -95,21 +95,20 @@ class GlobalConfig:
@pytest.fixture
def configure_loader_modules():
opts = salt.config.DEFAULT_MINION_OPTS.copy()
opts["grains"] = salt.loader.grains(opts)
def configure_loader_modules(minion_opts):
minion_opts["grains"] = salt.loader.grains(minion_opts)
ctx = {}
utils = salt.loader.utils(
opts,
minion_opts,
whitelist=["boto3", "args", "systemd", "path", "platform"],
context=ctx,
)
serializers = salt.loader.serializers(opts)
serializers = salt.loader.serializers(minion_opts)
funcs = funcs = salt.loader.minion_mods(
opts, context=ctx, utils=utils, whitelist=["boto_iot"]
minion_opts, context=ctx, utils=utils, whitelist=["boto_iot"]
)
salt_states = salt.loader.states(
opts=opts,
opts=minion_opts,
functions=funcs,
utils=utils,
whitelist=["boto_iot"],
@ -117,7 +116,7 @@ def configure_loader_modules():
)
return {
boto_iot: {
"__opts__": opts,
"__opts__": minion_opts,
"__salt__": funcs,
"__utils__": utils,
"__states__": salt_states,

15
tests/pytests/unit/states/test_boto_lambda.py
View file

@ -73,21 +73,20 @@ def global_config():
@pytest.fixture
def configure_loader_modules():
opts = salt.config.DEFAULT_MINION_OPTS.copy()
opts["grains"] = salt.loader.grains(opts)
def configure_loader_modules(minion_opts):
minion_opts["grains"] = salt.loader.grains(minion_opts)
ctx = {}
utils = salt.loader.utils(
opts,
minion_opts,
whitelist=["boto", "boto3", "args", "systemd", "path", "platform", "reg"],
context=ctx,
)
serializers = salt.loader.serializers(opts)
serializers = salt.loader.serializers(minion_opts)
funcs = salt.loader.minion_mods(
opts, context=ctx, utils=utils, whitelist=["boto_lambda"]
minion_opts, context=ctx, utils=utils, whitelist=["boto_lambda"]
)
salt_states = salt.loader.states(
opts=opts,
opts=minion_opts,
functions=funcs,
utils=utils,
whitelist=["boto_lambda"],
@ -95,7 +94,7 @@ def configure_loader_modules():
)
return {
boto_lambda: {
"__opts__": opts,
"__opts__": minion_opts,
"__salt__": funcs,
"__utils__": utils,
"__states__": salt_states,

13
tests/pytests/unit/states/test_boto_s3_bucket.py
View file

@ -180,20 +180,19 @@ def global_config():
@pytest.fixture
def configure_loader_modules():
opts = salt.config.DEFAULT_MINION_OPTS.copy()
def configure_loader_modules(minion_opts):
ctx = {}
utils = salt.loader.utils(
opts,
minion_opts,
whitelist=["boto", "boto3", "args", "systemd", "path", "platform", "reg"],
context=ctx,
)
serializers = salt.loader.serializers(opts)
serializers = salt.loader.serializers(minion_opts)
funcs = salt.loader.minion_mods(
opts, context=ctx, utils=utils, whitelist=["boto_s3_bucket"]
minion_opts, context=ctx, utils=utils, whitelist=["boto_s3_bucket"]
)
salt_states = salt.loader.states(
opts=opts,
opts=minion_opts,
functions=funcs,
utils=utils,
whitelist=["boto_s3_bucket"],
@ -201,7 +200,7 @@ def configure_loader_modules():
)
return {
boto_s3_bucket: {
"__opts__": opts,
"__opts__": minion_opts,
"__salt__": funcs,
"__utils__": utils,
"__states__": salt_states,

5
tests/pytests/unit/states/test_boto_sqs.py
View file

@ -12,10 +12,9 @@ from tests.support.mock import MagicMock, patch
@pytest.fixture
def configure_loader_modules():
opts = salt.config.DEFAULT_MINION_OPTS.copy()
def configure_loader_modules(minion_opts):
utils = salt.loader.utils(
opts,
minion_opts,
whitelist=["boto3", "yaml", "args", "systemd", "path", "platform"],
context={},
)

19
tests/pytests/unit/states/test_service.py
View file

@ -679,21 +679,22 @@ def test_mod_beacon(tmp_path):
@pytest.mark.skip_on_darwin(reason="service.running is currently failing on OSX")
@pytest.mark.skip_if_not_root
@pytest.mark.destructive_test
@pytest.mark.slow_test
def test_running_with_reload():
def test_running_with_reload(minion_opts):
"""
Test that a running service is properly reloaded
"""
opts = salt.config.DEFAULT_MINION_OPTS.copy()
opts["grains"] = salt.loader.grains(opts)
utils = salt.loader.utils(opts)
modules = salt.loader.minion_mods(opts, utils=utils)
# TODO: This is not a unit test, it interacts with the system. Move to functional.
minion_opts["grains"] = salt.loader.grains(minion_opts)
utils = salt.loader.utils(minion_opts)
modules = salt.loader.minion_mods(minion_opts, utils=utils)
service_name = "cron"
cmd_name = "crontab"
os_family = opts["grains"]["os_family"]
os_release = opts["grains"]["osrelease"]
os_family = minion_opts["grains"]["os_family"]
os_release = minion_opts["grains"]["osrelease"]
if os_family == "RedHat":
service_name = "crond"
elif os_family == "Arch":
@ -718,8 +719,8 @@ def test_running_with_reload():
post_srv_disable = True
try:
with patch.dict(service.__grains__, opts["grains"]), patch.dict(
service.__opts__, opts
with patch.dict(service.__grains__, minion_opts["grains"]), patch.dict(
service.__opts__, minion_opts
), patch.dict(service.__salt__, modules), patch.dict(
service.__utils__, utils
), patch.dict(

9
tests/pytests/unit/states/test_win_lgpo.py
View file

@ -14,13 +14,12 @@ from tests.support.mock import patch
@pytest.fixture
def configure_loader_modules():
opts = salt.config.DEFAULT_MINION_OPTS.copy()
utils = salt.loader.utils(opts)
modules = salt.loader.minion_mods(opts, utils=utils)
def configure_loader_modules(minion_opts):
utils = salt.loader.utils(minion_opts)
modules = salt.loader.minion_mods(minion_opts, utils=utils)
return {
win_lgpo: {
"__opts__": copy.deepcopy(opts),
"__opts__": copy.deepcopy(minion_opts),
"__salt__": modules,
"__utils__": utils,
}

11
tests/pytests/unit/states/test_zfs.py
View file

@ -25,11 +25,14 @@ def utils_patch():
@pytest.fixture
def configure_loader_modules():
opts = salt.config.DEFAULT_MINION_OPTS.copy()
utils = salt.loader.utils(opts, whitelist=["zfs"])
def configure_loader_modules(minion_opts):
utils = salt.loader.utils(minion_opts, whitelist=["zfs"])
zfs_obj = {
zfs: {"__opts__": opts, "__grains__": {"kernel": "SunOS"}, "__utils__": utils}
zfs: {
"__opts__": minion_opts,
"__grains__": {"kernel": "SunOS"},
"__utils__": utils,
},
}
return zfs_obj

7
tests/pytests/unit/states/test_zpool.py
View file

@ -25,12 +25,11 @@ def utils_patch():
@pytest.fixture
def configure_loader_modules():
opts = salt.config.DEFAULT_MINION_OPTS.copy()
utils = salt.loader.utils(opts, whitelist=["zfs"])
def configure_loader_modules(minion_opts):
utils = salt.loader.utils(minion_opts, whitelist=["zfs"])
zpool_obj = {
zpool: {
"__opts__": opts,
"__opts__": minion_opts,
"__grains__": {"kernel": "SunOS"},
"__utils__": utils,
}

49
tests/pytests/unit/test_beacons.py
View file

@ -4,22 +4,21 @@ unit tests for the beacon_module parameter
import logging
import salt.config
import salt.beacons
from tests.support.mock import MagicMock, call, patch
log = logging.getLogger(__name__)
def test_beacon_process():
def test_beacon_process(minion_opts):
"""
Test the process function in the beacon class
returns the correct information when an exception
occurs
"""
mock_opts = salt.config.DEFAULT_MINION_OPTS.copy()
mock_opts["id"] = "minion"
mock_opts["__role"] = "minion"
mock_opts["beacons"] = {
minion_opts["id"] = "minion"
minion_opts["__role"] = "minion"
minion_opts["beacons"] = {
"watch_apache": [
{"processes": {"apache2": "stopped"}},
{"beacon_module": "ps"},
@ -28,11 +27,11 @@ def test_beacon_process():
beacon_mock = MagicMock(side_effect=Exception("Global Thermonuclear War"))
beacon_mock.__globals__ = {}
beacon = salt.beacons.Beacon(mock_opts, [])
beacon = salt.beacons.Beacon(minion_opts, [])
found = "ps.beacon" in beacon.beacons
beacon.beacons["ps.beacon"] = beacon_mock
ret = beacon.process(mock_opts["beacons"], mock_opts["grains"])
ret = beacon.process(minion_opts["beacons"], minion_opts["grains"])
_expected = [
{
@ -45,37 +44,36 @@ def test_beacon_process():
assert ret == _expected
def test_beacon_process_invalid():
def test_beacon_process_invalid(minion_opts):
"""
Test the process function in the beacon class
when the configuration is invalid.
"""
mock_opts = salt.config.DEFAULT_MINION_OPTS.copy()
mock_opts["id"] = "minion"
mock_opts["__role"] = "minion"
minion_opts["id"] = "minion"
minion_opts["__role"] = "minion"
mock_opts["beacons"] = {"status": {}}
minion_opts["beacons"] = {"status": {}}
beacon = salt.beacons.Beacon(mock_opts, [])
beacon = salt.beacons.Beacon(minion_opts, [])
with patch.object(salt.beacons, "log") as log_mock, patch.object(
salt.beacons.log, "error"
) as log_error_mock:
ret = beacon.process(mock_opts["beacons"], mock_opts["grains"])
ret = beacon.process(minion_opts["beacons"], minion_opts["grains"])
log_error_mock.assert_called_with(
"Beacon %s configuration invalid, not running.\n%s",
"status",
"Configuration for status beacon must be a list.",
)
mock_opts["beacons"] = {"mybeacon": {}}
minion_opts["beacons"] = {"mybeacon": {}}
beacon = salt.beacons.Beacon(mock_opts, [])
beacon = salt.beacons.Beacon(minion_opts, [])
with patch.object(salt.beacons.log, "warning") as log_warn_mock, patch.object(
salt.beacons.log, "error"
) as log_error_mock:
ret = beacon.process(mock_opts["beacons"], mock_opts["grains"])
ret = beacon.process(minion_opts["beacons"], minion_opts["grains"])
log_warn_mock.assert_called_with(
"No validate function found for %s, running basic beacon validation.",
"mybeacon",
@ -83,21 +81,20 @@ def test_beacon_process_invalid():
log_error_mock.assert_called_with("Configuration for beacon must be a list.")
def test_beacon_module():
def test_beacon_module(minion_opts):
"""
Test that beacon_module parameter for beacon configuration
"""
mock_opts = salt.config.DEFAULT_MINION_OPTS.copy()
mock_opts["id"] = "minion"
mock_opts["__role"] = "minion"
mock_opts["beacons"] = {
minion_opts["id"] = "minion"
minion_opts["__role"] = "minion"
minion_opts["beacons"] = {
"watch_apache": [
{"processes": {"apache2": "stopped"}},
{"beacon_module": "ps"},
]
}
beacon = salt.beacons.Beacon(mock_opts, [])
ret = beacon.process(mock_opts["beacons"], mock_opts["grains"])
beacon = salt.beacons.Beacon(minion_opts, [])
ret = beacon.process(minion_opts["beacons"], minion_opts["grains"])
_expected = [
{
@ -122,5 +119,5 @@ def test_beacon_module():
)
]
with patch.object(beacon, "beacons", mocked) as patched:
beacon.process(mock_opts["beacons"], mock_opts["grains"])
beacon.process(minion_opts["beacons"], minion_opts["grains"])
patched[name].assert_has_calls(calls)

19
tests/pytests/unit/test_client.py
View file

@ -20,14 +20,7 @@ from tests.support.mock import MagicMock, patch
log = logging.getLogger(__name__)
@pytest.fixture
def master_config():
opts = salt.config.DEFAULT_MASTER_OPTS.copy()
opts["__role"] = "master"
return opts
def test_job_result_return_success(master_config):
def test_job_result_return_success(master_opts):
"""
Should return the `expected_return`, since there is a job with the right jid.
"""
@ -35,7 +28,7 @@ def test_job_result_return_success(master_config):
jid = "0815"
raw_return = {"id": "fake-id", "jid": jid, "data": "", "return": "fake-return"}
expected_return = {"fake-id": {"ret": "fake-return"}}
with client.LocalClient(mopts=master_config) as local_client:
with client.LocalClient(mopts=master_opts) as local_client:
local_client.event.get_event = MagicMock(return_value=raw_return)
local_client.returners = MagicMock()
ret = local_client.get_event_iter_returns(jid, minions)
@ -43,7 +36,7 @@ def test_job_result_return_success(master_config):
assert val == expected_return
def test_job_result_return_failure(master_config):
def test_job_result_return_failure(master_opts):
"""
We are _not_ getting a job return, because the jid is different. Instead we should
get a StopIteration exception.
@ -56,7 +49,7 @@ def test_job_result_return_failure(master_config):
"data": "",
"return": "fake-return",
}
with client.LocalClient(mopts=master_config) as local_client:
with client.LocalClient(mopts=master_opts) as local_client:
local_client.event.get_event = MagicMock()
local_client.event.get_event.side_effect = [raw_return, None]
local_client.returners = MagicMock()
@ -65,8 +58,8 @@ def test_job_result_return_failure(master_config):
next(ret)
def test_create_local_client(master_config):
with client.LocalClient(mopts=master_config) as local_client:
def test_create_local_client(master_opts):
with client.LocalClient(mopts=master_opts) as local_client:
assert isinstance(
local_client, client.LocalClient
), "LocalClient did not create a LocalClient instance"

5
tests/pytests/unit/utils/event/test_event_return.py
View file

@ -1,17 +1,16 @@
import pytest
from pytestshellutils.utils.processes import terminate_process
import salt.config
import salt.ext.tornado.ioloop
import salt.utils.event
import salt.utils.stringutils
@pytest.mark.slow_test
def test_event_return():
def test_event_return(master_opts):
evt = None
try:
evt = salt.utils.event.EventReturn(salt.config.DEFAULT_MASTER_OPTS.copy())
evt = salt.utils.event.EventReturn(master_opts)
evt.start()
except TypeError as exc:
if "object" in str(exc):

8
tests/pytests/unit/utils/jinja/test_custom_extensions.py
View file

@ -11,7 +11,6 @@ import re
import pytest
from jinja2 import DictLoader, Environment, exceptions
import salt.config
import salt.loader
# dateutils is needed so that the strftime jinja filter is loaded
@ -36,9 +35,8 @@ except ImportError:
@pytest.fixture
def minion_opts(tmp_path):
_opts = salt.config.DEFAULT_MINION_OPTS.copy()
_opts.update(
def minion_opts(tmp_path, minion_opts):
minion_opts.update(
{
"cachedir": str(tmp_path / "jinja-template-cache"),
"file_buffer_size": 1048576,
@ -54,7 +52,7 @@ def minion_opts(tmp_path):
),
}
)
return _opts
return minion_opts
@pytest.fixture()

17
tests/pytests/unit/utils/jinja/test_dot_notation_looup.py
View file

@ -2,7 +2,6 @@
Tests for salt.utils.jinja
"""
import salt.config
import salt.loader
# dateutils is needed so that the strftime jinja filter is loaded
@ -15,14 +14,12 @@ from salt.utils.jinja import SaltCacheLoader
from tests.support.mock import Mock, patch
def render(tmpl_str, context=None):
def render(tmpl_str, minion_opts, context=None):
functions = {
"mocktest.ping": lambda: True,
"mockgrains.get": lambda x: "jerry",
}
minion_opts = salt.config.DEFAULT_MINION_OPTS.copy()
_render = salt.loader.render(minion_opts, functions)
jinja = _render.get("jinja")
@ -30,29 +27,29 @@ def render(tmpl_str, context=None):
return jinja(tmpl_str, context=context or {}, argline="-s").read()
def test_normlookup():
def test_normlookup(minion_opts):
"""
Sanity-check the normal dictionary-lookup syntax for our stub function
"""
tmpl_str = """Hello, {{ salt['mocktest.ping']() }}."""
with patch.object(SaltCacheLoader, "file_client", Mock()):
ret = render(tmpl_str)
ret = render(tmpl_str, minion_opts)
assert ret == "Hello, True."
def test_dotlookup():
def test_dotlookup(minion_opts):
"""
Check calling a stub function using awesome dot-notation
"""
tmpl_str = """Hello, {{ salt.mocktest.ping() }}."""
with patch.object(SaltCacheLoader, "file_client", Mock()):
ret = render(tmpl_str)
ret = render(tmpl_str, minion_opts)
assert ret == "Hello, True."
def test_shadowed_dict_method():
def test_shadowed_dict_method(minion_opts):
"""
Check calling a stub function with a name that shadows a ``dict``
method name
@ -60,5 +57,5 @@ def test_shadowed_dict_method():
tmpl_str = """Hello, {{ salt.mockgrains.get('id') }}."""
with patch.object(SaltCacheLoader, "file_client", Mock()):
ret = render(tmpl_str)
ret = render(tmpl_str, minion_opts)
assert ret == "Hello, jerry."

8
tests/pytests/unit/utils/jinja/test_get_template.py
View file

@ -9,7 +9,6 @@ import os
import pytest
import salt.config
import salt.loader
# dateutils is needed so that the strftime jinja filter is loaded
@ -51,9 +50,8 @@ class MockFileClient:
@pytest.fixture
def minion_opts(tmp_path):
_opts = salt.config.DEFAULT_MINION_OPTS.copy()
_opts.update(
def minion_opts(tmp_path, minion_opts):
minion_opts.update(
{
"cachedir": str(tmp_path),
"file_buffer_size": 1048576,
@ -69,7 +67,7 @@ def minion_opts(tmp_path):
),
}
)
return _opts
return minion_opts
@pytest.fixture

18
tests/pytests/unit/utils/jinja/test_jinja_default_options.py
View file

@ -6,22 +6,18 @@ import os
import pytest
import salt.config
import salt.loader
# dateutils is needed so that the strftime jinja filter is loaded
import salt.utils.dateutils # pylint: disable=unused-import
import salt.utils.files
import salt.utils.json
import salt.utils.stringutils
import salt.utils.yaml
import salt.utils.files # pylint: disable=unused-import
import salt.utils.json # pylint: disable=unused-import
import salt.utils.stringutils # pylint: disable=unused-import
import salt.utils.yaml # pylint: disable=unused-import
from salt.utils.templates import render_jinja_tmpl
@pytest.fixture
def minion_opts(tmp_path):
_opts = salt.config.DEFAULT_MINION_OPTS.copy()
_opts.update(
def minion_opts(tmp_path, minion_opts):
minion_opts.update(
{
"cachedir": str(tmp_path / "jinja-template-cache"),
"file_buffer_size": 1048576,
@ -38,7 +34,7 @@ def minion_opts(tmp_path):
"jinja_env": {"line_comment_prefix": "##", "line_statement_prefix": "%"},
}
)
return _opts
return minion_opts
@pytest.fixture

18
tests/pytests/unit/utils/jinja/test_salt_cache_loader.py
View file

@ -8,23 +8,19 @@ import os
import pytest
from jinja2 import Environment, exceptions
import salt.config
import salt.loader
# dateutils is needed so that the strftime jinja filter is loaded
import salt.utils.dateutils # pylint: disable=unused-import
import salt.utils.files
import salt.utils.json
import salt.utils.stringutils
import salt.utils.yaml
import salt.utils.files # pylint: disable=unused-import
import salt.utils.json # pylint: disable=unused-import
import salt.utils.stringutils # pylint: disable=unused-import
import salt.utils.yaml # pylint: disable=unused-import
from salt.utils.jinja import SaltCacheLoader
from tests.support.mock import Mock, patch
@pytest.fixture
def minion_opts(tmp_path):
_opts = salt.config.DEFAULT_MINION_OPTS.copy()
_opts.update(
def minion_opts(tmp_path, minion_opts):
minion_opts.update(
{
"file_buffer_size": 1048576,
"cachedir": str(tmp_path),
@ -35,7 +31,7 @@ def minion_opts(tmp_path):
),
}
)
return _opts
return minion_opts
@pytest.fixture

8
tests/pytests/unit/utils/templates/test_jinja.py
View file

@ -5,15 +5,13 @@ import os
import re
import pytest
import salt.config
from salt.exceptions import SaltRenderError
from salt.utils.templates import render_jinja_tmpl
@pytest.fixture
def minion_opts(tmp_path):
_opts = salt.config.DEFAULT_MINION_OPTS.copy()
_opts.update(
def minion_opts(tmp_path, minion_opts):
minion_opts.update(
{
"cachedir": str(tmp_path / "jinja-template-cache"),
"file_buffer_size": 1048576,
@ -29,7 +27,7 @@ def minion_opts(tmp_path):
),
}
)
return _opts
return minion_opts
@pytest.fixture

37
tests/pytests/unit/utils/test_cache.py
View file

@ -5,11 +5,11 @@
Test the salt cache objects
"""
import pathlib
import time
import pytest
import salt.config
import salt.loader
import salt.payload
import salt.utils.cache as cache
@ -47,24 +47,15 @@ def test_ttl():
@pytest.fixture
def cache_dir(tmp_path):
cachedir = tmp_path / "cachedir"
cachedir.mkdir()
return cachedir
def cache_dir(minion_opts):
return pathlib.Path(minion_opts["cachedir"])
@pytest.fixture
def minion_config(cache_dir):
opts = salt.config.DEFAULT_MINION_OPTS.copy()
opts["cachedir"] = str(cache_dir)
return opts
def test_smoke_context(minion_config):
def test_smoke_context(minion_opts):
"""
Smoke test the context cache
"""
context_cache = cache.ContextCache(minion_config, "cache_test")
context_cache = cache.ContextCache(minion_opts, "cache_test")
data = {"a": "b"}
context_cache.cache_context(data.copy())
@ -106,7 +97,7 @@ def cache_mods_path(tmp_path, cache_mod_name):
yield _cache_mods_path
def test_context_wrapper(minion_config, cache_mods_path):
def test_context_wrapper(minion_opts, cache_mods_path):
"""
Test to ensure that a module which decorates itself
with a context cache can store and retrieve its contextual
@ -117,7 +108,7 @@ def test_context_wrapper(minion_config, cache_mods_path):
[str(cache_mods_path)],
tag="rawmodule",
virtual_enable=False,
opts=minion_config,
opts=minion_opts,
)
cache_test_func = loader["cache_mod.test_context_module"]
@ -126,7 +117,7 @@ def test_context_wrapper(minion_config, cache_mods_path):
assert cache_test_func()["called"] == 1
def test_set_cache(minion_config, cache_mods_path, cache_mod_name, cache_dir):
def test_set_cache(minion_opts, cache_mods_path, cache_mod_name, cache_dir):
"""
Tests to ensure the cache is written correctly
"""
@ -136,8 +127,8 @@ def test_set_cache(minion_config, cache_mods_path, cache_mod_name, cache_dir):
[str(cache_mods_path)],
tag="rawmodule",
virtual_enable=False,
opts=minion_config,
pack={"__context__": context, "__opts__": minion_config},
opts=minion_opts,
pack={"__context__": context, "__opts__": minion_opts},
)
cache_test_func = loader["cache_mod.test_context_module"]
@ -160,13 +151,13 @@ def test_set_cache(minion_config, cache_mods_path, cache_mod_name, cache_dir):
# Test cache de-serialize
cc = cache.ContextCache(
minion_config, "salt.loaded.ext.rawmodule.{}".format(cache_mod_name)
minion_opts, "salt.loaded.ext.rawmodule.{}".format(cache_mod_name)
)
retrieved_cache = cc.get_cache_context()
assert retrieved_cache == dict(context, called=1)
def test_refill_cache(minion_config, cache_mods_path):
def test_refill_cache(minion_opts, cache_mods_path):
"""
Tests to ensure that the context cache can rehydrate a wrapped function
"""
@ -175,8 +166,8 @@ def test_refill_cache(minion_config, cache_mods_path):
[str(cache_mods_path)],
tag="rawmodule",
virtual_enable=False,
opts=minion_config,
pack={"__context__": context, "__opts__": minion_config},
opts=minion_opts,
pack={"__context__": context, "__opts__": minion_opts},
)
cache_test_func = loader["cache_mod.test_compare_context"]

23
tests/pytests/unit/utils/test_pyobjects.py
View file

@ -2,7 +2,6 @@ import logging
import pytest
import salt.config
import salt.renderers.pyobjects as pyobjects
from salt.utils.odict import OrderedDict
from tests.support.mock import MagicMock
@ -10,28 +9,14 @@ from tests.support.mock import MagicMock
log = logging.getLogger(__name__)
@pytest.fixture
def cache_dir(tmp_path):
cachedir = tmp_path / "cachedir"
cachedir.mkdir()
return cachedir
@pytest.fixture
def minion_config(cache_dir):
opts = salt.config.DEFAULT_MINION_OPTS.copy()
opts["cachedir"] = str(cache_dir)
opts["file_client"] = "local"
opts["id"] = "testminion"
return opts
@pytest.fixture()
def configure_loader_modules(minion_config):
def configure_loader_modules(minion_opts):
minion_opts["file_client"] = "local"
minion_opts["id"] = "testminion"
pillar = MagicMock(return_value={})
return {
pyobjects: {
"__opts__": minion_config,
"__opts__": minion_opts,
"__pillar__": pillar,
"__salt__": {
"config.get": MagicMock(),

30
tests/pytests/unit/utils/test_reactor.py
View file

@ -6,31 +6,17 @@ from tests.support.mock import MagicMock, call, patch
@pytest.fixture
def minion_config():
opts = salt.config.DEFAULT_MINION_OPTS.copy()
opts["__role"] = "minion"
return opts
def minion_reactor(minion_opts):
return reactor.Reactor(minion_opts)
@pytest.fixture
def master_config():
opts = salt.config.DEFAULT_MASTER_OPTS.copy()
opts["__role"] = "master"
return opts
@pytest.fixture
def minion_reactor(minion_config):
return reactor.Reactor(minion_config)
@pytest.fixture
def master_reactor(master_config):
return reactor.Reactor(master_config)
def master_reactor(master_opts):
return reactor.Reactor(master_opts)
@pytest.mark.skip_on_windows(reason="Reactors unavailable on Windows")
def test_run_minion_reactor(minion_config, minion_reactor):
def test_run_minion_reactor(minion_opts, minion_reactor):
"""
Ensure that list_reactors() returns the correct list of reactor SLS
files for each tag.
@ -44,14 +30,14 @@ def test_run_minion_reactor(minion_config, minion_reactor):
assert os_nice_mock.mock_calls == []
# if reactor_niceness is set, os.nice should be called with that value
with patch.dict(minion_config, {"reactor_niceness": 9}):
with patch.dict(minion_opts, {"reactor_niceness": 9}):
minion_reactor.run()
calls = [call(9)]
os_nice_mock.assert_has_calls(calls)
@pytest.mark.skip_on_windows(reason="Reactors unavailable on Windows")
def test_run_master_reactor(master_config, master_reactor):
def test_run_master_reactor(master_opts, master_reactor):
"""
Ensure that list_reactors() returns the correct list of reactor SLS
files for each tag.
@ -65,7 +51,7 @@ def test_run_master_reactor(master_config, master_reactor):
assert os_nice_mock.mock_calls == []
# if reactor_niceness is set, os.nice should be called with that value
with patch.dict(master_config, {"reactor_niceness": 9}):
with patch.dict(master_opts, {"reactor_niceness": 9}):
master_reactor.run()
calls = [call(9)]
os_nice_mock.assert_has_calls(calls)