salt/tests/unit/utils/test_minions.py

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

# Import python libs
from __future__ import absolute_import, unicode_literals
import sys

# Import Salt Libs
import salt.utils.minions

# Import Salt Testing Libs
from tests.support.unit import TestCase, skipIf
from tests.support.mock import (
    patch,
    MagicMock,
)

NODEGROUPS = {
    'group1': 'L@host1,host2,host3',
    'group2': ['G@foo:bar', 'or', 'web1*'],
    'group3': ['N@group1', 'or', 'N@group2'],
    'group4': ['host4', 'host5', 'host6'],
}

EXPECTED = {
    'group1': ['L@host1,host2,host3'],
    'group2': ['G@foo:bar', 'or', 'web1*'],
    'group3': ['(', '(', 'L@host1,host2,host3', ')', 'or', '(', 'G@foo:bar', 'or', 'web1*', ')', ')'],
    'group4': ['L@host4,host5,host6'],
}


class MinionsTestCase(TestCase):
    '''
    TestCase for salt.utils.minions module functions
    '''
    def test_nodegroup_comp(self):
        '''
        Test a simple string nodegroup
        '''
        for nodegroup in NODEGROUPS:
            expected = EXPECTED[nodegroup]
            ret = salt.utils.minions.nodegroup_comp(nodegroup, NODEGROUPS)
            self.assertEqual(ret, expected)


