4c207ae082
This change allows to use network context without admin user Change-Id: Ia25148e08e33d80ea9d021d5e16f87153b902d94
447 lines
20 KiB
Python
447 lines
20 KiB
Python
# Copyright 2014: Mirantis Inc.
|
|
# 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
|
|
|
|
import ddt
|
|
from neutronclient.common import exceptions as neutron_exceptions
|
|
|
|
from rally.common import utils
|
|
|
|
from rally_openstack.common import consts
|
|
from rally_openstack.common.wrappers import network
|
|
from tests.unit import test
|
|
|
|
|
|
SVC = "rally_openstack.common.wrappers.network."
|
|
|
|
|
|
class Owner(utils.RandomNameGeneratorMixin):
|
|
task = {"uuid": "task-uuid"}
|
|
|
|
|
|
@ddt.ddt
|
|
class NeutronWrapperTestCase(test.TestCase):
|
|
def setUp(self):
|
|
super(NeutronWrapperTestCase, self).setUp()
|
|
self.owner = Owner()
|
|
self.owner.generate_random_name = mock.Mock()
|
|
clients = mock.MagicMock()
|
|
clients.credential.permission = consts.EndpointPermission.ADMIN
|
|
self.wrapper = network.NeutronWrapper(
|
|
clients, 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.services.network.net_utils.generate_cidr")
|
|
def test__generate_cidr(self, mock_generate_cidr):
|
|
cidrs = iter(range(5))
|
|
|
|
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):
|
|
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):
|
|
neutron_net = {"id": "foo_id",
|
|
"name": "foo_name",
|
|
"tenant_id": "foo_tenant",
|
|
"status": "foo_status",
|
|
"router:external": "foo_external",
|
|
"subnets": "foo_subnets"}
|
|
expected_net = {"id": "foo_id",
|
|
"name": "foo_name",
|
|
"tenant_id": "foo_tenant",
|
|
"status": "foo_status",
|
|
"external": "foo_external",
|
|
"router_id": None,
|
|
"subnets": "foo_subnets"}
|
|
self._nc.show_network.return_value = {"network": neutron_net}
|
|
net = self.wrapper.get_network(net_id="foo_id")
|
|
self.assertEqual(expected_net, net)
|
|
self._nc.show_network.assert_called_once_with("foo_id")
|
|
|
|
self._nc.show_network.side_effect = (
|
|
neutron_exceptions.NeutronClientException)
|
|
self.assertRaises(network.NetworkWrapperException,
|
|
self.wrapper.get_network,
|
|
net_id="foo_id")
|
|
|
|
self._nc.list_networks.return_value = {"networks": [neutron_net]}
|
|
net = self.wrapper.get_network(name="foo_name")
|
|
self.assertEqual(expected_net, net)
|
|
self._nc.list_networks.assert_called_once_with(name="foo_name")
|
|
|
|
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"
|
|
expected_pool = {"pool": {
|
|
"id": "pool_id",
|
|
"name": self.owner.generate_random_name.return_value,
|
|
"subnet_id": subnet,
|
|
"tenant_id": tenant}}
|
|
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,
|
|
"protocol": "HTTP",
|
|
"name": self.owner.generate_random_name.return_value}})
|
|
self.assertEqual(expected_pool, resultant_pool)
|
|
|
|
def test_create_network(self):
|
|
self._nc.create_network.return_value = {
|
|
"network": {"id": "foo_id",
|
|
"name": self.owner.generate_random_name.return_value,
|
|
"status": "foo_status"}}
|
|
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",
|
|
"name": self.owner.generate_random_name.return_value,
|
|
"status": "foo_status",
|
|
"external": False,
|
|
"tenant_id": "foo_tenant",
|
|
"router_id": None,
|
|
"subnets": []}, net)
|
|
|
|
def test_create_network_with_subnets(self):
|
|
subnets_num = 4
|
|
subnets_ids = iter(range(subnets_num))
|
|
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 = self.wrapper.create_network("foo_tenant",
|
|
subnets_num=subnets_num)
|
|
|
|
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",
|
|
"name": self.owner.generate_random_name.return_value,
|
|
"status": "foo_status",
|
|
"external": False,
|
|
"router_id": None,
|
|
"tenant_id": "foo_tenant",
|
|
"subnets": ["subnet-%d" % i
|
|
for i in range(subnets_num)]}, net)
|
|
self.assertEqual(
|
|
[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": ["8.8.8.8", "8.8.4.4"],
|
|
"cidr": mock.ANY}})
|
|
for i in range(subnets_num)],
|
|
self.wrapper.client.create_subnet.call_args_list
|
|
)
|
|
|
|
def test_create_network_with_router(self):
|
|
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 = 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",
|
|
"external": False,
|
|
"tenant_id": "foo_tenant",
|
|
"router_id": "foo_router",
|
|
"subnets": []}, net)
|
|
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
|
|
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 = 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",
|
|
"external": False,
|
|
"tenant_id": "foo_tenant",
|
|
"router_id": "foo_router",
|
|
"subnets": ["foo_subnet"] * subnets_num}, net)
|
|
self._nc.create_router.assert_called_once_with(
|
|
{"router": {"name": self.owner.generate_random_name.return_value,
|
|
"tenant_id": "foo_tenant"}})
|
|
self.assertEqual(
|
|
[
|
|
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)])
|
|
|
|
def test_delete_v1_pool(self):
|
|
pool = {"pool": {"id": "pool-id"}}
|
|
self.wrapper.delete_v1_pool(pool["pool"]["id"])
|
|
self.wrapper.client.delete_pool.assert_called_once_with("pool-id")
|
|
|
|
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):
|
|
|
|
subnets = ["foo_subnet", "bar_subnet"]
|
|
ports = [{"id": "foo_port", "device_owner": "network:router_interface",
|
|
"device_id": "rounttter"},
|
|
{"id": "bar_port", "device_owner": "network:dhcp"}]
|
|
self._nc.list_ports.return_value = ({"ports": ports})
|
|
self._nc.list_subnets.return_value = (
|
|
{"subnets": [{"id": id_} for id_ in subnets]})
|
|
|
|
self.wrapper.delete_network(
|
|
{"id": "foo_id", "router_id": "foo_router", "subnets": subnets,
|
|
"lb_pools": [], "name": "foo", "status": "x", "external": False})
|
|
|
|
self.assertEqual(self._nc.remove_gateway_router.mock_calls,
|
|
[mock.call("foo_router")])
|
|
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(
|
|
[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},
|
|
{"exception_type": neutron_exceptions.BadRequest,
|
|
"should_raise": False},
|
|
{"exception_type": KeyError,
|
|
"should_raise": True})
|
|
@ddt.unpack
|
|
def test_delete_network_with_router_throw_exception(
|
|
self, exception_type, should_raise):
|
|
# Ensure cleanup context still move forward even
|
|
# remove_interface_router throw NotFound/BadRequest exception
|
|
|
|
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"}]
|
|
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, 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:
|
|
self.wrapper.delete_network(
|
|
{"id": "foo_id", "name": "foo", "status": "xxx",
|
|
"router_id": "foo_router", "subnets": subnets,
|
|
"lb_pools": [], "external": False})
|
|
|
|
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(
|
|
[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._nc.remove_gateway_router.assert_called_once_with(
|
|
"foo_router")
|
|
|
|
def test_list_networks(self):
|
|
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()
|
|
|
|
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, self.wrapper.create_floating_ip)
|
|
|
|
self._nc.list_networks.return_value = {"networks": []}
|
|
self.assertRaises(network.NetworkWrapperException,
|
|
self.wrapper.create_floating_ip,
|
|
tenant_id="foo_tenant")
|
|
|
|
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)
|
|
|
|
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")
|
|
|
|
self.assertRaises(
|
|
network.NetworkWrapperException,
|
|
self.wrapper.create_floating_ip, tenant_id="foo_tenant",
|
|
ext_network="ext_net_2")
|
|
|
|
def test_delete_floating_ip(self):
|
|
self.wrapper.delete_floating_ip("fip_id")
|
|
self.wrapper.delete_floating_ip("fip_id", ignored_kwarg="bar")
|
|
self.assertEqual([mock.call("fip_id")] * 2,
|
|
self._nc.delete_floatingip.call_args_list)
|
|
|
|
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"}]}
|
|
self._nc.list_networks.return_value = {"networks": [{"id": "ext_id"}]}
|
|
|
|
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)
|
|
|
|
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},
|
|
"flavor_id": "bar"}})
|
|
|
|
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 = 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")
|
|
|
|
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"},
|
|
"flavor_id": "bar"}})
|
|
|
|
def test_create_port(self):
|
|
self._nc.create_port.return_value = {"port": "foo_port"}
|
|
|
|
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 = 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):
|
|
self._nc.list_extensions.return_value = (
|
|
{"extensions": [{"alias": "extension"}]})
|
|
self.assertTrue(self.wrapper.supports_extension("extension")[0])
|
|
|
|
self.wrapper.neutron._cached_supported_extensions = None
|
|
self._nc.list_extensions.return_value = (
|
|
{"extensions": [{"alias": "extension"}]})
|
|
self.assertFalse(self.wrapper.supports_extension("dummy-group")[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_wrap(self):
|
|
mock_clients = mock.Mock()
|
|
config = {"fakearg": "fake"}
|
|
owner = Owner()
|
|
|
|
mock_clients.services.return_value = {"foo": consts.Service.NEUTRON}
|
|
wrapper = network.wrap(mock_clients, owner, config)
|
|
self.assertIsInstance(wrapper, network.NeutronWrapper)
|
|
self.assertEqual(wrapper.owner, owner)
|
|
self.assertEqual(wrapper.config, config)
|