x
/
tobiko
Code Issues Proposed changes

Fix/refactor PortTest 

Change-Id: I56fd8959f27aea74c175891ec0a049b0adc60b29
This commit is contained in:
Federico Ressi 2020-11-27 15:10:27 +01:00
parent e68bc18773
commit 0cc9d75f32
4 changed files with 93 additions and 53 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
tobiko/openstack/neutron/__init__.py
View File

@ -72,6 +72,8 @@ default_nameservers = _nameservers.default_nameservers
find_port_ip_address = _port.find_port_ip_address
list_port_ip_addresses = _port.list_port_ip_addresses
find_device_ip_address = _port.find_device_ip_address
list_device_ip_addresses = _port.list_device_ip_addresses
NeutronNetworkFixture = _network.NeutronNetworkFixture
NoSuchNetwork = _network.NoSuchNetwork

8
tobiko/openstack/neutron/_client.py
View File

@ -114,7 +114,13 @@ def list_ports(client=None, **params):
return tobiko.select(ports)
def list_subnets(client=None, **params):
NeutronSubnetType = typing.Dict[str, typing.Any]
def list_subnets(client=None, ip_version: typing.Optional[int] = None,
**params) -> tobiko.Selection[NeutronSubnetType]:
if ip_version is not None:
params['ip_version'] = ip_version
subnets = neutron_client(client).list_subnets(**params)
if isinstance(subnets, collections.Mapping):
subnets = subnets['subnets']

73
tobiko/openstack/neutron/_port.py
View File

@ -13,31 +13,74 @@
# under the License.
from __future__ import absolute_import
import typing
import netaddr
import tobiko
from tobiko.openstack.neutron import _client
from tobiko.shell import ssh
from tobiko.shell import ping
def list_port_ip_addresses(port, subnet_id=None, ip_version=None,
check_connectivity=False, ssh_client=None):
selected_addresses = []
for fixed_ip in port['fixed_ips']:
if subnet_id and subnet_id != fixed_ip['subnet_id']:
continue
ip_address = netaddr.IPAddress(fixed_ip['ip_address'])
if ip_version and ip_version != ip_address.version:
continue
if check_connectivity and not ping.ping(
host=ip_address, ssh_client=ssh_client).received:
continue
selected_addresses.append(ip_address)
return tobiko.Selection(selected_addresses)
NeutronPortType = typing.Dict[str, typing.Any]
def find_port_ip_address(port, unique=False, **kwargs):
def list_port_ip_addresses(port: NeutronPortType,
subnet_id: typing.Optional[str] = None,
ip_version: typing.Optional[int] = None,
check_connectivity: bool = False,
ssh_client: ssh.SSHClientFixture = None) -> \
tobiko.Selection[netaddr.IPAddress]:
addresses = tobiko.Selection[netaddr.IPAddress](
netaddr.IPAddress(fixed_ip['ip_address'])
for fixed_ip in port['fixed_ips']
if subnet_id is None or subnet_id == fixed_ip['subnet_id'])
if ip_version:
addresses = addresses.with_attributes(version=ip_version)
if addresses and check_connectivity:
hosts = ping.list_reachable_hosts(addresses, ssh_client=ssh_client)
addresses = tobiko.Selection(netaddr.IPAddress(host) for host in hosts)
return addresses
def find_port_ip_address(port: NeutronPortType, unique: bool = False,
**kwargs) -> netaddr.IPAddress:
addresses = list_port_ip_addresses(port=port, **kwargs)
if unique:
return addresses.unique
else:
return addresses.first
def list_device_ip_addresses(device_id: str,
network_id: typing.Optional[str] = None,
ip_version: typing.Optional[int] = None,
check_connectivity: bool = False,
ssh_client: ssh.SSHClientFixture = None,
**subnet_params) -> \
tobiko.Selection[netaddr.IPAddress]:
ports = _client.list_ports(device_id=device_id,
network_id=network_id)
subnets = _client.list_subnets(network_id=network_id,
ip_version=ip_version,
**subnet_params)
addresses = tobiko.Selection[netaddr.IPAddress](
port_ip
for subnet in subnets
for port in ports
for port_ip in list_port_ip_addresses(port=port,
subnet_id=subnet['id'],
ip_version=ip_version))
if addresses and check_connectivity:
hosts = ping.list_reachable_hosts(addresses, ssh_client=ssh_client)
addresses = tobiko.Selection(netaddr.IPAddress(host) for host in hosts)
return addresses
def find_device_ip_address(device_id: str, unique: bool = False, **kwargs):
addresses = list_device_ip_addresses(device_id=device_id, **kwargs)
if unique:
return addresses.unique
else:
return addresses.first

