salt/tests/integration/ssh/test_state.py

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