diff --git a/tobiko/openstack/stacks/__init__.py b/tobiko/openstack/stacks/__init__.py index 9b0081b92..bd4a47984 100644 --- a/tobiko/openstack/stacks/__init__.py +++ b/tobiko/openstack/stacks/__init__.py @@ -52,9 +52,12 @@ NetworkWithNetMtuWriteStackFixture = ( _neutron.NetworkWithNetMtuWriteStackFixture) SecurityGroupsFixture = _neutron.SecurityGroupsFixture +get_external_network = _neutron.get_external_network +has_external_network = _neutron.has_external_network +skip_unless_has_external_network = _neutron.skip_unless_has_external_network get_floating_network = _neutron.get_floating_network has_floating_network = _neutron.has_floating_network -skip_if_missing_floating_network = _neutron.skip_if_missing_floating_network +skip_unless_has_floating_network = _neutron.skip_unless_has_floating_network ServerStackFixture = _nova.ServerStackFixture KeyPairStackFixture = _nova.KeyPairStackFixture diff --git a/tobiko/openstack/stacks/_neutron.py b/tobiko/openstack/stacks/_neutron.py index 2cb6e058c..d65645b3d 100644 --- a/tobiko/openstack/stacks/_neutron.py +++ b/tobiko/openstack/stacks/_neutron.py @@ -16,6 +16,7 @@ from __future__ import absolute_import import json +import typing import netaddr from oslo_log import log @@ -33,6 +34,8 @@ LOG = log.getLogger(__name__) CONF = config.CONF LOG = log.getLogger(__name__) +NeutronNetworkType = typing.Dict[str, typing.Any] + class ExternalNetworkStackFixture(heat.HeatStackFixture): @@ -42,15 +45,40 @@ class ExternalNetworkStackFixture(heat.HeatStackFixture): def external_name(self): return tobiko.tobiko_config().neutron.external_network - _external_network = None + subnet_enable_dhcp: typing.Optional[bool] = True + + _external_network: typing.Optional[NeutronNetworkType] = None @property - def external_network(self): - network = self._external_network - if network is None: - self._external_network = network = find_external_network( - name=self.external_name) or {} - return network + def external_network(self) -> typing.Optional[NeutronNetworkType]: + external_network = self._external_network + if external_network is None: + subnet_parameters = {} + if self.subnet_enable_dhcp is not None: + subnet_parameters['enable_dhcp'] = self.subnet_enable_dhcp + for network in list_external_networks(name=self.external_name): + if not network['subnets']: + LOG.debug(f"Network '{network['id']}' has any subnet") + continue + if subnet_parameters: + subnets = neutron.list_subnets(network_id=network['id'], + **subnet_parameters) + if not subnets: + LOG.debug(f"Network '{network['id']}' has any valid " + f"subnet: {subnet_parameters}") + continue + + network_dump = json.dumps(network, indent=4, sort_keys=True) + LOG.debug(f"Found external network for {self.fixture_name}:\n" + f"{network_dump}") + self._external_network = external_network = network + break + else: + LOG.warning("No external network found for " + f"'{self.fixture_name}':\n" + f" - name or ID: {self.external_name}\n" + f" - subnet attributes: {subnet_parameters}\n") + return external_network @property def external_id(self): @@ -74,6 +102,8 @@ class FloatingNetworkStackFixture(ExternalNetworkStackFixture): def external_name(self): return tobiko.tobiko_config().neutron.floating_network + subnet_enable_dhcp = None + @neutron.skip_if_missing_networking_extensions('port-security') class NetworkStackFixture(heat.HeatStackFixture): @@ -270,32 +300,41 @@ class SecurityGroupsFixture(heat.HeatStackFixture): template = _hot.heat_template_file('neutron/security_groups.yaml') -def find_external_network(name=None): - network = None - if name: +def list_external_networks(name: typing.Optional[str] = None) -> \ + tobiko.Selection[NeutronNetworkType]: + networks = tobiko.Selection[NeutronNetworkType]() + if name is not None: try: network = neutron.get_network(name) except neutron.NoSuchNetwork: - LOG.debug('No such network with ID %r', name) + LOG.error(f"No such network with ID '{name}'") + else: + networks.append(network) + if not networks: + network_params = {'router:external': True, "status": "ACTIVE"} + if name is not None: + network_params['name'] = name + networks += neutron.list_networks(**network_params) + if not networks and name: + raise ValueError("No such external network with name or ID " + f"'{name}'") + return networks - if not network: - params = {'router:external': True, "status": "ACTIVE"} - if name: - params['name'] = name - try: - network = neutron.find_network(**params) - except tobiko.ObjectNotFound as ex: - LOG.exception('No such network (%s):', - json.dumps(params, sort_keys=True)) - if name: - raise ValueError("No such external network with name or ID " - f"'{name}'") from ex - if network: - LOG.debug('Found external network %r:\n%s', - network['name'], json.dumps(network, indent=4, - sort_keys=True)) - return network +def get_external_network_id(): + return tobiko.setup_fixture(ExternalNetworkStackFixture).network_id + + +def get_external_network(): + return tobiko.setup_fixture(ExternalNetworkStackFixture).network_details + + +def has_external_network(): + return tobiko.setup_fixture(ExternalNetworkStackFixture).has_network + + +skip_unless_has_external_network = tobiko.skip_unless( + 'External network not found', has_external_network) def get_floating_network_id(): @@ -310,5 +349,5 @@ def has_floating_network(): return tobiko.setup_fixture(FloatingNetworkStackFixture).has_network -skip_if_missing_floating_network = tobiko.skip_unless( +skip_unless_has_floating_network = tobiko.skip_unless( 'Floating network not found', has_floating_network) diff --git a/tobiko/tests/functional/openstack/stacks/test_neutron.py b/tobiko/tests/functional/openstack/stacks/test_neutron.py index a4753cb42..d732130d2 100644 --- a/tobiko/tests/functional/openstack/stacks/test_neutron.py +++ b/tobiko/tests/functional/openstack/stacks/test_neutron.py @@ -102,15 +102,31 @@ class L3HaNetworkTest(NetworkTest): @keystone.skip_unless_has_keystone_credentials() +@stacks.skip_unless_has_external_network +class ExternalNetworkStackTest(testtools.TestCase): + + def test_get_external_network(self): + network = stacks.get_external_network() + self.assertTrue(network['id']) + self.assertIs(True, network['router:external']) + self.assertEqual('ACTIVE', network['status']) + subnets = neutron.list_subnets(network_id=network['id'], + enable_dhcp=True) + self.assertNotEqual([], subnets) + + def test_has_external_network(self): + self.assertIs(True, stacks.has_external_network()) + + +@keystone.skip_unless_has_keystone_credentials() +@stacks.skip_unless_has_floating_network class FloatingNetworkStackTest(testtools.TestCase): - @stacks.skip_if_missing_floating_network def test_get_floating_network(self): network = stacks.get_floating_network() self.assertTrue(network['id']) self.assertIs(True, network['router:external']) self.assertEqual('ACTIVE', network['status']) - @stacks.skip_if_missing_floating_network def test_has_floating_network(self): self.assertIs(True, stacks.has_floating_network()) diff --git a/tobiko/tests/scenario/neutron/test_port.py b/tobiko/tests/scenario/neutron/test_port.py index 77a5b45ce..d5dfd992b 100644 --- a/tobiko/tests/scenario/neutron/test_port.py +++ b/tobiko/tests/scenario/neutron/test_port.py @@ -110,6 +110,7 @@ class PortTest(testtools.TestCase): # --- Test opening ports on external network ---------------------------------- +@stacks.skip_unless_has_external_network class ExternalPortTest(PortTest): """Test Neutron ports"""