salt/tests/unit/utils/test_roster_matcher.py

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

"""
Test generic roster matching utility.
"""

# Import Python libs
from __future__ import absolute_import, print_function, unicode_literals

import os

# Import Salt Libs
import salt.config
import salt.loader
import salt.utils.roster_matcher

# Import Salt Testing Libs
from tests.support import mixins
from tests.support.runtests import RUNTIME_VARS
from tests.support.unit import TestCase, skipIf

EXPECTED = {
    "host1": {
        "host": "host1",
        "passwd": "test123",
        "minion_opts": {
            "escape_pods": 2,
            "halon_system_timeout": 30,
            "self_destruct_countdown": 60,
            "some_server": "foo.southeast.example.com",
        },
    },
    "host2": {
        "host": "host2",
        "passwd": "test123",
        "minion_opts": {
            "escape_pods": 2,
            "halon_system_timeout": 30,
            "self_destruct_countdown": 60,
            "some_server": "foo.southeast.example.com",
        },
    },
    "host3": {"host": "host3", "passwd": "test123", "minion_opts": {}},
    "host4": "host4.example.com",  # For testing get_data -> string_types branch
    "host5": None,  # For testing get_data -> False
}


class RosterMatcherTestCase(TestCase, mixins.LoaderModuleMockMixin):
    """
    Test the RosterMatcher Utility
    """

    def setup_loader_modules(self):
        opts = salt.config.master_config(
            os.path.join(RUNTIME_VARS.TMP_CONF_DIR, "master")
        )
        utils = salt.loader.utils(opts, whitelist=["json", "stringutils"])
        runner = salt.loader.runner(opts, utils=utils, whitelist=["salt"])
        return {
            salt.utils.roster_matcher: {
                "__utils__": utils,
                "__opts__": {
                    "ssh_list_nodegroups": {
                        "list_nodegroup": ["host5", "host1", "host2"],
                        "string_nodegroup": "host3,host5,host1,host4",
                    }
                },
                "__runner__": runner,
            }
        }

    def test_get_data(self):
        """
        Test that the get_data method returns the expected dictionaries.
        """
        # We don't care about tgt and tgt_type here.
        roster_matcher = salt.utils.roster_matcher.RosterMatcher(
            EXPECTED, "tgt", "tgt_type"
        )
        self.assertEqual(EXPECTED["host1"], roster_matcher.get_data("host1"))
        self.assertEqual(EXPECTED["host2"], roster_matcher.get_data("host2"))
        self.assertEqual(EXPECTED["host3"], roster_matcher.get_data("host3"))
        self.assertEqual({"host": EXPECTED["host4"]}, roster_matcher.get_data("host4"))

    def test_ret_glob_minions(self):
        """
        Test that we return minions matching a glob.
        """
        result = salt.utils.roster_matcher.targets(EXPECTED, "*[245]", "glob")
        self.assertNotIn("host1", result)
        self.assertIn("host2", result)
        self.assertNotIn("host3", result)
        self.assertIn("host4", result)
        self.assertNotIn("host5", result)

    def test_ret_pcre_minions(self):
        """
        Test that we return minions matching a regular expression.
        """
        result = salt.utils.roster_matcher.targets(EXPECTED, ".*[^23]$", "pcre")
        self.assertIn("host1", result)
        self.assertNotIn("host2", result)
        self.assertNotIn("host3", result)
        self.assertIn("host4", result)
        self.assertNotIn("host5", result)

    def test_ret_literal_list_minions(self):
        """
        Test that we return minions that are in a literal list.
        """
        result = salt.utils.roster_matcher.targets(
            EXPECTED, ["host1", "host2", "host5"], "list"
        )
        self.assertIn("host1", result)
        self.assertIn("host2", result)
        self.assertNotIn("host3", result)
        self.assertNotIn("host4", result)
        self.assertNotIn("host5", result)

    def test_ret_comma_delimited_string_minions(self):
        """
        Test that we return minions that are in a comma-delimited
        string of literal minion names.
        """
        result = salt.utils.roster_matcher.targets(
            EXPECTED, "host5,host3,host2", "list"
        )
        self.assertNotIn("host1", result)
        self.assertIn("host2", result)
        self.assertIn("host3", result)
        self.assertNotIn("host4", result)
        self.assertNotIn("host5", result)

    def test_ret_oops_minions(self):
        """
        Test that we return no minions when we try to use a matching
        method that is not defined.
        """
        result = salt.utils.roster_matcher.targets(EXPECTED, None, "xyzzy")
        self.assertEqual({}, result)

    def test_ret_literal_list_nodegroup_minions(self):
        """
        Test that we return minions that are in a nodegroup
        where the nodegroup expresses a literal list of minion names.
        """
        result = salt.utils.roster_matcher.targets(
            EXPECTED, "list_nodegroup", "nodegroup"
        )
        self.assertIn("host1", result)
        self.assertIn("host2", result)
        self.assertNotIn("host3", result)
        self.assertNotIn("host4", result)
        self.assertNotIn("host5", result)

    def test_ret_comma_delimited_string_nodegroup_minions(self):
        """
        Test that we return minions that are in a nodegroup
        where the nodegroup expresses a comma delimited string
        of minion names.
        """
        result = salt.utils.roster_matcher.targets(
            EXPECTED, "string_nodegroup", "nodegroup"
        )
        self.assertIn("host1", result)
        self.assertNotIn("host2", result)
        self.assertIn("host3", result)
        self.assertIn("host4", result)
        self.assertNotIn("host5", result)

    def test_ret_no_range_installed_minions(self):
        """
        Test that range matcher raises a Runtime Error if seco.range is not installed.
        """
        salt.utils.roster_matcher.HAS_RANGE = False
        with self.assertRaises(RuntimeError):
            salt.utils.roster_matcher.targets(EXPECTED, None, "range")

    @skipIf(not salt.utils.roster_matcher.HAS_RANGE, "seco.range is not installed")
    def test_ret_range_minions(self):
        """
        Test that range matcher raises a Runtime Error if seco.range is not installed.
        """
        self.fail("Not implemented")