63
tobiko/tests/scenario/neutron/test_port.py
View File

@ -14,6 +14,8 @@
# under the License.
from __future__ import absolute_import
import typing
import netaddr
from oslo_log import log
import testtools
@ -37,21 +39,17 @@ class PortTest(testtools.TestCase):
#: Resources stack with Nova server to send messages to
stack = tobiko.required_setup_fixture(stacks.CirrosServerStackFixture)
def test_port_ips(self, ip_version=None):
port = self.stack.port_details
port_ips = tobiko.Selection()
for subnet in neutron.list_subnets(
network_id=self.stack.network_stack.network_id):
if subnet['enable_dhcp']:
port_ips += neutron.list_port_ip_addresses(
port=port, subnet_id=subnet['id'], ip_version=ip_version)
else:
LOG.warning(f"Subnet '{subnet['id']}' has "
f" enable_dhcp={subnet['enable_dhcp']}")
if port_ips:
server_ips = ip.list_ip_addresses(scope='global',
ssh_client=self.stack.ssh_client)
self.assertEqual(set(port_ips), set(port_ips) & set(server_ips))
def test_port_ips(self, ip_version: typing.Optional[int] = None):
"""Checks port IPS has been assigned to server via DHCP protocol"""
device_ips = set(neutron.list_device_ip_addresses(
device_id=self.stack.server_id,
network_id=self.stack.network_stack.network_id,
enable_dhcp=True,
ip_version=ip_version))
if device_ips:
server_ips = set(ip.list_ip_addresses(
scope='global', ssh_client=self.stack.ssh_client))
self.assertEqual(device_ips, device_ips & server_ips)
elif ip_version:
self.skipTest(f"No port IPv{ip_version} addresses found")
else:
@ -62,36 +60,34 @@ class PortTest(testtools.TestCase):
self.stack.port_details['network_id'])
def test_port_subnets(self):
"""Checks port subnets"""
port_subnets = [fixed_ip['subnet_id']
for fixed_ip in self.stack.port_details['fixed_ips']]
network_subnets = self.stack.network_stack.network_details['subnets']
self.assertEqual(set(network_subnets), set(port_subnets))
def test_ping_subnet_gateways(self):
"""Checks server can ping its gateway IPs"""
network_id = self.stack.network_stack.network_id
subnets = neutron.list_subnets(network_id=network_id)
subnets = neutron.list_subnets(network_id=network_id,
enable_dhcp=True)
LOG.debug(f"Subnets with DHCP enabled are: {subnets}")
gateway_ips = [netaddr.IPAddress(subnet['gateway_ip'])
for subnet in subnets]
LOG.debug(f"Gateway IPs are: {gateway_ips}")
ping.assert_reachable_hosts(gateway_ips,
ssh_client=self.stack.ssh_client)
def test_ping_port(self, network_id=None, device_id=None,
ip_version=None):
network_id = network_id or self.stack.network_stack.network_id
device_id = device_id or self.stack.server_id
ports = neutron.list_ports(network_id=network_id,
device_id=device_id)
port_ips: tobiko.Selection[netaddr.IPAddress] = tobiko.Selection()
for port in ports:
self.assertEqual(network_id, port['network_id'])
self.assertEqual(device_id, port['device_id'])
port_ips.extend(neutron.list_port_ip_addresses(port=port))
if ip_version is not None:
port_ips = port_ips.with_attributes(version=ip_version)
def test_ping_port(self, network_id=None, device_id=None, ip_version=None):
"""Checks server can ping its own port"""
device_ips = neutron.list_device_ip_addresses(
device_id=device_id or self.stack.server_id,
network_id=network_id or self.stack.network_stack.network_id,
enable_dhcp=True, ip_version=ip_version)
server_ips = ip.list_ip_addresses(scope='global',
ssh_client=self.stack.ssh_client)
# Remove IPs that hasn't been assigned to server
port_ips = tobiko.Selection(set(port_ips) & set(server_ips))
port_ips = tobiko.Selection(set(device_ips) & set(server_ips))
if port_ips:
ping.assert_reachable_hosts(port_ips,
ssh_client=self.stack.ssh_client)
@ -100,13 +96,6 @@ class PortTest(testtools.TestCase):
else:
self.skipTest("No port IP addresses found")
@tobiko.retry_test_case(interval=30.)
def test_ping_inner_gateway_ip(self, ip_version=None):
if not self.stack.network_stack.has_gateway:
self.skip('Server network has no gateway router')
self.test_ping_port(device_id=self.stack.network_stack.gateway_id,
ip_version=ip_version)
# --- Test opening ports on external network ----------------------------------