Merge "Add wrapper to which command"

This commit is contained in:
Zuul 2022-07-08 10:38:04 +00:00 committed by Gerrit Code Review
commit 26051ea631
5 changed files with 176 additions and 4 deletions

View File

@ -86,7 +86,13 @@ ENV TOBIKO_REPORT_NAME=tobiko_results
ENV TOBIKO_PREVENT_CREATE=false
ENV TOBIKO_TEST_PATH=${TOBIKO_DIR}/tobiko/tests/unit
RUN ${INSTALL_PACKAGES} iperf3 iputils nmap-ncat findutils procps
RUN ${INSTALL_PACKAGES} \
findutils \
iperf3 \
iputils \
nmap-ncat \
procps \
which
# Write log files to report directory
RUN mkdir -p /etc/tobiko

View File

@ -16,13 +16,14 @@ from __future__ import absolute_import
import functools
import inspect
import typing
import unittest
import fixtures
import testtools
SKIP_CLASSES = unittest.SkipTest, testtools.TestCase.skipException
SkipException: typing.Type[Exception] = (
testtools.TestCase.skipException)
SkipException = unittest.SkipTest
SkipTarget = typing.Union[typing.Callable,
typing.Type[testtools.TestCase],
@ -42,7 +43,7 @@ def skip_test(reason: str,
reason += f'\nhttps://bugzilla.redhat.com/show_bug.cgi?id={bugzilla}\n'
if cause is not None:
reason += f"\n{cause}\n"
raise SkipException(reason) from cause
raise unittest.SkipTest(reason) from cause
def skip(reason: str,

View File

@ -32,6 +32,7 @@ from tobiko.shell.sh import _ssh
from tobiko.shell.sh import _systemctl
from tobiko.shell.sh import _uptime
from tobiko.shell.sh import _wc
from tobiko.shell.sh import _which
get_command_line = _cmdline.get_command_line
@ -129,3 +130,7 @@ get_uptime = _uptime.get_uptime
assert_file_size = _wc.assert_file_size
get_file_size = _wc.get_file_size
CommandNotFound = _which.CommandNotFound
SkipOnCommandNotFound = _which.SkipOnCommandNotFound
find_command = _which.find_command

61
tobiko/shell/sh/_which.py Normal file
View File

@ -0,0 +1,61 @@
# Copyright (c) 2021 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 functools
from oslo_log import log
import tobiko
from tobiko.shell.sh import _exception
from tobiko.shell.sh import _execute
from tobiko.shell.sh import _hostname
from tobiko.shell import ssh
LOG = log.getLogger(__name__)
class CommandNotFound(tobiko.ObjectNotFound):
message = "Command {command!r} not found on host {hostname!r}"
class SkipOnCommandNotFound(CommandNotFound, tobiko.SkipException):
pass
@functools.lru_cache()
def find_command(command: str,
ssh_client: ssh.SSHClientType = None,
sudo=False,
skip=False) -> str:
hostname = _hostname.get_hostname(ssh_client=ssh_client)
try:
result = _execute.execute(['which', command],
ssh_client=ssh_client,
sudo=sudo)
except _exception.ShellCommandFailed as ex:
if skip:
raise SkipOnCommandNotFound(command=command,
hostname=hostname) from ex
else:
raise CommandNotFound(command=command,
hostname=hostname) from ex
else:
command_path = result.stdout.strip()
LOG.debug(f"Command {command!r} found on host {hostname}: "
f"{command_path}")
return command_path

View File

@ -0,0 +1,99 @@
# 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 os
import testtools
import tobiko
from tobiko import config
from tobiko.openstack import keystone
from tobiko.openstack import stacks
from tobiko.shell import sh
from tobiko.shell import ssh
CONF = config.CONF
class WhichTest(testtools.TestCase):
@property
def ssh_client(self) -> ssh.SSHClientType:
return False
def test_find_command(self):
result = sh.find_command(command='which',
ssh_client=self.ssh_client)
self.assertEqual('which', os.path.basename(result))
def test_find_command_with_invalid(self):
ex = self.assertRaises(sh.CommandNotFound,
sh.find_command,
command='<invalid_command>',
ssh_client=self.ssh_client)
self.assertEqual('<invalid_command>', ex.command)
self.assertEqual(sh.get_hostname(ssh_client=self.ssh_client),
ex.hostname)
self.assertIsInstance(ex.__cause__, sh.ShellCommandFailed)
def test_find_command_with_skip(self):
result = sh.find_command(command='which',
ssh_client=self.ssh_client,
skip=True)
self.assertEqual('which', os.path.basename(result))
def test_find_command_with_invalid_and_skip(self):
ex = self.assertRaises(sh.SkipOnCommandNotFound,
sh.find_command,
command='<invalid_command>',
skip=True,
ssh_client=self.ssh_client)
self.assertIsInstance(ex, tobiko.SkipException)
self.assertIsInstance(ex, sh.CommandNotFound)
self.assertEqual('<invalid_command>', ex.command)
self.assertEqual(sh.get_hostname(ssh_client=self.ssh_client),
ex.hostname)
self.assertIsInstance(ex.__cause__, sh.ShellCommandFailed)
class ProxyJumpWhichTest(WhichTest):
def setUp(self):
super().setUp()
if ssh.ssh_proxy_client() is None:
self.skipTest('SSH proxy jump not configured')
@property
def ssh_client(self) -> ssh.SSHClientType:
return None
@keystone.skip_unless_has_keystone_credentials()
class SSHWhichTest(WhichTest):
server_stack = tobiko.required_fixture(
stacks.UbuntuMinimalServerStackFixture)
@property
def ssh_client(self):
return self.server_stack.ssh_client
@keystone.skip_unless_has_keystone_credentials()
class CirrosExecuteTest(SSHWhichTest):
server_stack = tobiko.required_fixture(stacks.CirrosServerStackFixture)