salt/tests/integration/ssh/test_state.py

# -*- coding: utf-8 -*-

from __future__ import absolute_import, print_function, unicode_literals

import logging
import os
import shutil
import threading
import time

import pytest
from salt.ext import six
from salt.ext.six.moves import range
from tests.support.case import SSHCase
from tests.support.runtests import RUNTIME_VARS

SSH_SLS = "ssh_state_tests"
SSH_SLS_FILE = "/tmp/test"

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,
            )

    @pytest.mark.slow_test(seconds=60)  # Test takes >30 and <=60 seconds
    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", ["/tmp/test"])
        self.assertTrue(check_file)

    @pytest.mark.slow_test(seconds=60)  # Test takes >30 and <=60 seconds
    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 /tmp/test is set to be changed\nNote: No changes made, actual changes may\nbe different due to other states.",
        )

        # 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", ["/tmp/test"])
        self.assertTrue(check_file)

    @pytest.mark.slow_test(seconds=30)  # Test takes >10 and <=30 seconds
    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

    @pytest.mark.slow_test(seconds=30)  # Test takes >10 and <=30 seconds
    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)

    @pytest.mark.slow_test(seconds=30)  # Test takes >10 and <=30 seconds
    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"]})

    @pytest.mark.slow_test(seconds=30)  # Test takes >10 and <=30 seconds
    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"])

    @pytest.mark.slow_test(seconds=30)  # Test takes >10 and <=30 seconds
    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")

    @pytest.mark.slow_test(seconds=30)  # Test takes >10 and <=30 seconds
    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"])

    @pytest.mark.slow_test(seconds=30)  # Test takes >10 and <=30 seconds
    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)

    @pytest.mark.slow_test(seconds=30)  # Test takes >10 and <=30 seconds
    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"])

    @pytest.mark.slow_test(seconds=60)  # Test takes >30 and <=60 seconds
    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)

    @pytest.mark.slow_test(seconds=60)  # Test takes >30 and <=60 seconds
    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)

    @pytest.mark.flaky(max_runs=4)
    @pytest.mark.slow_test(seconds=60)  # Test takes >30 and <=60 seconds
    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)

        if os.path.exists(SSH_SLS_FILE):
            os.remove(SSH_SLS_FILE)