salt/tests/integration/runners/test_cache.py

# -*- coding: utf-8 -*-
"""
Tests for the salt-run command
"""
from __future__ import absolute_import, print_function, unicode_literals

import logging

import pytest
from tests.support.case import ShellCase

log = logging.getLogger(__name__)


@pytest.mark.usefixtures("salt_sub_minion")
class ManageTest(ShellCase):
    """
    Test the manage runner
    """

    @pytest.mark.slow_test(seconds=10)  # Test takes >5 and <=10 seconds
    def test_cache(self):
        """
        Store, list, fetch, then flush data
        """
        # Store the data
        ret = self.run_run_plus(
            "cache.store",
            bank="cachetest/runner",
            key="test_cache",
            data="The time has come the walrus said",
        )
        # Make sure we can see the new key
        ret = self.run_run_plus("cache.list", bank="cachetest/runner")
        self.assertIn("test_cache", ret["return"])
        # Make sure we can see the new data
        ret = self.run_run_plus(
            "cache.fetch", bank="cachetest/runner", key="test_cache"
        )
        self.assertIn("The time has come the walrus said", ret["return"])
        # Make sure we can delete the data
        ret = self.run_run_plus(
            "cache.flush", bank="cachetest/runner", key="test_cache"
        )
        ret = self.run_run_plus("cache.list", bank="cachetest/runner")
        self.assertNotIn("test_cache", ret["return"])

    @pytest.mark.slow_test(seconds=5)  # Test takes >1 and <=5 seconds
    def test_cache_invalid(self):
        """
        Store, list, fetch, then flush data
        """
        # Store the data
        ret = self.run_run_plus("cache.store",)
        # Make sure we can see the new key
        expected = "Passed invalid arguments:"
        self.assertIn(expected, ret["return"])

    @pytest.mark.slow_test(seconds=5)  # Test takes >1 and <=5 seconds
    def test_grains(self):
        """
        Test cache.grains
        """
        # Store the data
        ret = self.run_run_plus("cache.grains", tgt="minion")

        self.assertIn("minion", ret["return"])

    @pytest.mark.slow_test(seconds=5)  # Test takes >1 and <=5 seconds
    def test_pillar(self):
        """
        Test cache.pillar
        """
        # Store the data
        ret = self.run_run_plus("cache.pillar", tgt="minion")

        assert "minion" in ret["return"]
        assert "sub_minion" not in ret["return"]

    @pytest.mark.slow_test(seconds=5)  # Test takes >1 and <=5 seconds
    def test_pillar_no_tgt(self):
        """
        Test cache.pillar when no tgt is
        supplied. This should return pillar
        data for all minions
        """
        # Store the data
        ret = self.run_run_plus("cache.pillar",)

        assert all(x in ret["return"] for x in ["minion", "sub_minion"])

    @pytest.mark.slow_test(seconds=5)  # Test takes >1 and <=5 seconds
    def test_pillar_minion_noexist(self):
        """
        Test cache.pillar when the target does not exist
        """
        ret = self.run_run_plus("cache.pillar", tgt="doesnotexist")

        assert "minion" not in ret["return"]
        assert "sub_minion" not in ret["return"]

    @pytest.mark.slow_test(seconds=5)  # Test takes >1 and <=5 seconds
    def test_pillar_minion_tgt_type_pillar(self):
        """
        Test cache.pillar when the target exists
        and tgt_type is pillar
        """
        ret = self.run_run_plus("cache.pillar", tgt="monty:python", tgt_type="pillar",)

        assert all(x in ret["return"] for x in ["minion", "sub_minion"])

    @pytest.mark.slow_test(seconds=5)  # Test takes >1 and <=5 seconds
    def test_mine(self):
        """
        Test cache.mine
        """
        # Store the data
        ret = self.run_run_plus("cache.mine", tgt="minion")

        self.assertIn("minion", ret["return"])