salt/tests/unit/utils/test_state.py
Erik Johnson 6db2beb6c0
Replace "pchanges" with "changes" to fix onchanges/prereq requisites
Since "pchanges" was never supported in the state compiler, and
"changes" is what these reqs always used, replacing "pchanges" with
"changes" will allow those requisites to work in test mode.

Conflicts:
  - salt/states/file.py
  - salt/states/linux_acl.py
  - salt/utils/napalm.py
  - tests/integration/modules/test_state.py
  - tests/unit/states/test_file.py
2019-04-18 10:56:30 -04:00

638 lines
27 KiB
Python

# -*- coding: utf-8 -*-
'''
Unit Tests for functions located in salt.utils.state.py.
'''
# Import python libs
from __future__ import absolute_import, print_function, unicode_literals
import copy
import textwrap
# Import Salt libs
from salt.ext import six
import salt.utils.odict
import salt.utils.state
# Import Salt Testing libs
from tests.support.unit import TestCase
class StateUtilTestCase(TestCase):
'''
Test case for state util.
'''
def test_check_result(self):
self.assertFalse(salt.utils.state.check_result(None),
'Failed to handle None as an invalid data type.')
self.assertFalse(salt.utils.state.check_result([]),
'Failed to handle an invalid data type.')
self.assertFalse(salt.utils.state.check_result({}),
'Failed to handle an empty dictionary.')
self.assertFalse(salt.utils.state.check_result({'host1': []}),
'Failed to handle an invalid host data structure.')
test_valid_state = {'host1': {'test_state': {'result': 'We have liftoff!'}}}
self.assertTrue(salt.utils.state.check_result(test_valid_state))
test_valid_false_states = {
'test1': salt.utils.odict.OrderedDict([
('host1',
salt.utils.odict.OrderedDict([
('test_state0', {'result': True}),
('test_state', {'result': False}),
])),
]),
'test2': salt.utils.odict.OrderedDict([
('host1',
salt.utils.odict.OrderedDict([
('test_state0', {'result': True}),
('test_state', {'result': True}),
])),
('host2',
salt.utils.odict.OrderedDict([
('test_state0', {'result': True}),
('test_state', {'result': False}),
])),
]),
'test3': ['a'],
'test4': salt.utils.odict.OrderedDict([
('asup', salt.utils.odict.OrderedDict([
('host1',
salt.utils.odict.OrderedDict([
('test_state0', {'result': True}),
('test_state', {'result': True}),
])),
('host2',
salt.utils.odict.OrderedDict([
('test_state0', {'result': True}),
('test_state', {'result': False}),
]))
]))
]),
'test5': salt.utils.odict.OrderedDict([
('asup', salt.utils.odict.OrderedDict([
('host1',
salt.utils.odict.OrderedDict([
('test_state0', {'result': True}),
('test_state', {'result': True}),
])),
('host2', salt.utils.odict.OrderedDict([]))
]))
])
}
for test, data in six.iteritems(test_valid_false_states):
self.assertFalse(
salt.utils.state.check_result(data),
msg='{0} failed'.format(test))
test_valid_true_states = {
'test1': salt.utils.odict.OrderedDict([
('host1',
salt.utils.odict.OrderedDict([
('test_state0', {'result': True}),
('test_state', {'result': True}),
])),
]),
'test3': salt.utils.odict.OrderedDict([
('host1',
salt.utils.odict.OrderedDict([
('test_state0', {'result': True}),
('test_state', {'result': True}),
])),
('host2',
salt.utils.odict.OrderedDict([
('test_state0', {'result': True}),
('test_state', {'result': True}),
])),
]),
'test4': salt.utils.odict.OrderedDict([
('asup', salt.utils.odict.OrderedDict([
('host1',
salt.utils.odict.OrderedDict([
('test_state0', {'result': True}),
('test_state', {'result': True}),
])),
('host2',
salt.utils.odict.OrderedDict([
('test_state0', {'result': True}),
('test_state', {'result': True}),
]))
]))
]),
'test2': salt.utils.odict.OrderedDict([
('host1',
salt.utils.odict.OrderedDict([
('test_state0', {'result': None}),
('test_state', {'result': True}),
])),
('host2',
salt.utils.odict.OrderedDict([
('test_state0', {'result': True}),
('test_state', {'result': 'abc'}),
]))
])
}
for test, data in six.iteritems(test_valid_true_states):
self.assertTrue(
salt.utils.state.check_result(data),
msg='{0} failed'.format(test))
test_invalid_true_ht_states = {
'test_onfail_simple2': (
salt.utils.odict.OrderedDict([
('host1',
salt.utils.odict.OrderedDict([
('test_vstate0', {'result': False}),
('test_vstate1', {'result': True}),
])),
]),
{
'test_vstate0': {
'__env__': 'base',
'__sls__': 'a',
'cmd': [salt.utils.odict.OrderedDict([('name', '/bin/true')]),
'run',
{'order': 10002}]},
'test_vstate1': {
'__env__': 'base',
'__sls__': 'a',
'cmd': [salt.utils.odict.OrderedDict([('name', '/bin/true')]),
salt.utils.odict.OrderedDict([
('onfail_stop', True),
('onfail',
[salt.utils.odict.OrderedDict([('cmd', 'test_vstate0')])])
]),
'run',
{'order': 10004}]},
}
),
'test_onfail_integ2': (
salt.utils.odict.OrderedDict([
('host1',
salt.utils.odict.OrderedDict([
('t_|-test_ivstate0_|-echo_|-run', {
'result': False}),
('cmd_|-test_ivstate0_|-echo_|-run', {
'result': False}),
('cmd_|-test_ivstate1_|-echo_|-run', {
'result': False}),
])),
]),
{
'test_ivstate0': {
'__env__': 'base',
'__sls__': 'a',
'cmd': [salt.utils.odict.OrderedDict([('name', '/bin/true')]),
'run',
{'order': 10002}],
't': [salt.utils.odict.OrderedDict([('name', '/bin/true')]),
'run',
{'order': 10002}]},
'test_ivstate1': {
'__env__': 'base',
'__sls__': 'a',
'cmd': [salt.utils.odict.OrderedDict([('name', '/bin/true')]),
salt.utils.odict.OrderedDict([
('onfail_stop', False),
('onfail',
[salt.utils.odict.OrderedDict([('cmd', 'test_ivstate0')])])
]),
'run',
{'order': 10004}]},
}
),
'test_onfail_integ3': (
salt.utils.odict.OrderedDict([
('host1',
salt.utils.odict.OrderedDict([
('t_|-test_ivstate0_|-echo_|-run', {
'result': True}),
('cmd_|-test_ivstate0_|-echo_|-run', {
'result': False}),
('cmd_|-test_ivstate1_|-echo_|-run', {
'result': False}),
])),
]),
{
'test_ivstate0': {
'__env__': 'base',
'__sls__': 'a',
'cmd': [salt.utils.odict.OrderedDict([('name', '/bin/true')]),
'run',
{'order': 10002}],
't': [salt.utils.odict.OrderedDict([('name', '/bin/true')]),
'run',
{'order': 10002}]},
'test_ivstate1': {
'__env__': 'base',
'__sls__': 'a',
'cmd': [salt.utils.odict.OrderedDict([('name', '/bin/true')]),
salt.utils.odict.OrderedDict([
('onfail_stop', False),
('onfail',
[salt.utils.odict.OrderedDict([('cmd', 'test_ivstate0')])])
]),
'run',
{'order': 10004}]},
}
),
'test_onfail_integ4': (
salt.utils.odict.OrderedDict([
('host1',
salt.utils.odict.OrderedDict([
('t_|-test_ivstate0_|-echo_|-run', {
'result': False}),
('cmd_|-test_ivstate0_|-echo_|-run', {
'result': False}),
('cmd_|-test_ivstate1_|-echo_|-run', {
'result': True}),
])),
]),
{
'test_ivstate0': {
'__env__': 'base',
'__sls__': 'a',
'cmd': [salt.utils.odict.OrderedDict([('name', '/bin/true')]),
'run',
{'order': 10002}],
't': [salt.utils.odict.OrderedDict([('name', '/bin/true')]),
'run',
{'order': 10002}]},
'test_ivstate1': {
'__env__': 'base',
'__sls__': 'a',
'cmd': [salt.utils.odict.OrderedDict([('name', '/bin/true')]),
salt.utils.odict.OrderedDict([
('onfail_stop', False),
('onfail',
[salt.utils.odict.OrderedDict([('cmd', 'test_ivstate0')])])
]),
'run',
{'order': 10004}]},
'test_ivstate2': {
'__env__': 'base',
'__sls__': 'a',
'cmd': [salt.utils.odict.OrderedDict([('name', '/bin/true')]),
salt.utils.odict.OrderedDict([
('onfail_stop', True),
('onfail',
[salt.utils.odict.OrderedDict([('cmd', 'test_ivstate0')])])
]),
'run',
{'order': 10004}]},
}
),
'test_onfail': (
salt.utils.odict.OrderedDict([
('host1',
salt.utils.odict.OrderedDict([
('test_state0', {'result': False}),
('test_state', {'result': True}),
])),
]),
None
),
'test_onfail_d': (
salt.utils.odict.OrderedDict([
('host1',
salt.utils.odict.OrderedDict([
('test_state0', {'result': False}),
('test_state', {'result': True}),
])),
]),
{}
)
}
for test, testdata in six.iteritems(test_invalid_true_ht_states):
data, ht = testdata
for t_ in [a for a in data['host1']]:
tdata = data['host1'][t_]
if '_|-' in t_:
t_ = t_.split('_|-')[1]
tdata['__id__'] = t_
self.assertFalse(
salt.utils.state.check_result(data, highstate=ht),
msg='{0} failed'.format(test))
test_valid_true_ht_states = {
'test_onfail_integ': (
salt.utils.odict.OrderedDict([
('host1',
salt.utils.odict.OrderedDict([
('cmd_|-test_ivstate0_|-echo_|-run', {
'result': False}),
('cmd_|-test_ivstate1_|-echo_|-run', {
'result': True}),
])),
]),
{
'test_ivstate0': {
'__env__': 'base',
'__sls__': 'a',
'cmd': [salt.utils.odict.OrderedDict([('name', '/bin/true')]),
'run',
{'order': 10002}]},
'test_ivstate1': {
'__env__': 'base',
'__sls__': 'a',
'cmd': [salt.utils.odict.OrderedDict([('name', '/bin/true')]),
salt.utils.odict.OrderedDict([
('onfail_stop', False),
('onfail',
[salt.utils.odict.OrderedDict([('cmd', 'test_ivstate0')])])
]),
'run',
{'order': 10004}]},
}
),
'test_onfail_intega3': (
salt.utils.odict.OrderedDict([
('host1',
salt.utils.odict.OrderedDict([
('t_|-test_ivstate0_|-echo_|-run', {
'result': True}),
('cmd_|-test_ivstate0_|-echo_|-run', {
'result': False}),
('cmd_|-test_ivstate1_|-echo_|-run', {
'result': True}),
])),
]),
{
'test_ivstate0': {
'__env__': 'base',
'__sls__': 'a',
'cmd': [salt.utils.odict.OrderedDict([('name', '/bin/true')]),
'run',
{'order': 10002}],
't': [salt.utils.odict.OrderedDict([('name', '/bin/true')]),
'run',
{'order': 10002}]},
'test_ivstate1': {
'__env__': 'base',
'__sls__': 'a',
'cmd': [salt.utils.odict.OrderedDict([('name', '/bin/true')]),
salt.utils.odict.OrderedDict([
('onfail_stop', False),
('onfail',
[salt.utils.odict.OrderedDict([('cmd', 'test_ivstate0')])])
]),
'run',
{'order': 10004}]},
}
),
'test_onfail_simple': (
salt.utils.odict.OrderedDict([
('host1',
salt.utils.odict.OrderedDict([
('test_vstate0', {'result': False}),
('test_vstate1', {'result': True}),
])),
]),
{
'test_vstate0': {
'__env__': 'base',
'__sls__': 'a',
'cmd': [salt.utils.odict.OrderedDict([('name', '/bin/true')]),
'run',
{'order': 10002}]},
'test_vstate1': {
'__env__': 'base',
'__sls__': 'a',
'cmd': [salt.utils.odict.OrderedDict([('name', '/bin/true')]),
salt.utils.odict.OrderedDict([
('onfail_stop', False),
('onfail',
[salt.utils.odict.OrderedDict([('cmd', 'test_vstate0')])])
]),
'run',
{'order': 10004}]},
}
), # order is different
'test_onfail_simple_rev': (
salt.utils.odict.OrderedDict([
('host1',
salt.utils.odict.OrderedDict([
('test_vstate0', {'result': False}),
('test_vstate1', {'result': True}),
])),
]),
{
'test_vstate0': {
'__env__': 'base',
'__sls__': 'a',
'cmd': [salt.utils.odict.OrderedDict([('name', '/bin/true')]),
'run',
{'order': 10002}]},
'test_vstate1': {
'__env__': 'base',
'__sls__': 'a',
'cmd': [salt.utils.odict.OrderedDict([('name', '/bin/true')]),
salt.utils.odict.OrderedDict([
('onfail',
[salt.utils.odict.OrderedDict([('cmd', 'test_vstate0')])])
]),
salt.utils.odict.OrderedDict([('onfail_stop', False)]),
'run',
{'order': 10004}]},
}
)
}
for test, testdata in six.iteritems(test_valid_true_ht_states):
data, ht = testdata
for t_ in [a for a in data['host1']]:
tdata = data['host1'][t_]
if '_|-' in t_:
t_ = t_.split('_|-')[1]
tdata['__id__'] = t_
self.assertTrue(
salt.utils.state.check_result(data, highstate=ht),
msg='{0} failed'.format(test))
test_valid_false_state = {'host1': {'test_state': {'result': False}}}
self.assertFalse(salt.utils.state.check_result(test_valid_false_state))
class UtilStateMergeSubreturnTestcase(TestCase):
'''
Test cases for salt.utils.state.merge_subreturn function.
'''
main_ret = {
'name': 'primary',
# result may be missing, as primarysalt.utils.state is still in progress
'comment': '',
'changes': {},
}
sub_ret = {
'name': 'secondary',
'result': True,
'comment': '',
'changes': {},
}
def test_merge_result(self):
# result not created if not needed
for no_effect_result in [True, None]:
m = copy.deepcopy(self.main_ret)
s = copy.deepcopy(self.sub_ret)
s['result'] = no_effect_result
res = salt.utils.state.merge_subreturn(m, s)
self.assertNotIn('result', res)
# False subresult is propagated to existing result
for original_result in [True, None, False]:
m = copy.deepcopy(self.main_ret)
m['result'] = original_result
s = copy.deepcopy(self.sub_ret)
s['result'] = False
res = salt.utils.state.merge_subreturn(m, s)
self.assertFalse(res['result'])
# False result cannot be overridden
for any_result in [True, None, False]:
m = copy.deepcopy(self.main_ret)
m['result'] = False
s = copy.deepcopy(self.sub_ret)
s['result'] = any_result
res = salt.utils.state.merge_subreturn(m, s)
self.assertFalse(res['result'])
def test_merge_changes(self):
# The main changes dict should always already exist,
# and there should always be a changes dict in the secondary.
primary_changes = {'old': None, 'new': 'my_resource'}
secondary_changes = {'old': None, 'new': ['alarm-1', 'alarm-2']}
# No changes case
m = copy.deepcopy(self.main_ret)
s = copy.deepcopy(self.sub_ret)
res = salt.utils.state.merge_subreturn(m, s)
self.assertDictEqual(res['changes'], {})
# New changes don't get rid of existing changes
m = copy.deepcopy(self.main_ret)
m['changes'] = copy.deepcopy(primary_changes)
s = copy.deepcopy(self.sub_ret)
s['changes'] = copy.deepcopy(secondary_changes)
res = salt.utils.state.merge_subreturn(m, s)
self.assertDictEqual(res['changes'], {
'old': None,
'new': 'my_resource',
'secondary': secondary_changes,
})
# The subkey parameter is respected
m = copy.deepcopy(self.main_ret)
m['changes'] = copy.deepcopy(primary_changes)
s = copy.deepcopy(self.sub_ret)
s['changes'] = copy.deepcopy(secondary_changes)
res = salt.utils.state.merge_subreturn(m, s, subkey='alarms')
self.assertDictEqual(res['changes'], {
'old': None,
'new': 'my_resource',
'alarms': secondary_changes,
})
def test_merge_comments(self):
main_comment_1 = 'First primary comment.'
main_comment_2 = 'Second primary comment.'
sub_comment_1 = 'First secondary comment,\nwhich spans two lines.'
sub_comment_2 = 'Second secondary comment: {0}'.format(
'some error\n And a traceback',
)
final_comment = textwrap.dedent('''\
First primary comment.
Second primary comment.
First secondary comment,
which spans two lines.
Second secondary comment: some error
And a traceback
'''.rstrip())
# Joining two strings
m = copy.deepcopy(self.main_ret)
m['comment'] = main_comment_1 + '\n' + main_comment_2
s = copy.deepcopy(self.sub_ret)
s['comment'] = sub_comment_1 + '\n' + sub_comment_2
res = salt.utils.state.merge_subreturn(m, s)
self.assertMultiLineEqual(res['comment'], final_comment)
# Joining string and a list
m = copy.deepcopy(self.main_ret)
m['comment'] = main_comment_1 + '\n' + main_comment_2
s = copy.deepcopy(self.sub_ret)
s['comment'] = [sub_comment_1, sub_comment_2]
res = salt.utils.state.merge_subreturn(m, s)
self.assertMultiLineEqual(res['comment'], final_comment)
# For tests where output is a list,
# also test that final joined output will match
# Joining list and a string
m = copy.deepcopy(self.main_ret)
m['comment'] = [main_comment_1, main_comment_2]
s = copy.deepcopy(self.sub_ret)
s['comment'] = sub_comment_1 + '\n' + sub_comment_2
res = salt.utils.state.merge_subreturn(m, s)
self.assertEqual(res['comment'], [
main_comment_1,
main_comment_2,
sub_comment_1 + '\n' + sub_comment_2,
])
self.assertMultiLineEqual('\n'.join(res['comment']), final_comment)
# Joining two lists
m = copy.deepcopy(self.main_ret)
m['comment'] = [main_comment_1, main_comment_2]
s = copy.deepcopy(self.sub_ret)
s['comment'] = [sub_comment_1, sub_comment_2]
res = salt.utils.state.merge_subreturn(m, s)
self.assertEqual(res['comment'], [
main_comment_1,
main_comment_2,
sub_comment_1,
sub_comment_2,
])
self.assertMultiLineEqual('\n'.join(res['comment']), final_comment)
def test_merge_empty_comments(self):
# Since the primarysalt.utils.state is in progress,
# the main comment may be empty, either '' or [].
# Note that [''] is a degenerate case and should never happen,
# hence the behavior is left unspecified in that case.
# The secondary comment should never be empty,
# because thatsalt.utils.state has already returned,
# so we leave the behavior unspecified in that case.
sub_comment_1 = 'Secondary comment about changes:'
sub_comment_2 = 'A diff that goes with the previous comment'
# No contributions from primary
final_comment = sub_comment_1 + '\n' + sub_comment_2
# Joining empty string and a string
m = copy.deepcopy(self.main_ret)
m['comment'] = ''
s = copy.deepcopy(self.sub_ret)
s['comment'] = sub_comment_1 + '\n' + sub_comment_2
res = salt.utils.state.merge_subreturn(m, s)
self.assertEqual(res['comment'], final_comment)
# Joining empty string and a list
m = copy.deepcopy(self.main_ret)
m['comment'] = ''
s = copy.deepcopy(self.sub_ret)
s['comment'] = [sub_comment_1, sub_comment_2]
res = salt.utils.state.merge_subreturn(m, s)
self.assertEqual(res['comment'], final_comment)
# For tests where output is a list,
# also test that final joined output will match
# Joining empty list and a string
m = copy.deepcopy(self.main_ret)
m['comment'] = []
s = copy.deepcopy(self.sub_ret)
s['comment'] = sub_comment_1 + '\n' + sub_comment_2
res = salt.utils.state.merge_subreturn(m, s)
self.assertEqual(res['comment'], [final_comment])
self.assertEqual('\n'.join(res['comment']), final_comment)
# Joining empty list and a list
m = copy.deepcopy(self.main_ret)
m['comment'] = []
s = copy.deepcopy(self.sub_ret)
s['comment'] = [sub_comment_1, sub_comment_2]
res = salt.utils.state.merge_subreturn(m, s)
self.assertEqual(res['comment'], [sub_comment_1, sub_comment_2])
self.assertEqual('\n'.join(res['comment']), final_comment)