x/tobiko
Code Issues Proposed changes

Use Subnet pools to avoid cidr concurrency issues

Browse Source 
This patch removes the tobiko.openstack.neutron._cidr module that was
used to generate CIDRs whenever tobiko was going to create a new subnet.
That method had problems when several tobiko workers running in parallel
tried to create subnets. Collisions happened sometimes, resulting in the
creation of two subnets with a common CIDR that could fail if they were
connected to a common router.

This patch uses the neutron subnet pool feature [1]. When Tobiko creates
a subnet, it will not calculate a free IPv4/IPv6 CIDR, but will
associate that subnet to common Subnet Pools, shared by the different
tests. When the subnet is created, it is neutron responsibiltiy to
allocate a free CIDR for that subnet, based on the created Subnet Pools.

The method that creates the Subnet Pools uses an
@oslo_concurrency.lockutils.synchronized decorator to guarantee that two
Tobiko workers do not create the same Subnet Pool at the same time.

[1] https://docs.openstack.org/neutron/latest/admin/config-subnet-pools.html

Change-Id: I3b09e468310e06f8c4b7d7dbec02e4b2e6f67530
This commit is contained in:
Eduardo Olivares 2023-04-12 12:01:26 +02:00
parent 2ad24dd2f0
commit a6e7a522df
15 changed files with 295 additions and 190 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
lower-constraints.txt
View File

@ -10,6 +10,7 @@ mock==3.0.5
netaddr==0.8.0
neutron-lib==2.7.0
openstacksdk==0.31.2
oslo.concurrency==3.26.0
oslo.config==8.4.0
oslo.log==4.4.0
packaging==20.4

1
requirements.txt
View File

@ -10,6 +10,7 @@ metalsmith>=1.6.2 # Apache-2.0
netaddr>=0.8.0 # BSD
neutron-lib>=2.7.0 # Apache-2.0
openstacksdk>=0.31.2 # Apache-2.0
oslo.concurrency>=3.26.0 # Apache-2.0
oslo.config>=8.4.0 # Apache-2.0
oslo.log>=4.4.0 # Apache-2.0
packaging>=20.4 # Apache-2.0

8
roles/tobiko-cleanup/tasks/main.yaml
View File

@ -19,6 +19,14 @@
until: result.rc == 0
ignore_errors: yes
- name: "cleanup subnet pools created by Tobiko tests"
shell: |
source {{ stackrc_file }}
openstack subnet pool list -f value -c 'Name' | \
grep "^tobiko\." | \
xargs -r openstack subnet pool delete
ignore_errors: yes
- name: "cleanup Glance images created by Tobiko tests"
shell: |
source {{ stackrc_file }}

5
tobiko/config.py
View File

@ -86,7 +86,10 @@ COMMON_GROUP_NAME = 'common'
COMMON_OPTIONS = [
cfg.StrOpt('shelves_dir',
default='~/.tobiko/cache/shelves',
help=("Default directory where to look for shelves.")),
help="Directory where to look for shelves."),
cfg.StrOpt('lock_dir',
default='~/.tobiko/cache/lock',
help="Directory where lock persistent files will be saved"),
]

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

@ -15,7 +15,6 @@ from __future__ import absolute_import
from tobiko.openstack.neutron import _agent
from tobiko.openstack.neutron import _client
from tobiko.openstack.neutron import _cidr
from tobiko.openstack.neutron import _extension
from tobiko.openstack.neutron import _floating_ip
from tobiko.openstack.neutron import _port
@ -24,6 +23,7 @@ from tobiko.openstack.neutron import _network
from tobiko.openstack.neutron import _router
from tobiko.openstack.neutron import _security_group
from tobiko.openstack.neutron import _subnet
from tobiko.openstack.neutron import _subnet_pool
SERVER = 'neutron-server'
@ -65,10 +65,6 @@ NeutronClientType = _client.NeutronClientType
neutron_client = _client.neutron_client
get_neutron_client = _client.get_neutron_client
new_ipv4_cidr = _cidr.new_ipv4_cidr
new_ipv6_cidr = _cidr.new_ipv6_cidr
list_subnet_cidrs = _cidr.list_subnet_cidrs
get_networking_extensions = _extension.get_networking_extensions
missing_networking_extensions = _extension.missing_networking_extensions
has_networking_extensions = _extension.has_networking_extensions
@ -140,6 +136,14 @@ SubnetType = _subnet.SubnetType
SubnetIdType = _subnet.SubnetIdType
NoSuchSubnet = _subnet.NoSuchSubnet
SubnetPoolType = _subnet_pool.SubnetPoolType
SubnetPoolIdType = _subnet_pool.SubnetPoolIdType
NoSuchSubnetPool = _subnet_pool.NoSuchSubnetPool
get_subnet_pool = _subnet_pool.get_subnet_pool
create_subnet_pool = _subnet_pool.create_subnet_pool
delete_subnet_pool = _subnet_pool.delete_subnet_pool
find_subnet_pool = _subnet_pool.find_subnet_pool
list_security_groups = _security_group.list_security_groups
get_security_group = _security_group.get_security_group
get_default_security_group = _security_group.get_default_security_group

