Browse Source

Restructure shell package

- Create base fixtures for process creation
- Separate process creation from execution completion

Change-Id: If07ce73d4836a7853d64035ee0632547f772b5f6
changes/79/667979/20
Federico Ressi 2 years ago
parent
commit
92c7c09634
  1. 6
      tobiko/shell/ping/_ping.py
  2. 20
      tobiko/shell/sh/__init__.py
  3. 326
      tobiko/shell/sh/_execute.py
  4. 15
      tobiko/shell/sh/_io.py
  5. 116
      tobiko/shell/sh/_local.py
  6. 300
      tobiko/shell/sh/_process.py
  7. 158
      tobiko/shell/sh/_ssh.py
  8. 33
      tobiko/tests/functional/shell/test_sh.py

6
tobiko/shell/ping/_ping.py

@ -221,8 +221,10 @@ def execute_ping(parameters, ssh_client=None, check=True, **params):
"to execute ping on a CirrOS image.")
command = get_ping_command(parameters)
result = sh.execute(command=command, ssh_client=ssh_client,
timeout=parameters.timeout, check=False, wait=True)
result = sh.execute(command=command,
ssh_client=ssh_client,
timeout=parameters.timeout,
expect_exit_status=None)
if check and result.exit_status and result.stderr:
handle_ping_command_error(error=str(result.stderr))

20
tobiko/shell/sh/__init__.py

@ -18,8 +18,13 @@ from __future__ import absolute_import
from tobiko.shell.sh import _command
from tobiko.shell.sh import _exception
from tobiko.shell.sh import _execute
from tobiko.shell.sh import _local
from tobiko.shell.sh import _process
from tobiko.shell.sh import _ssh
shell_command = _command.shell_command
ShellError = _exception.ShellError
ShellCommandFailed = _exception.ShellCommandFailed
ShellTimeoutExpired = _exception.ShellTimeoutExpired
@ -28,7 +33,16 @@ ShellProcessNotTeriminated = _exception.ShellProcessNotTeriminated
ShellStdinClosed = _exception.ShellStdinClosed
execute = _execute.execute
local_execute = _execute.local_execute
ssh_execute = _execute.ssh_execute
execute_process = _execute.execute_process
ShellExecuteResult = _execute.ShellExecuteResult
shell_command = _command.shell_command
local_execute = _local.local_execute
local_process = _local.local_process
LocalShellProcessFixture = _local.LocalShellProcessFixture
process = _process.process
ShellProcessFixture = _process.ShellProcessFixture
ssh_process = _ssh.ssh_process
ssh_execute = _ssh.ssh_execute
SSHShellProcessFixture = _ssh.SSHShellProcessFixture

326
tobiko/shell/sh/_execute.py

