Add functions to create and delete a network

Change-Id: Ica95f8835ef3659ccb6d8795750c0da51c0635bd
This commit is contained in:
Federico Ressi 2020-11-16 11:38:28 +01:00
parent f19c3c1688
commit c12d9624d7
5 changed files with 134 additions and 38 deletions

View File

@ -18,6 +18,7 @@ from tobiko.openstack.neutron import _client
from tobiko.openstack.neutron import _cidr
from tobiko.openstack.neutron import _extension
from tobiko.openstack.neutron import _port
from tobiko.openstack.neutron import _network
DHCP_AGENT = _agent.DHCP_AGENT
@ -37,8 +38,6 @@ NeutronClientFixture = _client.NeutronClientFixture
ServiceUnavailable = _client.ServiceUnavailable
neutron_client = _client.neutron_client
get_neutron_client = _client.get_neutron_client
find_network = _client.find_network
list_networks = _client.list_networks
find_subnet = _client.find_subnet
find_port = _client.find_port
list_ports = _client.list_ports
@ -46,7 +45,6 @@ list_subnets = _client.list_subnets
list_subnet_cidrs = _client.list_subnet_cidrs
list_agents = _client.list_agents
get_floating_ip = _client.get_floating_ip
get_network = _client.get_network
get_router = _client.get_router
get_port = _client.get_port
get_subnet = _client.get_subnet
@ -54,7 +52,6 @@ list_l3_agent_hosting_routers = _client.list_l3_agent_hosting_routers
find_l3_agent_hosting_router = _client.find_l3_agent_hosting_router
list_dhcp_agent_hosting_network = _client.list_dhcp_agent_hosting_network
NoSuchNetwork = _client.NoSuchNetwork
NoSuchPort = _client.NoSuchPort
NoSuchRouter = _client.NoSuchRouter
NoSuchSubnet = _client.NoSuchSubnet
@ -71,3 +68,11 @@ skip_if_missing_networking_extensions = (
find_port_ip_address = _port.find_port_ip_address
list_port_ip_addresses = _port.list_port_ip_addresses
NeutronNetworkFixture = _network.NeutronNetworkFixture
NoSuchNetwork = _network.NoSuchNetwork
create_network = _network.create_network
delete_network = _network.delete_network
get_network = _network.get_network
find_network = _network.find_network
list_networks = _network.list_networks

View File

@ -28,7 +28,7 @@ def new_ipv4_cidr(seed=None):
return tobiko.setup_fixture(IPv4CIDRGeneratorFixture).new_cidr(seed=seed)
def new_ipv6_cidr(seed):
def new_ipv6_cidr(seed=None):
return tobiko.setup_fixture(IPv6CIDRGeneratorFixture).new_cidr(seed=seed)

View File

@ -14,6 +14,7 @@
from __future__ import absolute_import
import collections
import typing
import netaddr
from neutronclient.v2_0 import client as neutronclient
@ -40,7 +41,12 @@ class NeutronClientManager(_client.OpenstackClientManager):
CLIENTS = NeutronClientManager()
def neutron_client(obj):
NeutronClientType = typing.Union[None,
neutronclient.Client,
NeutronClientFixture]
def neutron_client(obj: NeutronClientType) -> neutronclient.Client:
if not obj:
return get_neutron_client()
@ -55,8 +61,20 @@ def neutron_client(obj):
raise TypeError(message)
class HasNeutronClientFixture(tobiko.SharedFixture):
client: typing.Optional[neutronclient.Client] = None
def __init__(self, obj: NeutronClientType = None):
super(HasNeutronClientFixture, self).__init__()
self._obj = obj
def setup_fixture(self):
self.client = neutron_client(self._obj)
def get_neutron_client(session=None, shared=True, init_client=None,
manager=None):
manager=None) -> neutronclient.Client:
manager = manager or CLIENTS
client = manager.get_client(session=session, shared=shared,
init_client=init_client)
@ -67,19 +85,6 @@ def get_neutron_client(session=None, shared=True, init_client=None,
_RAISE_ERROR = object()
def find_network(client=None, unique=False, default=_RAISE_ERROR,
**attributes):
"""Look for a network matching some property values"""
networks = list_networks(client=client, **attributes)
if default is _RAISE_ERROR or networks:
if unique:
return networks.unique
else:
return networks.first
else:
return default
def find_port(client=None, unique=False, default=_RAISE_ERROR, **attributes):
"""Look for a port matching some property values"""
ports = list_ports(client=client, **attributes)
@ -104,11 +109,6 @@ def find_subnet(client=None, unique=False, default=_RAISE_ERROR, **attributes):
return default
def list_networks(client=None, **params):
networks = neutron_client(client).list_networks(**params)['networks']
return tobiko.select(networks)
def list_ports(client=None, **params):
ports = neutron_client(client).list_ports(**params)['ports']
return tobiko.select(ports)
@ -138,14 +138,6 @@ def get_floating_ip(floating_ip, client=None, **params):
return floating_ip['floatingip']
def get_network(network, client=None, **params):
try:
return neutron_client(client).show_network(network,
**params)['network']
except neutronclient.exceptions.NotFound as ex:
raise NoSuchNetwork(id=network) from ex
def get_port(port, client=None, **params):
try:
return neutron_client(client).show_port(port, **params)['port']
@ -196,10 +188,6 @@ def list_dhcp_agent_hosting_network(network, client=None, **params):
return tobiko.select(agents)
class NoSuchNetwork(tobiko.ObjectNotFound):
message = "No such network found for {id!r}"
class NoSuchPort(tobiko.ObjectNotFound):
message = "No such port found for {id!r}"

View File

@ -0,0 +1,88 @@
# 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 typing
import tobiko
from tobiko.openstack.neutron import _client
class NoSuchNetwork(tobiko.ObjectNotFound):
message = "No such network found for {id!r}"
NeutronNetworkType = typing.Dict[str, typing.Any]
def get_network(network, client=None, **params) -> NeutronNetworkType:
try:
return _client.neutron_client(
client).show_network(network, **params)['network']
except _client.neutronclient.exceptions.NotFound as ex:
raise NoSuchNetwork(id=network) from ex
def list_networks(client=None, **params) -> \
tobiko.Selection[NeutronNetworkType]:
networks = _client.neutron_client(client).list_networks(
**params)['networks']
return tobiko.select(networks)
_RAISE_ERROR = object()
def find_network(client=None, unique=False, default=_RAISE_ERROR,
**attributes) -> NeutronNetworkType:
"""Look for a network matching some property values"""
networks = list_networks(client=client, **attributes)
if default is _RAISE_ERROR or networks:
if unique:
return networks.unique
else:
return networks.first
else:
return default
def create_network(client=None, **params) -> NeutronNetworkType:
return _client.neutron_client(client).create_network(
body={'network': params})['network']
def delete_network(network, client=None):
return _client.neutron_client(client).delete_network(network=network)
class NeutronNetworkFixture(_client.HasNeutronClientFixture):
details: typing.Optional[NeutronNetworkType] = None
def __init__(self, name: typing.Optional[str] = None,
obj: _client.NeutronClientType = None):
super(NeutronNetworkFixture, self).__init__(obj=obj)
if name is None:
name = self.fixture_name
self.name: str = name
@property
def id(self):
return self.details['id']
def setup_fixture(self):
super(NeutronNetworkFixture, self).setup_fixture()
self.name = self.fixture_name
self.details = create_network(client=self.client, name=self.name)
self.addCleanup(delete_network, network=self.id, client=self.client)

View File

@ -31,7 +31,7 @@ CONF = config.CONF
@keystone.skip_unless_has_keystone_credentials()
class NeutronApiTestCase(testtools.TestCase):
class NeutronApiTest(testtools.TestCase):
"""Tests network creation"""
#: Stack of resources with a network with a gateway router
@ -85,6 +85,21 @@ class NeutronApiTestCase(testtools.TestCase):
else:
self.assertNotIn(self.stack.ipv6_subnet_id, network['subnets'])
def test_create_network(self):
network = neutron.create_network(name=self.id())
self.addCleanup(neutron.delete_network, network['id'])
self.assertIsInstance(network['id'], str)
self.assertNotEqual('', network['id'])
self.assertEqual(self.id(), network['name'])
observed = neutron.get_network(network['id'])
self.assertEqual(network['id'], observed['id'])
def test_delete_network(self):
network = neutron.create_network(name=self.id())
neutron.delete_network(network['id'])
self.assertRaises(neutron.NoSuchNetwork, neutron.get_network,
network['id'])
def test_get_router(self):
if not self.stack.has_gateway:
tobiko.skip("Stack {stack} has no gateway router",