salt/tests/integration/shell/test_call.py

# -*- coding: utf-8 -*-
"""
    :codeauthor: Pedro Algarvio (pedro@algarvio.me)


    tests.integration.shell.call
    ~~~~~~~~~~~~~~~~~~~~~~~~~~~~
"""
from __future__ import absolute_import

import logging
import os
import re
import shutil
import sys

import pytest
import salt.utils.files
import salt.utils.json
import salt.utils.platform
import salt.utils.yaml
from salt.ext import six
from tests.integration.utils import testprogram
from tests.support.case import ShellCase
from tests.support.helpers import flaky, with_tempfile
from tests.support.mixins import ShellCaseCommonTestsMixin
from tests.support.runtests import RUNTIME_VARS
from tests.support.unit import skipIf

log = logging.getLogger(__name__)


@pytest.mark.windows_whitelisted
class CallTest(ShellCase, testprogram.TestProgramCase, ShellCaseCommonTestsMixin):

    _call_binary_ = "salt-call"

    @skipIf(True, "SLOWTEST skip")
    def test_default_output(self):
        out = self.run_call("-l quiet test.fib 3")

        expect = ["local:", "    - 2"]
        self.assertEqual(expect, out[:-1])

    @skipIf(True, "SLOWTEST skip")
    def test_text_output(self):
        out = self.run_call("-l quiet --out txt test.fib 3")

        expect = ["local: (2"]

        self.assertEqual("".join(expect), "".join(out).rsplit(",", 1)[0])

    @skipIf(True, "SLOWTEST skip")
    def test_json_out_indent(self):
        out = self.run_call("test.ping -l quiet --out=json --out-indent=-1")
        self.assertIn('"local": true', "".join(out))

        out = self.run_call("test.ping -l quiet --out=json --out-indent=0")
        self.assertIn('"local": true', "".join(out))

        out = self.run_call("test.ping -l quiet --out=json --out-indent=1")
        self.assertIn('"local": true', "".join(out))

    @skipIf(True, "SLOWTEST skip")
    def test_local_sls_call(self):
        fileroot = os.path.join(RUNTIME_VARS.FILES, "file", "base")
        out = self.run_call(
            "--file-root {0} state.sls saltcalllocal".format(fileroot), local=True
        )
        self.assertIn("Name: test.echo", "".join(out))
        self.assertIn("Result: True", "".join(out))
        self.assertIn("hello", "".join(out))
        self.assertIn("Succeeded: 1", "".join(out))

    @with_tempfile()
    @skipIf(True, "SLOWTEST skip")
    def test_local_salt_call(self, name):
        """
        This tests to make sure that salt-call does not execute the
        function twice, see https://github.com/saltstack/salt/pull/49552
        """

        def _run_call(cmd):
            cmd = "--out=json " + cmd
            return salt.utils.json.loads("".join(self.run_call(cmd, local=True)))[
                "local"
            ]

        ret = _run_call('state.single file.append name={0} text="foo"'.format(name))
        ret = ret[next(iter(ret))]

        # Make sure we made changes
        assert ret["changes"]

        # 2nd sanity check: make sure that "foo" only exists once in the file
        with salt.utils.files.fopen(name) as fp_:
            contents = fp_.read()
        assert contents.count("foo") == 1, contents

    @skipIf(
        salt.utils.platform.is_windows() or salt.utils.platform.is_darwin(),
        "This test requires a supported master",
    )
    @skipIf(True, "SLOWTEST skip")
    def test_user_delete_kw_output(self):
        ret = self.run_call("-l quiet -d user.delete")
        assert "salt '*' user.delete name remove=True force=True" in "".join(ret)

    @skipIf(True, "SLOWTEST skip")
    def test_salt_documentation_too_many_arguments(self):
        """
        Test to see if passing additional arguments shows an error
        """
        data = self.run_call("-d virtualenv.create /tmp/ve", catch_stderr=True)
        self.assertIn(
            "You can only get documentation for one method at one time",
            "\n".join(data[1]),
        )

    @skipIf(True, "SLOWTEST skip")
    def test_issue_6973_state_highstate_exit_code(self):
        """
        If there is no tops/master_tops or state file matches
        for this minion, salt-call should exit non-zero if invoked with
        option --retcode-passthrough
        """
        src = os.path.join(RUNTIME_VARS.BASE_FILES, "top.sls")
        dst = os.path.join(RUNTIME_VARS.BASE_FILES, "top.sls.bak")
        shutil.move(src, dst)
        expected_comment = "No states found for this minion"
        try:
            stdout, retcode = self.run_call(
                "-l quiet --retcode-passthrough state.highstate", with_retcode=True
            )
        finally:
            shutil.move(dst, src)
        self.assertIn(expected_comment, "".join(stdout))
        self.assertNotEqual(0, retcode)

    @skipIf(sys.platform.startswith("win"), "This test does not apply on Win")
    @skipIf(True, "to be re-enabled when #23623 is merged")
    def test_return(self):
        self.run_call('cmd.run "echo returnTOmaster"')
        jobs = [a for a in self.run_run("jobs.list_jobs")]

        self.assertTrue(True in ["returnTOmaster" in j for j in jobs])
        # lookback jid
        first_match = [(i, j) for i, j in enumerate(jobs) if "returnTOmaster" in j][0]
        jid, idx = None, first_match[0]
        while idx > 0:
            jid = re.match("([0-9]+):", jobs[idx])
            if jid:
                jid = jid.group(1)
                break
            idx -= 1
        assert idx > 0
        assert jid
        master_out = [a for a in self.run_run("jobs.lookup_jid {0}".format(jid))]
        self.assertTrue(True in ["returnTOmaster" in a for a in master_out])

    @skipIf(salt.utils.platform.is_windows(), "Skip on Windows")
    @skipIf(True, "SLOWTEST skip")
    def test_syslog_file_not_found(self):
        """
        test when log_file is set to a syslog file that does not exist
        """
        old_cwd = os.getcwd()
        config_dir = os.path.join(RUNTIME_VARS.TMP, "log_file_incorrect")
        if not os.path.isdir(config_dir):
            os.makedirs(config_dir)

        os.chdir(config_dir)

        with salt.utils.files.fopen(self.get_config_file_path("minion"), "r") as fh_:
            minion_config = salt.utils.yaml.load(fh_.read())
            minion_config["log_file"] = "file:///dev/doesnotexist"
            with salt.utils.files.fopen(os.path.join(config_dir, "minion"), "w") as fh_:
                fh_.write(salt.utils.yaml.dump(minion_config, default_flow_style=False))
        ret = self.run_script(
            "salt-call",
            '--config-dir {0} cmd.run "echo foo"'.format(config_dir),
            timeout=120,
            catch_stderr=True,
            with_retcode=True,
        )
        try:
            if sys.version_info >= (3, 5, 4):
                self.assertIn("local:", ret[0])
                self.assertIn(
                    "[WARNING ] The log_file does not exist. Logging not setup correctly or syslog service not started.",
                    ret[1],
                )
                self.assertEqual(ret[2], 0)
            else:
                self.assertIn(
                    "Failed to setup the Syslog logging handler", "\n".join(ret[1])
                )
                self.assertEqual(ret[2], 2)
        finally:
            self.chdir(old_cwd)
            if os.path.isdir(config_dir):
                shutil.rmtree(config_dir)

    @skipIf(True, "This test is unreliable. Need to investigate why more deeply.")
    @flaky
    def test_issue_15074_output_file_append(self):
        output_file_append = os.path.join(RUNTIME_VARS.TMP, "issue-15074")
        try:
            # Let's create an initial output file with some data
            _ = self.run_script(
                "salt-call",
                "-c {0} --output-file={1} test.versions".format(
                    RUNTIME_VARS.TMP_MINION_CONF_DIR, output_file_append
                ),
                catch_stderr=True,
                with_retcode=True,
            )

            with salt.utils.files.fopen(output_file_append) as ofa:
                output = ofa.read()

            self.run_script(
                "salt-call",
                "-c {0} --output-file={1} --output-file-append test.versions".format(
                    self.config_dir, output_file_append
                ),
                catch_stderr=True,
                with_retcode=True,
            )
            with salt.utils.files.fopen(output_file_append) as ofa:
                self.assertEqual(ofa.read(), output + output)
        finally:
            if os.path.exists(output_file_append):
                os.unlink(output_file_append)

    @skipIf(True, "This test is unreliable. Need to investigate why more deeply.")
    @flaky
    def test_issue_14979_output_file_permissions(self):
        output_file = os.path.join(RUNTIME_VARS.TMP, "issue-14979")
        with salt.utils.files.set_umask(0o077):
            try:
                # Let's create an initial output file with some data
                self.run_script(
                    "salt-call",
                    "-c {0} --output-file={1} -l trace -g".format(
                        RUNTIME_VARS.TMP_MINION_CONF_DIR, output_file
                    ),
                    catch_stderr=True,
                    with_retcode=True,
                )
                try:
                    stat1 = os.stat(output_file)
                except OSError:
                    self.fail("Failed to generate output file, see log for details")

                # Let's change umask
                os.umask(0o777)  # pylint: disable=blacklisted-function

                self.run_script(
                    "salt-call",
                    "-c {0} --output-file={1} --output-file-append -g".format(
                        RUNTIME_VARS.TMP_MINION_CONF_DIR, output_file
                    ),
                    catch_stderr=True,
                    with_retcode=True,
                )
                try:
                    stat2 = os.stat(output_file)
                except OSError:
                    self.fail("Failed to generate output file, see log for details")
                self.assertEqual(stat1.st_mode, stat2.st_mode)
                # Data was appeneded to file
                self.assertTrue(stat1.st_size < stat2.st_size)

                # Let's remove the output file
                os.unlink(output_file)

                # Not appending data
                self.run_script(
                    "salt-call",
                    "-c {0} --output-file={1} -g".format(
                        RUNTIME_VARS.TMP_MINION_CONF_DIR, output_file
                    ),
                    catch_stderr=True,
                    with_retcode=True,
                )
                try:
                    stat3 = os.stat(output_file)
                except OSError:
                    self.fail("Failed to generate output file, see log for details")
                # Mode must have changed since we're creating a new log file
                self.assertNotEqual(stat1.st_mode, stat3.st_mode)
            finally:
                if os.path.exists(output_file):
                    os.unlink(output_file)

    @skipIf(sys.platform.startswith("win"), "This test does not apply on Win")
    @skipIf(True, "SLOWTEST skip")
    def test_42116_cli_pillar_override(self):
        ret = self.run_call(
            "state.apply issue-42116-cli-pillar-override "
            'pillar=\'{"myhost": "localhost"}\''
        )
        for line in ret:
            line = line.lstrip()
            if line == 'Comment: Command "ping -c 2 localhost" run':
                # Successful test
                break
        else:
            log.debug("salt-call output:\n\n%s", "\n".join(ret))
            self.fail("CLI pillar override not found in pillar data")

    @skipIf(True, "SLOWTEST skip")
    def test_pillar_items_masterless(self):
        """
        Test to ensure we get expected output
        from pillar.items with salt-call
        """
        get_items = self.run_call("pillar.items", local=True)
        exp_out = [
            "        - Lancelot",
            "        - Galahad",
            "        - Bedevere",
            "    monty:",
            "        python",
        ]
        for out in exp_out:
            self.assertIn(out, get_items)

    def tearDown(self):
        """
        Teardown method to remove installed packages
        """
        user = ""
        user_info = self.run_call(" grains.get username", local=True)
        if (
            user_info
            and isinstance(user_info, (list, tuple))
            and isinstance(user_info[-1], six.string_types)
        ):
            user = user_info[-1].strip()
        super(CallTest, self).tearDown()

    @skipIf(True, "SLOWTEST skip")
    def test_exit_status_unknown_argument(self):
        """
        Ensure correct exit status when an unknown argument is passed to salt-call.
        """

        call = testprogram.TestProgramSaltCall(
            name="unknown_argument", parent_dir=self._test_dir,
        )
        # Call setup here to ensure config and script exist
        call.setup()
        stdout, stderr, status = call.run(
            args=["--unknown-argument"], catch_stderr=True, with_retcode=True,
        )
        self.assert_exit_status(
            status, "EX_USAGE", message="unknown argument", stdout=stdout, stderr=stderr
        )

    @skipIf(True, "SLOWTEST skip")
    def test_masterless_highstate(self):
        """
        test state.highstate in masterless mode
        """
        ret = self.run_call("state.highstate", local=True)

        destpath = os.path.join(RUNTIME_VARS.TMP, "testfile")
        exp_out = [
            "    Function: file.managed",
            "      Result: True",
            "          ID: {0}".format(destpath),
        ]

        for out in exp_out:
            self.assertIn(out, ret)

        self.assertTrue(os.path.exists(destpath))

    @skipIf(True, "SLOWTEST skip")
    def test_exit_status_correct_usage(self):
        """
        Ensure correct exit status when salt-call starts correctly.
        """

        call = testprogram.TestProgramSaltCall(
            name="correct_usage", parent_dir=self._test_dir,
        )
        # Call setup here to ensure config and script exist
        call.setup()
        stdout, stderr, status = call.run(
            args=["--local", "test.true"], catch_stderr=True, with_retcode=True,
        )
        self.assert_exit_status(
            status, "EX_OK", message="correct usage", stdout=stdout, stderr=stderr
        )