class CkMinionsTestCase(TestCase):
    '''
    TestCase for salt.utils.minions.CkMinions class
    '''
    def setUp(self):
        self.ckminions = salt.utils.minions.CkMinions({'minion_data_cache': True})

    def test_spec_check(self):
        # Test spec-only rule
        auth_list = ['@runner']
        ret = self.ckminions.spec_check(auth_list, 'test.arg', {}, 'runner')
        self.assertTrue(ret)
        ret = self.ckminions.spec_check(auth_list, 'test.arg', {}, 'wheel')
        self.assertFalse(ret)
        ret = self.ckminions.spec_check(auth_list, 'testarg', {}, 'runner')
        mock_ret = {'error': {'name': 'SaltInvocationError',
                              'message': 'A command invocation error occurred: Check syntax.'}}
        self.assertDictEqual(mock_ret, ret)

        # Test spec in plural form
        auth_list = ['@runners']
        ret = self.ckminions.spec_check(auth_list, 'test.arg', {}, 'runner')
        self.assertTrue(ret)
        ret = self.ckminions.spec_check(auth_list, 'test.arg', {}, 'wheel')
        self.assertFalse(ret)

        # Test spec with module.function restriction
        auth_list = [{'@runner': 'test.arg'}]
        ret = self.ckminions.spec_check(auth_list, 'test.arg', {}, 'runner')
        self.assertTrue(ret)
        ret = self.ckminions.spec_check(auth_list, 'test.arg', {}, 'wheel')
        self.assertFalse(ret)
        ret = self.ckminions.spec_check(auth_list, 'tes.arg', {}, 'runner')
        self.assertFalse(ret)
        ret = self.ckminions.spec_check(auth_list, 'test.ar', {}, 'runner')
        self.assertFalse(ret)

        # Test function name is a regex
        auth_list = [{'@runner': 'test.arg.*some'}]
        ret = self.ckminions.spec_check(auth_list, 'test.arg', {}, 'runner')
        self.assertFalse(ret)
        ret = self.ckminions.spec_check(auth_list, 'test.argsome', {}, 'runner')
        self.assertTrue(ret)
        ret = self.ckminions.spec_check(auth_list, 'test.arg_aaa_some', {}, 'runner')
        self.assertTrue(ret)

        # Test a list of funcs
        auth_list = [{'@runner': ['test.arg', 'jobs.active']}]
        ret = self.ckminions.spec_check(auth_list, 'test.arg', {}, 'runner')
        self.assertTrue(ret)
        ret = self.ckminions.spec_check(auth_list, 'jobs.active', {}, 'runner')
        self.assertTrue(ret)
        ret = self.ckminions.spec_check(auth_list, 'test.active', {}, 'runner')
        self.assertFalse(ret)
        ret = self.ckminions.spec_check(auth_list, 'jobs.arg', {}, 'runner')
        self.assertFalse(ret)

        # Test args-kwargs rules
        auth_list = [{
                '@runner': {
                    'test.arg': {
                        'args': ['1', '2'],
                        'kwargs': {
                            'aaa': 'bbb',
                            'ccc': 'ddd'
                            }
                        }
                    }
                }]
        ret = self.ckminions.spec_check(auth_list, 'test.arg', {}, 'runner')
        self.assertFalse(ret)
        args = {
                'arg': ['1', '2'],
                'kwarg': {'aaa': 'bbb', 'ccc': 'ddd'}
                }
        ret = self.ckminions.spec_check(auth_list, 'test.arg', args, 'runner')
        self.assertTrue(ret)
        args = {
                'arg': ['1', '2', '3'],
                'kwarg': {'aaa': 'bbb', 'ccc': 'ddd'}
                }
        ret = self.ckminions.spec_check(auth_list, 'test.arg', args, 'runner')
        self.assertTrue(ret)
        args = {
                'arg': ['1', '2'],
                'kwarg': {'aaa': 'bbb', 'ccc': 'ddd', 'zzz': 'zzz'}
                }
        ret = self.ckminions.spec_check(auth_list, 'test.arg', args, 'runner')
        self.assertTrue(ret)
        args = {
                'arg': ['1', '2'],
                'kwarg': {'aaa': 'bbb', 'ccc': 'ddc'}
                }
        ret = self.ckminions.spec_check(auth_list, 'test.arg', args, 'runner')
        self.assertFalse(ret)
        args = {
                'arg': ['1', '2'],
                'kwarg': {'aaa': 'bbb'}
                }
        ret = self.ckminions.spec_check(auth_list, 'test.arg', args, 'runner')
        self.assertFalse(ret)
        args = {
                'arg': ['1', '3'],
                'kwarg': {'aaa': 'bbb', 'ccc': 'ddd'}
                }
        ret = self.ckminions.spec_check(auth_list, 'test.arg', args, 'runner')
        self.assertFalse(ret)
        args = {
                'arg': ['1'],
                'kwarg': {'aaa': 'bbb', 'ccc': 'ddd'}
                }
        ret = self.ckminions.spec_check(auth_list, 'test.arg', args, 'runner')
        self.assertFalse(ret)
        args = {
                'kwarg': {'aaa': 'bbb', 'ccc': 'ddd'}
                }
        ret = self.ckminions.spec_check(auth_list, 'test.arg', args, 'runner')
        self.assertFalse(ret)
        args = {
                'arg': ['1', '2'],
                }
        ret = self.ckminions.spec_check(auth_list, 'test.arg', args, 'runner')
        self.assertFalse(ret)

        # Test kwargs only
        auth_list = [{
                '@runner': {
                    'test.arg': {
                        'kwargs': {
                            'aaa': 'bbb',
                            'ccc': 'ddd'
                            }
                        }
                    }
                }]
        ret = self.ckminions.spec_check(auth_list, 'test.arg', {}, 'runner')
        self.assertFalse(ret)
        args = {
                'arg': ['1', '2'],
                'kwarg': {'aaa': 'bbb', 'ccc': 'ddd'}
                }
        ret = self.ckminions.spec_check(auth_list, 'test.arg', args, 'runner')
        self.assertTrue(ret)

        # Test args only
        auth_list = [{
                '@runner': {
                    'test.arg': {
                        'args': ['1', '2']
                        }
                    }
                }]
        ret = self.ckminions.spec_check(auth_list, 'test.arg', {}, 'runner')
        self.assertFalse(ret)
        args = {
                'arg': ['1', '2'],
                'kwarg': {'aaa': 'bbb', 'ccc': 'ddd'}
                }
        ret = self.ckminions.spec_check(auth_list, 'test.arg', args, 'runner')
        self.assertTrue(ret)

        # Test list of args
        auth_list = [{'@runner': [{'test.arg': [{'args': ['1', '2'],
                                                 'kwargs': {'aaa': 'bbb',
                                                            'ccc': 'ddd'
                                                            }
                                                 },
                                                {'args': ['2', '3'],
                                                 'kwargs': {'aaa': 'aaa',
                                                            'ccc': 'ccc'
                                                            }
                                                 }]
                                   }]
                      }]
        args = {
                'arg': ['1', '2'],
                'kwarg': {'aaa': 'bbb', 'ccc': 'ddd'}
                }
        ret = self.ckminions.spec_check(auth_list, 'test.arg', args, 'runner')
        self.assertTrue(ret)
        args = {
                'arg': ['2', '3'],
                'kwarg': {'aaa': 'aaa', 'ccc': 'ccc'}
                }
        ret = self.ckminions.spec_check(auth_list, 'test.arg', args, 'runner')
        self.assertTrue(ret)

        # Test @module form
        auth_list = ['@jobs']
        ret = self.ckminions.spec_check(auth_list, 'jobs.active', {}, 'runner')
        self.assertTrue(ret)
        ret = self.ckminions.spec_check(auth_list, 'jobs.active', {}, 'wheel')
        self.assertTrue(ret)
        ret = self.ckminions.spec_check(auth_list, 'test.arg', {}, 'runner')
        self.assertFalse(ret)
        ret = self.ckminions.spec_check(auth_list, 'job.arg', {}, 'runner')
        self.assertFalse(ret)

        # Test @module: function
        auth_list = [{'@jobs': 'active'}]
        ret = self.ckminions.spec_check(auth_list, 'jobs.active', {}, 'runner')
        self.assertTrue(ret)
        ret = self.ckminions.spec_check(auth_list, 'jobs.active', {}, 'wheel')
        self.assertTrue(ret)
        ret = self.ckminions.spec_check(auth_list, 'jobs.active_jobs', {}, 'runner')
        self.assertTrue(ret)
        ret = self.ckminions.spec_check(auth_list, 'jobs.activ', {}, 'runner')
        self.assertFalse(ret)

        # Test @module: [functions]
        auth_list = [{'@jobs': ['active', 'li']}]
        ret = self.ckminions.spec_check(auth_list, 'jobs.active', {}, 'runner')
        self.assertTrue(ret)
        ret = self.ckminions.spec_check(auth_list, 'jobs.list_jobs', {}, 'runner')
        self.assertTrue(ret)
        ret = self.ckminions.spec_check(auth_list, 'jobs.last_run', {}, 'runner')
        self.assertFalse(ret)

        # Test @module: function with args
        auth_list = [{'@jobs': {'active': {'args': ['1', '2'],
                                           'kwargs': {'a': 'b', 'c': 'd'}}}}]
        args = {'arg': ['1', '2'],
                'kwarg': {'a': 'b', 'c': 'd'}}
        ret = self.ckminions.spec_check(auth_list, 'jobs.active', args, 'runner')
        self.assertTrue(ret)
        ret = self.ckminions.spec_check(auth_list, 'jobs.active', {}, 'runner')
        self.assertFalse(ret)

    @patch('salt.utils.minions.CkMinions._pki_minions', MagicMock(return_value=['alpha', 'beta', 'gamma']))
    def test_auth_check(self):
        # Test function-only rule
        auth_list = ['test.ping']
        ret = self.ckminions.auth_check(auth_list, 'test.ping', None, 'alpha')
        self.assertTrue(ret)
        ret = self.ckminions.auth_check(auth_list, 'test.arg', None, 'alpha')
        self.assertFalse(ret)

        # Test minion and function
        auth_list = [{'alpha': 'test.ping'}]
        ret = self.ckminions.auth_check(auth_list, 'test.ping', None, 'alpha')
        self.assertTrue(ret)
        ret = self.ckminions.auth_check(auth_list, 'test.arg', None, 'alpha')
        self.assertFalse(ret)
        ret = self.ckminions.auth_check(auth_list, 'test.ping', None, 'beta')
        self.assertFalse(ret)

        # Test function list
        auth_list = [{'*': ['test.*', 'saltutil.cmd']}]
        ret = self.ckminions.auth_check(auth_list, 'test.arg', None, 'alpha')
        self.assertTrue(ret)
        ret = self.ckminions.auth_check(auth_list, 'test.ping', None, 'beta')
        self.assertTrue(ret)
        ret = self.ckminions.auth_check(auth_list, 'saltutil.cmd', None, 'gamma')
        self.assertTrue(ret)
        ret = self.ckminions.auth_check(auth_list, 'saltutil.running', None, 'gamma')
        self.assertFalse(ret)

        # Test an args and kwargs rule
        auth_list = [{
                'alpha': {
                    'test.arg': {
                        'args': ['1', '2'],
                        'kwargs': {
                            'aaa': 'bbb',
                            'ccc': 'ddd'
                            }
                        }
                    }
                }]
        ret = self.ckminions.auth_check(auth_list, 'test.arg', None, 'runner')
        self.assertFalse(ret)
        ret = self.ckminions.auth_check(auth_list, 'test.arg', [], 'runner')
        self.assertFalse(ret)
        args = ['1', '2', {'aaa': 'bbb', 'ccc': 'ddd', '__kwarg__': True}]
        ret = self.ckminions.auth_check(auth_list, 'test.arg', args, 'runner')
        self.assertTrue(ret)
        args = ['1', '2', '3', {'aaa': 'bbb', 'ccc': 'ddd', 'eee': 'fff', '__kwarg__': True}]
        ret = self.ckminions.auth_check(auth_list, 'test.arg', args, 'runner')
        self.assertTrue(ret)
        args = ['1', {'aaa': 'bbb', 'ccc': 'ddd', '__kwarg__': True}]
        ret = self.ckminions.auth_check(auth_list, 'test.arg', args, 'runner')
        self.assertFalse(ret)
        args = ['1', '2', {'aaa': 'bbb', '__kwarg__': True}]
        ret = self.ckminions.auth_check(auth_list, 'test.arg', args, 'runner')
        self.assertFalse(ret)
        args = ['1', '3', {'aaa': 'bbb', 'ccc': 'ddd', '__kwarg__': True}]
        ret = self.ckminions.auth_check(auth_list, 'test.arg', args, 'runner')
        self.assertFalse(ret)
        args = ['1', '2', {'aaa': 'bbb', 'ccc': 'fff', '__kwarg__': True}]
        ret = self.ckminions.auth_check(auth_list, 'test.arg', args, 'runner')
        self.assertFalse(ret)

        # Test kwargs only rule
        auth_list = [{
                'alpha': {
                    'test.arg': {
                        'kwargs': {
                            'aaa': 'bbb',
                            'ccc': 'ddd'
                            }
                        }
                    }
                }]
        args = ['1', '2', {'aaa': 'bbb', 'ccc': 'ddd', '__kwarg__': True}]
        ret = self.ckminions.auth_check(auth_list, 'test.arg', args, 'runner')
        self.assertTrue(ret)
        args = [{'aaa': 'bbb', 'ccc': 'ddd', 'eee': 'fff', '__kwarg__': True}]
        ret = self.ckminions.auth_check(auth_list, 'test.arg', args, 'runner')
        self.assertTrue(ret)

        # Test args only rule
        auth_list = [{
                'alpha': {
                    'test.arg': {
                        'args': ['1', '2'],
                        }
                    }
                }]
        args = ['1', '2', {'aaa': 'bbb', 'ccc': 'ddd', '__kwarg__': True}]
        ret = self.ckminions.auth_check(auth_list, 'test.arg', args, 'runner')
        self.assertTrue(ret)
        args = ['1', '2']
        ret = self.ckminions.auth_check(auth_list, 'test.arg', args, 'runner')
        self.assertTrue(ret)