@ -15,17 +15,10 @@
# under the License.
from __future__ import absolute_import
import fcntl
import subprocess
import os
from oslo_log import log
import paramiko
import six
import tobiko
from tobiko.shell import ssh
from tobiko.shell.sh import _command
from tobiko.shell.sh import _process
@ -35,9 +28,20 @@ LOG = log.getLogger(__name__)
DATA_TYPES = six.string_types + (six.binary_type, six.text_type)
def execute(command, environment=None, timeout=None, shell=None, check=True,
wait=None, stdin=True, stdout=True, stderr=True, ssh_client=None,
**kwargs):
class ShellExecuteResult(object):
def __init__(self, command=None, exit_status=None, stdin=None, stdout=None,
stderr=None):
self.command = str(command)
self.exit_status = int(exit_status)
self.stdin = stdin and str(stdin) or None
self.stdout = stdout and str(stdout) or None
self.stderr = stderr and str(stderr) or None
def execute(command, environment=None, timeout=None, shell=None,
stdin=True, stdout=True, stderr=True, ssh_client=None,
expect_exit_status=0, **kwargs):
"""Execute command inside a remote or local shell
:param command: command argument list
@ -57,283 +61,29 @@ def execute(command, environment=None, timeout=None, shell=None, check=True,
:raises ShellCommandError: when command execution terminates with non-zero
exit status.
"""
fixture = ShellExecuteFixture(
command, environment=environment, shell=shell, stdin=stdin,
stdout=stdout, stderr=stderr, timeout=timeout, check=check, wait=wait,
ssh_client=ssh_client, **kwargs)
return tobiko.setup_fixture(fixture).process
def local_execute(command, environment=None, shell=None, stdin=True,
stdout=True, stderr=True, timeout=None, check=True,
wait=None, **kwargs):
"""Execute command on local host using local shell"""
return execute(
command=command, environment=environment, shell=shell, stdin=stdin,
stdout=stdout, stderr=stderr, timeout=timeout, check=check, wait=wait,
ssh_client=False, **kwargs)
def ssh_execute(ssh_client, command, environment=None, shell=None, stdin=True,
stdout=True, stderr=True, timeout=None, check=True, wait=None,
**kwargs):
"""Execute command on local host using local shell"""
return execute(
command=command, environment=environment, shell=shell, stdin=stdin,
stdout=stdout, stderr=stderr, timeout=timeout, check=check, wait=wait,
ssh_client=ssh_client, **kwargs)
class ShellExecuteFixture(tobiko.SharedFixture):
command = None
shell = None
environment = {}
stdin = None
stderr = None
stdout = None
timeout = 120.
check = None
wait = None
process = None
process_parameters = None
def __init__(self, command=None, shell=None, environment=None, stdin=None,
stdout=None, stderr=None, timeout=None, check=None, wait=None,
ssh_client=None, **kwargs):
super(ShellExecuteFixture, self).__init__()
if ssh_client is not None:
self.ssh_client = ssh_client
else:
self.ssh_client = ssh_client = self.default_ssh_client
if shell is not None:
self.shell = shell = bool(shell) and _command.shell_command(shell)
elif not ssh_client:
self.shell = shell = self.default_shell_command
if command is None:
command = self.command
command = _command.shell_command(command)
if shell:
command = shell + [str(command)]
self.command = command
environment = environment or self.environment
if environment:
self.environment = dict(environment).update(environment)
if stdin is not None:
self.stdin = stdin
if stdout is not None:
self.stdout = stdout
if stderr is not None:
self.stderr = stderr
if timeout is not None:
self.timeout = timeout
if check is not None:
self.check = check
if wait is not None:
self.wait = wait
self.process_parameters = (self.process_parameters and
dict(self.process_parameters) or
{})
if kwargs:
self.process_parameters.update(kwargs)
@property
def default_shell_command(self):
from tobiko import config
CONF = config.CONF
return _command.shell_command(CONF.tobiko.shell.command)
@property
def default_ssh_client(self):
return ssh.ssh_proxy_client()
def setup_fixture(self):
self.setup_process()
def setup_process(self):
self.process = self.execute()
def execute(self, timeout=None, stdin=None, stdout=None, stderr=None,
check=None, ssh_client=None, wait=None, **kwargs):
command = self.command
environment = self.environment
if timeout is None:
timeout = self.timeout
LOG.debug("Execute command '%s' on local host (timeout=%r, "
"environment=%r)...",
command, timeout, environment)
if stdin is None:
stdin = self.stdin
if stdout is None:
stdout = self.stdout
if stderr is None:
stderr = self.stderr
if check is None:
check = self.check
if wait is None:
wait = self.wait
if ssh_client is None:
ssh_client = self.ssh_client
process_parameters = self.process_parameters
if kwargs:
process_parameters = dict(process_parameters, **kwargs)
process = self.create_process(command=command,
environment=environment,
timeout=timeout, stdin=stdin,
stdout=stdout, stderr=stderr,
ssh_client=ssh_client,
**process_parameters)
self.addCleanup(process.close)
process = _process.process(command=command,
environment=environment,
timeout=timeout,
shell=shell,
stdin=stdin,
stdout=stdout,
stderr=stderr,
ssh_client=ssh_client,
**kwargs)
return execute_process(process=process,
stdin=stdin,
expect_exit_status=expect_exit_status)
def execute_process(process, stdin, expect_exit_status):
with process:
if stdin and isinstance(stdin, DATA_TYPES):
process.send(data=stdin)
if wait or check:
if process.stdin:
process.stdin.close()
process.wait()
if check:
process.check_exit_status()
return process
def create_process(self, ssh_client, **kwargs):
if ssh_client:
return self.create_ssh_process(ssh_client=ssh_client, **kwargs)
else:
return self.create_local_process(**kwargs)
def create_local_process(self, command, environment, timeout, stdin,
stdout, stderr, **kwargs):
popen_params = {}
if stdin:
popen_params.update(stdin=subprocess.PIPE)
if stdout:
popen_params.update(stdout=subprocess.PIPE)
if stderr:
popen_params.update(stderr=subprocess.PIPE)
process = subprocess.Popen(command,
universal_newlines=True,
env=environment,
**popen_params)
if stdin:
set_non_blocking(process.stdin.fileno())
kwargs.update(stdin=process.stdin)
if stdout:
set_non_blocking(process.stdout.fileno())
kwargs.update(stdout=process.stdout)
if stderr:
set_non_blocking(process.stderr.fileno())
kwargs.update(stderr=process.stderr)
return LocalShellProcess(process=process, command=command,
timeout=timeout, **kwargs)
def create_ssh_process(self, command, environment, timeout, stdin, stdout,
stderr, ssh_client, **kwargs):
"""Execute command on a remote host using SSH client"""
if isinstance(ssh_client, ssh.SSHClientFixture):
# Connect to SSH server
ssh_client = ssh_client.connect()
if not isinstance(ssh_client, paramiko.SSHClient):
message = "Object {!r} is not an SSHClient".format(ssh_client)
raise TypeError(message)
LOG.debug("Execute command %r on remote host (timeout=%r)...",
str(command), timeout)
channel = ssh_client.get_transport().open_session()
if environment:
channel.update_environment(environment)
channel.exec_command(str(command))
if stdin:
kwargs.update(stdin=StdinSSHChannelFile(channel, 'wb'))
if stdout:
kwargs.update(stdout=StdoutSSHChannelFile(channel, 'rb'))
if stderr:
kwargs.update(stderr=StderrSSHChannelFile(channel, 'rb'))
return SSHShellProcess(channel=channel, command=command,
timeout=timeout, **kwargs)
def set_non_blocking(fd):
flag = fcntl.fcntl(fd, fcntl.F_GETFL)
fcntl.fcntl(fd, fcntl.F_SETFL, flag | os.O_NONBLOCK)
class LocalShellProcess(_process.ShellProcess):
def __init__(self, process=None, **kwargs):
super(LocalShellProcess, self).__init__(**kwargs)
self.process = process
def poll_exit_status(self):
return self.process.poll()
def kill(self):
self.process.kill()
class SSHChannelFile(paramiko.ChannelFile):
def fileno(self):
return self.channel.fileno()
class StdinSSHChannelFile(SSHChannelFile):
def close(self):
super(StdinSSHChannelFile, self).close()
self.channel.shutdown_write()
@property
def write_ready(self):
return self.channel.send_ready()
class StdoutSSHChannelFile(SSHChannelFile):
def fileno(self):
return self.channel.fileno()
def close(self):
super(StdoutSSHChannelFile, self).close()
self.channel.shutdown_read()
@property
def read_ready(self):
return self.channel.recv_ready()
class StderrSSHChannelFile(SSHChannelFile, paramiko.channel.ChannelStderrFile):
def fileno(self):
return self.channel.fileno()
@property
def read_ready(self):
return self.channel.recv_stderr_ready()
class SSHShellProcess(_process.ShellProcess):
def __init__(self, channel=None, **kwargs):
super(SSHShellProcess, self).__init__(**kwargs)
self.channel = channel
def poll_exit_status(self):
if self.channel.exit_status_ready():
return self.channel.recv_exit_status()
def close(self):
super(SSHShellProcess, self).close()
self.channel.close()
if expect_exit_status is not None:
process.check_exit_status(expect_exit_status)
return ShellExecuteResult(command=str(process.command),
exit_status=int(process.exit_status),
stdin=_process.str_from_stream(process.stdin),
stdout=_process.str_from_stream(process.stdout),
stderr=_process.str_from_stream(process.stderr))

