mirror of
https://github.com/valitydev/salt.git
synced 2024-11-07 17:09:03 +00:00
3273bbdab7
Conflicts: - doc/ref/configuration/master.rst - doc/ref/modules/all/index.rst - doc/topics/grains/index.rst - doc/topics/releases/2016.3.4.rst - doc/topics/spm/spm_formula.rst - doc/topics/tutorials/cron.rst - doc/topics/tutorials/index.rst - doc/topics/tutorials/stormpath.rst - salt/engines/slack.py - salt/log/handlers/fluent_mod.py - salt/modules/cyg.py - salt/modules/junos.py - salt/modules/namecheap_dns.py - salt/modules/namecheap_domains.py - salt/modules/namecheap_ns.py - salt/modules/namecheap_ssl.py - salt/modules/namecheap_users.py - salt/modules/reg.py - salt/modules/tomcat.py - salt/modules/vault.py - salt/modules/win_file.py - salt/modules/zpool.py - salt/output/highstate.py - salt/renderers/pass.py - salt/runners/cache.py - salt/states/boto_apigateway.py - salt/states/boto_iam.py - salt/states/boto_route53.py - salt/states/msteams.py - salt/states/reg.py - salt/states/win_iis.py - tests/integration/modules/test_cmdmod.py - tests/integration/states/test_user.py - tests/support/helpers.py - tests/unit/cloud/clouds/test_openstack.py - tests/unit/fileserver/test_gitfs.py - tests/unit/modules/test_junos.py - tests/unit/pillar/test_git.py - tests/unit/states/test_win_path.py - tests/unit/test_pillar.py - tests/unit/utils/test_format_call.py - tests/unit/utils/test_utils.py - tests/unit/utils/test_warnings.py
1182 lines
46 KiB
Python
1182 lines
46 KiB
Python
# -*- coding: utf-8 -*-
|
|
'''
|
|
:codeauthor: Rahul Handay <rahulha@saltstack.com>
|
|
'''
|
|
|
|
# Import Python libs
|
|
from __future__ import absolute_import, print_function, unicode_literals
|
|
import os
|
|
|
|
# Import Salt Testing Libs
|
|
from tests.support.mixins import LoaderModuleMockMixin
|
|
from tests.support.unit import TestCase, skipIf
|
|
from tests.support.mock import (
|
|
MagicMock,
|
|
patch,
|
|
mock_open,
|
|
NO_MOCK,
|
|
NO_MOCK_REASON
|
|
)
|
|
|
|
# Import Salt Libs
|
|
import salt.config
|
|
import salt.loader
|
|
import salt.utils.hashutils
|
|
import salt.utils.odict
|
|
import salt.utils.platform
|
|
import salt.utils.state
|
|
import salt.modules.state as state
|
|
from salt.exceptions import CommandExecutionError, SaltInvocationError
|
|
import salt.modules.config as config
|
|
from salt.ext import six
|
|
|
|
|
|
class MockState(object):
|
|
'''
|
|
Mock class
|
|
'''
|
|
def __init__(self):
|
|
pass
|
|
|
|
class State(object):
|
|
'''
|
|
Mock state class
|
|
'''
|
|
flag = None
|
|
|
|
def __init__(self,
|
|
opts,
|
|
pillar_override=False,
|
|
pillar_enc=None,
|
|
initial_pillar=None):
|
|
pass
|
|
|
|
def verify_data(self, data):
|
|
'''
|
|
Mock verify_data method
|
|
'''
|
|
data = data
|
|
if self.flag:
|
|
return True
|
|
else:
|
|
return False
|
|
|
|
@staticmethod
|
|
def call(data):
|
|
'''
|
|
Mock call method
|
|
'''
|
|
data = data
|
|
return list
|
|
|
|
@staticmethod
|
|
def call_high(data, orchestration_jid=None):
|
|
'''
|
|
Mock call_high method
|
|
'''
|
|
data = data
|
|
return True
|
|
|
|
@staticmethod
|
|
def call_template_str(data):
|
|
'''
|
|
Mock call_template_str method
|
|
'''
|
|
data = data
|
|
return True
|
|
|
|
@staticmethod
|
|
def _mod_init(data):
|
|
'''
|
|
Mock _mod_init method
|
|
'''
|
|
data = data
|
|
return True
|
|
|
|
def verify_high(self, data):
|
|
'''
|
|
Mock verify_high method
|
|
'''
|
|
data = data
|
|
if self.flag:
|
|
return True
|
|
else:
|
|
return -1
|
|
|
|
@staticmethod
|
|
def compile_high_data(data):
|
|
'''
|
|
Mock compile_high_data
|
|
'''
|
|
data = data
|
|
return [{"__id__": "ABC"}]
|
|
|
|
@staticmethod
|
|
def call_chunk(data, data1, data2):
|
|
'''
|
|
Mock call_chunk method
|
|
'''
|
|
data = data
|
|
data1 = data1
|
|
data2 = data2
|
|
return {'': 'ABC'}
|
|
|
|
@staticmethod
|
|
def call_chunks(data):
|
|
'''
|
|
Mock call_chunks method
|
|
'''
|
|
data = data
|
|
return True
|
|
|
|
@staticmethod
|
|
def call_listen(data, ret):
|
|
'''
|
|
Mock call_listen method
|
|
'''
|
|
data = data
|
|
ret = ret
|
|
return True
|
|
|
|
def requisite_in(self, data): # pylint: disable=unused-argument
|
|
return data, []
|
|
|
|
class HighState(object):
|
|
'''
|
|
Mock HighState class
|
|
'''
|
|
flag = False
|
|
opts = {'state_top': '',
|
|
'pillar': {}}
|
|
|
|
def __init__(self, opts, pillar_override=None, *args, **kwargs):
|
|
self.building_highstate = salt.utils.odict.OrderedDict
|
|
self.state = MockState.State(opts,
|
|
pillar_override=pillar_override)
|
|
|
|
def render_state(self, sls, saltenv, mods, matches, local=False):
|
|
'''
|
|
Mock render_state method
|
|
'''
|
|
sls = sls
|
|
saltenv = saltenv
|
|
mods = mods
|
|
matches = matches
|
|
local = local
|
|
if self.flag:
|
|
return {}, True
|
|
else:
|
|
return {}, False
|
|
|
|
@staticmethod
|
|
def get_top():
|
|
'''
|
|
Mock get_top method
|
|
'''
|
|
return "_top"
|
|
|
|
def verify_tops(self, data):
|
|
'''
|
|
Mock verify_tops method
|
|
'''
|
|
data = data
|
|
if self.flag:
|
|
return ["a", "b"]
|
|
else:
|
|
return []
|
|
|
|
@staticmethod
|
|
def top_matches(data):
|
|
'''
|
|
Mock top_matches method
|
|
'''
|
|
data = data
|
|
return ["a", "b", "c"]
|
|
|
|
@staticmethod
|
|
def push_active():
|
|
'''
|
|
Mock push_active method
|
|
'''
|
|
return True
|
|
|
|
@staticmethod
|
|
def compile_highstate():
|
|
'''
|
|
Mock compile_highstate method
|
|
'''
|
|
return "A"
|
|
|
|
@staticmethod
|
|
def compile_state_usage():
|
|
'''
|
|
Mock compile_state_usage method
|
|
'''
|
|
return "A"
|
|
|
|
@staticmethod
|
|
def pop_active():
|
|
'''
|
|
Mock pop_active method
|
|
'''
|
|
return True
|
|
|
|
@staticmethod
|
|
def compile_low_chunks():
|
|
'''
|
|
Mock compile_low_chunks method
|
|
'''
|
|
return True
|
|
|
|
def render_highstate(self, data):
|
|
'''
|
|
Mock render_highstate method
|
|
'''
|
|
data = data
|
|
if self.flag:
|
|
return ["a", "b"], True
|
|
else:
|
|
return ["a", "b"], False
|
|
|
|
@staticmethod
|
|
def call_highstate(exclude, cache, cache_name, force=None,
|
|
whitelist=None, orchestration_jid=None):
|
|
'''
|
|
Mock call_highstate method
|
|
'''
|
|
exclude = exclude
|
|
cache = cache
|
|
cache_name = cache_name
|
|
force = force
|
|
whitelist = whitelist
|
|
return True
|
|
|
|
|
|
class MockSerial(object):
|
|
'''
|
|
Mock Class
|
|
'''
|
|
def __init__(self):
|
|
pass
|
|
|
|
class Serial(object):
|
|
'''
|
|
Mock Serial class
|
|
'''
|
|
def __init__(self, data):
|
|
data = data
|
|
|
|
@staticmethod
|
|
def load(data):
|
|
'''
|
|
Mock load method
|
|
'''
|
|
data = data
|
|
return {"A": "B"}
|
|
|
|
@staticmethod
|
|
def dump(data, data1):
|
|
'''
|
|
Mock dump method
|
|
'''
|
|
data = data
|
|
data1 = data1
|
|
return True
|
|
|
|
|
|
class MockTarFile(object):
|
|
'''
|
|
Mock tarfile class
|
|
'''
|
|
path = os.sep + "tmp"
|
|
|
|
def __init__(self):
|
|
pass
|
|
|
|
@staticmethod
|
|
def open(data, data1):
|
|
'''
|
|
Mock open method
|
|
'''
|
|
data = data
|
|
data1 = data1
|
|
return MockTarFile
|
|
|
|
@staticmethod
|
|
def getmembers():
|
|
'''
|
|
Mock getmembers method
|
|
'''
|
|
return [MockTarFile]
|
|
|
|
@staticmethod
|
|
def extractall(data):
|
|
'''
|
|
Mock extractall method
|
|
'''
|
|
data = data
|
|
return True
|
|
|
|
@staticmethod
|
|
def close():
|
|
'''
|
|
Mock close method
|
|
'''
|
|
return True
|
|
|
|
|
|
@skipIf(NO_MOCK, NO_MOCK_REASON)
|
|
class StateTestCase(TestCase, LoaderModuleMockMixin):
|
|
'''
|
|
Test case for salt.modules.state
|
|
'''
|
|
|
|
def setup_loader_modules(self):
|
|
utils = salt.loader.utils(
|
|
salt.config.DEFAULT_MINION_OPTS,
|
|
whitelist=['state']
|
|
)
|
|
utils.keys()
|
|
patcher = patch('salt.modules.state.salt.state', MockState())
|
|
patcher.start()
|
|
self.addCleanup(patcher.stop)
|
|
return {
|
|
state: {
|
|
'__opts__': {
|
|
'cachedir': '/D',
|
|
'saltenv': None,
|
|
'__cli': 'salt',
|
|
},
|
|
'__utils__': utils,
|
|
'__salt__': {
|
|
'config.get': config.get,
|
|
}
|
|
},
|
|
config: {
|
|
'__opts__': {},
|
|
'__pillar__': {},
|
|
},
|
|
|
|
}
|
|
|
|
def test_running(self):
|
|
'''
|
|
Test of checking i fthe state function is already running
|
|
'''
|
|
self.assertEqual(state.running(True), [])
|
|
|
|
mock = MagicMock(side_effect=[[{"fun": "state.running", "pid": "4126",
|
|
"jid": "20150325123407204096"}], []])
|
|
with patch.dict(state.__salt__,
|
|
{'saltutil.is_running': mock}
|
|
):
|
|
self.assertListEqual(state.running(),
|
|
['The function "state.running"'
|
|
' is running as PID 4126 and '
|
|
'was started at 2015, Mar 25 12:34:07.'
|
|
'204096 with jid 20150325123407204096'])
|
|
|
|
self.assertListEqual(state.running(), [])
|
|
|
|
def test_low(self):
|
|
'''
|
|
Test of executing a single low data call
|
|
'''
|
|
mock = MagicMock(side_effect=[False, None, None])
|
|
with patch.object(state, '_check_queue', mock):
|
|
self.assertFalse(state.low({"state": "pkg", "fun": "installed",
|
|
"name": "vi"}))
|
|
|
|
MockState.State.flag = False
|
|
self.assertEqual(state.low({"state": "pkg", "fun": "installed",
|
|
"name": "vi"}), list)
|
|
|
|
MockState.State.flag = True
|
|
self.assertTrue(state.low({"state": "pkg", "fun": "installed",
|
|
"name": "vi"}))
|
|
|
|
def test_high(self):
|
|
'''
|
|
Test for checking the state system
|
|
'''
|
|
mock = MagicMock(side_effect=[False, None])
|
|
with patch.object(state, '_check_queue', mock):
|
|
self.assertFalse(state.high({"vim": {"pkg": ["installed"]}}))
|
|
|
|
mock = MagicMock(return_value={"test": True})
|
|
with patch.object(salt.utils.state, 'get_sls_opts', mock):
|
|
self.assertTrue(state.high({"vim": {"pkg": ["installed"]}}))
|
|
|
|
def test_template(self):
|
|
'''
|
|
Test of executing the information
|
|
stored in a template file on the minion
|
|
'''
|
|
mock = MagicMock(side_effect=[False, None, None])
|
|
with patch.object(state, '_check_queue', mock):
|
|
self.assertFalse(state.template('/home/salt/salt.sls'))
|
|
|
|
MockState.HighState.flag = True
|
|
self.assertTrue(state.template('/home/salt/salt.sls'))
|
|
|
|
MockState.HighState.flag = False
|
|
self.assertTrue(state.template('/home/salt/salt.sls'))
|
|
|
|
def test_template_str(self):
|
|
'''
|
|
Test for Executing the information
|
|
stored in a string from an sls template
|
|
'''
|
|
mock = MagicMock(side_effect=[False, None])
|
|
with patch.object(state, '_check_queue', mock):
|
|
self.assertFalse(state.template_str('Template String'))
|
|
|
|
self.assertTrue(state.template_str('Template String'))
|
|
|
|
def test_apply_(self):
|
|
'''
|
|
Test to apply states
|
|
'''
|
|
mock = MagicMock(return_value=True)
|
|
with patch.object(state, 'sls', mock):
|
|
self.assertTrue(state.apply_(True))
|
|
|
|
with patch.object(state, 'highstate', mock):
|
|
self.assertTrue(state.apply_(None))
|
|
|
|
def test_list_disabled(self):
|
|
'''
|
|
Test to list disabled states
|
|
'''
|
|
mock = MagicMock(return_value=["A", "B", "C"])
|
|
with patch.dict(state.__salt__, {'grains.get': mock}):
|
|
self.assertListEqual(state.list_disabled(), ["A", "B", "C"])
|
|
|
|
def test_enable(self):
|
|
'''
|
|
Test to Enable state function or sls run
|
|
'''
|
|
mock = MagicMock(return_value=["A", "B"])
|
|
with patch.dict(state.__salt__, {'grains.get': mock}):
|
|
mock = MagicMock(return_value=[])
|
|
with patch.dict(state.__salt__, {'grains.setval': mock}):
|
|
mock = MagicMock(return_value=[])
|
|
with patch.dict(state.__salt__, {'saltutil.refresh_modules':
|
|
mock}):
|
|
self.assertDictEqual(state.enable("A"),
|
|
{'msg': 'Info: A state enabled.',
|
|
'res': True})
|
|
|
|
self.assertDictEqual(state.enable("Z"),
|
|
{'msg': 'Info: Z state already '
|
|
'enabled.', 'res': True})
|
|
|
|
def test_disable(self):
|
|
'''
|
|
Test to disable state run
|
|
'''
|
|
mock = MagicMock(return_value=["C", "D"])
|
|
with patch.dict(state.__salt__, {'grains.get': mock}):
|
|
mock = MagicMock(return_value=[])
|
|
with patch.dict(state.__salt__, {'grains.setval': mock}):
|
|
mock = MagicMock(return_value=[])
|
|
with patch.dict(state.__salt__, {'saltutil.refresh_modules':
|
|
mock}):
|
|
self.assertDictEqual(state.disable("C"),
|
|
{'msg': 'Info: C state '
|
|
'already disabled.',
|
|
'res': True})
|
|
|
|
self.assertDictEqual(state.disable("Z"),
|
|
{'msg': 'Info: Z state '
|
|
'disabled.', 'res': True})
|
|
|
|
def test_clear_cache(self):
|
|
'''
|
|
Test to clear out cached state file
|
|
'''
|
|
mock = MagicMock(return_value=["A.cache.p", "B.cache.p", "C"])
|
|
with patch.object(os, 'listdir', mock):
|
|
mock = MagicMock(return_value=True)
|
|
with patch.object(os.path, 'isfile', mock):
|
|
mock = MagicMock(return_value=True)
|
|
with patch.object(os, 'remove', mock):
|
|
self.assertEqual(state.clear_cache(),
|
|
['A.cache.p',
|
|
'B.cache.p'])
|
|
|
|
def test_single(self):
|
|
'''
|
|
Test to execute single state function
|
|
'''
|
|
ret = {'pkg_|-name=vim_|-name=vim_|-installed': list}
|
|
mock = MagicMock(side_effect=["A", None, None, None, None])
|
|
with patch.object(state, '_check_queue', mock):
|
|
self.assertEqual(state.single("pkg.installed",
|
|
" name=vim"), "A")
|
|
|
|
self.assertEqual(state.single("pk", "name=vim"),
|
|
"Invalid function passed")
|
|
|
|
with patch.dict(state.__opts__, {"test": "install"}):
|
|
mock = MagicMock(return_value={"test": ""})
|
|
with patch.object(salt.utils.state, 'get_sls_opts', mock):
|
|
mock = MagicMock(return_value=True)
|
|
with patch.object(salt.utils, 'test_mode', mock):
|
|
self.assertRaises(SaltInvocationError,
|
|
state.single,
|
|
"pkg.installed",
|
|
"name=vim",
|
|
pillar="A")
|
|
|
|
MockState.State.flag = True
|
|
self.assertTrue(state.single("pkg.installed",
|
|
"name=vim"))
|
|
|
|
MockState.State.flag = False
|
|
self.assertDictEqual(state.single("pkg.installed",
|
|
"name=vim"), ret)
|
|
|
|
def test_show_top(self):
|
|
'''
|
|
Test to return the top data that the minion will use for a highstate
|
|
'''
|
|
mock = MagicMock(side_effect=["A", None, None])
|
|
with patch.object(state, '_check_queue', mock):
|
|
self.assertEqual(state.show_top(), "A")
|
|
|
|
MockState.HighState.flag = True
|
|
self.assertListEqual(state.show_top(), ['a', 'b'])
|
|
|
|
MockState.HighState.flag = False
|
|
self.assertListEqual(state.show_top(), ['a', 'b', 'c'])
|
|
|
|
def test_run_request(self):
|
|
'''
|
|
Test to Execute the pending state request
|
|
'''
|
|
mock = MagicMock(side_effect=[{},
|
|
{"name": "A"},
|
|
{"name": {'mods': "A",
|
|
'kwargs': {}}}])
|
|
with patch.object(state, 'check_request', mock):
|
|
self.assertDictEqual(state.run_request("A"), {})
|
|
|
|
self.assertDictEqual(state.run_request("A"), {})
|
|
|
|
mock = MagicMock(return_value=["True"])
|
|
with patch.object(state, 'apply_', mock):
|
|
mock = MagicMock(return_value="")
|
|
with patch.object(os, 'remove', mock):
|
|
self.assertListEqual(state.run_request("name"),
|
|
["True"])
|
|
|
|
def test_show_highstate(self):
|
|
'''
|
|
Test to retrieve the highstate data from the salt master
|
|
'''
|
|
mock = MagicMock(side_effect=["A", None, None])
|
|
with patch.object(state, '_check_queue', mock):
|
|
self.assertEqual(state.show_highstate(), "A")
|
|
|
|
self.assertRaises(SaltInvocationError,
|
|
state.show_highstate,
|
|
pillar="A")
|
|
|
|
self.assertEqual(state.show_highstate(), "A")
|
|
|
|
def test_show_lowstate(self):
|
|
'''
|
|
Test to list out the low data that will be applied to this minion
|
|
'''
|
|
mock = MagicMock(side_effect=["A", None])
|
|
with patch.object(state, '_check_queue', mock):
|
|
self.assertRaises(AssertionError, state.show_lowstate)
|
|
|
|
self.assertTrue(state.show_lowstate())
|
|
|
|
def test_show_state_usage(self):
|
|
'''
|
|
Test to list out the state usage that will be applied to this minion
|
|
'''
|
|
|
|
mock = MagicMock(side_effect=["A", None, None])
|
|
with patch.object(state, '_check_queue', mock):
|
|
self.assertEqual(state.show_state_usage(), "A")
|
|
|
|
self.assertRaises(SaltInvocationError,
|
|
state.show_state_usage,
|
|
pillar="A")
|
|
|
|
self.assertEqual(state.show_state_usage(), "A")
|
|
|
|
def test_sls_id(self):
|
|
'''
|
|
Test to call a single ID from the
|
|
named module(s) and handle all requisites
|
|
'''
|
|
mock = MagicMock(side_effect=["A", None, None, None])
|
|
with patch.object(state, '_check_queue', mock):
|
|
self.assertEqual(state.sls_id("apache", "http"), "A")
|
|
|
|
with patch.dict(state.__opts__, {"test": "A"}):
|
|
mock = MagicMock(
|
|
return_value={'test': True,
|
|
'saltenv': None}
|
|
)
|
|
with patch.object(salt.utils.state, 'get_sls_opts', mock):
|
|
mock = MagicMock(return_value=True)
|
|
with patch.object(salt.utils, 'test_mode', mock):
|
|
MockState.State.flag = True
|
|
MockState.HighState.flag = True
|
|
self.assertEqual(state.sls_id("apache", "http"), 2)
|
|
|
|
MockState.State.flag = False
|
|
self.assertDictEqual(state.sls_id("ABC", "http"),
|
|
{'': 'ABC'})
|
|
self.assertRaises(SaltInvocationError,
|
|
state.sls_id,
|
|
"DEF", "http")
|
|
|
|
def test_show_low_sls(self):
|
|
'''
|
|
Test to display the low data from a specific sls
|
|
'''
|
|
mock = MagicMock(side_effect=["A", None, None])
|
|
with patch.object(state, '_check_queue', mock):
|
|
self.assertEqual(state.show_low_sls("foo"), "A")
|
|
|
|
with patch.dict(state.__opts__, {"test": "A"}):
|
|
mock = MagicMock(
|
|
return_value={'test': True,
|
|
'saltenv': None}
|
|
)
|
|
with patch.object(salt.utils.state, 'get_sls_opts', mock):
|
|
MockState.State.flag = True
|
|
MockState.HighState.flag = True
|
|
self.assertEqual(state.show_low_sls("foo"), 2)
|
|
|
|
MockState.State.flag = False
|
|
self.assertListEqual(state.show_low_sls("foo"),
|
|
[{'__id__': 'ABC'}])
|
|
|
|
def test_show_sls(self):
|
|
'''
|
|
Test to display the state data from a specific sls
|
|
'''
|
|
mock = MagicMock(side_effect=["A", None, None, None])
|
|
with patch.object(state, '_check_queue', mock):
|
|
self.assertEqual(state.show_sls("foo"), "A")
|
|
|
|
with patch.dict(state.__opts__, {"test": "A"}):
|
|
mock = MagicMock(
|
|
return_value={'test': True,
|
|
'saltenv': None}
|
|
)
|
|
with patch.object(salt.utils.state, 'get_sls_opts', mock):
|
|
mock = MagicMock(return_value=True)
|
|
with patch.object(salt.utils, 'test_mode', mock):
|
|
self.assertRaises(SaltInvocationError,
|
|
state.show_sls,
|
|
"foo",
|
|
pillar="A")
|
|
|
|
MockState.State.flag = True
|
|
self.assertEqual(state.show_sls("foo"), 2)
|
|
|
|
MockState.State.flag = False
|
|
self.assertListEqual(state.show_sls("foo"),
|
|
['a', 'b'])
|
|
|
|
def test_top(self):
|
|
'''
|
|
Test to execute a specific top file
|
|
'''
|
|
ret = ['Pillar failed to render with the following messages:', 'E']
|
|
mock = MagicMock(side_effect=["A", None, None, None])
|
|
with patch.object(state, '_check_queue', mock):
|
|
self.assertEqual(state.top("reverse_top.sls"), "A")
|
|
|
|
mock = MagicMock(side_effect=[['E'], None, None])
|
|
with patch.object(state, '_get_pillar_errors', mock):
|
|
with patch.dict(state.__pillar__, {"_errors": ['E']}):
|
|
self.assertListEqual(state.top("reverse_top.sls"), ret)
|
|
|
|
with patch.dict(state.__opts__, {"test": "A"}):
|
|
mock = MagicMock(return_value={'test': True})
|
|
with patch.object(salt.utils.state, 'get_sls_opts', mock):
|
|
mock = MagicMock(return_value=True)
|
|
with patch.object(salt.utils, 'test_mode', mock):
|
|
self.assertRaises(SaltInvocationError,
|
|
state.top,
|
|
"reverse_top.sls",
|
|
pillar="A")
|
|
|
|
mock = MagicMock(
|
|
return_value
|
|
=
|
|
'salt://reverse_top.sls')
|
|
with patch.object(os.path, 'join', mock):
|
|
mock = MagicMock(return_value=True)
|
|
with patch.object(state, '_set_retcode',
|
|
mock):
|
|
self.assertTrue(
|
|
state.
|
|
top("reverse_top.sls "
|
|
"exclude=exclude.sls"))
|
|
|
|
def test_highstate(self):
|
|
'''
|
|
Test to retrieve the state data from the
|
|
salt master for the minion and execute it
|
|
'''
|
|
arg = "whitelist=sls1.sls"
|
|
mock = MagicMock(side_effect=[True, False, False, False])
|
|
with patch.object(state, '_disabled', mock):
|
|
self.assertDictEqual(state.highstate("whitelist=sls1.sls"),
|
|
{'comment': 'Disabled',
|
|
'name': 'Salt highstate run is disabled. '
|
|
'To re-enable, run state.enable highstate',
|
|
'result': 'False'})
|
|
|
|
mock = MagicMock(side_effect=["A", None, None])
|
|
with patch.object(state, '_check_queue', mock):
|
|
self.assertEqual(state.highstate("whitelist=sls1.sls"), "A")
|
|
|
|
with patch.dict(state.__opts__, {"test": "A"}):
|
|
mock = MagicMock(return_value={'test': True})
|
|
with patch.object(salt.utils.state, 'get_sls_opts', mock):
|
|
self.assertRaises(SaltInvocationError,
|
|
state.highstate,
|
|
"whitelist=sls1.sls",
|
|
pillar="A")
|
|
|
|
mock = MagicMock(return_value=True)
|
|
with patch.dict(state.__salt__,
|
|
{'config.option': mock}):
|
|
mock = MagicMock(return_value="A")
|
|
with patch.object(state, '_filter_running',
|
|
mock):
|
|
mock = MagicMock(return_value=True)
|
|
with patch.object(state, '_filter_running',
|
|
mock):
|
|
mock = MagicMock(return_value=True)
|
|
with patch.object(salt.payload, 'Serial',
|
|
mock):
|
|
with patch.object(os.path,
|
|
'join', mock):
|
|
with patch.object(
|
|
state,
|
|
'_set'
|
|
'_retcode',
|
|
mock):
|
|
self.assertTrue(state.
|
|
highstate
|
|
(arg))
|
|
|
|
def test_clear_request(self):
|
|
'''
|
|
Test to clear out the state execution request without executing it
|
|
'''
|
|
mock = MagicMock(return_value=True)
|
|
with patch.object(os.path, 'join', mock):
|
|
mock = MagicMock(return_value=True)
|
|
with patch.object(salt.payload, 'Serial', mock):
|
|
mock = MagicMock(side_effect=[False, True, True])
|
|
with patch.object(os.path, 'isfile', mock):
|
|
self.assertTrue(state.clear_request("A"))
|
|
|
|
mock = MagicMock(return_value=True)
|
|
with patch.object(os, 'remove', mock):
|
|
self.assertTrue(state.clear_request())
|
|
|
|
mock = MagicMock(return_value={})
|
|
with patch.object(state, 'check_request', mock):
|
|
self.assertFalse(state.clear_request("A"))
|
|
|
|
def test_check_request(self):
|
|
'''
|
|
Test to return the state request information
|
|
'''
|
|
mock = MagicMock(return_value=True)
|
|
with patch.object(os.path, 'join', mock), \
|
|
patch('salt.modules.state.salt.payload', MockSerial):
|
|
mock = MagicMock(side_effect=[True, True, False])
|
|
with patch.object(os.path, 'isfile', mock):
|
|
with patch('salt.utils.files.fopen', mock_open()):
|
|
self.assertDictEqual(state.check_request(), {'A': 'B'})
|
|
|
|
with patch('salt.utils.files.fopen', mock_open()):
|
|
self.assertEqual(state.check_request("A"), 'B')
|
|
|
|
self.assertDictEqual(state.check_request(), {})
|
|
|
|
def test_request(self):
|
|
'''
|
|
Test to request the local admin execute a state run
|
|
'''
|
|
mock = MagicMock(return_value=True)
|
|
with patch.object(state, 'apply_', mock):
|
|
mock = MagicMock(return_value=True)
|
|
with patch.object(os.path, 'join', mock):
|
|
mock = MagicMock(return_value=
|
|
{"test_run": "",
|
|
"mods": "",
|
|
"kwargs": ""})
|
|
with patch.object(state, 'check_request', mock):
|
|
mock = MagicMock(return_value=True)
|
|
with patch.object(os, 'umask', mock):
|
|
with patch.object(salt.utils.platform, 'is_windows', mock):
|
|
with patch.dict(state.__salt__, {'cmd.run': mock}):
|
|
with patch('salt.utils.files.fopen', mock_open()):
|
|
mock = MagicMock(return_value=True)
|
|
with patch.object(os, 'umask', mock):
|
|
self.assertTrue(state.request("A"))
|
|
|
|
def test_sls(self):
|
|
'''
|
|
Test to execute a set list of state files from an environment
|
|
'''
|
|
arg = "core,edit.vim dev"
|
|
ret = ['Pillar failed to render with the following messages:', 'E', '1']
|
|
mock = MagicMock(return_value=True)
|
|
with patch.object(state, 'running', mock):
|
|
with patch.dict(state.__context__, {"retcode": 1}):
|
|
self.assertEqual(state.sls("core,edit.vim dev"), True)
|
|
|
|
mock = MagicMock(side_effect=[True, True, True, True, True, True])
|
|
with patch.object(state, '_wait', mock):
|
|
mock = MagicMock(side_effect=[["A"], [], [], [], [], []])
|
|
with patch.object(state, '_disabled', mock):
|
|
with patch.dict(state.__context__, {"retcode": 1}):
|
|
self.assertEqual(
|
|
state.sls("core,edit.vim dev",
|
|
None,
|
|
None,
|
|
True),
|
|
["A"])
|
|
|
|
mock = MagicMock(side_effect=[['E', '1'], None, None, None, None])
|
|
with patch.object(state, '_get_pillar_errors', mock):
|
|
with patch.dict(state.__context__, {"retcode": 5}):
|
|
with patch.dict(state.__pillar__, {"_errors": ['E', '1']}):
|
|
self.assertListEqual(state.sls("core,edit.vim dev",
|
|
None,
|
|
None,
|
|
True), ret)
|
|
|
|
with patch.dict(state.__opts__, {"test": None}):
|
|
mock = MagicMock(return_value={"test": "",
|
|
"saltenv": None})
|
|
with patch.object(salt.utils.state, 'get_sls_opts', mock):
|
|
mock = MagicMock(return_value=True)
|
|
with patch.object(salt.utils,
|
|
'test_mode',
|
|
mock):
|
|
self.assertRaises(
|
|
SaltInvocationError,
|
|
state.sls,
|
|
"core,edit.vim dev",
|
|
None,
|
|
None,
|
|
True,
|
|
pillar="A")
|
|
|
|
mock = MagicMock(return_value="/D/cache.cache.p")
|
|
with patch.object(os.path,
|
|
'join',
|
|
mock):
|
|
mock = MagicMock(return_value=True)
|
|
with patch.object(os.path,
|
|
'isfile',
|
|
mock):
|
|
with patch(
|
|
'salt.utils.files.fopen',
|
|
mock_open()):
|
|
self.assertTrue(
|
|
state.sls(arg,
|
|
None,
|
|
None,
|
|
True,
|
|
cache
|
|
=True
|
|
)
|
|
)
|
|
|
|
MockState.HighState.flag = True
|
|
self.assertTrue(state.sls("core,edit"
|
|
".vim dev",
|
|
None,
|
|
None,
|
|
True)
|
|
)
|
|
|
|
MockState.HighState.flag = False
|
|
mock = MagicMock(return_value=True)
|
|
with patch.dict(state.__salt__,
|
|
{'config.option':
|
|
mock}):
|
|
mock = MagicMock(return_value=
|
|
True)
|
|
with patch.object(
|
|
state,
|
|
'_filter_'
|
|
'running',
|
|
mock):
|
|
self.sub_test_sls()
|
|
|
|
def test_get_test_value(self):
|
|
'''
|
|
Test _get_test_value when opts contains different values
|
|
'''
|
|
test_arg = 'test'
|
|
with patch.dict(state.__opts__, {test_arg: True}):
|
|
self.assertTrue(state._get_test_value(test=None),
|
|
msg='Failure when {0} is True in __opts__'.format(test_arg))
|
|
|
|
with patch.dict(config.__pillar__, {test_arg: 'blah'}):
|
|
self.assertFalse(state._get_test_value(test=None),
|
|
msg='Failure when {0} is blah in __opts__'.format(test_arg))
|
|
|
|
with patch.dict(config.__pillar__, {test_arg: 'true'}):
|
|
self.assertFalse(state._get_test_value(test=None),
|
|
msg='Failure when {0} is true in __opts__'.format(test_arg))
|
|
|
|
with patch.dict(config.__opts__, {test_arg: False}):
|
|
self.assertFalse(state._get_test_value(test=None),
|
|
msg='Failure when {0} is False in __opts__'.format(test_arg))
|
|
|
|
with patch.dict(config.__opts__, {}):
|
|
self.assertFalse(state._get_test_value(test=None),
|
|
msg='Failure when {0} does not exist in __opts__'.format(test_arg))
|
|
|
|
with patch.dict(config.__pillar__, {test_arg: None}):
|
|
self.assertEqual(state._get_test_value(test=None), None,
|
|
msg='Failure when {0} is None in __opts__'.format(test_arg))
|
|
|
|
with patch.dict(config.__pillar__, {test_arg: True}):
|
|
self.assertTrue(state._get_test_value(test=None),
|
|
msg='Failure when {0} is True in __pillar__'.format(test_arg))
|
|
|
|
with patch.dict(config.__pillar__, {'master': {test_arg: True}}):
|
|
self.assertTrue(state._get_test_value(test=None),
|
|
msg='Failure when {0} is True in master __pillar__'.format(test_arg))
|
|
|
|
with patch.dict(config.__pillar__, {'master': {test_arg: False}}):
|
|
with patch.dict(config.__pillar__, {test_arg: True}):
|
|
self.assertTrue(state._get_test_value(test=None),
|
|
msg='Failure when {0} is False in master __pillar__ and True in pillar'.format(test_arg))
|
|
|
|
with patch.dict(config.__pillar__, {'master': {test_arg: True}}):
|
|
with patch.dict(config.__pillar__, {test_arg: False}):
|
|
self.assertFalse(state._get_test_value(test=None),
|
|
msg='Failure when {0} is True in master __pillar__ and False in pillar'.format(test_arg))
|
|
|
|
with patch.dict(state.__opts__, {'test': False}):
|
|
self.assertFalse(state._get_test_value(test=None),
|
|
msg='Failure when {0} is False in __opts__'.format(test_arg))
|
|
|
|
with patch.dict(state.__opts__, {'test': False}):
|
|
with patch.dict(config.__pillar__, {'master': {test_arg: True}}):
|
|
self.assertTrue(state._get_test_value(test=None),
|
|
msg='Failure when {0} is False in __opts__'.format(test_arg))
|
|
|
|
with patch.dict(state.__opts__, {}):
|
|
self.assertTrue(state._get_test_value(test=True),
|
|
msg='Failure when test is True as arg')
|
|
|
|
def sub_test_sls(self):
|
|
'''
|
|
Sub function of test_sls
|
|
'''
|
|
mock = MagicMock(return_value=True)
|
|
with patch.object(os.path, 'join', mock):
|
|
with patch.object(os, 'umask', mock):
|
|
mock = MagicMock(return_value=False)
|
|
with patch.object(salt.utils.platform, 'is_windows', mock):
|
|
mock = MagicMock(return_value=True)
|
|
with patch.object(os, 'umask', mock):
|
|
with patch.object(state, '_set_retcode', mock):
|
|
with patch.dict(state.__opts__,
|
|
{"test": True}):
|
|
with patch('salt.utils.files.fopen', mock_open()):
|
|
self.assertTrue(state.sls("core,edit"
|
|
".vim dev",
|
|
None,
|
|
None,
|
|
True))
|
|
|
|
def test_pkg(self):
|
|
'''
|
|
Test to execute a packaged state run
|
|
'''
|
|
tar_file = os.sep + os.path.join('tmp', 'state_pkg.tgz')
|
|
mock = MagicMock(side_effect=[False, True, True, True, True, True,
|
|
True, True, True, True, True])
|
|
mock_json_loads_true = MagicMock(return_value=[True])
|
|
mock_json_loads_dictlist = MagicMock(return_value=[{"test": ""}])
|
|
with patch.object(os.path, 'isfile', mock), \
|
|
patch('salt.modules.state.tarfile', MockTarFile), \
|
|
patch.object(salt.utils, 'json', mock_json_loads_dictlist):
|
|
self.assertEqual(state.pkg(tar_file, "", "md5"), {})
|
|
|
|
mock = MagicMock(side_effect=[False, 0, 0, 0, 0])
|
|
with patch.object(salt.utils.hashutils, 'get_hash', mock):
|
|
# Verify hash
|
|
self.assertDictEqual(state.pkg(tar_file, "", "md5"), {})
|
|
|
|
# Verify file outside intended root
|
|
self.assertDictEqual(state.pkg(tar_file, 0, "md5"), {})
|
|
|
|
MockTarFile.path = ""
|
|
with patch('salt.utils.files.fopen', mock_open()), \
|
|
patch.object(salt.utils.json, 'loads', mock_json_loads_true):
|
|
self.assertEqual(state.pkg(tar_file, 0, "md5"), True)
|
|
|
|
MockTarFile.path = ""
|
|
if six.PY2:
|
|
with patch('salt.utils.files.fopen', mock_open()), \
|
|
patch.dict(state.__utils__, {'state.check_result': MagicMock(return_value=True)}):
|
|
self.assertTrue(state.pkg(tar_file, 0, "md5"))
|
|
else:
|
|
with patch('salt.utils.files.fopen', mock_open()):
|
|
self.assertTrue(state.pkg(tar_file, 0, "md5"))
|
|
|
|
def test_lock_saltenv(self):
|
|
'''
|
|
Tests lock_saltenv in each function which accepts saltenv on the CLI
|
|
'''
|
|
lock_msg = 'lock_saltenv is enabled, saltenv cannot be changed'
|
|
empty_list_mock = MagicMock(return_value=[])
|
|
with patch.dict(state.__opts__, {'lock_saltenv': True}), \
|
|
patch.dict(state.__salt__, {'grains.get': empty_list_mock}), \
|
|
patch.object(state, 'running', empty_list_mock):
|
|
|
|
# Test high
|
|
with self.assertRaisesRegex(CommandExecutionError, lock_msg):
|
|
state.high(
|
|
[{"vim": {"pkg": ["installed"]}}], saltenv='base')
|
|
|
|
# Test template
|
|
with self.assertRaisesRegex(CommandExecutionError, lock_msg):
|
|
state.template('foo', saltenv='base')
|
|
|
|
# Test template_str
|
|
with self.assertRaisesRegex(CommandExecutionError, lock_msg):
|
|
state.template_str('foo', saltenv='base')
|
|
|
|
# Test apply_ with SLS
|
|
with self.assertRaisesRegex(CommandExecutionError, lock_msg):
|
|
state.apply_('foo', saltenv='base')
|
|
|
|
# Test apply_ with Highstate
|
|
with self.assertRaisesRegex(CommandExecutionError, lock_msg):
|
|
state.apply_(saltenv='base')
|
|
|
|
# Test highstate
|
|
with self.assertRaisesRegex(CommandExecutionError, lock_msg):
|
|
state.highstate(saltenv='base')
|
|
|
|
# Test sls
|
|
with self.assertRaisesRegex(CommandExecutionError, lock_msg):
|
|
state.sls('foo', saltenv='base')
|
|
|
|
# Test top
|
|
with self.assertRaisesRegex(CommandExecutionError, lock_msg):
|
|
state.top('foo.sls', saltenv='base')
|
|
|
|
# Test show_highstate
|
|
with self.assertRaisesRegex(CommandExecutionError, lock_msg):
|
|
state.show_highstate(saltenv='base')
|
|
|
|
# Test show_lowstate
|
|
with self.assertRaisesRegex(CommandExecutionError, lock_msg):
|
|
state.show_lowstate(saltenv='base')
|
|
|
|
# Test sls_id
|
|
with self.assertRaisesRegex(CommandExecutionError, lock_msg):
|
|
state.sls_id('foo', 'bar', saltenv='base')
|
|
|
|
# Test show_low_sls
|
|
with self.assertRaisesRegex(CommandExecutionError, lock_msg):
|
|
state.show_low_sls('foo', saltenv='base')
|
|
|
|
# Test show_sls
|
|
with self.assertRaisesRegex(CommandExecutionError, lock_msg):
|
|
state.show_sls('foo', saltenv='base')
|
|
|
|
# Test show_top
|
|
with self.assertRaisesRegex(CommandExecutionError, lock_msg):
|
|
state.show_top(saltenv='base')
|
|
|
|
# Test single
|
|
with self.assertRaisesRegex(CommandExecutionError, lock_msg):
|
|
state.single('foo.bar', name='baz', saltenv='base')
|
|
|
|
# Test pkg
|
|
with self.assertRaisesRegex(CommandExecutionError, lock_msg):
|
|
state.pkg(
|
|
'/tmp/salt_state.tgz',
|
|
'760a9353810e36f6d81416366fc426dc',
|
|
'md5',
|
|
saltenv='base')
|
|
|
|
def test_get_pillar_errors_CC(self):
|
|
'''
|
|
Test _get_pillar_errors function.
|
|
CC: External clean, Internal clean
|
|
:return:
|
|
'''
|
|
for int_pillar, ext_pillar in [({'foo': 'bar'}, {'fred': 'baz'}),
|
|
({'foo': 'bar'}, None),
|
|
({}, {'fred': 'baz'})]:
|
|
with patch('salt.modules.state.__pillar__', int_pillar):
|
|
for opts, res in [({'force': True}, None),
|
|
({'force': False}, None),
|
|
({}, None)]:
|
|
assert res == state._get_pillar_errors(kwargs=opts, pillar=ext_pillar)
|
|
|
|
def test_get_pillar_errors_EC(self):
|
|
'''
|
|
Test _get_pillar_errors function.
|
|
EC: External erroneous, Internal clean
|
|
:return:
|
|
'''
|
|
errors = ['failure', 'everywhere']
|
|
for int_pillar, ext_pillar in [({'foo': 'bar'}, {'fred': 'baz', '_errors': errors}),
|
|
({}, {'fred': 'baz', '_errors': errors})]:
|
|
with patch('salt.modules.state.__pillar__', int_pillar):
|
|
for opts, res in [({'force': True}, None),
|
|
({'force': False}, errors),
|
|
({}, errors)]:
|
|
assert res == state._get_pillar_errors(kwargs=opts, pillar=ext_pillar)
|
|
|
|
def test_get_pillar_errors_EE(self):
|
|
'''
|
|
Test _get_pillar_errors function.
|
|
CC: External erroneous, Internal erroneous
|
|
:return:
|
|
'''
|
|
errors = ['failure', 'everywhere']
|
|
for int_pillar, ext_pillar in [({'foo': 'bar', '_errors': errors}, {'fred': 'baz', '_errors': errors})]:
|
|
with patch('salt.modules.state.__pillar__', int_pillar):
|
|
for opts, res in [({'force': True}, None),
|
|
({'force': False}, errors),
|
|
({}, errors)]:
|
|
assert res == state._get_pillar_errors(kwargs=opts, pillar=ext_pillar)
|
|
|
|
def test_get_pillar_errors_CE(self):
|
|
'''
|
|
Test _get_pillar_errors function.
|
|
CC: External clean, Internal erroneous
|
|
:return:
|
|
'''
|
|
errors = ['failure', 'everywhere']
|
|
for int_pillar, ext_pillar in [({'foo': 'bar', '_errors': errors}, {'fred': 'baz'}),
|
|
({'foo': 'bar', '_errors': errors}, None)]:
|
|
with patch('salt.modules.state.__pillar__', int_pillar):
|
|
for opts, res in [({'force': True}, None),
|
|
({'force': False}, errors),
|
|
({}, errors)]:
|
|
assert res == state._get_pillar_errors(kwargs=opts, pillar=ext_pillar)
|