From c12d9624d716488d865e747fa41144e50ecb9e81 Mon Sep 17 00:00:00 2001 From: Federico Ressi Date: Mon, 16 Nov 2020 11:38:28 +0100 Subject: [PATCH] Add functions to create and delete a network Change-Id: Ica95f8835ef3659ccb6d8795750c0da51c0635bd --- tobiko/openstack/neutron/__init__.py | 13 ++- tobiko/openstack/neutron/_cidr.py | 2 +- tobiko/openstack/neutron/_client.py | 52 +++++------ tobiko/openstack/neutron/_network.py | 88 +++++++++++++++++++ .../functional/openstack/test_neutron.py | 17 +++- 5 files changed, 134 insertions(+), 38 deletions(-) create mode 100644 tobiko/openstack/neutron/_network.py diff --git a/tobiko/openstack/neutron/__init__.py b/tobiko/openstack/neutron/__init__.py index dba2e872e..d249b7348 100644 --- a/tobiko/openstack/neutron/__init__.py +++ b/tobiko/openstack/neutron/__init__.py @@ -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 diff --git a/tobiko/openstack/neutron/_cidr.py b/tobiko/openstack/neutron/_cidr.py index 13678ed49..6e48cc101 100644 --- a/tobiko/openstack/neutron/_cidr.py +++ b/tobiko/openstack/neutron/_cidr.py @@ -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) diff --git a/tobiko/openstack/neutron/_client.py b/tobiko/openstack/neutron/_client.py index 9f1a15ed3..ab2132136 100644 --- a/tobiko/openstack/neutron/_client.py +++ b/tobiko/openstack/neutron/_client.py @@ -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}" diff --git a/tobiko/openstack/neutron/_network.py b/tobiko/openstack/neutron/_network.py new file mode 100644 index 000000000..8aa92c5db --- /dev/null +++ b/tobiko/openstack/neutron/_network.py @@ -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) diff --git a/tobiko/tests/functional/openstack/test_neutron.py b/tobiko/tests/functional/openstack/test_neutron.py index bb757fe55..685ff8c30 100644 --- a/tobiko/tests/functional/openstack/test_neutron.py +++ b/tobiko/tests/functional/openstack/test_neutron.py @@ -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",