15
tobiko/shell/sh/_io.py

@ -94,8 +94,17 @@ class ShellReadable(ShellIOBase):
def read(self, size=None):
size = size or self.buffer_size
chunk = self.delegate.read(size)
self._data_chunks.append(chunk)
try:
chunk = self.delegate.read(size)
except IOError:
chunk = None
try:
self.close()
except Exception:
pass
if chunk:
self._data_chunks.append(chunk)
return chunk
@property
@ -110,6 +119,8 @@ class ShellWritable(ShellIOBase):
return True
def write(self, data):
if not isinstance(data, six.binary_type):
data = data.encode()
witten_bytes = self.delegate.write(data)
if witten_bytes is None:
witten_bytes = len(data)

116
tobiko/shell/sh/_local.py

@ -0,0 +1,116 @@
# Copyright (c) 2019 Red Hat, Inc.
#
# All Rights Reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may
# not use this file except in compliance with the License. You may obtain
# a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations
# under the License.
from __future__ import absolute_import
import fcntl
import os
import subprocess
from oslo_log import log
from tobiko.shell.sh import _io
from tobiko.shell.sh import _execute
from tobiko.shell.sh import _process
LOG = log.getLogger(__name__)
def local_execute(command, environment=None, timeout=None, shell=None,
stdin=True, stdout=True, stderr=True, expect_exit_status=0,
**kwargs):
"""Execute command on local host using local shell"""
process = local_process(command=command,
environment=environment,
timeout=timeout,
shell=shell,
stdin=stdin,
stdout=stdout,
stderr=stderr,
**kwargs)
return _execute.execute_process(process=process,
stdin=stdin,
expect_exit_status=expect_exit_status)
def local_process(command, environment=None, current_dir=None, timeout=None,
shell=None, stdin=None, stdout=None, stderr=True):
return LocalShellProcessFixture(
command=command, environment=environment, current_dir=current_dir,
timeout=timeout, shell=shell, stdin=stdin, stdout=stdout,
stderr=stderr)
class LocalShellProcessFixture(_process.ShellProcessFixture):
def create_process(self):
parameters = self.parameters
popen_params = {}
if parameters.stdin:
popen_params.update(stdin=subprocess.PIPE)
if parameters.stdout:
popen_params.update(stdout=subprocess.PIPE)
if parameters.stderr:
popen_params.update(stderr=subprocess.PIPE)
return subprocess.Popen(
args=self.command,
bufsize=parameters.buffer_size,
shell=False,
cwd=parameters.current_dir,
env=merge_dictionaries(os.environ, parameters.environment),
universal_newlines=False,
**popen_params)
def setup_stdin(self):
set_non_blocking(self.process.stdin.fileno())
self.stdin = _io.ShellStdin(delegate=self.process.stdin,
buffer_size=self.parameters.buffer_size)
def setup_stdout(self):
set_non_blocking(self.process.stdout.fileno())
self.stdout = _io.ShellStdout(delegate=self.process.stdout,
buffer_size=self.parameters.buffer_size)
def setup_stderr(self):
set_non_blocking(self.process.stderr.fileno())
self.stderr = _io.ShellStderr(delegate=self.process.stderr,
buffer_size=self.parameters.buffer_size)
def poll_exit_status(self):
return self.process.poll()
def kill(self):
try:
self.process.kill()
except Exception:
LOG.exception('Failed killing subprocess')
@property
def pid(self):
return self.process.pid
def set_non_blocking(fd):
flag = fcntl.fcntl(fd, fcntl.F_GETFL)
fcntl.fcntl(fd, fcntl.F_SETFL, flag | os.O_NONBLOCK)
def merge_dictionaries(*dictionaries):
merged = {}
for d in dictionaries:
if d:
merged.update(d)
return merged

