# -*- 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)