mirror of
https://github.com/valitydev/salt.git
synced 2024-11-07 17:09:03 +00:00
961 lines
37 KiB
Python
961 lines
37 KiB
Python
# -*- coding: utf-8 -*-
|
|
'''
|
|
:codeauthor: :email:`Rahul Handay <rahulha@saltstack.com>`
|
|
'''
|
|
|
|
# Import Python libs
|
|
from __future__ import absolute_import
|
|
import os
|
|
|
|
# Import Salt Testing Libs
|
|
from salttesting import TestCase, skipIf
|
|
from salt.exceptions import SaltInvocationError
|
|
from salttesting.helpers import ensure_in_syspath
|
|
from salttesting.mock import (
|
|
MagicMock,
|
|
patch,
|
|
mock_open,
|
|
NO_MOCK,
|
|
NO_MOCK_REASON
|
|
)
|
|
|
|
ensure_in_syspath('../../')
|
|
|
|
# Import Salt Libs
|
|
import salt.utils
|
|
from salt.modules import state
|
|
|
|
# Globals
|
|
state.__salt__ = {}
|
|
state.__context__ = {}
|
|
state.__opts__ = {}
|
|
state.__pillar__ = {}
|
|
|
|
|
|
class MockState(object):
|
|
'''
|
|
Mock class
|
|
'''
|
|
def __init__(self):
|
|
pass
|
|
|
|
class State(object):
|
|
'''
|
|
Mock state class
|
|
'''
|
|
flag = None
|
|
|
|
def __init__(self, opts, pillar=False):
|
|
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):
|
|
'''
|
|
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
|
|
|
|
class HighState(object):
|
|
'''
|
|
Mock HighState class
|
|
'''
|
|
flag = False
|
|
opts = {'state_top': ""}
|
|
|
|
def __init__(self, opts, pillar=None, *args, **kwargs):
|
|
self.state = MockState.State(opts,
|
|
pillar=pillar)
|
|
|
|
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 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):
|
|
'''
|
|
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 = "/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
|
|
|
|
|
|
class MockJson(object):
|
|
'''
|
|
Mock json class
|
|
'''
|
|
flag = None
|
|
|
|
def __init__(self):
|
|
pass
|
|
|
|
def load(self, data, object_hook=None):
|
|
'''
|
|
Mock load method
|
|
'''
|
|
data = data
|
|
object_hook = object_hook
|
|
if self.flag:
|
|
return [True]
|
|
else:
|
|
return [{"test": ""}]
|
|
|
|
|
|
@skipIf(NO_MOCK, NO_MOCK_REASON)
|
|
@patch('salt.modules.state.salt.state', MockState())
|
|
class StateTestCase(TestCase):
|
|
'''
|
|
Test case for salt.modules.state
|
|
'''
|
|
|
|
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(state, '_get_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
|
|
'''
|
|
with patch.dict(state.__opts__, {"cachedir": "/D/"}):
|
|
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(state, '_get_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):
|
|
with patch.dict(state.__opts__, {"cachedir": "/D/"}):
|
|
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_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})
|
|
with patch.object(state, '_get_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})
|
|
with patch.object(state, '_get_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})
|
|
with patch.object(state, '_get_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=[False, True, True])
|
|
with patch.object(state, '_check_pillar', 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(state, '_get_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(state, '_get_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.dict(state.__opts__,
|
|
{"cachedir": "D"}):
|
|
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
|
|
'''
|
|
with patch.dict(state.__opts__, {"cachedir": "D"}):
|
|
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"))
|
|
|
|
@patch('salt.modules.state.salt.payload', MockSerial)
|
|
def test_check_request(self):
|
|
'''
|
|
Test to return the state request information
|
|
'''
|
|
with patch.dict(state.__opts__, {"cachedir": "D"}):
|
|
mock = MagicMock(return_value=True)
|
|
with patch.object(os.path, 'join', mock):
|
|
mock = MagicMock(side_effect=[True, True, False])
|
|
with patch.object(os.path, 'isfile', mock):
|
|
with patch('salt.utils.fopen', mock_open()):
|
|
self.assertDictEqual(state.check_request(), {'A': 'B'})
|
|
|
|
with patch('salt.utils.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):
|
|
with patch.dict(state.__opts__, {"cachedir": "D"}):
|
|
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, 'is_windows', mock):
|
|
with patch.dict(state.__salt__,
|
|
{'cmd.run': mock}):
|
|
with patch('salt.utils.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,
|
|
None,
|
|
True),
|
|
["A"])
|
|
|
|
mock = MagicMock(side_effect=[False,
|
|
True,
|
|
True,
|
|
True,
|
|
True])
|
|
with patch.object(state, '_check_pillar', mock):
|
|
with patch.dict(state.__context__, {"retcode": 5}):
|
|
with patch.dict(state.__pillar__, {"_errors": "E1"}):
|
|
self.assertListEqual(state.sls("core,edit.vim dev",
|
|
None,
|
|
None,
|
|
None,
|
|
True), ret)
|
|
|
|
with patch.dict(state.__opts__, {"test": None}):
|
|
mock = MagicMock(return_value={"test": ""})
|
|
with patch.object(state, '_get_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,
|
|
None,
|
|
True,
|
|
pillar="A")
|
|
|
|
with patch.dict(
|
|
state.__opts__,
|
|
{"cachedir": "/D/"}):
|
|
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.fopen',
|
|
mock_open()):
|
|
self.assertTrue(
|
|
state.sls(arg,
|
|
None,
|
|
None,
|
|
None,
|
|
True,
|
|
cache
|
|
=True
|
|
)
|
|
)
|
|
|
|
MockState.HighState.flag = True
|
|
self.assertTrue(state.sls("core,edit"
|
|
".vim dev",
|
|
None,
|
|
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):
|
|
with patch.dict(
|
|
state.
|
|
__opts__,
|
|
{"cachedir":
|
|
"/D/"}):
|
|
self.sub_test_sls()
|
|
|
|
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, '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.fopen', mock_open()):
|
|
self.assertTrue(state.sls("core,edit"
|
|
".vim dev",
|
|
None,
|
|
None,
|
|
None,
|
|
True))
|
|
|
|
@patch('salt.modules.state.tarfile', MockTarFile)
|
|
@patch('salt.modules.state.json', MockJson())
|
|
def test_pkg(self):
|
|
'''
|
|
Test to execute a packaged state run
|
|
'''
|
|
mock = MagicMock(side_effect=[False, True, True, True, True, True])
|
|
with patch.object(os.path, 'isfile', mock):
|
|
self.assertEqual(state.pkg("/tmp/state_pkg.tgz", "", "md5"), {})
|
|
|
|
mock = MagicMock(side_effect=[False, 0, 0, 0, 0])
|
|
with patch.object(salt.utils, 'get_hash', mock):
|
|
self.assertDictEqual(state.pkg("/tmp/state_pkg.tgz", "", "md5"),
|
|
{})
|
|
|
|
self.assertDictEqual(state.pkg("/tmp/state_pkg.tgz", 0, "md5"),
|
|
{})
|
|
|
|
MockTarFile.path = ""
|
|
MockJson.flag = True
|
|
with patch('salt.utils.fopen', mock_open()):
|
|
self.assertListEqual(state.pkg("/tmp/state_pkg.tgz",
|
|
0,
|
|
"md5"),
|
|
[True])
|
|
|
|
MockTarFile.path = ""
|
|
MockJson.flag = False
|
|
with patch('salt.utils.fopen', mock_open()):
|
|
self.assertTrue(state.pkg("/tmp/state_pkg.tgz",
|
|
0, "md5"))
|
|
|
|
if __name__ == '__main__':
|
|
from integration import run_tests
|
|
run_tests(StateTestCase, needs_daemon=False)
|