@skipIf(sys.version_info < (2, 7), 'Python 2.7 needed for dictionary equality assertions')
class TargetParseTestCase(TestCase):

    def test_parse_grains_target(self):
        '''
        Ensure proper parsing for grains
        '''
        g_tgt = 'G@a:b'
        ret = salt.utils.minions.parse_target(g_tgt)
        self.assertDictEqual(ret, {'engine': 'G', 'pattern': 'a:b', 'delimiter': None})

    def test_parse_grains_pcre_target(self):
        '''
        Ensure proper parsing for grains PCRE matching
        '''
        p_tgt = 'P@a:b'
        ret = salt.utils.minions.parse_target(p_tgt)
        self.assertDictEqual(ret, {'engine': 'P', 'pattern': 'a:b', 'delimiter': None})

    def test_parse_pillar_pcre_target(self):
        '''
        Ensure proper parsing for pillar PCRE matching
        '''
        j_tgt = 'J@a:b'
        ret = salt.utils.minions.parse_target(j_tgt)
        self.assertDictEqual(ret, {'engine': 'J', 'pattern': 'a:b', 'delimiter': None})

    def test_parse_list_target(self):
        '''
        Ensure proper parsing for list matching
        '''
        l_tgt = 'L@a:b'
        ret = salt.utils.minions.parse_target(l_tgt)
        self.assertDictEqual(ret, {'engine': 'L', 'pattern': 'a:b', 'delimiter': None})

    def test_parse_nodegroup_target(self):
        '''
        Ensure proper parsing for pillar matching
        '''
        n_tgt = 'N@a:b'
        ret = salt.utils.minions.parse_target(n_tgt)
        self.assertDictEqual(ret, {'engine': 'N', 'pattern': 'a:b', 'delimiter': None})

    def test_parse_subnet_target(self):
        '''
        Ensure proper parsing for subnet matching
        '''
        s_tgt = 'S@a:b'
        ret = salt.utils.minions.parse_target(s_tgt)
        self.assertDictEqual(ret, {'engine': 'S', 'pattern': 'a:b', 'delimiter': None})

    def test_parse_minion_pcre_target(self):
        '''
        Ensure proper parsing for minion PCRE matching
        '''
        e_tgt = 'E@a:b'
        ret = salt.utils.minions.parse_target(e_tgt)
        self.assertDictEqual(ret, {'engine': 'E', 'pattern': 'a:b', 'delimiter': None})

    def test_parse_range_target(self):
        '''
        Ensure proper parsing for range matching
        '''
        r_tgt = 'R@a:b'
        ret = salt.utils.minions.parse_target(r_tgt)
        self.assertDictEqual(ret, {'engine': 'R', 'pattern': 'a:b', 'delimiter': None})

    def test_parse_multiword_target(self):
        '''
        Ensure proper parsing for multi-word targets

        Refs https://github.com/saltstack/salt/issues/37231
        '''
        mw_tgt = 'G@a:b c'
        ret = salt.utils.minions.parse_target(mw_tgt)
        self.assertEqual(ret['pattern'], 'a:b c')


