salt/tests/integration/modules/test_grains.py

# -*- coding: utf-8 -*-
"""
Test the grains module
"""

from __future__ import absolute_import, print_function, unicode_literals

import logging
import os
import pprint
import time

import pytest
from salt.ext.six.moves import range
from tests.support.case import ModuleCase
from tests.support.unit import skipIf

log = logging.getLogger(__name__)


@pytest.mark.windows_whitelisted
class TestModulesGrains(ModuleCase):
    """
    Test the grains module
    """

    @pytest.mark.slow_test(seconds=30)  # Test takes >10 and <=30 seconds
    def test_items(self):
        """
        grains.items
        """
        opts = self.minion_opts
        self.assertEqual(
            self.run_function("grains.items")["test_grain"],
            opts["grains"]["test_grain"],
        )

    @pytest.mark.slow_test(seconds=30)  # Test takes >10 and <=30 seconds
    def test_item(self):
        """
        grains.item
        """
        opts = self.minion_opts
        self.assertEqual(
            self.run_function("grains.item", ["test_grain"])["test_grain"],
            opts["grains"]["test_grain"],
        )

    @pytest.mark.slow_test(seconds=30)  # Test takes >10 and <=30 seconds
    def test_ls(self):
        """
        grains.ls
        """
        check_for = (
            "cpu_flags",
            "cpu_model",
            "cpuarch",
            "domain",
            "fqdn",
            "fqdns",
            "gid",
            "groupname",
            "host",
            "kernel",
            "kernelrelease",
            "kernelversion",
            "localhost",
            "mem_total",
            "num_cpus",
            "os",
            "os_family",
            "path",
            "pid",
            "ps",
            "pythonpath",
            "pythonversion",
            "saltpath",
            "saltversion",
            "uid",
            "username",
            "virtual",
        )
        lsgrains = self.run_function("grains.ls")
        os = self.run_function("grains.get", ["os"])
        for grain in check_for:
            if os == "Windows" and grain in ["cpu_flags", "gid", "groupname", "uid"]:
                continue
            self.assertTrue(grain in lsgrains)

    @skipIf(
        os.environ.get("TRAVIS_PYTHON_VERSION", None) is not None,
        "Travis environment can't keep up with salt refresh",
    )
    @pytest.mark.slow_test(seconds=30)  # Test takes >10 and <=30 seconds
    def test_set_val(self):
        """
        test grains.set_val
        """
        self.assertEqual(
            self.run_function("grains.setval", ["setgrain", "grainval"]),
            {"setgrain": "grainval"},
        )
        time.sleep(5)
        ret = self.run_function("grains.item", ["setgrain"])
        if not ret:
            # Sleep longer, sometimes test systems get bogged down
            time.sleep(20)
            ret = self.run_function("grains.item", ["setgrain"])
        self.assertTrue(ret)

    @pytest.mark.slow_test(seconds=30)  # Test takes >10 and <=30 seconds
    def test_get(self):
        """
        test grains.get
        """
        self.assertEqual(self.run_function("grains.get", ["level1:level2"]), "foo")

    @pytest.mark.slow_test(seconds=120)  # Test takes >60 and <=120 seconds
    def test_get_core_grains(self):
        """
        test to ensure some core grains are returned
        """
        grains = ("os", "os_family", "osmajorrelease", "osrelease", "osfullname", "id")
        os = self.run_function("grains.get", ["os"])

        for grain in grains:
            get_grain = self.run_function("grains.get", [grain])
            log.debug("Value of '%s' grain: '%s'", grain, get_grain)
            if os == "Arch" and grain in ["osmajorrelease"]:
                self.assertEqual(get_grain, "")
                continue
            if os == "Windows" and grain in ["osmajorrelease"]:
                self.assertEqual(get_grain, "")
                continue

            self.assertTrue(get_grain)

    @pytest.mark.slow_test(seconds=60)  # Test takes >30 and <=60 seconds
    def test_get_grains_int(self):
        """
        test to ensure int grains
        are returned as integers
        """
        grains = ["num_cpus", "mem_total", "num_gpus", "uid"]
        os = self.run_function("grains.get", ["os"])
        for grain in grains:
            get_grain = self.run_function("grains.get", [grain])
            if os == "Windows" and grain in ["uid"]:
                self.assertEqual(get_grain, "")
                continue
            self.assertIsInstance(
                get_grain, int, msg="grain: {0} is not an int or empty".format(grain)
            )


