Fix support for current_dir parameter in ssh_execute
Change-Id: I1a5bb0f63cd74b7637232a4a2221e921c20471a1
This commit is contained in:
parent
7c0c433a0d
commit
9f03249dad
|
@ -89,6 +89,7 @@ class SSHShellProcessFixture(_process.ShellProcessFixture):
|
|||
tobiko.check_valid_type(ssh_client, ssh.SSHClientFixture)
|
||||
tobiko.check_valid_type(parameters, SSHShellProcessParameters)
|
||||
environment = parameters.environment
|
||||
current_dir = parameters.current_dir
|
||||
|
||||
for attempt in tobiko.retry(
|
||||
timeout=self.parameters.timeout,
|
||||
|
@ -98,6 +99,7 @@ class SSHShellProcessFixture(_process.ShellProcessFixture):
|
|||
|
||||
timeout = attempt.time_left
|
||||
details = (f"command='{command}', "
|
||||
f"current_dir='{current_dir}', "
|
||||
f"login={ssh_client.login}, "
|
||||
f"timeout={timeout}, "
|
||||
f"attempt={attempt}, "
|
||||
|
@ -111,6 +113,8 @@ class SSHShellProcessFixture(_process.ShellProcessFixture):
|
|||
f"{name}={shlex.quote(value)}"
|
||||
for name, value in self.environment.items())
|
||||
command = variables + " " + command
|
||||
if current_dir is not None:
|
||||
command = f"cd {current_dir} && {command}"
|
||||
process.exec_command(command)
|
||||
LOG.debug(f"Remote process created. ({details})")
|
||||
return process
|
||||
|
|
|
@ -15,6 +15,7 @@
|
|||
# under the License.
|
||||
from __future__ import absolute_import
|
||||
|
||||
import os
|
||||
import typing
|
||||
|
||||
import testtools
|
||||
|
@ -42,8 +43,13 @@ class ExecuteTest(testtools.TestCase):
|
|||
else:
|
||||
return SSH_EXPECTED_SHELL
|
||||
|
||||
def test_succeed(self, command='true', stdin=None, stdout=None,
|
||||
stderr=None, expect_exit_status=0, **kwargs):
|
||||
def test_succeed(self,
|
||||
command: sh.ShellCommandType = 'true',
|
||||
stdin: str = None,
|
||||
stdout: str = None,
|
||||
stderr: str = None,
|
||||
expect_exit_status: typing.Optional[int] = 0,
|
||||
**kwargs):
|
||||
process = self.execute(command=command,
|
||||
stdin=stdin,
|
||||
stdout=bool(stdout),
|
||||
|
@ -89,6 +95,13 @@ class ExecuteTest(testtools.TestCase):
|
|||
def test_succeed_with_no_exit_status(self):
|
||||
self.test_succeed(command='false', expect_exit_status=None)
|
||||
|
||||
def test_succeed_with_current_dir(self):
|
||||
temp_file = self.make_temporary()
|
||||
self.execute(command=f"echo '{self.id()}' > '{temp_file}'")
|
||||
self.test_succeed(command=f"cat './{os.path.basename(temp_file)}'",
|
||||
current_dir=os.path.dirname(temp_file),
|
||||
stdout=f"{self.id()}\n")
|
||||
|
||||
def test_fails(self, command='false', exit_status=None, stdin=None,
|
||||
stdout=None, stderr=None, expect_exit_status=0,
|
||||
**kwargs):
|
||||
|
@ -162,6 +175,17 @@ class ExecuteTest(testtools.TestCase):
|
|||
self.assertIsNone(ex.stderr)
|
||||
self.assertEqual(timeout, ex.timeout)
|
||||
|
||||
def make_temporary(self,
|
||||
directory=False,
|
||||
add_cleanup=True) -> str:
|
||||
command = sh.shell_command('mktemp')
|
||||
if directory:
|
||||
command += '-d'
|
||||
temporary_path = self.execute(command=command).stdout.strip()
|
||||
if add_cleanup:
|
||||
self.addCleanup(sh.execute, f"rm -fR '{temporary_path}'")
|
||||
return temporary_path
|
||||
|
||||
def execute(self, **kwargs):
|
||||
return sh.execute(**kwargs)
|
||||
|
||||
|
|
Loading…
Reference in New Issue