salt/tests/integration/ssh/test_state.py

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

# Import Python libs
from __future__ import absolute_import, print_function, unicode_literals
import os
import shutil
import threading
import time
import logging

# Import Salt Testing Libs
from tests.support.case import SSHCase
from tests.support.helpers import flaky
from tests.support.paths import TMP

# Import Salt Libs
from salt.ext import six
from salt.ext.six.moves import range  # pylint: disable=redefined-builtin

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)

    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)

    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)

    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

    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)

    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']})

    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'])

    def test_show_highstate(self):
        '''
        state.show_highstate with salt-ssh
        '''
        high = self.run_function('state.show_highstate')
        destpath = os.path.join(TMP, 'testfile')
        self.assertIsInstance(high, dict)
        self.assertIn(destpath, high)
        self.assertEqual(high[destpath]['__env__'], 'base')

    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'])

    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)

    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'])

    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)

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