# -*- coding: utf-8 -*- """ Integration tests for the saltutil module. """ from __future__ import absolute_import, print_function, unicode_literals import os import shutil import textwrap import time import pytest import salt.utils.files import salt.utils.stringutils from tests.support.case import ModuleCase from tests.support.runtests import RUNTIME_VARS from tests.support.unit import skipIf @pytest.mark.windows_whitelisted @pytest.mark.usefixtures("salt_sub_minion") class SaltUtilModuleTest(ModuleCase): """ Testcase for the saltutil execution module """ @classmethod def setUpClass(cls): # Whell functions, on a minion, must run with the master running # along side the minion. # We copy the master config to the minion's configuration directory just # for this test since the test suite master and minion(s) do not share the # same configuration directory src = os.path.join(RUNTIME_VARS.TMP_CONF_DIR, "master") dst = os.path.join(RUNTIME_VARS.TMP_MINION_CONF_DIR, "master") shutil.copyfile(src, dst) cls.copied_master_config_file = dst @classmethod def tearDownClass(cls): if os.path.exists(cls.copied_master_config_file): os.unlink(cls.copied_master_config_file) cls.copied_master_config_file = None def setUp(self): self.run_function("saltutil.refresh_pillar") # Tests for the wheel function @pytest.mark.slow_test(seconds=30) # Test takes >10 and <=30 seconds def test_wheel_just_function(self): """ Tests using the saltutil.wheel function when passing only a function. """ # Wait for the pillar refresh to kick in, so that grains are ready to go time.sleep(3) ret = self.run_function("saltutil.wheel", ["minions.connected"]) self.assertIn("minion", ret["return"]) self.assertIn("sub_minion", ret["return"]) @pytest.mark.slow_test(seconds=30) # Test takes >10 and <=30 seconds def test_wheel_with_arg(self): """ Tests using the saltutil.wheel function when passing a function and an arg. """ ret = self.run_function("saltutil.wheel", ["key.list", "minion"]) self.assertEqual(ret["return"], {}) @pytest.mark.slow_test(seconds=30) # Test takes >10 and <=30 seconds def test_wheel_no_arg_raise_error(self): """ Tests using the saltutil.wheel function when passing a function that requires an arg, but one isn't supplied. """ self.assertRaises(TypeError, "saltutil.wheel", ["key.list"]) @pytest.mark.slow_test(seconds=30) # Test takes >10 and <=30 seconds def test_wheel_with_kwarg(self): """ Tests using the saltutil.wheel function when passing a function and a kwarg. This function just generates a key pair, but doesn't do anything with it. We just need this for testing purposes. """ ret = self.run_function("saltutil.wheel", ["key.gen"], keysize=1024) self.assertIn("pub", ret["return"]) self.assertIn("priv", ret["return"]) @pytest.mark.windows_whitelisted class SyncGrainsTest(ModuleCase): @pytest.mark.slow_test(seconds=30) # Test takes >10 and <=30 seconds def test_sync_grains(self): ret = self.run_function("saltutil.sync_grains") self.assertEqual(ret, []) @pytest.mark.windows_whitelisted class SaltUtilSyncModuleTest(ModuleCase): """ Testcase for the saltutil sync execution module """ def setUp(self): whitelist = { "modules": [], } self.run_function("saltutil.sync_all", extmod_whitelist=whitelist) def tearDown(self): self.run_function("saltutil.sync_all") @pytest.mark.slow_test(seconds=60) # Test takes >30 and <=60 seconds def test_sync_all(self): """ Test syncing all ModuleCase """ expected_return = { "engines": [], "clouds": [], "grains": [], "beacons": [], "utils": [], "returners": [], "modules": [ "modules.depends_versioned", "modules.depends_versionless", "modules.mantest", "modules.override_test", "modules.runtests_decorators", "modules.runtests_helpers", "modules.salttest", ], "renderers": [], "log_handlers": [], "matchers": [], "states": [], "sdb": [], "proxymodules": [], "executors": [], "output": [], "thorium": [], "serializers": [], } ret = self.run_function("saltutil.sync_all") self.assertEqual(ret, expected_return) @pytest.mark.slow_test(seconds=60) # Test takes >30 and <=60 seconds def test_sync_all_whitelist(self): """ Test syncing all ModuleCase with whitelist """ expected_return = { "engines": [], "clouds": [], "grains": [], "beacons": [], "utils": [], "returners": [], "modules": ["modules.salttest"], "renderers": [], "log_handlers": [], "matchers": [], "states": [], "sdb": [], "proxymodules": [], "executors": [], "output": [], "thorium": [], "serializers": [], } ret = self.run_function( "saltutil.sync_all", extmod_whitelist={"modules": ["salttest"]} ) self.assertEqual(ret, expected_return) @pytest.mark.slow_test(seconds=60) # Test takes >30 and <=60 seconds def test_sync_all_blacklist(self): """ Test syncing all ModuleCase with blacklist """ expected_return = { "engines": [], "clouds": [], "grains": [], "beacons": [], "utils": [], "returners": [], "modules": [ "modules.mantest", "modules.override_test", "modules.runtests_helpers", "modules.salttest", ], "renderers": [], "log_handlers": [], "matchers": [], "states": [], "sdb": [], "proxymodules": [], "executors": [], "output": [], "thorium": [], "serializers": [], } ret = self.run_function( "saltutil.sync_all", extmod_blacklist={ "modules": [ "runtests_decorators", "depends_versioned", "depends_versionless", ] }, ) self.assertEqual(ret, expected_return) @pytest.mark.slow_test(seconds=60) # Test takes >30 and <=60 seconds def test_sync_all_blacklist_and_whitelist(self): """ Test syncing all ModuleCase with whitelist and blacklist """ expected_return = { "engines": [], "clouds": [], "grains": [], "beacons": [], "utils": [], "returners": [], "executors": [], "modules": [], "renderers": [], "log_handlers": [], "matchers": [], "states": [], "sdb": [], "proxymodules": [], "output": [], "thorium": [], "serializers": [], } ret = self.run_function( "saltutil.sync_all", extmod_whitelist={"modules": ["runtests_decorators"]}, extmod_blacklist={"modules": ["runtests_decorators"]}, ) self.assertEqual(ret, expected_return) @skipIf(True, "Pillar refresh test is flaky. Skipping for now.") @pytest.mark.windows_whitelisted class SaltUtilSyncPillarTest(ModuleCase): """ Testcase for the saltutil sync pillar module """ @pytest.mark.flaky(max_runs=4) def test_pillar_refresh(self): """ test pillar refresh module """ pillar_key = "itworked" pre_pillar = self.run_function("pillar.raw") self.assertNotIn(pillar_key, pre_pillar.get(pillar_key, "didnotwork")) with salt.utils.files.fopen( os.path.join(RUNTIME_VARS.TMP_PILLAR_TREE, "add_pillar.sls"), "w" ) as fp: fp.write(salt.utils.stringutils.to_str("{0}: itworked".format(pillar_key))) with salt.utils.files.fopen( os.path.join(RUNTIME_VARS.TMP_PILLAR_TREE, "top.sls"), "w" ) as fp: fp.write( textwrap.dedent( """\ base: '*': - add_pillar """ ) ) self.run_function("saltutil.refresh_pillar") pillar = False timeout = time.time() + 30 while not pillar: post_pillar = self.run_function("pillar.raw") try: self.assertIn(pillar_key, post_pillar.get(pillar_key, "didnotwork")) pillar = True except AssertionError: if time.time() > timeout: self.assertIn(pillar_key, post_pillar.get(pillar_key, "didnotwork")) continue post_pillar = self.run_function("pillar.raw") self.assertIn(pillar_key, post_pillar.get(pillar_key, "didnotwork")) def tearDown(self): for filename in os.listdir(RUNTIME_VARS.TMP_PILLAR_TREE): os.remove(os.path.join(RUNTIME_VARS.TMP_PILLAR_TREE, filename))