x
/
tobiko
Code Issues Proposed changes

Improve ping integration error handling 

Change-Id: Id55bc937d62a9f2aab44281761083e27cc878792
This commit is contained in:
Federico Ressi 2020-10-30 10:29:36 +01:00
parent 8d7ee27a8a
commit 6276909ac2
11 changed files with 186 additions and 61 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

9
tobiko/common/_exception.py
View File

@ -75,12 +75,17 @@ class TobikoException(Exception):
class_name=type(self).__name__,
message=self.message)
def __eq__(self, other):
return type(self) == type(other) and str(self) == str(other)
def __hash__(self):
return hash(type(self)) + hash(str(self))
def check_valid_type(obj, *valid_types):
if not isinstance(obj, valid_types):
types_str = ", ".join(str(t) for t in valid_types)
message = ("Object {!r} is not of a valid type ({!s})").format(
obj, types_str)
message = f"Object {obj!r} is not of a valid type ({types_str})"
raise TypeError(message)
return obj

10
tobiko/shell/ping/__init__.py
View File

@ -26,12 +26,14 @@ from tobiko.shell.ping import _statistics
assert_reachable_hosts = _assert.assert_reachable_hosts
assert_unreachable_hosts = _assert.assert_unreachable_hosts
PingException = _exception.PingException
PingError = _exception.PingError
LocalPingError = _exception.LocalPingError
BadAddressPingError = _exception.BadAddressPingError
UnknowHostError = _exception.UnknowHostError
LocalPingError = _exception.LocalPingError
ConnectPingError = _exception.ConnectPingError
PingFailed = _exception.PingFailed
PingError = _exception.PingError
PingException = _exception.PingException
SendToPingError = _exception.SendToPingError
UnknowHostError = _exception.UnknowHostError
skip_if_missing_fragment_ping_option = (
_interface.skip_if_missing_fragment_ping_option)

2
tobiko/shell/ping/_exception.py
View File

@ -45,7 +45,7 @@ class UnknowHostError(PingError):
class BadAddressPingError(PingError):
"""Raised when passing wrong address to ping command"""
message = "bad address ({address!r})"
message = "bad address: {address}"
class PingFailed(PingError, tobiko.FailureException):

2
tobiko/shell/ping/_interface.py
View File

