x/tobiko
Code Issues Proposed changes

Use ping.assert_reachable_hosts function to ping hosts

Browse Source 
Change-Id: I7d6ed0f1660333c2593367378c7f289cbc493ea2
This commit is contained in:
Federico Ressi 2021-07-01 12:35:15 +02:00
parent 188825386d
commit 5395792300
4 changed files with 24 additions and 27 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

31
tobiko/tests/functional/openstack/stacks/test_cirros.py
View File

@ -38,20 +38,21 @@ class CirrosServerStackTest(testtools.TestCase):
nameservers_filenames: typing.Optional[typing.Sequence[str]] = None
@property
def peer_ssh_client(self):
return self.stack.ssh_client
def test_ping_floating_ip(self):
"""Test connectivity to floating IP address"""
ping.ping_until_received(
self.stack.floating_ip_address).assert_replied()
ping.assert_reachable_hosts([self.stack.floating_ip_address])
def test_ping_fixed_ipv4(self):
ping.ping_until_received(
self.get_fixed_ip(ip_version=4),
ssh_client=self.stack.ssh_client).assert_replied()
ping.assert_reachable_hosts([self.get_fixed_ip(ip_version=4)],
ssh_client=self.peer_ssh_client)
def test_ping_fixed_ipv6(self):
ping.ping_until_received(
self.get_fixed_ip(ip_version=6),
ssh_client=self.stack.ssh_client).assert_replied()
ping.assert_reachable_hosts([self.get_fixed_ip(ip_version=6)],
ssh_client=self.peer_ssh_client)
def get_fixed_ip(self, ip_version: int):
try:
@ -141,19 +142,13 @@ class CirrosPeerServerStackTest(CirrosServerStackTest):
#: Stack of resources with an HTTP server
stack = tobiko.required_setup_fixture(stacks.CirrosPeerServerStackFixture)
@property
def peer_ssh_client(self):
return self.stack.peer_stack.ssh_client
def test_ping_floating_ip(self):
self.skipTest(f"Server '{self.stack.server_id}' has any floating IP")
def test_ping_fixed_ipv4(self):
ping.ping_until_received(
self.get_fixed_ip(ip_version=4),
ssh_client=self.stack.peer_stack.ssh_client).assert_replied()
def test_ping_fixed_ipv6(self):
ping.ping_until_received(
self.get_fixed_ip(ip_version=6),
ssh_client=self.stack.peer_stack.ssh_client).assert_replied()
class HttpServerStackTest(CirrosPeerServerStackTest):

10
tobiko/tests/scenario/neutron/test_floating_ip.py
View File

@ -49,8 +49,7 @@ class FloatingIPTest(testtools.TestCase):
def test_ping(self):
"""Test ICMP connectivity to floating IP address"""
ping.ping_until_received(
self.stack.floating_ip_address).assert_replied()
ping.assert_reachable_hosts([self.stack.floating_ip_address])
# --- test port-security extension ---------------------------------------
@ -155,10 +154,9 @@ class FloatingIPWithPortSecurityTest(FloatingIPTest):
# Wait for server instance to get ready by logging in
self.stack.ssh_client.connect()
# Check can't reach secured port via floating IP
ping.ping(self.stack.floating_ip_address,
count=5,
check=False).assert_not_replied()
# Expect port security to throw away ICMP packages
ping.assert_unreachable_hosts([self.stack.floating_ip_address],
count=5, check=False)
@ping.skip_if_missing_fragment_ping_option
@neutron.skip_if_missing_networking_extensions('net-mtu')

6
tobiko/tests/scenario/neutron/test_network.py
View File

@ -43,9 +43,9 @@ class NetworkTest(testtools.TestCase):
def test_ping(self):
"""Test ICMP connectivity to floating IP address"""
ping.ping_until_received(
self.stack.ip_address,
ssh_client=self.stack.peer_stack.ssh_client).assert_replied()
ping.assert_reachable_hosts(
[self.stack.ip_address],
ssh_client=self.stack.peer_stack.ssh_client)
# --- test l3_ha extension ------------------------------------------------

4
tobiko/tests/scenario/neutron/test_qos.py
View File

@ -21,6 +21,7 @@ import tobiko
from tobiko.openstack import stacks
from tobiko.openstack import topology
from tobiko.shell import iperf
from tobiko.shell import ping
from tobiko.tripleo import containers
from tobiko.tripleo import overcloud
@ -45,6 +46,9 @@ class QoSNetworkTest(testtools.TestCase):
# than 16.1
self.skipTest("QoS not supported in this setup")
def test_ping(self):
ping.assert_reachable_hosts([self.server.floating_ip_address],)
def test_network_qos_policy_id(self):
"""Verify network policy ID"""
self.assertEqual(self.policy.qos_policy_id,