is a systems management software for data center automation, cloud orchestration, server provisioning, configuration management and more. Community version.
| test_pre_flight.py (salt-3002.1) | : | test_pre_flight.py (salt-3002.2) |
|---|
|
| # -*- coding: utf-8 -*- | | |
| """ | | """ |
| Test for ssh_pre_flight roster option | | Test for ssh_pre_flight roster option |
| """ | | """ |
|
| from __future__ import absolute_import, print_function, unicode_literals | | |
| | |
| import os | | import os |
| | |
| import salt.utils.files | | import salt.utils.files |
| from tests.support.case import SSHCase | | from tests.support.case import SSHCase |
| from tests.support.helpers import slowTest | | from tests.support.helpers import slowTest |
| from tests.support.runtests import RUNTIME_VARS | | from tests.support.runtests import RUNTIME_VARS |
| | |
| class SSHPreFlightTest(SSHCase): | | class SSHPreFlightTest(SSHCase): |
| """ | | """ |
| Test ssh_pre_flight roster option | | Test ssh_pre_flight roster option |
| """ | | """ |
| | |
| def setUp(self): | | def setUp(self): |
|
| super(SSHPreFlightTest, self).setUp() | | super().setUp() |
| self.roster = os.path.join(RUNTIME_VARS.TMP, "pre_flight_roster") | | self.roster = os.path.join(RUNTIME_VARS.TMP, "pre_flight_roster") |
| self.data = { | | self.data = { |
| "ssh_pre_flight": os.path.join(RUNTIME_VARS.TMP, "ssh_pre_flight.sh"
) | | "ssh_pre_flight": os.path.join(RUNTIME_VARS.TMP, "ssh_pre_flight.sh"
) |
| } | | } |
| self.test_script = os.path.join( | | self.test_script = os.path.join( |
| RUNTIME_VARS.TMP, "test-pre-flight-script-worked.txt" | | RUNTIME_VARS.TMP, "test-pre-flight-script-worked.txt" |
| ) | | ) |
| | |
| def _create_roster(self): | | def _create_roster(self): |
| self.custom_roster(self.roster, self.data) | | self.custom_roster(self.roster, self.data) |
| | |
| with salt.utils.files.fopen(self.data["ssh_pre_flight"], "w") as fp_: | | with salt.utils.files.fopen(self.data["ssh_pre_flight"], "w") as fp_: |
|
| fp_.write("touch {0}".format(self.test_script)) | | fp_.write("touch {}".format(self.test_script)) |
| | |
| @slowTest | | @slowTest |
| def test_ssh_pre_flight(self): | | def test_ssh_pre_flight(self): |
| """ | | """ |
| test ssh when ssh_pre_flight is set | | test ssh when ssh_pre_flight is set |
| ensure the script runs successfully | | ensure the script runs successfully |
| """ | | """ |
| self._create_roster() | | self._create_roster() |
|
| ret = self.run_function("test.ping", roster_file=self.roster) | | assert self.run_function("test.ping", roster_file=self.roster) |
| | |
| assert os.path.exists(self.test_script) | | assert os.path.exists(self.test_script) |
| | |
| @slowTest | | @slowTest |
| def test_ssh_run_pre_flight(self): | | def test_ssh_run_pre_flight(self): |
| """ | | """ |
| test ssh when --pre-flight is passed to salt-ssh | | test ssh when --pre-flight is passed to salt-ssh |
| to ensure the script runs successfully | | to ensure the script runs successfully |
| """ | | """ |
| self._create_roster() | | self._create_roster() |
| # make sure we previously ran a command so the thin dir exists | | # make sure we previously ran a command so the thin dir exists |
| self.run_function("test.ping", wipe=False) | | self.run_function("test.ping", wipe=False) |
| assert not os.path.exists(self.test_script) | | assert not os.path.exists(self.test_script) |
| | |
|
| ret = self.run_function( | | assert self.run_function( |
| "test.ping", ssh_opts="--pre-flight", roster_file=self.roster, wipe=
False | | "test.ping", ssh_opts="--pre-flight", roster_file=self.roster, wipe=
False |
| ) | | ) |
| assert os.path.exists(self.test_script) | | assert os.path.exists(self.test_script) |
| | |
|
| | @slowTest |
| | def test_ssh_run_pre_flight_failure(self): |
| | """ |
| | test ssh_pre_flight when there is a failure |
| | in the script. |
| | """ |
| | self._create_roster() |
| | with salt.utils.files.fopen(self.data["ssh_pre_flight"], "w") as fp_: |
| | fp_.write("exit 2") |
| | |
| | ret = self.run_function( |
| | "test.ping", ssh_opts="--pre-flight", roster_file=self.roster, wipe= |
| | False |
| | ) |
| | assert ret["retcode"] == 2 |
| | |
| def tearDown(self): | | def tearDown(self): |
| """ | | """ |
| make sure to clean up any old ssh directories | | make sure to clean up any old ssh directories |
| """ | | """ |
| files = [self.roster, self.data["ssh_pre_flight"], self.test_script] | | files = [self.roster, self.data["ssh_pre_flight"], self.test_script] |
| for fp_ in files: | | for fp_ in files: |
| if os.path.exists(fp_): | | if os.path.exists(fp_): |
| os.remove(fp_) | | os.remove(fp_) |
| | | |
| End of changes. 7 change blocks. |
|---|
| 6 lines changed or deleted | | 20 lines changed or added |
|---|
"Fossies" - the Fresh Open Source Software Archive 