Update Neutron Client

- implement CIDR management
- workaround neutron client bug listing subnets

Change-Id: I4619999de1d5c459af4338f93a7bc7bea9406b8b
Federico Ressi 4 years ago
parent 8a7b39d050
commit d9e9bcab86

@ -14,6 +14,7 @@
from __future__ import absolute_import
from tobiko.openstack.neutron import _client
from tobiko.openstack.neutron import _cidr
from tobiko.openstack.neutron import _extension
@ -29,6 +30,9 @@ show_network = _client.show_network
show_router = _client.show_router
show_subnet = _client.show_subnet
new_ipv4_cidr = _cidr.new_ipv4_cidr
new_ipv6_cidr = _cidr.new_ipv6_cidr
get_networking_extensions = _extension.get_networking_extensions
missing_networking_extensions = _extension.missing_networking_extensions
has_networking_extensions = _extension.has_networking_extensions

@ -0,0 +1,99 @@
# 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 netaddr
import tobiko
from tobiko.openstack.neutron import _client
def new_ipv4_cidr():
return tobiko.setup_fixture(IPv4CIDRGeneratorFixture).new_cidr()
def new_ipv6_cidr():
return tobiko.setup_fixture(IPv6CIDRGeneratorFixture).new_cidr()
class CIDRGeneratorFixture(tobiko.SharedFixture):
cidr = None
prefixlen = None
client = None
config = None
cidr_generator = None
subnet_cidrs = 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):
def setup_config(self):
from tobiko import config
CONF = config.CONF
self.config = CONF.tobiko.neutron
def setup_subnet_cidrs(self):
client = _client.neutron_client(self.client)
self.subnet_cidrs = set(_client.list_subnet_cidrs(client=client))
def setup_cidr_generator(self):
cidr = netaddr.IPNetwork(self.cidr)
prefixlen = int(self.prefixlen)
self.cidr_generator = cidr.subnet(prefixlen)
def new_cidr(self):
for cidr in self.cidr_generator:
if cidr not in self.subnet_cidrs:
return cidr
raise NoSuchCIDRLeft(cidr=self.cidr, prefixlen=self.prefixlen)
class IPv4CIDRGeneratorFixture(CIDRGeneratorFixture):
def cidr(self):
return self.config.ipv4_cidr
def prefixlen(self):
return self.config.ipv4_prefixlen
class IPv6CIDRGeneratorFixture(CIDRGeneratorFixture):
def cidr(self):
return self.config.ipv6_cidr
def prefixlen(self):
return self.config.ipv6_prefixlen
class NoSuchCIDRLeft(tobiko.TobikoException):
message = ("No such subnet CIDR left "
"(CIDR={cidr!s}, prefixlen={prefixlen!s})")

@ -13,6 +13,8 @@
# under the License.
from __future__ import absolute_import
import collections
import netaddr
from neutronclient.v2_0 import client as neutronclient
@ -82,7 +84,9 @@ def list_networks(show=False, client=None, **params):
def list_subnets(show=False, client=None, **params):
subnets = neutron_client(client).list_subnets(**params)['subnets']
subnets = neutron_client(client).list_subnets(**params)
if isinstance(subnets, collections.Mapping):
subnets = subnets['subnets']
if show:
subnets = [show_subnet(s['id'], client=client) for s in subnets]
return subnets

@ -94,7 +94,7 @@ class NeutronApiTestCase(testtools.TestCase):
subnet = neutron.show_subnet(self.stack.ipv4_subnet_id)
self.assertEqual(self.stack.ipv4_subnet_id, subnet['id'])
self.assertEqual(str(self.stack.ipv4_cidr), subnet['cidr'])
self.assertEqual(self.stack.ipv4_subnet_details, subnet)
def test_show_ipv6_subnet(self):
if not self.stack.has_ipv6:
@ -102,4 +102,4 @@ class NeutronApiTestCase(testtools.TestCase):
subnet = neutron.show_subnet(self.stack.ipv6_subnet_id)
self.assertEqual(self.stack.ipv6_subnet_id, subnet['id'])
self.assertEqual(str(self.stack.ipv6_cidr), subnet['cidr'])
self.assertEqual(self.stack.ipv6_subnet_details, subnet)