openstack
/
rally-openstack
Code Issues Proposed changes

Merge "Introduce NeutronService helper"

This commit is contained in:
Zuul 2020-05-25 14:25:57 +00:00 committed by Gerrit Code Review
parent 86bbea9db1 f89f962057
commit 918a2c4b02
14 changed files with 3481 additions and 945 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

0
rally_openstack/common/services/network/__init__.py Normal file
View File

57
rally_openstack/common/services/network/net_utils.py Normal file
View File

@ -0,0 +1,57 @@
# All Rights Reserved.
#
# 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.
import netaddr
from rally.common import logging
from rally.common import utils
LOG = logging.getLogger(__name__)
_IPv4_START_CIDR = "10.2.0.0/24"
_IPv6_START_CIDR = "dead:beaf::/64"
_IPv4_CIDR_INCR = utils.RAMInt()
_IPv6_CIDR_INCR = utils.RAMInt()
def get_ip_version(ip):
return netaddr.IPNetwork(ip).version
def generate_cidr(ip_version=None, start_cidr=None):
"""Generate next CIDR for network or subnet, without IP overlapping.
This is process and thread safe, because `cidr_incr' points to
value stored directly in RAM. This guarantees that CIDRs will be
serial and unique even under hard multiprocessing/threading load.
:param ip_version: version of IP to take default value for start_cidr
:param start_cidr: start CIDR str
"""
if start_cidr is None:
if ip_version == 6:
start_cidr = _IPv6_START_CIDR
else:
start_cidr = _IPv4_START_CIDR
ip_version = get_ip_version(start_cidr)
if ip_version == 4:
cidr = str(netaddr.IPNetwork(start_cidr).next(next(_IPv4_CIDR_INCR)))
else:
cidr = str(netaddr.IPNetwork(start_cidr).next(next(_IPv6_CIDR_INCR)))
LOG.debug("CIDR generated: %s" % cidr)
return ip_version, cidr

1358
rally_openstack/common/services/network/neutron.py Normal file
View File

File diff suppressed because it is too large Load Diff

244
rally_openstack/common/wrappers/network.py
View File