142
tobiko/openstack/neutron/_cidr.py
View File

@ -1,142 +0,0 @@
# Copyright 2019 Red Hat
#
# 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 os
import random
import netaddr
from netaddr.strategy import ipv4
from netaddr.strategy import ipv6
import tobiko
from tobiko.openstack.neutron import _client
def new_ipv4_cidr(seed=None):
return tobiko.setup_fixture(IPv4CIDRGeneratorFixture).new_cidr(seed=seed)
def new_ipv6_cidr(seed=None):
return tobiko.setup_fixture(IPv6CIDRGeneratorFixture).new_cidr(seed=seed)
class CIDRGeneratorFixture(tobiko.SharedFixture):
cidr = None
prefixlen = None
client = None
config = None
cidr_generator = None
def __init__(self, cidr=None, prefixlen=None, client=None):
super(CIDRGeneratorFixture, self).__init__()
if cidr:
self.cidr = cidr
if prefixlen:
self.prefixlen = prefixlen
if client:
self.client = client
def setup_fixture(self):
self.setup_config()
self.setup_client()
def setup_config(self):
from tobiko import config
CONF = config.CONF
self.config = CONF.tobiko.neutron
def setup_client(self):
self.client = _client.neutron_client(self.client)
def new_cidr(self, seed):
used_cidrs = set(list_subnet_cidrs(client=self.client))
for cidr in random_subnets(cidr=self.cidr, prefixlen=self.prefixlen,
seed=seed):
if cidr not in used_cidrs:
return cidr
raise NoSuchCIDRLeft(cidr=self.cidr, prefixlen=self.prefixlen)
class IPv4CIDRGeneratorFixture(CIDRGeneratorFixture):
@property
def cidr(self):
return netaddr.IPNetwork(self.config.ipv4_cidr)
@property
def prefixlen(self):
return int(self.config.ipv4_prefixlen)
class IPv6CIDRGeneratorFixture(CIDRGeneratorFixture):
@property
def cidr(self):
return netaddr.IPNetwork(self.config.ipv6_cidr)
@property
def prefixlen(self):
return int(self.config.ipv6_prefixlen)
class NoSuchCIDRLeft(tobiko.TobikoException):
message = ("No such subnet CIDR left "
"(CIDR={cidr!s}, prefixlen={prefixlen!s})")
def random_subnets(cidr, prefixlen, seed=None):
"""
A generator that divides up this IPNetwork's subnet into smaller
subnets based on a specified CIDR prefix.
:param prefixlen: a CIDR prefix indicating size of subnets to be
returned.
:return: an iterator containing random IPNetwork subnet objects.
"""
version = cidr.version
module = {4: ipv4, 6: ipv6}[version]
width = module.width
if not 0 <= cidr.prefixlen <= width:
message = "CIDR prefix /{!r} invalid for IPv{!s}!".format(
prefixlen, cidr.version)
raise ValueError(message)
if not cidr.prefixlen <= prefixlen:
# Don't return anything.
raise StopIteration
# Calculate number of subnets to be returned.
max_subnets = 2 ** (width - cidr.prefixlen) // 2 ** (width - prefixlen)
base_subnet = module.int_to_str(cidr.first)
i = 0
rand = random.Random(hash(seed) ^ os.getpid())
while True:
subnet = netaddr.IPNetwork('%s/%d' % (base_subnet, prefixlen), version)
subnet.value += (subnet.size * rand.randrange(0, max_subnets))
subnet.prefixlen = prefixlen
i += 1
yield subnet
def list_subnet_cidrs(client: _client.NeutronClientType = None,
**params) -> tobiko.Selection[netaddr.IPNetwork]:
from tobiko.openstack.neutron import _subnet
subnets = _subnet.list_subnets(client=client, **params)
return tobiko.select(netaddr.IPNetwork(subnet['cidr'])
for subnet in subnets)

