x
/
tobiko
Code Issues Proposed changes

Update Neutron client 

Change-Id: I974dd94e5635f06c3c9185d7d758dfd3d13f1aa5
This commit is contained in:
Federico Ressi 2022-06-02 15:31:02 +02:00
parent b23ece4a1c
commit c1ab286885
6 changed files with 48 additions and 11 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -56,6 +56,7 @@ 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
@ -123,7 +124,6 @@ get_subnet = _subnet.get_subnet
get_subnet_id = _subnet.get_subnet_id
find_subnet = _subnet.find_subnet
list_subnets = _subnet.list_subnets
list_subnet_cidrs = _subnet.list_subnet_cidrs
SubnetType = _subnet.SubnetType
SubnetIdType = _subnet.SubnetIdType
NoSuchSubnet = _subnet.NoSuchSubnet

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

@ -22,7 +22,6 @@ from netaddr.strategy import ipv6
import tobiko
from tobiko.openstack.neutron import _client
from tobiko.openstack.neutron import _subnet
def new_ipv4_cidr(seed=None):
@ -63,7 +62,7 @@ class CIDRGeneratorFixture(tobiko.SharedFixture):
self.client = _client.neutron_client(self.client)
def new_cidr(self, seed):
used_cidrs = set(_subnet.list_subnet_cidrs(client=self.client))
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:
@ -133,3 +132,11 @@ def random_subnets(cidr, prefixlen, seed=None):
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)

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

@ -19,7 +19,6 @@ import netaddr
import tobiko
from tobiko.openstack.neutron import _client
from tobiko.openstack.neutron import _subnet
NetworkType = typing.Dict[str, typing.Any]
@ -101,6 +100,7 @@ def delete_network(network: NetworkIdType,
def list_network_nameservers(network_id: typing.Optional[str] = None,
ip_version: typing.Optional[int] = None) -> \
tobiko.Selection[netaddr.IPAddress]:
from tobiko.openstack.neutron import _subnet
subnets = _subnet.list_subnets(network_id=network_id)
nameservers = tobiko.Selection[netaddr.IPAddress](
netaddr.IPAddress(nameserver)

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

@ -92,12 +92,33 @@ def delete_port(port: PortIdType,
raise NoSuchPort(id=port_id) from ex
DeviceType = typing.Dict[str, typing.Any]
DeviceIdType = typing.Union[str, DeviceType]
def list_ports(client: _client.NeutronClientType = None,
device: DeviceIdType = None,
network: _network.NetworkIdType = None,
subnet: _subnet.SubnetIdType = None,
**params) -> tobiko.Selection[PortType]:
if device is not None:
params.setdefault('device_id', get_device_id(device))
if network is not None:
params.setdefault('network_id', _network.get_network_id(network))
if subnet is not None:
subnet_id = _subnet.get_subnet_id(subnet)
params.setdefault('fixed_ips', f'subnet_id={subnet_id}')
ports = _client.neutron_client(client).list_ports(**params)['ports']
return tobiko.select(ports)
def get_device_id(device: DeviceIdType) -> str:
if isinstance(device, str):
return device
else:
return device['id']
def find_port(client: _client.NeutronClientType = None,
unique=False,
default: PortType = None,

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

@ -21,6 +21,7 @@ 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]
@ -48,6 +49,8 @@ def get_subnet(subnet: SubnetIdType,
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:
@ -57,6 +60,15 @@ def create_subnet(client: _client.NeutronClientType = None,
else:
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:
@ -83,21 +95,18 @@ def delete_subnet(subnet: SubnetIdType,
def list_subnets(client: _client.NeutronClientType = None,
ip_version: int = None,
network: _network.NetworkIdType = None,
**params) -> tobiko.Selection[SubnetType]:
if ip_version is not None:
params['ip_version'] = ip_version
if network is not None:
params.setdefault('network_id', _network.get_network_id(network))
subnets = _client.neutron_client(client).list_subnets(**params)
if isinstance(subnets, abc.Mapping):
subnets = subnets['subnets']
return tobiko.select(subnets)
def list_subnet_cidrs(client: _client.NeutronClientType = None,
**params) -> tobiko.Selection[netaddr.IPNetwork]:
return tobiko.select(netaddr.IPNetwork(subnet['cidr'])
for subnet in list_subnets(client=client, **params))
def find_subnet(client: _client.NeutronClientType = None,
unique=False,
default: SubnetType = None,

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

@ -51,7 +51,7 @@ class ExternalNetworkStackFixture(heat.HeatStackFixture):
def external_network(self) -> typing.Optional[neutron.NetworkType]:
external_network = self._external_network
if external_network is None:
subnet_parameters = {}
subnet_parameters: typing.Dict[str, typing.Any] = {}
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):