salt/tests/unit/test_daemons.py

# -*- coding: utf-8 -*-
'''
    :codeauthor: :email:`Bo Maryniuk <bo@suse.de>`
'''

# Import python libs
from __future__ import absolute_import
import multiprocessing

# Import Salt Testing libs
from tests.support.unit import TestCase, skipIf
from tests.support.mock import patch, MagicMock, NO_MOCK, NO_MOCK_REASON
from tests.support.mixins import SaltClientTestCaseMixin

# Import Salt libs
import salt.cli.daemons as daemons


class LoggerMock(object):
    '''
    Logger data collector
    '''

    def __init__(self):
        '''
        init
        :return:
        '''
        self.reset()

    def reset(self):
        '''
        Reset values

        :return:
        '''
        self.messages = list()

    def info(self, data):
        '''
        Collects the data from the logger of info type.

        :param data:
        :return:
        '''
        self.messages.append({'message': data, 'type': 'info'})

    def warning(self, data):
        '''
        Collects the data from the logger of warning type.

        :param data:
        :return:
        '''
        self.messages.append({'message': data, 'type': 'warning'})

    def has_message(self, msg, log_type=None):
        '''
        Check if log has message.

        :param data:
        :return:
        '''
        for data in self.messages:
            if (data['type'] == log_type or not log_type) and data['message'].find(msg) > -1:
                return True

        return False


def _master_exec_test(child_pipe):
    def _create_master():
        '''
        Create master instance
        :return:
        '''
        obj = daemons.Master()
        obj.config = {'user': 'dummy', 'hash_type': alg}
        for attr in ['start_log_info', 'prepare', 'shutdown', 'master']:
            setattr(obj, attr, MagicMock())

        return obj

    _logger = LoggerMock()
    ret = True
    with patch('salt.cli.daemons.check_user', MagicMock(return_value=True)):
        with patch('salt.cli.daemons.log', _logger):
            for alg in ['md5', 'sha1']:
                _create_master().start()
                ret = ret and _logger.messages \
                      and _logger.has_message('Do not use {alg}'.format(alg=alg),
                                              log_type='warning')

            _logger.reset()

            for alg in ['sha224', 'sha256', 'sha384', 'sha512']:
                _create_master().start()
                ret = ret and _logger.messages \
                      and not _logger.has_message('Do not use ')
            child_pipe.send(ret)
            child_pipe.close()


def _minion_exec_test(child_pipe):
    def _create_minion():
        '''
        Create minion instance
        :return:
        '''
        obj = daemons.Minion()
        obj.config = {'user': 'dummy', 'hash_type': alg}
        for attr in ['start_log_info', 'prepare', 'shutdown']:
            setattr(obj, attr, MagicMock())
        setattr(obj, 'minion', MagicMock(restart=False))

        return obj

    ret = True
    _logger = LoggerMock()
    with patch('salt.cli.daemons.check_user', MagicMock(return_value=True)):
        with patch('salt.cli.daemons.log', _logger):
            for alg in ['md5', 'sha1']:
                _create_minion().start()
                ret = ret and _logger.messages \
                      and _logger.has_message('Do not use {alg}'.format(alg=alg),
                                              log_type='warning')
            _logger.reset()

            for alg in ['sha224', 'sha256', 'sha384', 'sha512']:
                _create_minion().start()
                ret = ret and _logger.messages \
                      and not _logger.has_message('Do not use ')

        child_pipe.send(ret)
        child_pipe.close()


def _proxy_exec_test(child_pipe):
    def _create_proxy_minion():
        '''
        Create proxy minion instance
        :return:
        '''
        obj = daemons.ProxyMinion()
        obj.config = {'user': 'dummy', 'hash_type': alg}
        for attr in ['minion', 'start_log_info', 'prepare', 'shutdown', 'tune_in']:
            setattr(obj, attr, MagicMock())

        obj.minion.restart = False
        return obj

    ret = True
    _logger = LoggerMock()
    with patch('salt.cli.daemons.check_user', MagicMock(return_value=True)):
        with patch('salt.cli.daemons.log', _logger):
            for alg in ['md5', 'sha1']:
                _create_proxy_minion().start()
                ret = ret and _logger.messages \
                      and _logger.has_message('Do not use {alg}'.format(alg=alg),
                                              log_type='warning')

            _logger.reset()

            for alg in ['sha224', 'sha256', 'sha384', 'sha512']:
                _create_proxy_minion().start()
                ret = ret and _logger.messages \
                      and not _logger.has_message('Do not use ')
    child_pipe.send(ret)
    child_pipe.close()


def _syndic_exec_test(child_pipe):
    def _create_syndic():
        '''
        Create syndic instance
        :return:
        '''
        obj = daemons.Syndic()
        obj.config = {'user': 'dummy', 'hash_type': alg}
        for attr in ['syndic', 'start_log_info', 'prepare', 'shutdown']:
            setattr(obj, attr, MagicMock())

        return obj

    ret = True
    _logger = LoggerMock()
    with patch('salt.cli.daemons.check_user', MagicMock(return_value=True)):
        with patch('salt.cli.daemons.log', _logger):
            for alg in ['md5', 'sha1']:
                _create_syndic().start()
                ret = ret and _logger.messages \
                      and _logger.has_message('Do not use {alg}'.format(alg=alg),
                                              log_type='warning')

            _logger.reset()

            for alg in ['sha224', 'sha256', 'sha384', 'sha512']:
                _create_syndic().start()
                ret = ret and _logger.messages \
                      and not _logger.has_message('Do not use ')

    child_pipe.send(ret)
    child_pipe.close()


@skipIf(NO_MOCK, NO_MOCK_REASON)
class DaemonsStarterTestCase(TestCase, SaltClientTestCaseMixin):
    '''
    Unit test for the daemons starter classes.
    '''

    def _multiproc_exec_test(self, exec_test):
        m_parent, m_child = multiprocessing.Pipe()
        p_ = multiprocessing.Process(target=exec_test, args=(m_child,))
        p_.start()
        self.assertTrue(m_parent.recv())
        p_.join()

    def test_master_daemon_hash_type_verified(self):
        '''
        Verify if Master is verifying hash_type config option.

        :return:
        '''
        self._multiproc_exec_test(_master_exec_test)

    def test_minion_daemon_hash_type_verified(self):
        '''
        Verify if Minion is verifying hash_type config option.

        :return:
        '''
        self._multiproc_exec_test(_minion_exec_test)

    def test_proxy_minion_daemon_hash_type_verified(self):
        '''
        Verify if ProxyMinion is verifying hash_type config option.

        :return:
        '''
        self._multiproc_exec_test(_proxy_exec_test)

    def test_syndic_daemon_hash_type_verified(self):
        '''
        Verify if Syndic is verifying hash_type config option.

        :return:
        '''
        self._multiproc_exec_test(_syndic_exec_test)