@pytest.mark.windows_whitelisted
class GrainsAppendTestCase(ModuleCase):
    """
    Tests written specifically for the grains.append function.
    """

    GRAIN_KEY = "salttesting-grain-key"
    GRAIN_VAL = "my-grain-val"

    def setUp(self):
        # Start off with an empty list
        self.run_function("grains.setval", [self.GRAIN_KEY, []])
        if not self.wait_for_grain(self.GRAIN_KEY, []):
            raise Exception("Failed to set grain")
        self.addCleanup(self.cleanup_grain)

    def cleanup_grain(self):
        self.run_function("grains.setval", [self.GRAIN_KEY, []])
        if not self.wait_for_grain(self.GRAIN_KEY, []):
            raise Exception("Failed to set grain")

    @pytest.mark.slow_test(seconds=60)  # Test takes >30 and <=60 seconds
    def test_grains_append(self):
        """
        Tests the return of a simple grains.append call.
        """
        ret = self.run_function("grains.append", [self.GRAIN_KEY, self.GRAIN_VAL])
        self.assertEqual(ret[self.GRAIN_KEY], [self.GRAIN_VAL])

    @pytest.mark.slow_test(seconds=120)  # Test takes >60 and <=120 seconds
    def test_grains_append_val_already_present(self):
        """
        Tests the return of a grains.append call when the value is already
        present in the grains list.
        """
        msg = "The val {0} was already in the list " "salttesting-grain-key".format(
            self.GRAIN_VAL
        )

        # First, make sure the test grain is present
        ret = self.run_function("grains.append", [self.GRAIN_KEY, self.GRAIN_VAL])
        self.assertEqual(ret[self.GRAIN_KEY], [self.GRAIN_VAL])

        # Now try to append again
        ret = self.run_function("grains.append", [self.GRAIN_KEY, self.GRAIN_VAL])
        self.assertTrue(self.wait_for_grain(self.GRAIN_KEY, [self.GRAIN_VAL]))
        if not ret or isinstance(ret, dict):
            # Sleep for a bit, sometimes the second "append" runs too quickly
            time.sleep(5)
            ret = self.run_function("grains.append", [self.GRAIN_KEY, self.GRAIN_VAL])

        assert msg == ret

    @pytest.mark.flaky(max_runs=4)
    @pytest.mark.slow_test(seconds=120)  # Test takes >60 and <=120 seconds
    def test_grains_append_val_is_list(self):
        """
        Tests the return of a grains.append call when val is passed in as a list.
        """
        # Start off with an empty list, don't know if the flaky decorator runs the setUp function or not...
        self.run_function("grains.setval", [self.GRAIN_KEY, []])
        second_grain = self.GRAIN_VAL + "-2"
        ret = self.run_function(
            "grains.append", [self.GRAIN_KEY, [self.GRAIN_VAL, second_grain]]
        )
        self.assertEqual(ret[self.GRAIN_KEY], [self.GRAIN_VAL, second_grain])

    @pytest.mark.slow_test(seconds=120)  # Test takes >60 and <=120 seconds
    def test_grains_append_call_twice(self):
        """
        Tests the return of a grains.append call when the value is already present
        but also ensure the grain is not listed twice.
        """
        # First, add the test grain.
        append_1 = self.run_function("grains.append", [self.GRAIN_KEY, self.GRAIN_VAL])

        # Call the function again, which results in a string message, as tested in
        # test_grains_append_val_already_present above.
        append_2 = self.run_function("grains.append", [self.GRAIN_KEY, self.GRAIN_VAL])

        # Now make sure the grain doesn't show up twice.
        grains = self.run_function("grains.items")
        count = 0
        for grain in grains:
            if grain == self.GRAIN_KEY:
                count += 1

        # We should only have hit the grain key once.
        self.assertEqual(
            count,
            1,
            msg="Count did not match({}!=1) while looking for key '{}'.\nFirst append return:\n{}\nSecond append return:\n{}".format(
                count,
                self.GRAIN_KEY,
                pprint.pformat(append_1),
                pprint.pformat(append_2),
            ),
        )

    def wait_for_grain(self, key, val, timeout=60, sleep=0.3):
        start = time.time()
        while time.time() - start <= timeout:
            ret = self.run_function("grains.get", [key])
            if ret == val:
                return True
            time.sleep(sleep)
        return False

    def test_grains_remove_add(self):
        second_grain = self.GRAIN_VAL + "-2"
        ret = self.run_function("grains.get", [self.GRAIN_KEY])
        self.assertEqual(ret, [])

        for i in range(10):
            ret = self.run_function("grains.setval", [self.GRAIN_KEY, []])
            self.assertEqual(ret[self.GRAIN_KEY], [])
            self.wait_for_grain(self.GRAIN_KEY, [])
            ret = self.run_function("grains.append", [self.GRAIN_KEY, self.GRAIN_VAL])
            self.assertEqual(ret[self.GRAIN_KEY], [self.GRAIN_VAL])
            self.assertEqual(ret[self.GRAIN_KEY], [self.GRAIN_VAL])

            ret = self.run_function("grains.setval", [self.GRAIN_KEY, []])
            self.wait_for_grain(self.GRAIN_KEY, [])
            self.assertTrue(self.wait_for_grain(self.GRAIN_KEY, []))
            ret = self.run_function(
                "grains.append", [self.GRAIN_KEY, [self.GRAIN_VAL, second_grain]]
            )
            self.assertEqual(ret[self.GRAIN_KEY], [self.GRAIN_VAL, second_grain])
            self.assertEqual(ret[self.GRAIN_KEY], [self.GRAIN_VAL, second_grain])