# -*- coding: utf-8 -*- """ :codeauthor: Thayne Harbaugh (tharbaug@adobe.com) tests.integration.shell.saltcli ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ :NOTE: this was named ``saltcli`` rather than ``salt`` because ``salt`` conflates in the python importer with the expected ``salt`` namespace and breaks imports. """ from __future__ import absolute_import import logging import os import pytest import salt.defaults.exitcodes import salt.utils.files import salt.utils.path from tests.integration.utils import testprogram from tests.support.case import ShellCase from tests.support.helpers import slowTest from tests.support.runtests import RUNTIME_VARS log = logging.getLogger(__name__) @pytest.mark.windows_whitelisted class SaltTest(testprogram.TestProgramCase): """ Various integration tests for the salt executable. """ # pylint: disable=invalid-name @slowTest def test_exit_status_unknown_argument(self): """ Ensure correct exit status when an unknown argument is passed to salt-run. """ runner = testprogram.TestProgramSalt( name="run-unknown_argument", parent_dir=self._test_dir, ) # Call setup here to ensure config and script exist runner.setup() stdout, stderr, status = runner.run( args=["--unknown-argument"], catch_stderr=True, with_retcode=True, ) self.assert_exit_status( status, "EX_USAGE", message="unknown argument", stdout=stdout, stderr=stderr ) # runner.shutdown() should be unnecessary since the start-up should fail @slowTest def test_exit_status_correct_usage(self): """ Ensure correct exit status when salt-run starts correctly. """ runner = testprogram.TestProgramSalt( name="run-correct_usage", parent_dir=self._test_dir, ) # Call setup here to ensure config and script exist runner.setup() stdout, stderr, status = runner.run( args=["*", "-h"], catch_stderr=True, with_retcode=True, ) self.assert_exit_status( status, "EX_OK", message="correct usage", stdout=stdout, stderr=stderr ) @pytest.mark.windows_whitelisted class RetcodeTestCase(ShellCase): """ Tests to ensure that we set non-zero retcodes when execution fails """ # Hard-coding these instead of substituting values from # salt.defaults.exitcodes will give us a heads-up in the event that someone # tries to do something daft like change these values. error_status = salt.defaults.exitcodes.EX_GENERIC state_compiler_error = salt.defaults.exitcodes.EX_STATE_COMPILER_ERROR state_failure = salt.defaults.exitcodes.EX_STATE_FAILURE def _salt(self, command): cmdline = "minion " + command return self.run_salt(cmdline, with_retcode=True)[1] def _salt_call(self, command, retcode_passthrough=False): cmdline = "--retcode-passthrough " if retcode_passthrough else "" cmdline += command return self.run_call(cmdline, with_retcode=True)[1] def _test_error(self, salt_call=False): """ Tests retcode when various error conditions are triggered """ _run = self._salt_call if salt_call else self._salt retcode = _run("test.raise_exception TypeError") assert retcode == self.error_status, retcode retcode = _run("test.raise_exception salt.exceptions.CommandNotFoundError") assert retcode == self.error_status, retcode retcode = _run("test.raise_exception salt.exceptions.CommandExecutionError") assert retcode == self.error_status, retcode retcode = _run("test.raise_exception salt.exceptions.SaltInvocationError") assert retcode == self.error_status, retcode retcode = _run( "test.raise_exception " 'OSError 2 "No such file or directory" /tmp/foo.txt' ) assert retcode == self.error_status, retcode retcode = _run('test.echo "{foo: bar, result: False}"') assert retcode == self.error_status, retcode retcode = _run('test.echo "{foo: bar, success: False}"') assert retcode == self.error_status, retcode @slowTest def test_zero_exit_code(self): """ Test that a zero exit code is set when there are no errors and there is no explicit False result set in the return data. """ retcode = self._salt("test.ping") assert retcode == 0, retcode retcode = self._salt_call("test.ping") assert retcode == 0, retcode @slowTest def test_context_retcode(self): """ Test that a nonzero retcode set in the context dunder will cause the salt CLI to set a nonzero retcode. """ # test.retcode will set the retcode in the context dunder retcode = self._salt("test.retcode 0") assert retcode == 0, retcode retcode = self._salt("test.retcode 42") assert retcode == self.error_status, retcode # Test salt-call, making sure to also confirm the behavior of # retcode_passthrough. retcode = self._salt_call("test.retcode 0") assert retcode == 0, retcode retcode = self._salt_call("test.retcode 42") assert retcode == self.error_status, retcode retcode = self._salt_call("test.retcode 42", retcode_passthrough=True) assert retcode == 42, retcode # Test a state run that exits with one or more failures retcode = self._salt_call("state.single test.fail_without_changes foo") assert retcode == self.error_status, retcode retcode = self._salt_call( "state.single test.fail_without_changes foo", retcode_passthrough=True ) assert retcode == self.state_failure, retcode # Test a state compiler error retcode = self._salt_call("state.apply thisslsfiledoesnotexist") assert retcode == self.error_status, retcode retcode = self._salt_call( "state.apply thisslsfiledoesnotexist", retcode_passthrough=True ) assert retcode == self.state_compiler_error, retcode @slowTest def test_salt_error(self): """ Test that we return the expected retcode when a minion function raises an exception. """ self._test_error() self._test_error(salt_call=True) @slowTest def test_missing_minion(self): """ Test that a minion which doesn't respond results in a nonzeo exit code """ good = salt.utils.path.join(self.master_opts["pki_dir"], "minions", "minion") bad = salt.utils.path.join(self.master_opts["pki_dir"], "minions", "minion2") try: # Copy the key with salt.utils.files.fopen(good, "rb") as fhr, salt.utils.files.fopen( bad, "wb" ) as fhw: fhw.write(fhr.read()) retcode = self.run_script( "salt", "-c {0} -t 5 minion2 test.ping".format(RUNTIME_VARS.TMP_CONF_DIR), with_retcode=True, timeout=60, )[1] assert retcode == salt.defaults.exitcodes.EX_GENERIC, retcode finally: # Now get rid of it try: os.remove(bad) except OSError as exc: if exc.errno != os.errno.ENOENT: log.error( "Failed to remove %s, this may affect other tests: %s", bad, exc )