From 0cc9d75f32c38b241c304ca47a9407e4129379be Mon Sep 17 00:00:00 2001 From: Federico Ressi Date: Fri, 27 Nov 2020 15:10:27 +0100 Subject: [PATCH] Fix/refactor PortTest Change-Id: I56fd8959f27aea74c175891ec0a049b0adc60b29 --- tobiko/openstack/neutron/__init__.py | 2 + tobiko/openstack/neutron/_client.py | 8 ++- tobiko/openstack/neutron/_port.py | 73 +++++++++++++++++----- tobiko/tests/scenario/neutron/test_port.py | 63 ++++++++----------- 4 files changed, 93 insertions(+), 53 deletions(-) diff --git a/tobiko/openstack/neutron/__init__.py b/tobiko/openstack/neutron/__init__.py index b05036d3e..9c3a64570 100644 --- a/tobiko/openstack/neutron/__init__.py +++ b/tobiko/openstack/neutron/__init__.py @@ -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 diff --git a/tobiko/openstack/neutron/_client.py b/tobiko/openstack/neutron/_client.py index ab2132136..b9ebb71c3 100644 --- a/tobiko/openstack/neutron/_client.py +++ b/tobiko/openstack/neutron/_client.py @@ -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'] diff --git a/tobiko/openstack/neutron/_port.py b/tobiko/openstack/neutron/_port.py index 5d068ba7b..45d5624fb 100644 --- a/tobiko/openstack/neutron/_port.py +++ b/tobiko/openstack/neutron/_port.py @@ -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 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, unique=False, **kwargs): +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 diff --git a/tobiko/tests/scenario/neutron/test_port.py b/tobiko/tests/scenario/neutron/test_port.py index 04bb37288..3a744c087 100644 --- a/tobiko/tests/scenario/neutron/test_port.py +++ b/tobiko/tests/scenario/neutron/test_port.py @@ -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 ----------------------------------