@ -123,7 +123,7 @@ class PingInterface(object):
def get_ping_command(self, parameters):
host = parameters.host
if not host:
raise ValueError("Ping host destination hasn't been specified")
raise ValueError(f"Invalid destination host: '{host}'")
command = sh.shell_command([self.get_ping_executable(parameters)] +
self.get_ping_options(parameters) +

18
tobiko/shell/ping/_ping.py
View File

@ -291,12 +291,13 @@ def handle_ping_command_error(error):
if error:
prefix = 'ping: '
if error.startswith('ping: '):
text = error[len(prefix):]
handle_ping_bad_address_error(text)
handle_ping_local_error(text)
handle_ping_send_to_error(text)
handle_ping_unknow_host_error(text)
raise _exception.PingError(details=text)
error = error[len(prefix):]
handle_ping_bad_address_error(error)
handle_ping_local_error(error)
handle_ping_connect_error(error)
handle_ping_send_to_error(error)
handle_ping_unknow_host_error(error)
raise _exception.PingError(details=error)
def handle_ping_bad_address_error(text):
@ -333,6 +334,11 @@ def handle_ping_unknow_host_error(text):
details = text[len(prefix):].strip()
raise _exception.UnknowHostError(details=details)
prefix = 'unreachable-host: '
if text.startswith(prefix):
details = text[len(prefix):].strip()
raise _exception.UnknowHostError(details=details)
suffix = ': Name or service not known'
if text.endswith(suffix):
details = text[:-len(suffix)].strip()

4
tobiko/shell/ping/_statistics.py
View File

@ -60,9 +60,9 @@ def parse_ping_header(line_it):
header_fields = [(f[:-1] if f.endswith(':') else f)
for f in header_fields]
destination = header_fields[2]
if destination[0] == '(':
while destination and destination[0] == '(':
destination = destination[1:]
if destination[-1] == ')':
while destination and destination[-1] == ')':
destination = destination[:-1]
destination = netaddr.IPAddress(destination)

4
tobiko/shell/ssh/__init__.py
View File

@ -19,6 +19,7 @@ from tobiko.shell.ssh import _config
from tobiko.shell.ssh import _client
from tobiko.shell.ssh import _command
from tobiko.shell.ssh import _forward
from tobiko.shell.ssh import _skip
SSHHostConfig = _config.SSHHostConfig
@ -37,3 +38,6 @@ get_port_forward_url = _forward.get_forward_url
get_forward_port_address = _forward.get_forward_port_address
SSHTunnelForwarderFixture = _forward.SSHTunnelForwarderFixture
SSHTunnelForwarder = _forward.SSHTunnelForwarder
has_ssh_proxy_jump = _skip.has_ssh_proxy_jump
skip_unless_has_ssh_proxy_jump = _skip.skip_unless_has_ssh_proxy_jump

26
tobiko/shell/ssh/_skip.py Normal file
View File

@ -0,0 +1,26 @@
# Copyright (c) 2020 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 tobiko
def has_ssh_proxy_jump():
return bool(tobiko.tobiko_config().ssh.proxy_jump)
skip_unless_has_ssh_proxy_jump = tobiko.skip_unless(
"SSH proxy jump not configured", has_ssh_proxy_jump)

0
tobiko/tests/functional/shell/fixtures.py → tobiko/tests/functional/shell/_fixtures.py
View File

5
tobiko/tests/functional/shell/test_ip.py
View File

@ -24,7 +24,7 @@ import tobiko
from tobiko.openstack import stacks
from tobiko.shell import ip
from tobiko.shell import ssh
from tobiko.tests.functional.shell import fixtures
from tobiko.tests.functional.shell import _fixtures
class IpTest(testtools.TestCase):
@ -38,7 +38,8 @@ class IpTest(testtools.TestCase):
ubuntu_stack = tobiko.required_setup_fixture(
stacks.UbuntuServerStackFixture)
namespace = tobiko.required_setup_fixture(fixtures.NetworkNamespaceFixture)
namespace = tobiko.required_setup_fixture(
_fixtures.NetworkNamespaceFixture)
def test_list_ip_addresses(self, ip_version=None, scope=None,
**execute_params):

167
tobiko/tests/functional/shell/test_ping.py
View File

@ -15,106 +15,187 @@
# under the License.
from __future__ import absolute_import
import os
import typing # noqa
import netaddr
import testtools
import tobiko
from tobiko import config
from tobiko.openstack import stacks
from tobiko.shell import ip
from tobiko.shell import ping
from tobiko.tests.functional.shell import fixtures
from tobiko.shell import sh
from tobiko.shell import ssh
from tobiko.tests.functional.shell import _fixtures
CONF = config.CONF
SshClientType = typing.Union[bool, None, ssh.SSHClientFixture]
class PingTest(testtools.TestCase):
namespace = tobiko.required_setup_fixture(fixtures.NetworkNamespaceFixture)
ssh_client: SshClientType = False
def test_ping_recheable_address(self):
result = ping.ping('127.0.0.1', count=3)
@property
def execute_params(self):
return dict(ssh_client=self.ssh_client)
def test_ping_reachable_address(self):
result = ping.ping('127.0.0.1', count=3,
**self.execute_params)
self.assertIsNone(result.source)
self.assertEqual(netaddr.IPAddress('127.0.0.1'), result.destination)
result.assert_transmitted()
result.assert_replied()
def test_ping_reachable_hostname(self):
result = ping.ping('example.org', count=3)
result = ping.ping('localhost', count=3, **self.execute_params)
self.assertIsNone(result.source)
# self.assertIsNotNone(result.destination)
self.assertIsNotNone(result.destination)
result.assert_transmitted()
result.assert_replied()
def test_ping_unreachable_address(self):
result = ping.ping('1.2.3.4', count=3)
result = ping.ping('1.2.3.4', count=3, check=False,
**self.execute_params)
self.assertIsNone(result.source)
self.assertEqual(netaddr.IPAddress('1.2.3.4'), result.destination)
result.assert_transmitted()
self.assertIn(result.destination, [netaddr.IPAddress('1.2.3.4'),
None])
if result.destination is not None:
result.assert_transmitted()
result.assert_not_replied()
def test_ping_invalid_ip(self):
try:
result = ping.ping('0.1.2.3', count=1,
**self.execute_params)
except ping.PingError as ex:
self.assertIn(ex, [
ping.ConnectPingError(details='Invalid argument'),
ping.ConnectPingError(details='Network is unreachable'),
ping.SendToPingError(details='No route to host'),
])
else:
self.assertIsNone(result.source)
self.assertEqual(netaddr.IPAddress('0.1.2.3'),
result.destination)
result.assert_transmitted()
result.assert_not_replied()
def test_ping_unreachable_hostname(self):
ex = self.assertRaises(ping.UnknowHostError, ping.ping,
'unreachable-host', count=3)
if ex.details:
self.assertEqual('unreachable-host', ex.details)
ex = self.assertRaises(ping.PingError, ping.ping,
'unreachable-host', count=3,
**self.execute_params)
self.assertIn(ex, [
ping.UnknowHostError(details='unreachable-host'),
ping.UnknowHostError(
details='Temporary failure in name resolution'),
ping.BadAddressPingError(address='unreachable-host'),
ping.UnknowHostError(
details='Name or service not known')
])
def test_ping_until_received(self):
result = ping.ping_until_received('127.0.0.1', count=3)
result = ping.ping_until_received('127.0.0.1', count=3,
**self.execute_params)
self.assertIsNone(result.source)
self.assertEqual(netaddr.IPAddress('127.0.0.1'), result.destination)
result.assert_transmitted()
result.assert_replied()
def test_ping_until_received_unreachable(self):
ex = self.assertRaises(ping.PingFailed, ping.ping_until_received,
'1.2.3.4', count=3, timeout=6)
self.assertEqual(6, ex.timeout)
self.assertEqual(0, ex.count)
self.assertEqual(3, ex.expected_count)
self.assertEqual('received', ex.message_type)
ex = self.assertRaises(ping.PingError, ping.ping_until_received,
'1.2.3.4', count=3, timeout=6,
**self.execute_params)
self.assertIn(ex, [
ping.PingFailed(timeout=6, count=0, expected_count=3,
message_type='received'),
ping.ConnectPingError(details='Network is unreachable')])
def test_ping_until_unreceived_recheable(self):
def test_ping_until_unreceived_reachable(self):
ex = self.assertRaises(ping.PingFailed, ping.ping_until_unreceived,
'127.0.0.1', count=3, timeout=6)
'127.0.0.1', count=3, timeout=6,
**self.execute_params)
self.assertEqual(6, ex.timeout)
self.assertEqual(0, ex.count)
self.assertEqual(3, ex.expected_count)
self.assertEqual('unreceived', ex.message_type)
def test_ping_until_unreceived_unrecheable(self):
result = ping.ping_until_unreceived('1.2.3.4', count=3,
check=False)
def test_ping_until_unreceived_unreachable(self):
result = ping.ping_until_unreceived('1.2.3.4', count=3, check=False,
**self.execute_params)
self.assertIsNone(result.source)
self.assertEqual(netaddr.IPAddress('1.2.3.4'), result.destination)
result.assert_transmitted()
if result.destination is None:
result.assert_not_transmitted()
else:
self.assertEqual(result.destination, netaddr.IPAddress('1.2.3.4'))
result.assert_transmitted()
result.assert_not_replied()
def test_ping_reachable_with_timeout(self):
ex = self.assertRaises(ping.PingFailed, ping.ping, '127.0.0.1',
count=20, timeout=1.)
count=20, timeout=1.,
**self.execute_params)
self.assertEqual(1., ex.timeout)
self.assertEqual(20, ex.expected_count)
self.assertEqual('transmitted', ex.message_type)
def test_ping_hosts(self, ssh_client=None, network_namespace=None,
**params):
if not os.path.isfile('/sbin/ip'):
self.skip("'/sbin/ip' command not found")
ips = ip.list_ip_addresses(ssh_client=ssh_client,
network_namespace=network_namespace)
reachable_ips, unrecheable_ips = ping.ping_hosts(
ips, ssh_client=ssh_client, network_namespace=network_namespace,
**params)
def test_ping_hosts(self):
try:
sh.execute('[ -x /sbin/ip ]', ssh_client=self.ssh_client)
except sh.ShellCommandFailed:
self.skipTest("'/sbin/ip' command not found")
ips = ip.list_ip_addresses(**self.execute_params)
reachable_ips, unreachable_ips = ping.ping_hosts(
ips, **self.execute_params)
expected_reachable = [i for i in ips if i in reachable_ips]
self.assertEqual(expected_reachable, reachable_ips)
expected_unreachable = [i for i in ips if i not in reachable_ips]
self.assertEqual(expected_unreachable, unrecheable_ips)
self.assertEqual(expected_unreachable, unreachable_ips)
def test_ping_hosts_from_network_namespace(self):
self.test_ping_hosts(
ssh_client=self.namespace.ssh_client,
network_namespace=self.namespace.network_namespace)
@ssh.skip_unless_has_ssh_proxy_jump
class ProxyPingTest(PingTest):
ssh_client = None
class NamespacePingTest(PingTest):
namespace = tobiko.required_setup_fixture(
_fixtures.NetworkNamespaceFixture)
@property
def ssh_client(self):
return self.namespace.ssh_client
@property
def network_namespace(self):
return self.namespace.network_namespace
@property
def execute_params(self):
return dict(ssh_client=self.ssh_client,
network_namespace=self.network_namespace)
class CirrosPingTest(PingTest):
stack = tobiko.required_setup_fixture(stacks.CirrosServerStackFixture)
@property
def ssh_client(self):
return self.stack.ssh_client
class CentosPingTest(CirrosPingTest):
stack = tobiko.required_setup_fixture(stacks.CentosServerStackFixture)
class UbuntuPingTest(CirrosPingTest):
stack = tobiko.required_setup_fixture(stacks.UbuntuServerStackFixture)