# -*- coding: utf-8 -*- from __future__ import absolute_import, print_function, unicode_literals import glob import logging import os import shutil import threading import time from salt.ext import six from salt.ext.six.moves import range from tests.support.case import SSHCase from tests.support.helpers import flaky, slowTest from tests.support.runtests import RUNTIME_VARS SSH_SLS = "ssh_state_tests" SSH_SLS_FILE = "/tmp/salt_test_file" log = logging.getLogger(__name__) class SSHStateTest(SSHCase): """ testing the state system with salt-ssh """ def _check_dict_ret(self, ret, val, exp_ret, equal=True): self.assertIsInstance(ret, dict) for key, value in ret.items(): self.assertIsInstance(value, dict) if equal: self.assertEqual(value[val], exp_ret) else: self.assertNotEqual(value[val], exp_ret) def _check_request(self, empty=False): check = self.run_function("state.check_request", wipe=False) if empty: self.assertFalse(bool(check), "bool({0}) is not False".format(check)) else: self._check_dict_ret( ret=check["default"]["test_run"]["local"]["return"], val="__sls__", exp_ret=SSH_SLS, ) @slowTest def test_state_apply(self): """ test state.apply with salt-ssh """ ret = self.run_function("state.apply", [SSH_SLS]) self._check_dict_ret(ret=ret, val="__sls__", exp_ret=SSH_SLS) check_file = self.run_function("file.file_exists", [SSH_SLS_FILE]) self.assertTrue(check_file) @slowTest def test_state_sls_id(self): """ test state.sls_id with salt-ssh """ # check state.sls_id with test=True ret = self.run_function("state.sls_id", ["ssh-file-test", SSH_SLS, "test=True"]) self._check_dict_ret( ret=ret, val="comment", exp_ret=( "The file {} is set to be changed\n" "Note: No changes made, actual changes may\n" "be different due to other states." ).format(SSH_SLS_FILE), ) # check state.sls_id without test=True ret = self.run_function("state.sls_id", ["ssh-file-test", SSH_SLS]) self._check_dict_ret(ret=ret, val="__sls__", exp_ret=SSH_SLS) # make sure the other id in the state was not run self._check_dict_ret(ret=ret, val="__id__", exp_ret="second_id", equal=False) check_file = self.run_function("file.file_exists", [SSH_SLS_FILE]) self.assertTrue(check_file) @slowTest def test_state_sls_wrong_id(self): """ test state.sls_id when id does not exist """ # check state.sls_id with test=True ret = self.run_function("state.sls_id", ["doesnotexist", SSH_SLS]) assert "No matches for ID" in ret @slowTest def test_state_sls_id_with_pillar(self): """ test state.sls_id with pillar data """ self.run_function( "state.sls_id", ["ssh-file-test", SSH_SLS, 'pillar=\'{"test_file_suffix": "_pillar"}\''], ) check_file = self.run_function( "file.file_exists", ["/tmp/salt_test_file_pillar"] ) self.assertTrue(check_file) @slowTest def test_state_show_sls(self): """ test state.show_sls with salt-ssh """ ret = self.run_function("state.show_sls", [SSH_SLS]) self._check_dict_ret(ret=ret, val="__sls__", exp_ret=SSH_SLS) check_file = self.run_function("file.file_exists", [SSH_SLS_FILE], wipe=False) self.assertFalse(check_file) @slowTest def test_state_show_top(self): """ test state.show_top with salt-ssh """ ret = self.run_function("state.show_top") self.assertEqual(ret, {"base": ["core", "master_tops_test"]}) @slowTest def test_state_single(self): """ state.single with salt-ssh """ ret_out = {"name": "itworked", "result": True, "comment": "Success!"} single = self.run_function( "state.single", ["test.succeed_with_changes name=itworked"] ) self.assertIsInstance(single, dict) for key, value in six.iteritems(single): self.assertIsInstance(value, dict) self.assertEqual(value["name"], ret_out["name"]) self.assertEqual(value["result"], ret_out["result"]) self.assertEqual(value["comment"], ret_out["comment"]) @slowTest def test_show_highstate(self): """ state.show_highstate with salt-ssh """ high = self.run_function("state.show_highstate") destpath = os.path.join(RUNTIME_VARS.TMP, "testfile") self.assertIsInstance(high, dict) self.assertIn(destpath, high) self.assertEqual(high[destpath]["__env__"], "base") @slowTest def test_state_high(self): """ state.high with salt-ssh """ ret_out = {"name": "itworked", "result": True, "comment": "Success!"} high = self.run_function( "state.high", ['"{"itworked": {"test": ["succeed_with_changes"]}}"'] ) self.assertIsInstance(high, dict) for key, value in six.iteritems(high): self.assertIsInstance(value, dict) self.assertEqual(value["name"], ret_out["name"]) self.assertEqual(value["result"], ret_out["result"]) self.assertEqual(value["comment"], ret_out["comment"]) @slowTest def test_show_lowstate(self): """ state.show_lowstate with salt-ssh """ low = self.run_function("state.show_lowstate") self.assertIsInstance(low, list) self.assertIsInstance(low[0], dict) @slowTest def test_state_low(self): """ state.low with salt-ssh """ ret_out = {"name": "itworked", "result": True, "comment": "Success!"} low = self.run_function( "state.low", ['"{"state": "test", "fun": "succeed_with_changes", "name": "itworked"}"'], ) self.assertIsInstance(low, dict) for key, value in six.iteritems(low): self.assertIsInstance(value, dict) self.assertEqual(value["name"], ret_out["name"]) self.assertEqual(value["result"], ret_out["result"]) self.assertEqual(value["comment"], ret_out["comment"]) @slowTest def test_state_request_check_clear(self): """ test state.request system with salt-ssh while also checking and clearing request """ request = self.run_function("state.request", [SSH_SLS], wipe=False) self._check_dict_ret(ret=request, val="__sls__", exp_ret=SSH_SLS) self._check_request() clear = self.run_function("state.clear_request", wipe=False) self._check_request(empty=True) @slowTest def test_state_run_request(self): """ test state.request system with salt-ssh while also running the request later """ request = self.run_function("state.request", [SSH_SLS], wipe=False) self._check_dict_ret(ret=request, val="__sls__", exp_ret=SSH_SLS) run = self.run_function("state.run_request", wipe=False) check_file = self.run_function("file.file_exists", [SSH_SLS_FILE], wipe=False) self.assertTrue(check_file) @flaky @slowTest def test_state_running(self): """ test state.running with salt-ssh """ def _run_in_background(): self.run_function("state.sls", ["running"], wipe=False) bg_thread = threading.Thread(target=_run_in_background) bg_thread.start() expected = 'The function "state.pkg" is running as' state_ret = [] for _ in range(30): time.sleep(5) get_sls = self.run_function("state.running", wipe=False) state_ret.append(get_sls) if expected in " ".join(get_sls): # We found the expected return break else: self.fail( "Did not find '{0}' in state.running return: {1}".format( expected, state_ret ) ) # make sure we wait until the earlier state is complete future = time.time() + 120 while True: if expected not in " ".join(self.run_function("state.running", wipe=False)): break if time.time() > future: self.fail( "state.pkg is still running overtime. Test did not clean up correctly." ) def tearDown(self): """ make sure to clean up any old ssh directories """ salt_dir = self.run_function("config.get", ["thin_dir"], wipe=False) self.assertIsInstance(salt_dir, six.string_types) if os.path.exists(salt_dir): shutil.rmtree(salt_dir) for test_file_path in glob.glob(SSH_SLS_FILE + "*"): os.remove(test_file_path)