10
tobiko/openstack/neutron/_subnet.py
View File

@ -21,7 +21,6 @@ import netaddr
import tobiko
from tobiko.openstack.neutron import _client
from tobiko.openstack.neutron import _network
from tobiko.openstack.neutron import _cidr
SubnetType = typing.Dict[str, typing.Any]
@ -50,7 +49,6 @@ def create_subnet(client: _client.NeutronClientType = None,
network: _network.NetworkIdType = None,
add_cleanup=True,
ip_version=4,
cidr: str = None,
**params) -> SubnetType:
if 'network_id' not in params:
if network is None:
@ -61,14 +59,6 @@ def create_subnet(client: _client.NeutronClientType = None,
network_id = _network.get_network_id(network)
params['network_id'] = network_id
params['ip_version'] = ip_version
if cidr is None:
if ip_version == 4:
cidr = _cidr.new_ipv4_cidr()
elif ip_version == 6:
cidr = _cidr.new_ipv6_cidr()
else:
raise ValueError(f'Invalid ip version: {ip_version}')
params['cidr'] = cidr
subnet = _client.neutron_client(client).create_subnet(
body={'subnet': params})['subnet']
if add_cleanup:

96
tobiko/openstack/neutron/_subnet_pool.py Normal file
View File

@ -0,0 +1,96 @@
# Copyright 2023 Red Hat
#
# 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
from collections import abc
import typing
import tobiko
from tobiko.openstack.neutron import _client
SubnetPoolType = typing.Dict[str, typing.Any]
SubnetPoolIdType = typing.Union[str, SubnetPoolType]
def get_subnet_pool_id(subnet_pool: SubnetPoolIdType) -> str:
if isinstance(subnet_pool, str):
return subnet_pool
else:
return subnet_pool['id']
def get_subnet_pool(subnet_pool: SubnetPoolIdType,
client: _client.NeutronClientType = None,
**params) -> SubnetPoolType:
subnet_pool_id = get_subnet_pool_id(subnet_pool)
try:
return _client.neutron_client(client).show_subnetpool(
subnet_pool_id, **params)['subnetpool']
except _client.NotFound as ex:
raise NoSuchSubnetPool from ex
def create_subnet_pool(client: _client.NeutronClientType = None,
add_cleanup: bool = True,
**params) -> SubnetPoolType:
subnet_pool = _client.neutron_client(client).create_subnetpool(
body={'subnetpool': params})['subnetpool']
if add_cleanup:
tobiko.add_cleanup(
cleanup_subnet_pool, subnet_pool=subnet_pool, client=client)
return subnet_pool
def cleanup_subnet_pool(subnet_pool: SubnetPoolIdType,
client: _client.NeutronClientType = None):
try:
delete_subnet_pool(subnet_pool=subnet_pool, client=client)
except NoSuchSubnetPool:
pass
def delete_subnet_pool(subnet_pool: SubnetPoolIdType,
client: _client.NeutronClientType = None):
subnet_pool_id = get_subnet_pool_id(subnet_pool)
try:
_client.neutron_client(client).delete_subnetpool(subnet_pool_id)
except _client.NotFound as ex:
raise NoSuchSubnetPool from ex
def list_subnet_pools(client: _client.NeutronClientType = None,
**params) -> tobiko.Selection[SubnetPoolType]:
subnet_pools = _client.neutron_client(client).list_subnetpools(**params)
if isinstance(subnet_pools, abc.Mapping):
subnet_pools = subnet_pools['subnetpools']
return tobiko.select(subnet_pools)
def find_subnet_pool(client: _client.NeutronClientType = None,
unique=False,
**params) -> SubnetPoolType:
"""Look for a subnet pool matching some values"""
subnet_pools = list_subnet_pools(client=client, **params)
if subnet_pools:
if unique:
return subnet_pools.unique
else:
return subnet_pools.first
else:
raise NoSuchSubnetPool
class NoSuchSubnetPool(tobiko.ObjectNotFound):
message = "No such subnet pool found"

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

@ -33,7 +33,7 @@ OPTIONS = [
default=None,
help="List of nameservers IPv4 addresses"),
cfg.StrOpt('ipv6_cidr',
default='2001:db8::/48',
default='fc00::/48',
help="The CIDR block to allocate IPv6 subnets from"),
cfg.IntOpt('ipv6_prefixlen',
default=64,

124
tobiko/openstack/stacks/_neutron.py
View File

@ -16,9 +16,11 @@
from __future__ import absolute_import
import json
import os
import typing
import netaddr
from oslo_concurrency import lockutils
from oslo_log import log
import tobiko
@ -33,6 +35,7 @@ from tobiko.shell import ssh
CONF = config.CONF
LOG = log.getLogger(__name__)
LOCK_DIR = os.path.expanduser(CONF.tobiko.common.lock_dir)
class ExternalNetworkStackFixture(heat.HeatStackFixture):
@ -272,9 +275,112 @@ def ensure_router_interface(
add_cleanup=add_cleanup)
@neutron.skip_if_missing_networking_extensions('subnet_allocation')
class SubnetPoolFixture(tobiko.SharedFixture):
"""Neutron Subnet Pool Fixture.
A subnet pool is a dependency of network fixtures with either IPv4 or
IPv6 subnets (or both). The CIDRs for those subnets are obtained from this
resource.
NOTE: this fixture does not represent a heat stack, but it is under the
stacks module until a decision is taken on where this kind of fixtures are
located
"""
name: typing.Optional[str] = None
prefixes: list = [CONF.tobiko.neutron.ipv4_cidr]
default_prefixlen: int = CONF.tobiko.neutron.ipv4_prefixlen
_subnet_pool: typing.Optional[neutron.SubnetPoolType] = None
def __init__(self,
name: typing.Optional[str] = None,
prefixes: typing.Optional[list] = None,
default_prefixlen: typing.Optional[int] = None):
self.name = name or self.fixture_name
if prefixes:
self.prefixes = prefixes
if default_prefixlen:
self.default_prefixlen = default_prefixlen
super().__init__()
@property
def ip_version(self):
valid_versions = (4, 6)
for valid_version in valid_versions:
if len(self.prefixes) > 0 and all(
netaddr.IPNetwork(prefix).version == valid_version
for prefix in self.prefixes):
return valid_version
# return None when neither IPv4 nor IPv6 (or when both)
return None
def setup_fixture(self):
if config.get_bool_env('TOBIKO_PREVENT_CREATE'):
LOG.debug("SubnetPoolFixture should have been already created: %r",
self.subnet_pool)
else:
self.try_create_subnet_pool()
if self.subnet_pool:
tobiko.addme_to_shared_resource(__name__, self.name)
@lockutils.synchronized(
'create_subnet_pool', external=True, lock_path=LOCK_DIR)
def try_create_subnet_pool(self):
if not self.subnet_pool:
self._subnet_pool = neutron.create_subnet_pool(
name=self.name, prefixes=self.prefixes,
default_prefixlen=self.default_prefixlen, add_cleanup=False)
def cleanup_fixture(self):
n_tests_using_resource = len(tobiko.removeme_from_shared_resource(
__name__, self.name))
if n_tests_using_resource == 0:
self._cleanup_subnet_pool()
else:
LOG.info('Subnet Pool %r not deleted because %d tests '
'are using it still.',
self.name, n_tests_using_resource)
def _cleanup_subnet_pool(self):
sp_id = self.subnet_pool_id
if sp_id:
self._subnet_pool = None
LOG.debug('Deleting Subnet Pool %r (%r)...',
self.name, sp_id)
neutron.delete_subnet_pool(sp_id)
LOG.debug('Subnet Pool %r (%r) deleted.', self.name, sp_id)
@property
def subnet_pool_id(self):
if self.subnet_pool:
return self._subnet_pool['id']
@property
def subnet_pool(self):
if not self._subnet_pool:
try:
self._subnet_pool = neutron.find_subnet_pool(name=self.name)
except neutron.NoSuchSubnetPool:
LOG.debug("Subnet Pool %r not found.", self.name)
self._subnet_pool = None
return self._subnet_pool
class SubnetPoolIPv6Fixture(SubnetPoolFixture):
prefixes: list = [CONF.tobiko.neutron.ipv6_cidr]
default_prefixlen: int = CONF.tobiko.neutron.ipv6_prefixlen
@neutron.skip_if_missing_networking_extensions('port-security')
class NetworkStackFixture(heat.HeatStackFixture):
"""Heat stack for creating internal network with a router to external"""
subnet_pools_ipv4_stack = (tobiko.required_fixture(SubnetPoolFixture)
if bool(CONF.tobiko.neutron.ipv4_cidr)
else None)
subnet_pools_ipv6_stack = (tobiko.required_fixture(SubnetPoolIPv6Fixture)
if bool(CONF.tobiko.neutron.ipv6_cidr)
else None)
#: Heat template file
template = _hot.heat_template_file('neutron/network.yaml')
@ -287,24 +393,18 @@ class NetworkStackFixture(heat.HeatStackFixture):
"""Whenever to setup IPv4 subnet"""
return bool(CONF.tobiko.neutron.ipv4_cidr)
@property
def ipv4_cidr(self):
if self.has_ipv4:
return neutron.new_ipv4_cidr(seed=self.fixture_name)
else:
return None
@property
def has_ipv6(self):
"""Whenever to setup IPv6 subnet"""
return bool(CONF.tobiko.neutron.ipv6_cidr)
@property
def ipv6_cidr(self):
if self.has_ipv6:
return neutron.new_ipv6_cidr(seed=self.fixture_name)
else:
return None
def subnet_pool_ipv4_id(self):
return self.subnet_pools_ipv4_stack.subnet_pool_id
@property
def subnet_pool_ipv6_id(self):
return self.subnet_pools_ipv6_stack.subnet_pool_id
@property
def network_value_specs(self):

16
tobiko/openstack/stacks/neutron/network.yaml
View File

@ -27,17 +27,13 @@ parameters:
type: boolean
default: false
ipv4_cidr:
description: IPv4 subnet CIDR to be assigned to new network
subnet_pool_ipv4_id:
description: IPv4 Subnet Pool ID
type: string
constraints:
- custom_constraint: net_cidr
ipv6_cidr:
description: IPv6 subnet CIDR to be assigned to new network
subnet_pool_ipv6_id:
description: IPv6 Subnet Pool ID
type: string
constraints:
- custom_constraint: net_cidr
ipv4_dns_nameservers:
description: IPv4 nameservers IP addresses
@ -125,7 +121,7 @@ resources:
properties:
network: {get_resource: _network}
ip_version: 4
cidr: {get_param: ipv4_cidr}
subnetpool: {get_param: subnet_pool_ipv4_id}
dns_nameservers: {get_param: ipv4_dns_nameservers}
_ipv6_subnet:
@ -134,7 +130,7 @@ resources:
properties:
network: {get_resource: _network}
ip_version: 6
cidr: {get_param: ipv6_cidr}
subnetpool: {get_param: subnet_pool_ipv6_id}
dns_nameservers: {get_param: ipv6_dns_nameservers}
ipv6_address_mode: {get_param: ipv6_address_mode}
ipv6_ra_mode: {get_param: ipv6_ra_mode}

2
tobiko/tests/faults/neutron/test_agents.py
View File

@ -529,7 +529,7 @@ class L3AgentTest(BaseAgentTest):
'ipv6_ra_mode')
ipv6_address_mode = self.stack.network_stack.ipv6_subnet_details.get(
'ipv6_address_mode')
if not self.stack.network_stack.ipv6_cidr:
if not self.stack.network_stack.ipv6_subnet_details.get('cidr'):
return False
if (ipv6_ra_mode not in stateless_modes or
ipv6_address_mode not in stateless_modes):

8
tobiko/tests/functional/openstack/neutron/test_router.py
View File

@ -95,9 +95,15 @@ class RouterTest(testtools.TestCase):
def test_add_router_interface_with_subnet(self):
network = neutron.create_network(name=self.id())
subnet_pool_range = "172.168.111.0/24"
subnet_pool_default_prefixlen = 26
subnet_pool = neutron.create_subnet_pool(
name=self.id(),
prefixes=[subnet_pool_range],
default_prefixlen=subnet_pool_default_prefixlen)
subnet = neutron.create_subnet(name=self.id(),
network=network,
cidr=neutron.new_ipv4_cidr(),
subnetpool_id=subnet_pool['id'],
ip_version=4)
router = self.test_create_router()
interface = neutron.add_router_interface(router=router, subnet=subnet)

31
tobiko/tests/functional/openstack/neutron/test_subnet.py
View File

@ -40,13 +40,26 @@ class SubnetTest(testtools.TestCase):
self.assertIn(self.stack.ipv6_subnet_id, subnets_ids)
def test_list_subnet_cidrs(self):
subnets_cidrs = neutron.list_subnet_cidrs()
def _find_prefix(cidr, prefixes):
for prefix in prefixes:
if cidr in prefix:
return prefix
return None
if self.stack.has_ipv4:
cidr = netaddr.IPNetwork(self.stack.ipv4_subnet_details['cidr'])
self.assertIn(cidr, subnets_cidrs)
prefixes = [
netaddr.IPNetwork(prefix)
for prefix in
self.stack.subnet_pools_ipv4_stack.subnet_pool['prefixes']]
self.assertIsNotNone(_find_prefix(cidr, prefixes))
if self.stack.has_ipv6:
cidr = netaddr.IPNetwork(self.stack.ipv6_subnet_details['cidr'])
self.assertIn(cidr, subnets_cidrs)
prefixes = [
netaddr.IPNetwork(prefix)
for prefix in
self.stack.subnet_pools_ipv6_stack.subnet_pool['prefixes']]
self.assertIsNotNone(_find_prefix(cidr, prefixes))
def test_get_ipv4_subnet(self):
if not self.stack.has_ipv4:
@ -66,12 +79,18 @@ class SubnetTest(testtools.TestCase):
def test_create_subnet(self):
network = neutron.create_network(name=self.id())
cidr = neutron.new_ipv4_cidr()
subnet_pool_range = "192.168.0.0/16"
subnet_pool_default_prefixlen = 24
subnet_pool = neutron.create_subnet_pool(
name=self.id(),
prefixes=[subnet_pool_range],
default_prefixlen=subnet_pool_default_prefixlen)
subnet = neutron.create_subnet(network=network,
ip_version=4,
cidr=cidr)
subnetpool_id=subnet_pool['id'])
self.assertEqual(network['id'], subnet['network_id'])
self.assertEqual(str(cidr), subnet['cidr'])
self.assertIn(netaddr.IPNetwork(subnet['cidr']),
netaddr.IPNetwork(subnet_pool_range))
self.assertEqual(subnet['id'], neutron.get_subnet(subnet=subnet)['id'])
return subnet