300
tobiko/shell/sh/_process.py

@ -20,6 +20,9 @@ import time
from oslo_log import log
import tobiko
from tobiko.shell.sh import _command
from tobiko.shell.sh import _exception
from tobiko.shell.sh import _io
@ -27,70 +30,166 @@ from tobiko.shell.sh import _io
LOG = log.getLogger(__name__)
class Timeout(object):
def process(command=None, environment=None, timeout=None, shell=None,
stdin=None, stdout=None, stderr=None, ssh_client=None, **kwargs):
kwargs.update(command=command, environment=environment, timeout=timeout,
shell=shell, stdin=stdin, stdout=stdout, stderr=stderr)
try:
from tobiko.shell.sh import _ssh
from tobiko.shell import ssh
except ImportError:
if ssh_client:
raise
else:
if ssh_client is None:
ssh_client = ssh.ssh_proxy_client()
if ssh_client:
return _ssh.ssh_process(ssh_client=ssh_client, **kwargs)
from tobiko.shell.sh import _local
return _local.local_process(**kwargs)
class Parameters(object):
def __init__(self, **kwargs):
cls = type(self)
for name, value in kwargs.items():
if value is not None:
if not hasattr(cls, name):
raise ValueError('Invalid parameter: {!s}'.format(name))
setattr(self, name, value)
class ShellProcessParameters(Parameters):
command = None
environment = None
current_dir = None
timeout = None
shell = None
stdin = False
stdout = True
stderr = True
buffer_size = io.DEFAULT_BUFFER_SIZE
poll_interval = 1.
timeout = float('inf')
def __init__(self, timeout=None, start_time=None):
if timeout is None:
timeout = float('inf')
else:
timeout = float(timeout)
self.timeout = timeout
start_time = start_time and float(start_time) or time.time()
self.start_time = start_time
self.end_time = start_time + timeout
class ShellProcessFixture(tobiko.SharedFixture):
def __float__(self):
return self.timeout
parameters = None
command = None
timeout = None
process = None
stdin = None
stdout = None
stderr = None
def time_left(self, now=None):
now = now or time.time()
return self.end_time - now
def __init__(self, **kwargs):
super(ShellProcessFixture, self).__init__()
self.parameters = self.init_parameters(**kwargs)
def is_expired(self, now=None):
raise self.time_left(now=now) <= 0.
def init_parameters(self, **kwargs):
return ShellProcessParameters(**kwargs)
def execute(self):
return tobiko.setup_fixture(self)
class ShellProcess(object):
def setup_fixture(self):
parameters = self.parameters
buffer_size = io.DEFAULT_BUFFER_SIZE
stdin = None
stdout = None
stderr = None
poll_time = 0.1
self.setup_command()
if parameters.timeout:
self.setup_timeout()
self.setup_process()
if parameters.stdin:
self.setup_stdin()
if parameters.stdout:
self.setup_stdout()
if parameters.stderr:
self.setup_stderr()
def __init__(self, command, timeout=None, stdin=None, stdout=None,
stderr=None, buffer_size=None, poll_time=None):
def setup_command(self):
command = _command.shell_command(self.parameters.command)
shell = self.parameters.shell
if shell:
if shell is True:
shell = default_shell_command()
else:
shell = _command.shell_command(shell)
command = shell + [str(command)]
else:
command = _command.shell_command(command)
self.command = command
self.timeout = Timeout(timeout)
if buffer_size is not None:
self.buffer_size = max(64, int(buffer_size))
if stdin:
self.stdin = _io.ShellStdin(stdin, buffer_size=self.buffer_size)
if stdout:
self.stdout = _io.ShellStdout(stdout, buffer_size=self.buffer_size)
if stderr:
self.stderr = _io.ShellStderr(stderr, buffer_size=self.buffer_size)
if poll_time is not None:
self.poll_time = max(0., float(poll_time))
def __enter__(self):
return self
def __exit__(self, _exception_type, _exception_value, _traceback):
self.close()
def close(self):
if self.is_running:
self.kill()
for f in _io.select_opened_files([self.stdin,
self.stdout,
self.stderr]):
f.close()
def setup_timeout(self):
self.timeout = ShellProcessTimeout(self.parameters.timeout)
def setup_process(self):
self.process = self.create_process()
self.addCleanup(self.close)
def setup_stdin(self):
raise NotImplementedError
def setup_stdout(self):
raise NotImplementedError
def setup_stderr(self):
raise NotImplementedError
def create_process(self):
raise NotImplementedError
def close_stdin(self):
stdin = self.stdin
if stdin is not None:
try:
stdin.closed or stdin.close()
except Exception:
LOG.exception("Error closing STDIN stream: %r", self.stdin)
def close_stdout(self):
stdout = self.stdout
if stdout is not None:
try:
stdout.closed or stdout.close()
except Exception:
LOG.exception("Error closing STDOUT stream: %r", self.stdout)
def close_stderr(self):
stderr = self.stderr
if stderr is not None:
try:
stderr.closed or stderr.close()
except Exception:
LOG.exception("Error closing STDERR stream: %r", self.stderr)
def close(self, timeout=None):
self.close_stdin()
try:
# Drain all incoming data from STDOUT and STDERR
self.wait(timeout=timeout)
finally:
# Avoid leaving zombie processes
self.timeout = None
self.close_stdout()
self.close_stderr()
if self.is_running:
self.kill()
def __getattr__(self, name):
try:
# Get attributes from parameters class
return getattr(self.parameters, name)
except AttributeError:
message = "object {!r} has not attribute {!r}".format(self, name)
raise AttributeError(message)
def kill(self):
pass
raise NotImplementedError
def poll_exit_status(self):
raise NotImplementedError
@ -101,25 +200,25 @@ class ShellProcess(object):
@property
def is_running(self):
return self.poll_exit_status() is None
return self.exit_status is None
def check_is_running(self):
exit_status = self.poll_exit_status()
if exit_status is not None:
raise _exception.ShellProcessTeriminated(
command=self.command,
command=str(self.command),
exit_status=int(exit_status),
stdin=self.stdin,
stdout=self.stdout,
stderr=self.stderr)
stdin=str_from_stream(self.stdin),
stdout=str_from_stream(self.stdout),
stderr=str_from_stream(self.stderr))
def check_stdin_is_opened(self):
if self.stdin.closed:
raise _exception.ShellStdinClosed(
command=self.command,
stdin=self.stdin,
stdout=self.stdout,
stderr=self.stderr)
command=str(self.command),
stdin=str_from_stream(self.stdin),
stdout=str_from_stream(self.stdout),
stderr=str_from_stream(self.stderr))
def send(self, data, timeout=None):
self.comunicate(stdin=data, timeout=timeout, wait=False)
@ -129,13 +228,13 @@ class ShellProcess(object):
def comunicate(self, stdin=None, stdout=True, stderr=True, timeout=None,
wait=True):
timeout = Timeout(timeout=timeout)
timeout = ShellProcessTimeout(timeout=timeout)
# Avoid waiting for data in the first loop
poll_time = 0.
poll_interval = 0.
poll_files = _io.select_opened_files([stdin and self.stdin,
stdout and self.stdout,
stderr and self.stderr])
buffer_size = self.parameters.buffer_size
while wait or stdin or poll_files:
self.check_timeout(timeout=timeout)
if stdin:
@ -145,14 +244,14 @@ class ShellProcess(object):
wait = wait and self.is_running
read_ready, write_ready = _io.select_files(files=poll_files,
timeout=poll_time)
timeout=poll_interval)
if read_ready or write_ready:
# Avoid waiting for data the next time
poll_time = 0.
poll_interval = 0.
else:
# Wait for data in the following loops
poll_time = min(self.poll_time,
self.check_timeout(timeout=timeout))
poll_interval = min(self.poll_interval,
self.check_timeout(timeout=timeout))
if self.stdin in write_ready:
# Write data to remote STDIN
@ -167,14 +266,14 @@ class ShellProcess(object):
if self.stdout in read_ready:
# Read data from remote STDOUT
chunk = self.stdout.read(self.buffer_size)
chunk = self.stdout.read(buffer_size)
if not chunk:
LOG.debug("STDOUT channel closed by peer on %r", self)
self.stdout.close()
if self.stderr in read_ready:
# Read data from remote STDERR
chunk = self.stderr.read(self.buffer_size)
chunk = self.stderr.read(buffer_size)
if not chunk:
LOG.debug("STDERR channel closed by peer on %r", self)
self.stderr.close()
@ -196,11 +295,11 @@ class ShellProcess(object):
time_left = min(time_left, timeout.time_left(now=now))
if time_left <= 0.:
ex = _exception.ShellTimeoutExpired(
command=self.command,
command=str(self.command),
timeout=timeout.timeout,
stdin=self.stdin,
stdout=self.stdout,
stderr=self.stderr)
stdin=str_from_stream(self.stdin),
stdout=str_from_stream(self.stdout),
stderr=str_from_stream(self.stderr))
LOG.debug("%s", ex)
raise ex
return time_left
@ -210,7 +309,7 @@ class ShellProcess(object):
if exit_status is None:
time_left = self.check_timeout()
ex = _exception.ShellProcessNotTeriminated(
command=self.command,
command=str(self.command),
time_left=time_left,
stdin=self.stdin,
stdout=self.stdout,
@ -221,11 +320,11 @@ class ShellProcess(object):
exit_status = int(exit_status)
if expected_status != exit_status:
ex = _exception.ShellCommandFailed(
command=self.command,
command=str(self.command),
exit_status=exit_status,
stdin=self.stdin,
stdout=self.stdout,
stderr=self.stderr)
stdin=str_from_stream(self.stdin),
stdout=str_from_stream(self.stdout),
stderr=str_from_stream(self.stderr))
LOG.debug("%s", ex)
raise ex
@ -237,5 +336,44 @@ class ShellProcess(object):
self.stdin, self.stdout, self.stderr)
def clamp(left, value, right):
return max(left, min(value, right))
def merge_dictionaries(*dictionaries):
merged = {}
for d in dictionaries:
if d:
merged.update(d)
return merged
class ShellProcessTimeout(object):
timeout = float('inf')
def __init__(self, timeout=None, start_time=None):
if timeout is None:
timeout = float('inf')
else:
timeout = float(timeout)
self.timeout = timeout
start_time = start_time and float(start_time) or time.time()
self.start_time = start_time
self.end_time = start_time + timeout
def __float__(self):
return self.timeout
def time_left(self, now=None):
now = now or time.time()
return self.end_time - now
def is_expired(self, now=None):
raise self.time_left(now=now) <= 0.
def str_from_stream(stream):
return stream and str(stream) or None
def default_shell_command():
from tobiko import config
CONF = config.CONF
return _command.shell_command(CONF.tobiko.shell.command)