@ -14,26 +14,22 @@
# under the License.
import abc
import itertools
import netaddr
from neutronclient.common import exceptions as neutron_exceptions
from rally.common import cfg
from rally.common import logging
from rally.common import utils
from rally import exceptions
from rally_openstack.common import consts
from rally_openstack.common.services.network import net_utils
from rally_openstack.common.services.network import neutron
LOG = logging.getLogger(__name__)
CONF = cfg.CONF
cidr_incr = utils.RAMInt()
ipv6_cidr_incr = utils.RAMInt()
def generate_cidr(start_cidr="10.2.0.0/24"):
"""Generate next CIDR for network or subnet, without IP overlapping.
@ -44,11 +40,7 @@ def generate_cidr(start_cidr="10.2.0.0/24"):
:param start_cidr: start CIDR str
:returns: next available CIDR str
"""
if netaddr.IPNetwork(start_cidr).version == 4:
cidr = str(netaddr.IPNetwork(start_cidr).next(next(cidr_incr)))
else:
cidr = str(netaddr.IPNetwork(start_cidr).next(next(ipv6_cidr_incr)))
LOG.debug("CIDR generated: %s" % cidr)
ip_version, cidr = net_utils.generate_cidr(start_cidr=start_cidr)
return cidr
@ -69,7 +61,7 @@ class NetworkWrapper(object, metaclass=abc.ABCMeta):
START_IPV6_CIDR = "dead:beaf::/64"
SERVICE_IMPL = None
def __init__(self, clients, owner, config=None):
def __init__(self, clients, owner, config=None, atomics=None):
"""Returns available network wrapper instance.
:param clients: rally.plugins.openstack.osclients.Clients instance
@ -124,10 +116,22 @@ class NeutronWrapper(NetworkWrapper):
LB_METHOD = "ROUND_ROBIN"
LB_PROTOCOL = "HTTP"
def __init__(self, *args, **kwargs):
super(NeutronWrapper, self).__init__(*args, **kwargs)
class _SingleClientWrapper(object):
def neutron(_self):
return self.client
self.neutron = neutron.NeutronService(
clients=_SingleClientWrapper(),
name_generator=self.owner.generate_random_name,
atomic_inst=getattr(self.owner, "_atomic_actions", [])
)
@property
def external_networks(self):
return self.client.list_networks(**{
"router:external": True})["networks"]
return self.neutron.list_networks(router_external=True)
@property
def ext_gw_mode_enabled(self):
@ -135,25 +139,30 @@ class NeutronWrapper(NetworkWrapper):
Without this extension, we can't pass the enable_snat parameter.
"""
return any(e["alias"] == "ext-gw-mode"
for e in self.client.list_extensions()["extensions"])
return self.neutron.supports_extension("ext-gw-mode", silent=True)
def get_network(self, net_id=None, name=None):
net = None
try:
if net_id:
net = self.client.show_network(net_id)["network"]
net = self.neutron.get_network(net_id)
else:
for net in self.client.list_networks(name=name)["networks"]:
break
networks = self.neutron.list_networks(name=name)
if networks:
net = networks[0]
except neutron_exceptions.NeutronClientException:
pass
if net:
return {"id": net["id"],
"name": net["name"],
"tenant_id": net["tenant_id"],
"tenant_id": net.get("tenant_id",
net.get("project_id", None)),
"status": net["status"],
"external": net["router:external"],
"subnets": net["subnets"],
"external": net.get("router:external", False),
"subnets": net.get("subnets", []),
"router_id": None}
except (TypeError, neutron_exceptions.NeutronClientException):
else:
raise NetworkWrapperException(
"Network not found: %s" % (name or net_id))
@ -164,14 +173,12 @@ class NeutronWrapper(NetworkWrapper):
:param **kwargs: POST /v2.0/routers request options
:returns: neutron router dict
"""
kwargs["name"] = self.owner.generate_random_name()
kwargs.pop("name", None)
if "tenant_id" in kwargs and "project_id" not in kwargs:
kwargs["project_id"] = kwargs.pop("tenant_id")
if external and "external_gateway_info" not in kwargs:
for net in self.external_networks:
kwargs["external_gateway_info"] = {"network_id": net["id"]}
if self.ext_gw_mode_enabled:
kwargs["external_gateway_info"]["enable_snat"] = True
return self.client.create_router({"router": kwargs})["router"]
return self.neutron.create_router(
discover_external_gw=external, **kwargs)
def create_v1_pool(self, tenant_id, subnet_id, **kwargs):
"""Create LB Pool (v1).
@ -194,9 +201,10 @@ class NeutronWrapper(NetworkWrapper):
def _generate_cidr(self, ip_version=4):
# TODO(amaretskiy): Generate CIDRs unique for network, not cluster
return generate_cidr(
ip_version, cidr = net_utils.generate_cidr(
start_cidr=self.start_cidr if ip_version == 4
else self.start_ipv6_cidr)
return cidr
def _create_network_infrastructure(self, tenant_id, **kwargs):
"""Create network.
@ -218,51 +226,34 @@ class NeutronWrapper(NetworkWrapper):
See above for recognized keyword args.
:returns: dict, network data
"""
network_args = {"network": kwargs.get("network_create_args", {})}
network_args["network"].update({
"tenant_id": tenant_id,
"name": self.owner.generate_random_name()})
network = self.client.create_network(network_args)["network"]
network_args = dict(kwargs.get("network_create_args", {}))
network_args["project_id"] = tenant_id
router = None
router_args = dict(kwargs.get("router_create_args", {}))
add_router = kwargs.get("add_router", False)
if router_args or add_router:
router_args["external"] = (
router_args.get("external", False) or add_router)
router_args["tenant_id"] = tenant_id
router = self.create_router(**router_args)
if not (router_args or add_router):
router_args = None
else:
router_args["project_id"] = tenant_id
router_args["discover_external_gw"] = router_args.pop(
"external", False) or add_router
subnet_create_args = {"project_id": tenant_id}
if "dns_nameservers" in kwargs:
subnet_create_args["dns_nameservers"] = kwargs["dns_nameservers"]
dualstack = kwargs.get("dualstack", False)
subnets = []
subnets_num = kwargs.get("subnets_num", 0)
ip_versions = itertools.cycle(
[self.SUBNET_IP_VERSION, self.SUBNET_IPV6_VERSION]
if dualstack else [self.SUBNET_IP_VERSION])
for i in range(subnets_num):
ip_version = next(ip_versions)
subnet_args = {
"subnet": {
"tenant_id": tenant_id,
"network_id": network["id"],
"name": self.owner.generate_random_name(),
"ip_version": ip_version,
"cidr": self._generate_cidr(ip_version),
"enable_dhcp": True,
"dns_nameservers": (
kwargs.get("dns_nameservers", ["8.8.8.8", "8.8.4.4"])
if ip_version == 4
else kwargs.get("dns_nameservers",
["dead:beaf::1", "dead:beaf::2"]))
}
}
subnet = self.client.create_subnet(subnet_args)["subnet"]
subnets.append(subnet)
if router:
self.client.add_interface_router(router["id"],
{"subnet_id": subnet["id"]})
net_topo = self.neutron.create_network_topology(
network_create_args=network_args,
router_create_args=router_args,
subnet_create_args=subnet_create_args,
subnets_dualstack=kwargs.get("dualstack", False),
subnets_count=kwargs.get("subnets_num", 0)
)
network = net_topo["network"]
subnets = net_topo["subnets"]
if net_topo["routers"]:
router = net_topo["routers"][0]
else:
router = None
return {
"network": {
@ -309,50 +300,33 @@ class NeutronWrapper(NetworkWrapper):
self.client.delete_pool(pool_id)
def delete_network(self, network):
"""Delete network
if network["router_id"]:
self.client.remove_gateway_router(network["router_id"])
:param network: network object returned by create_network method
"""
for port in self.client.list_ports(network_id=network["id"])["ports"]:
if port["device_owner"] in (
"network:router_interface",
"network:router_interface_distributed",
"network:ha_router_replicated_interface",
"network:router_gateway"):
try:
self.client.remove_interface_router(
port["device_id"], {"port_id": port["id"]})
except (neutron_exceptions.BadRequest,
neutron_exceptions.NotFound):
# Some neutron plugins don't use router as
# the device ID. Also, some plugin doesn't allow
# to update the ha rotuer interface as there is
# an internal logic to update the interface/data model
# instead.
pass
else:
try:
self.client.delete_port(port["id"])
except neutron_exceptions.PortNotFoundClient:
# port is auto-removed
pass
router = {"id": network["router_id"]} if network["router_id"] else None
# delete_network_topology uses only IDs, but let's transmit as much as
# possible info
topo = {
"network": {
"id": network["id"],
"name": network["name"],
"status": network["status"],
"subnets": network["subnets"],
"router:external": network["external"]
},
"subnets": [{"id": s} for s in network["subnets"]],
"routers": [router] if router else []
}
for subnet in self.client.list_subnets(
network_id=network["id"])["subnets"]:
self._delete_subnet(subnet["id"])
responce = self.client.delete_network(network["id"])
if network["router_id"]:
self.client.delete_router(network["router_id"])
return responce
self.neutron.delete_network_topology(topo)
def _delete_subnet(self, subnet_id):
self.client.delete_subnet(subnet_id)
self.neutron.delete_subnet(subnet_id)
def list_networks(self):
return self.client.list_networks()["networks"]
return self.neutron.list_networks()
def create_port(self, network_id, **kwargs):
"""Create neutron port.
@ -361,9 +335,7 @@ class NeutronWrapper(NetworkWrapper):
:param **kwargs: POST /v2.0/ports request options
:returns: neutron port dict
"""
kwargs["network_id"] = network_id
kwargs["name"] = self.owner.generate_random_name()
return self.client.create_port({"port": kwargs})["port"]
return self.neutron.create_port(network_id=network_id, **kwargs)
def create_floating_ip(self, ext_network=None,
tenant_id=None, port_id=None, **kwargs):
@ -377,34 +349,13 @@ class NeutronWrapper(NetworkWrapper):
"""
if not tenant_id:
raise ValueError("Missed tenant_id")
if type(ext_network) is dict:
net_id = ext_network["id"]
elif ext_network:
ext_net = self.get_network(name=ext_network)
if not ext_net["external"]:
raise NetworkWrapperException("Network is not external: %s"
% ext_network)
net_id = ext_net["id"]
else:
ext_networks = self.external_networks
if not ext_networks:
raise NetworkWrapperException(
"Failed to allocate floating IP: "
"no external networks found")
net_id = ext_networks[0]["id"]
kwargs = {"floatingip": {"floating_network_id": net_id,
"tenant_id": tenant_id}}
if not CONF.openstack.pre_newton_neutron:
descr = self.owner.generate_random_name()
kwargs["floatingip"]["description"] = descr
if port_id:
kwargs["floatingip"]["port_id"] = port_id
fip = self.client.create_floatingip(kwargs)["floatingip"]
try:
fip = self.neutron.create_floatingip(
floating_network=ext_network, project_id=tenant_id,
port_id=port_id)
except (exceptions.NotFoundException,
exceptions.GetResourceFailure) as e:
raise NetworkWrapperException(str(e)) from None
return {"id": fip["id"], "ip": fip["floating_ip_address"]}
def delete_floating_ip(self, fip_id, **kwargs):
@ -413,7 +364,7 @@ class NeutronWrapper(NetworkWrapper):
:param fip_id: int floating IP id
:param **kwargs: for compatibility, not used here
"""
self.client.delete_floatingip(fip_id)
self.neutron.delete_floatingip(fip_id)
def supports_extension(self, extension):
"""Check whether a neutron extension is supported
@ -422,11 +373,12 @@ class NeutronWrapper(NetworkWrapper):
:returns: result tuple
:rtype: (bool, string)
"""
extensions = self.client.list_extensions().get("extensions", [])
if any(ext.get("alias") == extension for ext in extensions):
return True, ""
try:
self.neutron.supports_extension(extension)
except exceptions.NotFoundException as e:
return False, str(e)
return False, "Neutron driver does not support %s" % extension
return True, ""
def wrap(clients, owner, config=None):

61
rally_openstack/task/cleanup/resources.py
View File

@ -20,6 +20,7 @@ from rally.task import utils as task_utils
from rally_openstack.common.services.identity import identity
from rally_openstack.common.services.image import glance_v2
from rally_openstack.common.services.image import image
from rally_openstack.common.services.network import neutron
from rally_openstack.task.cleanup import base
@ -201,13 +202,11 @@ _neutron_order = get_order(300)
@base.resource(service=None, resource=None, admin_required=True)
class NeutronMixin(SynchronizedDeletion, base.ResourceManager):
# Neutron has the best client ever, so we need to override everything
def supports_extension(self, extension):
exts = self._manager().list_extensions().get("extensions", [])
if any(ext.get("alias") == extension for ext in exts):
return True
return False
@property
def _neutron(self):
return neutron.NeutronService(
self._admin_required and self.admin or self.user)
def _manager(self):
client = self._admin_required and self.admin or self.user
@ -220,7 +219,10 @@ class NeutronMixin(SynchronizedDeletion, base.ResourceManager):
return self.raw_resource["name"]
def delete(self):
delete_method = getattr(self._manager(), "delete_%s" % self._resource)
key = "delete_%s" % self._resource
delete_method = getattr(
self._neutron, key, getattr(self._manager(), key)
)
delete_method(self.id())
@property
@ -242,7 +244,7 @@ class NeutronMixin(SynchronizedDeletion, base.ResourceManager):
class NeutronLbaasV1Mixin(NeutronMixin):
def list(self):
if self.supports_extension("lbaas"):
if self._neutron.supports_extension("lbaas", silent=True):
return super(NeutronLbaasV1Mixin, self).list()
return []
@ -268,7 +270,7 @@ class NeutronV1Pool(NeutronLbaasV1Mixin):
class NeutronLbaasV2Mixin(NeutronMixin):
def list(self):
if self.supports_extension("lbaasv2"):
if self._neutron.supports_extension("lbaasv2", silent=True):
return super(NeutronLbaasV2Mixin, self).list()
return []
@ -373,7 +375,7 @@ class OctaviaHealthMonitors(OctaviaMixIn):
admin_required=True, perform_for_admin_only=True)
class NeutronBgpvpn(NeutronMixin):
def list(self):
if self.supports_extension("bgpvpn"):
if self._neutron.supports_extension("bgpvpn", silent=True):
return self._manager().list_bgpvpns()["bgpvpns"]
return []
@ -414,20 +416,23 @@ class NeutronPort(NeutronMixin):
# NOTE(andreykurilin): port is the kind of resource that can be created
# automatically. In this case it doesn't have name field which matches
# our resource name templates.
ROUTER_INTERFACE_OWNERS = ("network:router_interface",
"network:router_interface_distributed",
"network:ha_router_replicated_interface")
ROUTER_GATEWAY_OWNER = "network:router_gateway"
def __init__(self, *args, **kwargs):
super(NeutronPort, self).__init__(*args, **kwargs)
self._cache = {}
@property
def ROUTER_INTERFACE_OWNERS(self):
return self._neutron.ROUTER_INTERFACE_OWNERS
@property
def ROUTER_GATEWAY_OWNER(self):
return self._neutron.ROUTER_GATEWAY_OWNER
def _get_resources(self, resource):
if resource not in self._cache:
resources = getattr(self._manager(), "list_%s" % resource)()
self._cache[resource] = [r for r in resources[resource]
resources = getattr(self._neutron, "list_%s" % resource)()
self._cache[resource] = [r for r in resources
if r["tenant_id"] == self.tenant_uuid]
return self._cache[resource]
@ -455,23 +460,11 @@ class NeutronPort(NeutronMixin):
self.raw_resource.get("name", ""))
def delete(self):
device_owner = self.raw_resource["device_owner"]
if (device_owner in self.ROUTER_INTERFACE_OWNERS
or device_owner == self.ROUTER_GATEWAY_OWNER):
if device_owner == self.ROUTER_GATEWAY_OWNER:
self._manager().remove_gateway_router(
self.raw_resource["device_id"])
self._manager().remove_interface_router(
self.raw_resource["device_id"], {"port_id": self.id()})
else:
from neutronclient.common import exceptions as neutron_exceptions
try:
self._manager().delete_port(self.id())
except neutron_exceptions.PortNotFoundClient:
# Port can be already auto-deleted, skip silently
LOG.debug("Port %s was not deleted. Skip silently because "
"port can be already auto-deleted." % self.id())
found = self._neutron.delete_port(self.raw_resource)
if not found:
# Port can be already auto-deleted, skip silently
LOG.debug(f"Port {self.id()} was not deleted. Skip silently "
f"because port can be already auto-deleted.")
@base.resource("neutron", "subnet", order=next(_neutron_order),

308
rally_openstack/task/scenarios/neutron/utils.py
View File

@ -13,7 +13,6 @@
# License for the specific language governing permissions and limitations
# under the License.
import netaddr
import random
from rally.common import cfg
@ -22,7 +21,7 @@ from rally import exceptions
from rally.task import atomic
from rally.task import utils
from rally_openstack.common.wrappers import network as network_wrapper
from rally_openstack.common.services.network import neutron
from rally_openstack.task import scenario
@ -32,7 +31,20 @@ CONF = cfg.CONF
LOG = logging.getLogger(__name__)
class NeutronScenario(scenario.OpenStackScenario):
class NeutronBaseScenario(scenario.OpenStackScenario):
"""Base class for Neutron scenarios with basic atomic actions."""
def __init__(self, *args, **kwargs):
super(NeutronBaseScenario, self).__init__(*args, **kwargs)
if hasattr(self, "_clients"):
self.neutron = neutron.NeutronService(
clients=self._clients,
name_generator=self.generate_random_name,
atomic_inst=self.atomic_actions()
)
class NeutronScenario(NeutronBaseScenario):
"""Base class for Neutron scenarios with basic atomic actions."""
# TODO(rkiran): modify in case LBaaS-v2 requires
@ -51,12 +63,11 @@ class NeutronScenario(scenario.OpenStackScenario):
:param kwargs: dict, network options
:returns: str, Neutron network-id
"""
networks = self._list_networks()
for net in networks:
if (net["name"] == network) or (net["id"] == network):
return net["id"]
raise exceptions.NotFoundException(
message="Network %s not found." % network)
try:
return self.neutron.find_network(network)["id"]
except exceptions.GetResourceFailure:
raise exceptions.NotFoundException(
message="Network %s not found." % network)
@property
def _ext_gw_mode_enabled(self):
@ -64,39 +75,32 @@ class NeutronScenario(scenario.OpenStackScenario):
Without this extension, we can't pass the enable_snat parameter.
"""
return any(
e["alias"] == "ext-gw-mode"
for e in self.clients("neutron").list_extensions()["extensions"])
return self.neutron.supports_extension("ext-gw-mode", silent=True)
@atomic.action_timer("neutron.create_network")
def _create_network(self, network_create_args):
"""Create neutron network.
:param network_create_args: dict, POST /v2.0/networks request options
:returns: neutron network dict
"""
network_create_args["name"] = self.generate_random_name()
return self.clients("neutron").create_network(
{"network": network_create_args})
network_create_args.pop("name", None)
return {"network": self.neutron.create_network(**network_create_args)}
@atomic.action_timer("neutron.list_networks")
def _list_networks(self, **kwargs):
"""Return user networks list.
:param kwargs: network list options
"""
return self.clients("neutron").list_networks(**kwargs)["networks"]
return self.neutron.list_networks(**kwargs)
@atomic.action_timer("neutron.list_agents")
def _list_agents(self, **kwargs):
"""Fetches agents.
:param kwargs: neutron agent list options
:returns: user agents list
"""
return self.clients("neutron").list_agents(**kwargs)["agents"]
return self.neutron.list_agents(**kwargs)
@atomic.action_timer("neutron.update_network")
def _update_network(self, network, network_update_args):
"""Update the network.
@ -107,11 +111,9 @@ class NeutronScenario(scenario.OpenStackScenario):
:returns: updated neutron network dict
"""
network_update_args["name"] = self.generate_random_name()
body = {"network": network_update_args}
return self.clients("neutron").update_network(
network["network"]["id"], body)
return {"network": self.neutron.update_network(
network["network"]["id"], **network_update_args)}
@atomic.action_timer("neutron.show_network")
def _show_network(self, network, **kwargs):
"""show network details.
@ -119,18 +121,16 @@ class NeutronScenario(scenario.OpenStackScenario):
:param kwargs: dict, POST /v2.0/networks show options
:returns: details of the network
"""
return self.clients("neutron").show_network(
network["network"]["id"], **kwargs)
network = self.neutron.get_network(network["network"]["id"], **kwargs)
return {"network": network}
@atomic.action_timer("neutron.delete_network")
def _delete_network(self, network):
"""Delete neutron network.
:param network: Network object
"""
self.clients("neutron").delete_network(network["id"])
self.neutron.delete_network(network["id"])
@atomic.action_timer("neutron.create_subnet")
def _create_subnet(self, network, subnet_create_args, start_cidr=None):
"""Create neutron subnet.
@ -138,27 +138,17 @@ class NeutronScenario(scenario.OpenStackScenario):
:param subnet_create_args: POST /v2.0/subnets request options
:returns: neutron subnet dict
"""
network_id = network["network"]["id"]
if not subnet_create_args.get("cidr"):
start_cidr = start_cidr or "10.2.0.0/24"
subnet_create_args["cidr"] = (
network_wrapper.generate_cidr(start_cidr=start_cidr))
subnet_create_args.pop("name", None)
subnet_create_args["network_id"] = network["network"]["id"]
subnet_create_args["start_cidr"] = start_cidr
subnet_create_args["network_id"] = network_id
subnet_create_args["name"] = self.generate_random_name()
subnet_create_args["ip_version"] = netaddr.IPNetwork(
subnet_create_args["cidr"]).version
return {"subnet": self.neutron.create_subnet(**subnet_create_args)}
return self.clients("neutron").create_subnet(
{"subnet": subnet_create_args})
@atomic.action_timer("neutron.list_subnets")
def _list_subnets(self):
"""Returns user subnetworks list."""
return self.clients("neutron").list_subnets()["subnets"]
return self.neutron.list_subnets()
@atomic.action_timer("neutron.show_subnet")
def _show_subnet(self, subnet, **kwargs):
"""show subnet details.
@ -166,10 +156,8 @@ class NeutronScenario(scenario.OpenStackScenario):
:param kwargs: Optional additional arguments for subnet show
:returns: details of the subnet
"""
return self.clients("neutron").show_subnet(subnet["subnet"]["id"],
**kwargs)
return {"subnet": self.neutron.get_subnet(subnet["subnet"]["id"])}
@atomic.action_timer("neutron.update_subnet")
def _update_subnet(self, subnet, subnet_update_args):
"""Update the neutron subnet.
@ -180,46 +168,35 @@ class NeutronScenario(scenario.OpenStackScenario):
:returns: updated neutron subnet dict
"""
subnet_update_args["name"] = self.generate_random_name()
body = {"subnet": subnet_update_args}
return self.clients("neutron").update_subnet(
subnet["subnet"]["id"], body)
return {"subnet": self.neutron.update_subnet(
subnet["subnet"]["id"], **subnet_update_args)}
@atomic.action_timer("neutron.delete_subnet")
def _delete_subnet(self, subnet):
"""Delete neutron subnet
:param subnet: Subnet object
"""
self.clients("neutron").delete_subnet(subnet["subnet"]["id"])
self.neutron.delete_subnet(subnet["subnet"]["id"])
@atomic.action_timer("neutron.create_router")
def _create_router(self, router_create_args, external_gw=False):
"""Create neutron router.
:param router_create_args: POST /v2.0/routers request options
:returns: neutron router dict
"""
router_create_args["name"] = self.generate_random_name()
router_create_args.pop("name", None)
if ("tenant_id" in router_create_args
and "project_id" not in router_create_args):
router_create_args["project_id"] = router_create_args.pop(
"tenant_id")
if external_gw:
for network in self._list_networks():
if network.get("router:external"):
external_network = network
gw_info = {"network_id": external_network["id"]}
if self._ext_gw_mode_enabled:
gw_info["enable_snat"] = True
router_create_args.setdefault("external_gateway_info",
gw_info)
return {"router": self.neutron.create_router(
discover_external_gw=external_gw, **router_create_args)}
return self.clients("neutron").create_router(
{"router": router_create_args})
@atomic.action_timer("neutron.list_routers")
def _list_routers(self):
"""Returns user routers list."""
return self.clients("neutron").list_routers()["routers"]
return self.neutron.list_routers()
@atomic.action_timer("neutron.show_router")
def _show_router(self, router, **kwargs):
"""Show information of a given router.
@ -227,18 +204,16 @@ class NeutronScenario(scenario.OpenStackScenario):
:kwargs: dict, POST /v2.0/routers show options
:return: details of the router
"""
return self.clients("neutron").show_router(
router["router"]["id"], **kwargs)
return {"router": self.neutron.get_router(
router["router"]["id"], **kwargs)}
@atomic.action_timer("neutron.delete_router")
def _delete_router(self, router):
"""Delete neutron router
:param router: Router object
"""
self.clients("neutron").delete_router(router["router"]["id"])
self.neutron.delete_router(router["router"]["id"])
@atomic.action_timer("neutron.update_router")
def _update_router(self, router, router_update_args):
"""Update the neutron router.
@ -249,11 +224,9 @@ class NeutronScenario(scenario.OpenStackScenario):
:returns: updated neutron router dict
"""
router_update_args["name"] = self.generate_random_name()
body = {"router": router_update_args}
return self.clients("neutron").update_router(
router["router"]["id"], body)
return {"router": self.neutron.update_router(
router["router"]["id"], **router_update_args)}
@atomic.action_timer("neutron.create_port")
def _create_port(self, network, port_create_args):
"""Create neutron port.
@ -261,16 +234,13 @@ class NeutronScenario(scenario.OpenStackScenario):
:param port_create_args: POST /v2.0/ports request options
:returns: neutron port dict
"""
port_create_args["network_id"] = network["network"]["id"]
port_create_args["name"] = self.generate_random_name()
return self.clients("neutron").create_port({"port": port_create_args})
return {"port": self.neutron.create_port(
network_id=network["network"]["id"], **port_create_args)}
@atomic.action_timer("neutron.list_ports")
def _list_ports(self):
"""Return user ports list."""
return self.clients("neutron").list_ports()["ports"]
return self.neutron.list_ports()
@atomic.action_timer("neutron.show_port")
def _show_port(self, port, **params):
"""Return user port details.
@ -278,9 +248,8 @@ class NeutronScenario(scenario.OpenStackScenario):
:param params: neutron port show options
:returns: neutron port dict
"""
return self.clients("neutron").show_port(port["port"]["id"], **params)
return {"port": self.neutron.get_port(port["port"]["id"], **params)}
@atomic.action_timer("neutron.update_port")
def _update_port(self, port, port_update_args):
"""Update the neutron port.
@ -291,16 +260,15 @@ class NeutronScenario(scenario.OpenStackScenario):
:returns: updated neutron port dict
"""
port_update_args["name"] = self.generate_random_name()
body = {"port": port_update_args}
return self.clients("neutron").update_port(port["port"]["id"], body)
return {"port": self.neutron.update_port(port["port"]["id"],
**port_update_args)}
@atomic.action_timer("neutron.delete_port")
def _delete_port(self, port):
"""Delete neutron port.
:param port: Port object
"""
self.clients("neutron").delete_port(port["port"]["id"])
self.neutron.delete_port(port["port"]["id"])
@logging.log_deprecated_args(
"network_create_args is deprecated; use the network context instead",
@ -356,10 +324,16 @@ class NeutronScenario(scenario.OpenStackScenario):
:parm subnet_cidr_start: str, start value for subnets CIDR
:returns: tuple of result network and subnets list
"""
network = self._create_network(network_create_args or {})
subnets = self._create_subnets(network, subnet_create_args,
subnet_cidr_start, subnets_per_network)
return network, subnets
subnet_create_args = dict(subnet_create_args or {})
subnet_create_args["start_cidr"] = subnet_cidr_start
net_topo = self.neutron.create_network_topology(
network_create_args=(network_create_args or {}),
subnet_create_args=subnet_create_args,
subnets_count=subnets_per_network
)
subnets = [{"subnet": s} for s in net_topo["subnets"]]
return {"network": net_topo["network"]}, subnets
def _create_network_structure(self, network_create_args=None,
subnet_create_args=None,
@ -375,41 +349,39 @@ class NeutronScenario(scenario.OpenStackScenario):
:param router_create_args: dict, POST /v2.0/routers request options
:returns: tuple of (network, subnets, routers)
"""
network = self._create_network(network_create_args or {})
subnets = self._create_subnets(network, subnet_create_args,
subnet_cidr_start,
subnets_per_network)
routers = []
for subnet in subnets:
router = self._create_router(router_create_args or {})
self._add_interface_router(subnet["subnet"],
router["router"])
routers.append(router)
subnet_create_args = dict(subnet_create_args or {})
subnet_create_args["start_cidr"] = subnet_cidr_start
return (network, subnets, routers)
net_topo = self.neutron.create_network_topology(
network_create_args=(network_create_args or {}),
router_create_args=(router_create_args or {}),
router_per_subnet=True,
subnet_create_args=subnet_create_args,
subnets_count=subnets_per_network
)
return ({"network": net_topo["network"]},
[{"subnet": s} for s in net_topo["subnets"]],
[{"router": r} for r in net_topo["routers"]])
@atomic.action_timer("neutron.add_interface_router")
def _add_interface_router(self, subnet, router):
"""Connect subnet to router.
:param subnet: dict, neutron subnet
:param router: dict, neutron router
"""
self.clients("neutron").add_interface_router(
router["id"], {"subnet_id": subnet["id"]})
self.neutron.add_interface_to_router(router_id=router["id"],
subnet_id=subnet["id"])
@atomic.action_timer("neutron.remove_interface_router")
def _remove_interface_router(self, subnet, router):
"""Remove subnet from router
:param subnet: dict, neutron subnet
:param router: dict, neutron router
"""
self.clients("neutron").remove_interface_router(
router["id"], {"subnet_id": subnet["id"]})
self.neutron.remove_interface_from_router(
router_id=router["id"], subnet_id=subnet["id"])
@atomic.action_timer("neutron.add_gateway_router")
def _add_gateway_router(self, router, ext_net, enable_snat=None):
"""Set the external network gateway for a router.
@ -417,21 +389,18 @@ class NeutronScenario(scenario.OpenStackScenario):
:param ext_net: external network for the gateway
:param enable_snat: True if enable snat, None to avoid update
"""
gw_info = {"network_id": ext_net["network"]["id"]}
if enable_snat is not None:
if self._ext_gw_mode_enabled:
gw_info["enable_snat"] = enable_snat
self.clients("neutron").add_gateway_router(
router["router"]["id"], gw_info)
self.neutron.add_gateway_to_router(
router_id=router["router"]["id"],
network_id=ext_net["network"]["id"],
enable_snat=enable_snat
)
@atomic.action_timer("neutron.remove_gateway_router")
def _remove_gateway_router(self, router):
"""Removes an external network gateway from the specified router.
:param router: dict, neutron router
"""
self.clients("neutron").remove_gateway_router(
router["router"]["id"])
self.neutron.remove_gateway_from_router(router["router"]["id"])
@atomic.action_timer("neutron.create_pool")
def _create_lb_pool(self, subnet_id, **pool_create_args):
@ -533,7 +502,6 @@ class NeutronScenario(scenario.OpenStackScenario):
body = {"vip": vip_update_args}
return self.clients("neutron").update_vip(vip["vip"]["id"], body)
@atomic.action_timer("neutron.create_floating_ip")
def _create_floatingip(self, floating_network, **floating_ip_args):
"""Create floating IP with floating_network.
@ -541,40 +509,21 @@ class NeutronScenario(scenario.OpenStackScenario):
:param floating_ip_args: dict, POST /floatingips create options
:returns: dict, neutron floating IP
"""
from neutronclient.common import exceptions as ne
floating_network_id = self._get_network_id(
floating_network)
args = {"floating_network_id": floating_network_id}
if not CONF.openstack.pre_newton_neutron:
args["description"] = self.generate_random_name()
args.update(floating_ip_args)
try:
return self.clients("neutron").create_floatingip(
{"floatingip": args})
except ne.BadRequest as e:
error = "%s" % e
if "Unrecognized attribute" in error and "'description'" in error:
LOG.info("It looks like you have Neutron API of pre-Newton "
"OpenStack release. Setting "
"openstack.pre_newton_neutron option via Rally "
"configuration should fix an issue.")
raise
return {"floatingip": self.neutron.create_floatingip(
floating_network=floating_network, **floating_ip_args)}
@atomic.action_timer("neutron.list_floating_ips")
def _list_floating_ips(self, **kwargs):
"""Return floating IPs list."""
return self.clients("neutron").list_floatingips(**kwargs)
return {"floatingips": self.neutron.list_floatingips(**kwargs)}
@atomic.action_timer("neutron.delete_floating_ip")
def _delete_floating_ip(self, floating_ip):
"""Delete floating IP.
:param dict, floating IP object
"""
return self.clients("neutron").delete_floatingip(floating_ip["id"])
return self.neutron.delete_floatingip(floating_ip["id"])
@atomic.action_timer("neutron.associate_floating_ip")
def _associate_floating_ip(self, floatingip, port):
"""Associate floating IP with port.
@ -582,20 +531,18 @@ class NeutronScenario(scenario.OpenStackScenario):
:param port: port dict
:returns: updated floating IP dict
"""
return self.clients("neutron").update_floatingip(
floatingip["id"],
{"floatingip": {"port_id": port["id"]}})["floatingip"]
return self.neutron.associate_floatingip(
port_id=port["id"],
floatingip_id=floatingip["id"])
@atomic.action_timer("neutron.dissociate_floating_ip")
def _dissociate_floating_ip(self, floatingip):
"""Dissociate floating IP from ports.
:param floatingip: floating IP dict
:returns: updated floating IP dict
"""
return self.clients("neutron").update_floatingip(
floatingip["id"],
{"floatingip": {"port_id": None}})["floatingip"]
return self.neutron.dissociate_floatingip(
floatingip_id=floatingip["id"])
@atomic.action_timer("neutron.create_healthmonitor")
def _create_v1_healthmonitor(self, **healthmonitor_create_args):
@ -648,7 +595,6 @@ class NeutronScenario(scenario.OpenStackScenario):
return self.clients("neutron").update_health_monitor(
healthmonitor["health_monitor"]["id"], body)
@atomic.action_timer("neutron.create_security_group")
def _create_security_group(self, **security_group_create_args):
"""Create Neutron security-group.
@ -657,24 +603,21 @@ class NeutronScenario(scenario.OpenStackScenario):
:returns: dict, neutron security-group
"""
security_group_create_args["name"] = self.generate_random_name()
return self.clients("neutron").create_security_group(
{"security_group": security_group_create_args})
return {"security_group": self.neutron.create_security_group(
**security_group_create_args)}
@atomic.action_timer("neutron.delete_security_group")
def _delete_security_group(self, security_group):
"""Delete Neutron security group.
:param security_group: dict, neutron security_group
"""
return self.clients("neutron").delete_security_group(
return self.neutron.delete_security_group(
security_group["security_group"]["id"])
@atomic.action_timer("neutron.list_security_groups")
def _list_security_groups(self, **kwargs):
"""Return list of Neutron security groups."""
return self.clients("neutron").list_security_groups(**kwargs)
return {"security_groups": self.neutron.list_security_groups(**kwargs)}
@atomic.action_timer("neutron.show_security_group")
def _show_security_group(self, security_group, **kwargs):
"""Show security group details.
@ -682,10 +625,9 @@ class NeutronScenario(scenario.OpenStackScenario):
:param kwargs: Optional additional arguments for security_group show
:returns: security_group details
"""
return self.clients("neutron").show_security_group(
security_group["security_group"]["id"], **kwargs)
return {"security_group": self.neutron.get_security_group(
security_group["security_group"]["id"], **kwargs)}
@atomic.action_timer("neutron.update_security_group")
def _update_security_group(self, security_group,
**security_group_update_args):
"""Update Neutron security-group.
@ -696,9 +638,9 @@ class NeutronScenario(scenario.OpenStackScenario):
:returns: dict, updated neutron security-group
"""
security_group_update_args["name"] = self.generate_random_name()
body = {"security_group": security_group_update_args}
return self.clients("neutron").update_security_group(
security_group["security_group"]["id"], body)
return {"security_group": self.neutron.update_security_group(
security_group["security_group"]["id"],
**security_group_update_args)}
def update_loadbalancer_resource(self, lb):
try:
@ -857,7 +799,6 @@ class NeutronScenario(scenario.OpenStackScenario):
return self.clients("neutron").list_bgpvpn_router_assocs(
bgpvpn["bgpvpn"]["id"], **kwargs)
@atomic.action_timer("neutron.create_security_group_rule")
def _create_security_group_rule(self, security_group_id,
**security_group_rule_args):
"""Create Neutron security-group-rule.
@ -867,25 +808,19 @@ class NeutronScenario(scenario.OpenStackScenario):
/v2.0/security-group-rules request options
:returns: dict, neutron security-group-rule
"""
security_group_rule_args["security_group_id"] = security_group_id
if "direction" not in security_group_rule_args:
security_group_rule_args["direction"] = "ingress"
if "protocol" not in security_group_rule_args:
security_group_rule_args["protocol"] = "tcp"
return {"security_group_rule": self.neutron.create_security_group_rule(
security_group_id, **security_group_rule_args
)}
return self.clients("neutron").create_security_group_rule(
{"security_group_rule": security_group_rule_args})
@atomic.action_timer("neutron.list_security_group_rules")
def _list_security_group_rules(self, **kwargs):
"""List all security group rules.
:param kwargs: Optional additional arguments for roles list
:return: list of security group rules
"""
return self.clients("neutron").list_security_group_rules(**kwargs)
result = self.neutron.list_security_group_rules(**kwargs)
return {"security_group_rules": result}
@atomic.action_timer("neutron.show_security_group_rule")
def _show_security_group_rule(self, security_group_rule, **kwargs):
"""Show information of a given security group rule.
@ -893,17 +828,15 @@ class NeutronScenario(scenario.OpenStackScenario):
:param kwargs: Optional additional arguments for roles list
:return: details of security group rule
"""
return self.clients("neutron").show_security_group_rule(
security_group_rule, **kwargs)
return {"security_group_rule": self.neutron.get_security_group_rule(
security_group_rule, **kwargs)}
@atomic.action_timer("neutron.delete_security_group_rule")
def _delete_security_group_rule(self, security_group_rule):
"""Delete a given security group rule.
:param security_group_rule: id of security group rule
"""
self.clients("neutron").delete_security_group_rule(
security_group_rule)
self.neutron.delete_security_group_rule(security_group_rule)
@atomic.action_timer("neutron.delete_trunk")
def _delete_trunk(self, trunk_port):
@ -918,10 +851,6 @@ class NeutronScenario(scenario.OpenStackScenario):
def _list_trunks(self, **kwargs):
return self.clients("neutron").list_trunks(**kwargs)["trunks"]
@atomic.action_timer("neutron.list_ports_by_device_id")
def _list_ports_by_device_id(self, device_id):
return self.clients("neutron").list_ports(device_id=device_id)
@atomic.action_timer("neutron.list_subports_by_trunk")
def _list_subports_by_trunk(self, trunk_id):
return self.clients("neutron").trunk_get_subports(trunk_id)
@ -930,3 +859,6 @@ class NeutronScenario(scenario.OpenStackScenario):
def _add_subports_to_trunk(self, trunk_id, subports):
return self.clients("neutron").trunk_add_subports(
trunk_id, {"sub_ports": subports})
def _list_ports_by_device_id(self, device_id):
return self.neutron.list_ports(device_id=device_id)

78
rally_openstack/task/scenarios/nova/utils.py
View File

@ -23,13 +23,15 @@ from rally.task import utils
from rally_openstack.common.services.image import image as image_service
from rally_openstack.task import scenario
from rally_openstack.task.scenarios.cinder import utils as cinder_utils
from rally_openstack.task.scenarios.neutron import utils as neutron_utils
CONF = cfg.CONF
LOG = logging.getLogger(__file__)
class NovaScenario(scenario.OpenStackScenario):
class NovaScenario(neutron_utils.NeutronBaseScenario,
scenario.OpenStackScenario):
"""Base class for Nova scenarios with basic atomic actions."""
@atomic.action_timer("nova.list_servers")
@ -633,37 +635,18 @@ class NovaScenario(scenario.OpenStackScenario):
:param fixed_address: The fixedIP address the FloatingIP is to be
associated with (optional)
"""
with atomic.ActionTimer(self, "neutron.list_ports"):
ports = self.clients("neutron").list_ports(device_id=server.id)
port = ports["ports"][0]
if isinstance(address, dict):
floating_ip = self.neutron.associate_floatingip(
device_id=server.id, fixed_ip_address=fixed_address,
floatingip_id=address["id"])
else:
floating_ip = self.neutron.associate_floatingip(
device_id=server.id, fixed_ip_address=fixed_address,
floating_ip_address=address)
fip = address
if not isinstance(address, dict):
LOG.warning(
"The argument 'address' of "
"NovaScenario._associate_floating_ip method accepts a "
"dict-like representation of floating ip. Transmitting a "
"string with just an IP is deprecated.")
with atomic.ActionTimer(self, "neutron.list_floating_ips"):
all_fips = self.clients("neutron").list_floatingips(
tenant_id=self.context["tenant"]["id"])
filtered_fip = [f for f in all_fips["floatingips"]
if f["floating_ip_address"] == address]
if not filtered_fip:
raise exceptions.NotFoundException(
"There is no floating ip with '%s' address." % address)
fip = filtered_fip[0]
# the first case: fip object is returned from network wrapper
# the second case: from neutronclient directly
fip_ip = fip.get("ip", fip.get("floating_ip_address", None))
fip_update_dict = {"port_id": port["id"]}
if fixed_address:
fip_update_dict["fixed_ip_address"] = fixed_address
self.clients("neutron").update_floatingip(
fip["id"], {"floatingip": fip_update_dict}
)
utils.wait_for(server,
is_ready=self.check_ip_address(fip_ip),
is_ready=self.check_ip_address(
floating_ip["floating_ip_address"]),
update_resource=utils.get_from_manager())
# Update server data
server.addresses = server.manager.get(server.id).addresses
@ -675,32 +658,19 @@ class NovaScenario(scenario.OpenStackScenario):
:param server: The :class:`Server` to add an IP to.
:param address: The dict-like representation of FloatingIP to remove
"""
fip = address
if not isinstance(fip, dict):
LOG.warning(
"The argument 'address' of "
"NovaScenario._dissociate_floating_ip method accepts a "
"dict-like representation of floating ip. Transmitting a "
"string with just an IP is deprecated.")
with atomic.ActionTimer(self, "neutron.list_floating_ips"):
all_fips = self.clients("neutron").list_floatingips(
tenant_id=self.context["tenant"]["id"]
)
filtered_fip = [f for f in all_fips["floatingips"]
if f["floating_ip_address"] == address]
if not filtered_fip:
raise exceptions.NotFoundException(
"There is no floating ip with '%s' address." % address)
fip = filtered_fip[0]
self.clients("neutron").update_floatingip(
fip["id"], {"floatingip": {"port_id": None}}
)
# the first case: fip object is returned from network wrapper
# the second case: from neutronclient directly
fip_ip = fip.get("ip", fip.get("floating_ip_address", None))
if isinstance(address, dict):
floating_ip = self.neutron.dissociate_floatingip(
floatingip_id=address["id"]
)
else:
floating_ip = self.neutron.dissociate_floatingip(
floating_ip_address=address
)
utils.wait_for(
server,
is_ready=self.check_ip_address(fip_ip, must_exist=False),
is_ready=self.check_ip_address(
floating_ip["floating_ip_address"], must_exist=False),
update_resource=utils.get_from_manager()
)
# Update server data

0
tests/unit/common/services/network/__init__.py Normal file
View File

42
tests/unit/common/services/network/test_net_utils.py Normal file
View File

@ -0,0 +1,42 @@
# All Rights Reserved.
#
# 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 unittest import mock
from rally_openstack.common.services.network import net_utils
from tests.unit import test
PATH = "rally_openstack.common.services.network.net_utils"
class FunctionsTestCase(test.TestCase):
def test_generate_cidr(self):
with mock.patch("%s._IPv4_CIDR_INCR" % PATH, iter(range(1, 4))):
self.assertEqual((4, "10.2.1.0/24"), net_utils.generate_cidr())
self.assertEqual((4, "10.2.2.0/24"), net_utils.generate_cidr())
self.assertEqual((4, "10.2.3.0/24"), net_utils.generate_cidr())
with mock.patch("%s._IPv4_CIDR_INCR" % PATH, iter(range(1, 4))):
start_cidr = "1.1.0.0/26"
self.assertEqual(
(4, "1.1.0.64/26"),
net_utils.generate_cidr(start_cidr=start_cidr))
self.assertEqual(
(4, "1.1.0.128/26"),
net_utils.generate_cidr(start_cidr=start_cidr))
self.assertEqual(
(4, "1.1.0.192/26"),
net_utils.generate_cidr(start_cidr=start_cidr))

1237
tests/unit/common/services/network/test_neutron.py Normal file
View File

File diff suppressed because it is too large Load Diff

439
tests/unit/common/wrappers/test_network.py
View File

@ -35,84 +35,87 @@ class Owner(utils.RandomNameGeneratorMixin):
@ddt.ddt
class NeutronWrapperTestCase(test.TestCase):
def setUp(self):
super(NeutronWrapperTestCase, self).setUp()
self.owner = Owner()
self.owner.generate_random_name = mock.Mock()
super(NeutronWrapperTestCase, self).setUp()
def get_wrapper(self, *skip_cidrs, **kwargs):
return network.NeutronWrapper(mock.Mock(), self.owner, config=kwargs)
self.wrapper = network.NeutronWrapper(mock.MagicMock(),
self.owner,
config={})
self._nc = self.wrapper.neutron.client
def test_SUBNET_IP_VERSION(self):
self.assertEqual(4, network.NeutronWrapper.SUBNET_IP_VERSION)
@mock.patch("rally_openstack.common.wrappers.network.generate_cidr")
@mock.patch(
"rally_openstack.common.services.network.net_utils.generate_cidr")
def test__generate_cidr(self, mock_generate_cidr):
cidrs = iter(range(5))
mock_generate_cidr.side_effect = (
lambda start_cidr: start_cidr + next(cidrs)
)
service = self.get_wrapper(start_cidr=3)
self.assertEqual(3, service._generate_cidr())
self.assertEqual(4, service._generate_cidr())
self.assertEqual(5, service._generate_cidr())
self.assertEqual(6, service._generate_cidr())
self.assertEqual(7, service._generate_cidr())
self.assertEqual([mock.call(start_cidr=3)] * 5,
mock_generate_cidr.mock_calls)
def fake_gen_cidr(ip_version=None, start_cidr=None):
return 4, 3 + next(cidrs)
mock_generate_cidr.side_effect = fake_gen_cidr
self.assertEqual(3, self.wrapper._generate_cidr())
self.assertEqual(4, self.wrapper._generate_cidr())
self.assertEqual(5, self.wrapper._generate_cidr())
self.assertEqual(6, self.wrapper._generate_cidr())
self.assertEqual(7, self.wrapper._generate_cidr())
self.assertEqual([mock.call(start_cidr=self.wrapper.start_cidr)] * 5,
mock_generate_cidr.call_args_list)
def test_external_networks(self):
wrap = self.get_wrapper()
wrap.client.list_networks.return_value = {"networks": "foo_networks"}
self.assertEqual("foo_networks", wrap.external_networks)
wrap.client.list_networks.assert_called_once_with(
self._nc.list_networks.return_value = {"networks": "foo_networks"}
self.assertEqual("foo_networks", self.wrapper.external_networks)
self._nc.list_networks.assert_called_once_with(
**{"router:external": True})
def test_get_network(self):
wrap = self.get_wrapper()
neutron_net = {"id": "foo_id",
"name": self.owner.generate_random_name.return_value,
"name": "foo_name",
"tenant_id": "foo_tenant",
"status": "foo_status",
"router:external": "foo_external",
"subnets": "foo_subnets"}
expected_net = {"id": "foo_id",
"name": self.owner.generate_random_name.return_value,
"name": "foo_name",
"tenant_id": "foo_tenant",
"status": "foo_status",
"external": "foo_external",
"router_id": None,
"subnets": "foo_subnets"}
wrap.client.show_network.return_value = {"network": neutron_net}
net = wrap.get_network(net_id="foo_id")
self._nc.show_network.return_value = {"network": neutron_net}
net = self.wrapper.get_network(net_id="foo_id")
self.assertEqual(expected_net, net)
wrap.client.show_network.assert_called_once_with("foo_id")
self._nc.show_network.assert_called_once_with("foo_id")
wrap.client.show_network.side_effect = (
self._nc.show_network.side_effect = (
neutron_exceptions.NeutronClientException)
self.assertRaises(network.NetworkWrapperException, wrap.get_network,
self.assertRaises(network.NetworkWrapperException,
self.wrapper.get_network,
net_id="foo_id")
wrap.client.list_networks.return_value = {"networks": [neutron_net]}
net = wrap.get_network(name="foo_name")
self._nc.list_networks.return_value = {"networks": [neutron_net]}
net = self.wrapper.get_network(name="foo_name")
self.assertEqual(expected_net, net)
wrap.client.list_networks.assert_called_once_with(name="foo_name")
self._nc.list_networks.assert_called_once_with(name="foo_name")
wrap.client.list_networks.return_value = {"networks": []}
self.assertRaises(network.NetworkWrapperException, wrap.get_network,
self._nc.list_networks.return_value = {"networks": []}
self.assertRaises(network.NetworkWrapperException,
self.wrapper.get_network,
name="foo_name")
def test_create_v1_pool(self):
subnet = "subnet_id"
tenant = "foo_tenant"
service = self.get_wrapper()
expected_pool = {"pool": {
"id": "pool_id",
"name": self.owner.generate_random_name.return_value,
"subnet_id": subnet,
"tenant_id": tenant}}
service.client.create_pool.return_value = expected_pool
resultant_pool = service.create_v1_pool(tenant, subnet)
service.client.create_pool.assert_called_once_with({
self.wrapper.client.create_pool.return_value = expected_pool
resultant_pool = self.wrapper.create_v1_pool(tenant, subnet)
self.wrapper.client.create_pool.assert_called_once_with({
"pool": {"lb_method": "ROUND_ROBIN",
"subnet_id": subnet,
"tenant_id": tenant,
@ -121,13 +124,12 @@ class NeutronWrapperTestCase(test.TestCase):
self.assertEqual(expected_pool, resultant_pool)
def test_create_network(self):
service = self.get_wrapper()
service.client.create_network.return_value = {
self._nc.create_network.return_value = {
"network": {"id": "foo_id",
"name": self.owner.generate_random_name.return_value,
"status": "foo_status"}}
net = service.create_network("foo_tenant")
service.client.create_network.assert_called_once_with({
net = self.wrapper.create_network("foo_tenant")
self._nc.create_network.assert_called_once_with({
"network": {"tenant_id": "foo_tenant",
"name": self.owner.generate_random_name.return_value}})
self.assertEqual({"id": "foo_id",
@ -140,22 +142,18 @@ class NeutronWrapperTestCase(test.TestCase):
def test_create_network_with_subnets(self):
subnets_num = 4
service = self.get_wrapper()
subnets_cidrs = iter(range(subnets_num))
subnets_ids = iter(range(subnets_num))
service._generate_cidr = mock.Mock(
side_effect=lambda v: "cidr-%d" % next(subnets_cidrs))
service.client.create_subnet = mock.Mock(
side_effect=lambda i: {
"subnet": {"id": "subnet-%d" % next(subnets_ids)}})
service.client.create_network.return_value = {
self._nc.create_subnet.side_effect = lambda i: {
"subnet": {"id": "subnet-%d" % next(subnets_ids)}}
self._nc.create_network.return_value = {
"network": {"id": "foo_id",
"name": self.owner.generate_random_name.return_value,
"status": "foo_status"}}
net = service.create_network("foo_tenant", subnets_num=subnets_num)
net = self.wrapper.create_network("foo_tenant",
subnets_num=subnets_num)
service.client.create_network.assert_called_once_with({
self._nc.create_network.assert_called_once_with({
"network": {"tenant_id": "foo_tenant",
"name": self.owner.generate_random_name.return_value}})
self.assertEqual({"id": "foo_id",
@ -167,25 +165,24 @@ class NeutronWrapperTestCase(test.TestCase):
"subnets": ["subnet-%d" % i
for i in range(subnets_num)]}, net)
self.assertEqual(
service.client.create_subnet.mock_calls,
[mock.call({"subnet":
{"name": self.owner.generate_random_name.return_value,
"enable_dhcp": True,
"network_id": "foo_id",
"tenant_id": "foo_tenant",
"ip_version": service.SUBNET_IP_VERSION,
"ip_version": self.wrapper.SUBNET_IP_VERSION,
"dns_nameservers": ["8.8.8.8", "8.8.4.4"],
"cidr": "cidr-%d" % i}})
for i in range(subnets_num)])
"cidr": mock.ANY}})
for i in range(subnets_num)],
self.wrapper.client.create_subnet.call_args_list
)
def test_create_network_with_router(self):
service = self.get_wrapper()
service.create_router = mock.Mock(return_value={"id": "foo_router"})
service.client.create_network.return_value = {
self._nc.create_router.return_value = {"router": {"id": "foo_router"}}
self._nc.create_network.return_value = {
"network": {"id": "foo_id",
"name": self.owner.generate_random_name.return_value,
"status": "foo_status"}}
net = service.create_network("foo_tenant", add_router=True)
net = self.wrapper.create_network("foo_tenant", add_router=True)
self.assertEqual({"id": "foo_id",
"name": self.owner.generate_random_name.return_value,
"status": "foo_status",
@ -193,23 +190,25 @@ class NeutronWrapperTestCase(test.TestCase):
"tenant_id": "foo_tenant",
"router_id": "foo_router",
"subnets": []}, net)
service.create_router.assert_called_once_with(external=True,
tenant_id="foo_tenant")
self._nc.create_router.assert_called_once_with({
"router": {
"name": self.owner.generate_random_name(),
"tenant_id": "foo_tenant"
}
})
def test_create_network_with_router_and_subnets(self):
subnets_num = 4
service = self.get_wrapper()
service._generate_cidr = mock.Mock(return_value="foo_cidr")
service.create_router = mock.Mock(return_value={"id": "foo_router"})
service.client.create_subnet = mock.Mock(
return_value={"subnet": {"id": "foo_subnet"}})
service.client.create_network.return_value = {
self.wrapper._generate_cidr = mock.Mock(return_value="foo_cidr")
self._nc.create_router.return_value = {"router": {"id": "foo_router"}}
self._nc.create_subnet.return_value = {"subnet": {"id": "foo_subnet"}}
self._nc.create_network.return_value = {
"network": {"id": "foo_id",
"name": self.owner.generate_random_name.return_value,
"status": "foo_status"}}
net = service.create_network("foo_tenant", add_router=True,
subnets_num=subnets_num,
dns_nameservers=["foo_nameservers"])
net = self.wrapper.create_network(
"foo_tenant", add_router=True, subnets_num=subnets_num,
dns_nameservers=["foo_nameservers"])
self.assertEqual({"id": "foo_id",
"name": self.owner.generate_random_name.return_value,
"status": "foo_status",
@ -217,76 +216,70 @@ class NeutronWrapperTestCase(test.TestCase):
"tenant_id": "foo_tenant",
"router_id": "foo_router",
"subnets": ["foo_subnet"] * subnets_num}, net)
service.create_router.assert_called_once_with(external=True,
tenant_id="foo_tenant")
self._nc.create_router.assert_called_once_with(
{"router": {"name": self.owner.generate_random_name.return_value,
"tenant_id": "foo_tenant"}})
self.assertEqual(
service.client.create_subnet.mock_calls,
[mock.call({"subnet":
{"name": self.owner.generate_random_name.return_value,
"enable_dhcp": True,
"network_id": "foo_id",
"tenant_id": "foo_tenant",
"ip_version": service.SUBNET_IP_VERSION,
"dns_nameservers": ["foo_nameservers"],
"cidr": "foo_cidr"}})] * subnets_num)
self.assertEqual(service.client.add_interface_router.mock_calls,
[
mock.call(
{"subnet": {
"name": self.owner.generate_random_name.return_value,
"network_id": "foo_id",
"tenant_id": "foo_tenant",
"ip_version": self.wrapper.SUBNET_IP_VERSION,
"dns_nameservers": ["foo_nameservers"],
"cidr": mock.ANY
}}
)
] * subnets_num,
self._nc.create_subnet.call_args_list,
)
self.assertEqual(self._nc.add_interface_router.call_args_list,
[mock.call("foo_router", {"subnet_id": "foo_subnet"})
for i in range(subnets_num)])
@mock.patch("rally_openstack.common.wrappers.network.NeutronWrapper"
".supports_extension", return_value=(False, ""))
def test_delete_network(self, mock_neutron_wrapper_supports_extension):
service = self.get_wrapper()
service.client.list_ports.return_value = {"ports": []}
service.client.list_subnets.return_value = {"subnets": []}
service.client.delete_network.return_value = "foo_deleted"
result = service.delete_network({"id": "foo_id", "router_id": None,
"subnets": []})
self.assertEqual("foo_deleted", result)
self.assertEqual([], service.client.remove_gateway_router.mock_calls)
self.assertEqual(
[], service.client.remove_interface_router.mock_calls)
self.assertEqual([], service.client.delete_router.mock_calls)
self.assertEqual([], service.client.delete_subnet.mock_calls)
service.client.delete_network.assert_called_once_with("foo_id")
def test_delete_v1_pool(self):
service = self.get_wrapper()
pool = {"pool": {"id": "pool-id"}}
service.delete_v1_pool(pool["pool"]["id"])
service.client.delete_pool.assert_called_once_with("pool-id")
self.wrapper.delete_v1_pool(pool["pool"]["id"])
self.wrapper.client.delete_pool.assert_called_once_with("pool-id")
@mock.patch("rally_openstack.common.wrappers.network.NeutronWrapper"
".supports_extension", return_value=(True, ""))
def test_delete_network_with_dhcp_and_router_and_ports_and_subnets(
self, mock_neutron_wrapper_supports_extension):
def test_delete_network(self):
self._nc.list_ports.return_value = {"ports": []}
self._nc.list_subnets.return_value = {"subnets": []}
self._nc.delete_network.return_value = "foo_deleted"
self.wrapper.delete_network(
{"id": "foo_id", "router_id": None, "subnets": [], "name": "x",
"status": "y", "external": False})
self.assertFalse(self._nc.remove_gateway_router.called)
self.assertFalse(self._nc.remove_interface_router.called)
self.assertFalse(self._nc.client.delete_router.called)
self.assertFalse(self._nc.client.delete_subnet.called)
self._nc.delete_network.assert_called_once_with("foo_id")
def test_delete_network_with_router_and_ports_and_subnets(self):
service = self.get_wrapper()
agents = ["foo_agent", "bar_agent"]
subnets = ["foo_subnet", "bar_subnet"]
ports = [{"id": "foo_port", "device_owner": "network:router_interface",
"device_id": "rounttter"},
{"id": "bar_port", "device_owner": "network:dhcp"}]
service.client.list_dhcp_agent_hosting_networks.return_value = (
{"agents": [{"id": agent_id} for agent_id in agents]})
service.client.list_ports.return_value = ({"ports": ports})
service.client.list_subnets.return_value = (
self._nc.list_ports.return_value = ({"ports": ports})
self._nc.list_subnets.return_value = (
{"subnets": [{"id": id_} for id_ in subnets]})
service.client.delete_network.return_value = "foo_deleted"
result = service.delete_network(
self.wrapper.delete_network(
{"id": "foo_id", "router_id": "foo_router", "subnets": subnets,
"lb_pools": []})
"lb_pools": [], "name": "foo", "status": "x", "external": False})
self.assertEqual("foo_deleted", result)
self.assertEqual(service.client.remove_gateway_router.mock_calls,
self.assertEqual(self._nc.remove_gateway_router.mock_calls,
[mock.call("foo_router")])
service.client.delete_port.assert_called_once_with(ports[1]["id"])
service.client.remove_interface_router.assert_called_once_with(
self._nc.delete_port.assert_called_once_with(ports[1]["id"])
self._nc.remove_interface_router.assert_called_once_with(
ports[0]["device_id"], {"port_id": ports[0]["id"]})
self.assertEqual(service.client.delete_subnet.mock_calls,
[mock.call(subnet_id) for subnet_id in subnets])
service.client.delete_network.assert_called_once_with("foo_id")
self.assertEqual(
[mock.call(subnet_id) for subnet_id in subnets],
self._nc.delete_subnet.call_args_list
)
self._nc.delete_network.assert_called_once_with("foo_id")
@ddt.data({"exception_type": neutron_exceptions.NotFound,
"should_raise": False},
@ -295,193 +288,153 @@ class NeutronWrapperTestCase(test.TestCase):
{"exception_type": KeyError,
"should_raise": True})
@ddt.unpack
@mock.patch("rally_openstack.common.wrappers.network.NeutronWrapper"
".supports_extension", return_value=(True, ""))
def test_delete_network_with_router_throw_exception(
self, mock_neutron_wrapper_supports_extension, exception_type,
should_raise):
self, exception_type, should_raise):
# Ensure cleanup context still move forward even
# remove_interface_router throw NotFound/BadRequest exception
service = self.get_wrapper()
service.client.remove_interface_router.side_effect = exception_type
agents = ["foo_agent", "bar_agent"]
self._nc.remove_interface_router.side_effect = exception_type
subnets = ["foo_subnet", "bar_subnet"]
ports = [{"id": "foo_port", "device_owner": "network:router_interface",
"device_id": "rounttter"},
{"id": "bar_port", "device_owner": "network:dhcp"}]
service.client.list_dhcp_agent_hosting_networks.return_value = (
{"agents": [{"id": agent_id} for agent_id in agents]})
service.client.list_ports.return_value = ({"ports": ports})
service.client.delete_network.return_value = "foo_deleted"
service.client.list_subnets.return_value = {"subnets": [
self._nc.list_ports.return_value = {"ports": ports}
self._nc.list_subnets.return_value = {"subnets": [
{"id": id_} for id_ in subnets]}
if should_raise:
self.assertRaises(exception_type, service.delete_network,
{"id": "foo_id", "router_id": "foo_router",
"subnets": subnets, "lb_pools": []})
self.assertNotEqual(service.client.delete_subnet.mock_calls,
[mock.call(subnet_id) for subnet_id in
subnets])
self.assertFalse(service.client.delete_network.called)
self.assertRaises(
exception_type, self.wrapper.delete_network,
{"id": "foo_id", "name": "foo", "router_id": "foo_router",
"subnets": subnets, "lb_pools": [], "status": "xxx",
"external": False})
self.assertFalse(self._nc.delete_subnet.called)
self.assertFalse(self._nc.delete_network.called)
else:
result = service.delete_network(
{"id": "foo_id", "router_id": "foo_router", "subnets": subnets,
"lb_pools": []})
self.wrapper.delete_network(
{"id": "foo_id", "name": "foo", "status": "xxx",
"router_id": "foo_router", "subnets": subnets,
"lb_pools": [], "external": False})
self.assertEqual("foo_deleted", result)
service.client.delete_port.assert_called_once_with(ports[1]["id"])
service.client.remove_interface_router.assert_called_once_with(
self._nc.delete_port.assert_called_once_with(ports[1]["id"])
self._nc.remove_interface_router.assert_called_once_with(
ports[0]["device_id"], {"port_id": ports[0]["id"]})
self.assertEqual(service.client.delete_subnet.mock_calls,
[mock.call(subnet_id) for subnet_id in subnets])
service.client.delete_network.assert_called_once_with("foo_id")
self.assertEqual(
[mock.call(subnet_id) for subnet_id in subnets],
self._nc.delete_subnet.call_args_list
)
self._nc.delete_network.assert_called_once_with("foo_id")
self.assertEqual(service.client.remove_gateway_router.mock_calls,
[mock.call("foo_router")])
self._nc.remove_gateway_router.assert_called_once_with(
"foo_router")
def test_list_networks(self):
service = self.get_wrapper()
service.client.list_networks.return_value = {"networks": "foo_nets"}
self.assertEqual("foo_nets", service.list_networks())
service.client.list_networks.assert_called_once_with()
self._nc.list_networks.return_value = {"networks": "foo_nets"}
self.assertEqual("foo_nets", self.wrapper.list_networks())
self._nc.list_networks.assert_called_once_with()
@mock.patch(SVC + "NeutronWrapper.external_networks")
def test_create_floating_ip(self, mock_neutron_wrapper_external_networks):
wrap = self.get_wrapper()
wrap.create_port = mock.Mock(return_value={"id": "port_id"})
wrap.client.create_floatingip = mock.Mock(
return_value={"floatingip": {"id": "fip_id",
"floating_ip_address": "fip_ip"}})
def test_create_floating_ip(self):
self._nc.create_port.return_value = {"port": {"id": "port_id"}}
self._nc.create_floatingip.return_value = {
"floatingip": {"id": "fip_id", "floating_ip_address": "fip_ip"}}
self.assertRaises(ValueError, wrap.create_floating_ip)
self.assertRaises(ValueError, self.wrapper.create_floating_ip)
mock_neutron_wrapper_external_networks.__get__ = lambda *args: []
self._nc.list_networks.return_value = {"networks": []}
self.assertRaises(network.NetworkWrapperException,
wrap.create_floating_ip, tenant_id="foo_tenant")
self.wrapper.create_floating_ip,
tenant_id="foo_tenant")
mock_neutron_wrapper_external_networks.__get__ = (
lambda *args: [{"id": "ext_id"}]
)
fip = wrap.create_floating_ip(tenant_id="foo_tenant",
port_id="port_id")
self._nc.list_networks.return_value = {"networks": [{"id": "ext_id"}]}
fip = self.wrapper.create_floating_ip(
tenant_id="foo_tenant", port_id="port_id")
self.assertEqual({"id": "fip_id", "ip": "fip_ip"}, fip)
wrap.get_network = mock.Mock(
return_value={"id": "foo_net", "external": True})
wrap.create_floating_ip(tenant_id="foo_tenant", ext_network="ext_net",
port_id="port_id")
self._nc.list_networks.return_value = {"networks": [
{"id": "ext_net_id", "name": "ext_net", "router:external": True}]}
self.wrapper.create_floating_ip(
tenant_id="foo_tenant", ext_network="ext_net", port_id="port_id")
wrap.get_network = mock.Mock(
return_value={"id": "foo_net", "external": False})
wrap.create_floating_ip(tenant_id="foo_tenant", port_id="port_id")
self.assertRaises(network.NetworkWrapperException,
wrap.create_floating_ip, tenant_id="foo_tenant",
ext_network="ext_net")
self.assertRaises(
network.NetworkWrapperException,
self.wrapper.create_floating_ip, tenant_id="foo_tenant",
ext_network="ext_net_2")
def test_delete_floating_ip(self):
wrap = self.get_wrapper()
wrap.delete_floating_ip("fip_id")
wrap.delete_floating_ip("fip_id", ignored_kwarg="bar")
self.wrapper.delete_floating_ip("fip_id")
self.wrapper.delete_floating_ip("fip_id", ignored_kwarg="bar")
self.assertEqual([mock.call("fip_id")] * 2,
wrap.client.delete_floatingip.mock_calls)
self._nc.delete_floatingip.call_args_list)
@mock.patch(SVC + "NeutronWrapper.external_networks")
def test_create_router(self, mock_neutron_wrapper_external_networks):
wrap = self.get_wrapper()
wrap.client.create_router.return_value = {"router": "foo_router"}
wrap.client.list_extensions.return_value = {
def test_create_router(self):
self._nc.create_router.return_value = {"router": "foo_router"}
self._nc.list_extensions.return_value = {
"extensions": [{"alias": "ext-gw-mode"}]}
mock_neutron_wrapper_external_networks.__get__ = (
lambda *args: [{"id": "ext_id"}]
)
self._nc.list_networks.return_value = {"networks": [{"id": "ext_id"}]}
router = wrap.create_router()
wrap.client.create_router.assert_called_once_with(
router = self.wrapper.create_router()
self._nc.create_router.assert_called_once_with(
{"router": {"name": self.owner.generate_random_name.return_value}})
self.assertEqual("foo_router", router)
router = wrap.create_router(external=True, foo="bar")
wrap.client.create_router.assert_called_with(
self.wrapper.create_router(external=True, flavor_id="bar")
self._nc.create_router.assert_called_with(
{"router": {"name": self.owner.generate_random_name.return_value,
"external_gateway_info": {
"network_id": "ext_id",
"enable_snat": True},
"foo": "bar"}})
"flavor_id": "bar"}})
@mock.patch(SVC + "NeutronWrapper.external_networks")
def test_create_router_without_ext_gw_mode_extension(
self, mock_neutron_wrapper_external_networks):
wrap = self.get_wrapper()
wrap.client.create_router.return_value = {"router": "foo_router"}
wrap.client.list_extensions.return_value = {"extensions": []}
mock_neutron_wrapper_external_networks.__get__ = (
lambda *args: [{"id": "ext_id"}]
)
def test_create_router_without_ext_gw_mode_extension(self):
self._nc.create_router.return_value = {"router": "foo_router"}
self._nc.list_extensions.return_value = {"extensions": []}
self._nc.list_networks.return_value = {"networks": [{"id": "ext_id"}]}
router = wrap.create_router()
wrap.client.create_router.assert_called_once_with(
router = self.wrapper.create_router()
self._nc.create_router.assert_called_once_with(
{"router": {"name": self.owner.generate_random_name.return_value}})
self.assertEqual(router, "foo_router")
router = wrap.create_router(external=True, foo="bar")
wrap.client.create_router.assert_called_with(
self.wrapper.create_router(external=True, flavor_id="bar")
self._nc.create_router.assert_called_with(
{"router": {"name": self.owner.generate_random_name.return_value,
"external_gateway_info": {"network_id": "ext_id"},
"foo": "bar"}})
"flavor_id": "bar"}})
def test_create_port(self):
wrap = self.get_wrapper()
wrap.client.create_port.return_value = {"port": "foo_port"}
self._nc.create_port.return_value = {"port": "foo_port"}
port = wrap.create_port("foo_net")
wrap.client.create_port.assert_called_once_with(
port = self.wrapper.create_port("foo_net")
self._nc.create_port.assert_called_once_with(
{"port": {"network_id": "foo_net",
"name": self.owner.generate_random_name.return_value}})
self.assertEqual("foo_port", port)
port = wrap.create_port("foo_net", foo="bar")
wrap.client.create_port.assert_called_with(
port = self.wrapper.create_port("foo_net", foo="bar")
self.wrapper.client.create_port.assert_called_with(
{"port": {"network_id": "foo_net",
"name": self.owner.generate_random_name.return_value,
"foo": "bar"}})
def test_supports_extension(self):
wrap = self.get_wrapper()
wrap.client.list_extensions.return_value = (
self._nc.list_extensions.return_value = (
{"extensions": [{"alias": "extension"}]})
self.assertTrue(wrap.supports_extension("extension")[0])
self.assertTrue(self.wrapper.supports_extension("extension")[0])
wrap.client.list_extensions.return_value = (
self.wrapper.neutron._cached_supported_extensions = None
self._nc.list_extensions.return_value = (
{"extensions": [{"alias": "extension"}]})
self.assertFalse(wrap.supports_extension("dummy-group")[0])
self.assertFalse(self.wrapper.supports_extension("dummy-group")[0])
wrap.client.list_extensions.return_value = {}
self.assertFalse(wrap.supports_extension("extension")[0])
self.wrapper.neutron._cached_supported_extensions = None
self._nc.list_extensions.return_value = {"extensions": []}
self.assertFalse(self.wrapper.supports_extension("extension")[0])
class FunctionsTestCase(test.TestCase):
def test_generate_cidr(self):
with mock.patch("rally_openstack.common.wrappers.network.cidr_incr",
iter(range(1, 4))):
self.assertEqual("10.2.1.0/24", network.generate_cidr())
self.assertEqual("10.2.2.0/24", network.generate_cidr())
self.assertEqual("10.2.3.0/24", network.generate_cidr())
with mock.patch("rally_openstack.common.wrappers.network.cidr_incr",
iter(range(1, 4))):
start_cidr = "1.1.0.0/26"
self.assertEqual("1.1.0.64/26", network.generate_cidr(start_cidr))
self.assertEqual("1.1.0.128/26", network.generate_cidr(start_cidr))
self.assertEqual("1.1.0.192/26", network.generate_cidr(start_cidr))
def test_wrap(self):
mock_clients = mock.Mock()
mock_clients.nova().networks.list.return_value = []
config = {"fakearg": "fake"}
owner = Owner()

24
tests/unit/task/cleanup/test_resources.py
View File

@ -149,16 +149,6 @@ class NeutronMixinTestCase(test.TestCase):
neut.user = mock.MagicMock()
self.assertEqual(neut.user.neutron.return_value, neut._manager())
@mock.patch("%s.NeutronMixin._manager" % BASE)
def test_supports_extension(self, mock__manager):
mock__manager().list_extensions.return_value = {
"extensions": [{"alias": "foo"}, {"alias": "bar"}]
}
neut = self.get_neutron_mixin()
self.assertTrue(neut.supports_extension("foo"))
self.assertTrue(neut.supports_extension("bar"))
self.assertFalse(neut.supports_extension("foobar"))
def test_id(self):
neut = self.get_neutron_mixin()
neut.raw_resource = {"id": "test"}
@ -200,11 +190,12 @@ class NeutronLbaasV1MixinTestCase(test.TestCase):
def get_neutron_lbaasv1_mixin(self, extensions=None):
if extensions is None:
extensions = []
neut = resources.NeutronLbaasV1Mixin()
user = mock.MagicMock()
neut = resources.NeutronLbaasV1Mixin(user=user)
neut._service = "neutron"
neut._resource = "some_resource"
neut._manager = mock.Mock()
neut._manager().list_extensions.return_value = {
user.neutron.return_value.list_extensions.return_value = {
"extensions": [{"alias": ext} for ext in extensions]
}
return neut
@ -234,11 +225,13 @@ class NeutronLbaasV2MixinTestCase(test.TestCase):
def get_neutron_lbaasv2_mixin(self, extensions=None):
if extensions is None:
extensions = []
neut = resources.NeutronLbaasV2Mixin()
user = mock.MagicMock()
neut = resources.NeutronLbaasV2Mixin(user=user)
neut._service = "neutron"
neut._resource = "some_resource"
neut._manager = mock.Mock()
neut._manager().list_extensions.return_value = {
user.neutron.return_value.list_extensions.return_value = {
"extensions": [{"alias": ext} for ext in extensions]
}
return neut
@ -310,7 +303,8 @@ class NeutronBgpvpnTestCase(test.TestCase):
admin = mock.Mock()
neut = resources.NeutronBgpvpn(admin=admin)
neut._manager = mock.Mock()
neut._manager().list_extensions.return_value = {
nc = admin.neutron.return_value
nc.list_extensions.return_value = {
"extensions": [{"alias": ext} for ext in extensions]
}
return neut

550
tests/unit/task/scenarios/neutron/test_utils.py
View File

File diff suppressed because it is too large Load Diff

28
tests/unit/task/scenarios/nova/test_utils.py
View File

@ -534,8 +534,10 @@ class NovaScenarioTestCase(test.ScenarioTestCase):
"nova.get_console_url_server")
def test__associate_floating_ip(self):
nova_scenario = utils.NovaScenario(context=self.context)
neutronclient = nova_scenario.clients("neutron")
clients = mock.MagicMock()
nova_scenario = utils.NovaScenario(context=self.context,
clients=clients)
neutronclient = clients.neutron.return_value
neutronclient.list_ports.return_value = {"ports": [{"id": "p1"},
{"id": "p2"}]}
@ -568,8 +570,10 @@ class NovaScenarioTestCase(test.ScenarioTestCase):
"nova.associate_floating_ip", count=2)
def test__associate_floating_ip_deprecated_behavior(self):
nova_scenario = utils.NovaScenario(context=self.context)
neutronclient = nova_scenario.clients("neutron")
clients = mock.MagicMock()
nova_scenario = utils.NovaScenario(context=self.context,
clients=clients)
neutronclient = clients.neutron.return_value
neutronclient.list_ports.return_value = {"ports": [{"id": "p1"},
{"id": "p2"}]}
@ -587,7 +591,7 @@ class NovaScenarioTestCase(test.ScenarioTestCase):
)
neutronclient.list_floatingips.assert_called_once_with(
tenant_id="fake_tenant")
floating_ip_address=fip_ip)
# it is an old behavior. let's check that it was not called
self.assertFalse(self.server.add_floating_ip.called)
@ -596,8 +600,10 @@ class NovaScenarioTestCase(test.ScenarioTestCase):
"nova.associate_floating_ip")
def test__dissociate_floating_ip(self):
nova_scenario = utils.NovaScenario(context=self.context)
neutronclient = nova_scenario.clients("neutron")
clients = mock.MagicMock()
nova_scenario = utils.NovaScenario(context=self.context,
clients=clients)
neutronclient = clients.neutron.return_value
fip_ip = "172.168.0.1"
fip_id = "some"
@ -628,8 +634,10 @@ class NovaScenarioTestCase(test.ScenarioTestCase):
"nova.dissociate_floating_ip", count=2)
def test__disassociate_floating_ip_deprecated_behavior(self):
nova_scenario = utils.NovaScenario(context=self.context)
neutronclient = nova_scenario.clients("neutron")
clients = mock.MagicMock()
nova_scenario = utils.NovaScenario(context=self.context,
clients=clients)
neutronclient = clients.neutron.return_value
fip_id = "fip1"
fip_ip = "172.168.0.1"
@ -645,7 +653,7 @@ class NovaScenarioTestCase(test.ScenarioTestCase):
)
neutronclient.list_floatingips.assert_called_once_with(
tenant_id="fake_tenant")
floating_ip_address=fip_ip)
self._test_atomic_action_timer(nova_scenario.atomic_actions(),
"nova.dissociate_floating_ip")