25
tobiko/tests/functional/openstack/stacks/test_neutron.py
View File

@ -163,6 +163,8 @@ class RouterInterfaceTest(testtools.TestCase):
required_fixtures = [router_stack, network_stack]
subnet_index = 201
@classmethod
def tearDownClass(cls) -> None:
for fixture in cls.required_fixtures:
@ -171,10 +173,30 @@ class RouterInterfaceTest(testtools.TestCase):
except Exception:
LOG.exception(f'Error cleaning up fixture: {fixture.fixture}')
def _get_subnet_pool(self, ip_version=4):
if ip_version == 4:
subnet_pool_range = f"172.168.{self.subnet_index}.0/24"
subnet_pool_default_prefixlen = 26
elif ip_version == 6:
subnet_pool_range = f"2003:{self.subnet_index}::/64"
subnet_pool_default_prefixlen = 68
else:
raise ValueError
subnet_pool = neutron.create_subnet_pool(
name=self.id(),
prefixes=[subnet_pool_range],
default_prefixlen=subnet_pool_default_prefixlen)
self.subnet_index += 1
return subnet_pool
def test_ensure_router_interface_with_subnet(self,
ip_version=4):
network = neutron.create_network()
subnet_pool = self._get_subnet_pool(ip_version)
subnet = neutron.create_subnet(network=network,
subnetpool_id=subnet_pool['id'],
ip_version=ip_version)
self._test_ensure_router(subnet=subnet)
@ -189,7 +211,8 @@ class RouterInterfaceTest(testtools.TestCase):
def test_ensure_router_interface_with_network(self):
network = neutron.create_network()
neutron.create_subnet(network=network)
subnet_pool = self._get_subnet_pool()
neutron.create_subnet(network=network, subnetpool_id=subnet_pool['id'])
self._test_ensure_router(network=network)
def test_ensure_router_interface_with_routed_network(self):