158
tobiko/shell/sh/_ssh.py

@ -0,0 +1,158 @@
# Copyright (c) 2019 Red Hat, Inc.
#
# All Rights Reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may
# not use this file except in compliance with the License. You may obtain
# a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations
# under the License.
from __future__ import absolute_import
from oslo_log import log
import paramiko
from tobiko.shell.sh import _io
from tobiko.shell.sh import _local
from tobiko.shell.sh import _process
from tobiko.shell.sh import _execute
from tobiko.shell import ssh
LOG = log.getLogger(__name__)
def ssh_execute(ssh_client, command, environment=None, timeout=None,
stdin=True, stdout=True, stderr=True, shell=None,
expect_exit_status=0, **kwargs):
"""Execute command on local host using local shell"""
process = ssh_process(command=command,
environment=environment,
timeout=timeout,
shell=shell,
stdin=stdin,
stdout=stdout,
stderr=stderr,
ssh_client=ssh_client,
**kwargs)
return _execute.execute_process(process=process,
stdin=stdin,
expect_exit_status=expect_exit_status)
def ssh_process(command, environment=None, current_dir=None, timeout=None,
shell=None, stdin=None, stdout=None, stderr=None,
ssh_client=None):
if ssh_client is None:
ssh_client = ssh.ssh_proxy_client()
if ssh_client:
return SSHShellProcessFixture(
command=command, environment=environment, current_dir=current_dir,
timeout=timeout, shell=shell, stdin=stdin, stdout=stdout,
stderr=stderr, ssh_client=ssh_client)
else:
return _local.local_process(
command=command, environment=environment, current_dir=current_dir,
timeout=timeout, shell=shell, stdin=stdin, stdout=stdout,
stderr=stderr)
class SSHShellProcessParameters(_process.ShellProcessParameters):
ssh_client = None
class SSHShellProcessFixture(_process.ShellProcessFixture):
def init_parameters(self, **kwargs):
return SSHShellProcessParameters(**kwargs)
def create_process(self):
"""Execute command on a remote host using SSH client"""
parameters = self.parameters
assert isinstance(parameters, SSHShellProcessParameters)
ssh_client = self.ssh_client
if isinstance(ssh_client, ssh.SSHClientFixture):
# Connect to SSH server
ssh_client = ssh_client.connect()
process = ssh_client.get_transport().open_session()
command = str(self.command)
LOG.debug("Execute command %r on remote host (timeout=%r)...",
command, self.timeout)
if parameters.environment:
process.update_environment(parameters.environment)
process.exec_command(command)
return process
def setup_stdin(self):
self.stdin = _io.ShellStdin(
delegate=StdinSSHChannelFile(self.process, 'wb'),
buffer_size=self.parameters.buffer_size)
def setup_stdout(self):
self.stdout = _io.ShellStdout(
delegate=StdoutSSHChannelFile(self.process, 'rb'),
buffer_size=self.parameters.buffer_size)
def setup_stderr(self):
self.stderr = _io.ShellStderr(
delegate=StderrSSHChannelFile(self.process, 'rb'),
buffer_size=self.parameters.buffer_size)
def poll_exit_status(self):
if self.process.exit_status_ready():
return self.process.recv_exit_status()
else:
return None
def kill(self):
self.process.close()
class SSHChannelFile(paramiko.ChannelFile):
def fileno(self):
return self.channel.fileno()
class StdinSSHChannelFile(SSHChannelFile):
def close(self):
super(StdinSSHChannelFile, self).close()
self.channel.shutdown_write()
@property
def write_ready(self):
return self.channel.send_ready()
class StdoutSSHChannelFile(SSHChannelFile):
def fileno(self):
return self.channel.fileno()
def close(self):
super(StdoutSSHChannelFile, self).close()
self.channel.shutdown_read()
@property
def read_ready(self):
return self.channel.recv_ready()
class StderrSSHChannelFile(SSHChannelFile, paramiko.channel.ChannelStderrFile):
def fileno(self):
return self.channel.fileno()
@property
def read_ready(self):
return self.channel.recv_stderr_ready()