class NodegroupCompTest(TestCase):
    '''
    Test nodegroup comparisons found in
    salt.utils.minions.nodgroup_comp()
    '''

    def test_simple_nodegroup(self):
        '''
        Smoke test a very simple nodegroup. No recursion.
        '''
        simple_nodegroup = {'group1': 'L@foo.domain.com,bar.domain.com,baz.domain.com or bl*.domain.com'}

        ret = salt.utils.minions.nodegroup_comp('group1', simple_nodegroup)
        expected_ret = ['L@foo.domain.com,bar.domain.com,baz.domain.com', 'or', 'bl*.domain.com']
        self.assertListEqual(ret, expected_ret)

    def test_simple_expression_nodegroup(self):
        '''
        Smoke test a nodegroup with a simple expression. No recursion.
        '''
        simple_nodegroup = {'group1': '[foo,bar,baz].domain.com'}

        ret = salt.utils.minions.nodegroup_comp('group1', simple_nodegroup)
        expected_ret = ['E@[foo,bar,baz].domain.com']
        self.assertListEqual(ret, expected_ret)

    def test_simple_recurse(self):
        '''
        Test a case where one nodegroup contains a second nodegroup
        '''
        referenced_nodegroups = {
                'group1': 'L@foo.domain.com,bar.domain.com,baz.domain.com or bl*.domain.com',
                'group2': 'G@os:Debian and N@group1'
                }

        ret = salt.utils.minions.nodegroup_comp('group2', referenced_nodegroups)
        expected_ret = [
                '(',
                'G@os:Debian',
                'and',
                '(',
                'L@foo.domain.com,bar.domain.com,baz.domain.com',
                'or',
                'bl*.domain.com',
                ')',
                ')'
                ]
        self.assertListEqual(ret, expected_ret)

    def test_circular_nodegroup_reference(self):
        '''
        Test to see what happens if A refers to B
        and B in turn refers back to A
        '''
        referenced_nodegroups = {
                'group1': 'N@group2',
                'group2': 'N@group1'
                }

        # If this works, it should also print an error to the console
        ret = salt.utils.minions.nodegroup_comp('group1', referenced_nodegroups)
        self.assertEqual(ret, [])