33
tobiko/tests/functional/shell/test_sh.py

@ -33,7 +33,7 @@ class ExecuteTest(testtools.TestCase):
def test_succeed(self, command='true', stdin=None, stdout=None,
stderr=None, **kwargs):
process = self.execute(command,
process = self.execute(command=command,
stdin=stdin,
stdout=bool(stdout),
stderr=bool(stderr),
@ -75,22 +75,24 @@ class ExecuteTest(testtools.TestCase):
def test_fails(self, command='false', exit_status=None, stdin=None,
stdout=None, stderr=None, **kwargs):
ex = self.assertRaises(sh.ShellCommandFailed, self.execute, command,
ex = self.assertRaises(sh.ShellCommandFailed,
self.execute,
command=command,
stdin=stdin,
stdout=bool(stdout),
stderr=bool(stderr),
**kwargs)
self.assertEqual(self.expected_command(command), ex.command)
if stdin:
self.assertEqual(stdin, str(ex.stdin))
self.assertEqual(stdin, ex.stdin)
else:
self.assertIsNone(ex.stdin)
if stdout:
self.assertEqual(stdout, str(ex.stdout))
self.assertEqual(stdout, ex.stdout)
else:
self.assertIsNone(ex.stdout)
if stderr:
self.assertEqual(stderr, str(ex.stderr))
self.assertEqual(stderr, ex.stderr)
else:
self.assertIsNone(ex.stderr)
if exit_status:
@ -116,7 +118,9 @@ class ExecuteTest(testtools.TestCase):
def test_timeout_expires(self, command='sleep 5', timeout=0.1, stdin=None,
stdout=None, stderr=None, **kwargs):
ex = self.assertRaises(sh.ShellTimeoutExpired, self.execute, command,
ex = self.assertRaises(sh.ShellTimeoutExpired,
self.execute,
command=command,
timeout=timeout,
stdin=stdin,
stdout=bool(stdout),
@ -137,22 +141,23 @@ class ExecuteTest(testtools.TestCase):
self.assertIsNone(ex.stderr)
self.assertEqual(timeout, ex.timeout)
def execute(self, command, **kwargs):
def execute(self, **kwargs):
kwargs.setdefault('shell', self.shell)
kwargs.setdefault('ssh_client', self.ssh_client)
return sh.execute(command, **kwargs)
return sh.execute(**kwargs)
def expected_command(self, command):
command = sh.shell_command(command)
shell = sh.shell_command(self.shell)
return shell + [str(command)]
if self.shell:
command = sh.shell_command(self.shell) + [str(command)]
return str(command)
class LocalExecuteTest(ExecuteTest):
def execute(self, command, **kwargs):
def execute(self, **kwargs):
kwargs.setdefault('shell', self.shell)
return sh.local_execute(command, **kwargs)
return sh.local_execute(**kwargs)
class SSHExecuteTest(ExecuteTest):
@ -164,9 +169,9 @@ class SSHExecuteTest(ExecuteTest):
def ssh_client(self):
return self.server_stack.ssh_client
def execute(self, command, **kwargs):
def execute(self, **kwargs):
kwargs.setdefault('shell', self.shell)
return sh.ssh_execute(self.ssh_client, command, **kwargs)
return sh.ssh_execute(ssh_client=self.ssh_client, **kwargs)
class ExecuteWithSSHCommandTest(ExecuteTest):

Loading…
Cancel
Save