42683e1784
in python3.x, there is no method named xrange(), it has replaced by range(),which is equal to xrange() in python2.x. so we must fix this issue.we use six module to fix this issue. Change-Id: I9cee3f0df2d8abbf6f66d7c5a3a0fed98df622e8 Closes-Bug: #1268439
550 lines
26 KiB
Python
550 lines
26 KiB
Python
# vim: tabstop=4 shiftwidth=4 softtabstop=4
|
|
|
|
# Copyright 2013 Big Switch Networks, 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.
|
|
#
|
|
# Adapted from neutron.tests.unit.test_l3_plugin
|
|
# @author: Sumit Naiksatam, sumitnaiksatam@gmail.com
|
|
#
|
|
|
|
import contextlib
|
|
import copy
|
|
|
|
from mock import patch
|
|
from oslo.config import cfg
|
|
from six.moves import xrange
|
|
from webob import exc
|
|
|
|
from neutron.common.test_lib import test_config
|
|
from neutron import context
|
|
from neutron.extensions import l3
|
|
from neutron.manager import NeutronManager
|
|
from neutron.openstack.common import uuidutils
|
|
from neutron.plugins.bigswitch.extensions import routerrule
|
|
from neutron.tests.unit.bigswitch import fake_server
|
|
from neutron.tests.unit.bigswitch import test_base
|
|
from neutron.tests.unit import test_api_v2
|
|
from neutron.tests.unit import test_extension_extradhcpopts as test_extradhcp
|
|
from neutron.tests.unit import test_l3_plugin
|
|
|
|
|
|
_uuid = uuidutils.generate_uuid
|
|
|
|
|
|
class RouterRulesTestExtensionManager(object):
|
|
|
|
def get_resources(self):
|
|
l3.RESOURCE_ATTRIBUTE_MAP['routers'].update(
|
|
routerrule.EXTENDED_ATTRIBUTES_2_0['routers'])
|
|
return l3.L3.get_resources()
|
|
|
|
def get_actions(self):
|
|
return []
|
|
|
|
def get_request_extensions(self):
|
|
return []
|
|
|
|
|
|
class DHCPOptsTestCase(test_base.BigSwitchTestBase,
|
|
test_extradhcp.TestExtraDhcpOpt):
|
|
|
|
def setUp(self, plugin=None):
|
|
self.setup_patches()
|
|
self.setup_config_files()
|
|
super(test_extradhcp.ExtraDhcpOptDBTestCase,
|
|
self).setUp(plugin=self._plugin_name)
|
|
|
|
|
|
class RouterDBTestCase(test_base.BigSwitchTestBase,
|
|
test_l3_plugin.L3NatDBIntTestCase):
|
|
|
|
def setUp(self):
|
|
self.setup_patches()
|
|
self.setup_config_files()
|
|
ext_mgr = RouterRulesTestExtensionManager()
|
|
super(RouterDBTestCase, self).setUp(plugin=self._plugin_name,
|
|
ext_mgr=ext_mgr)
|
|
cfg.CONF.set_default('allow_overlapping_ips', False)
|
|
self.plugin_obj = NeutronManager.get_plugin()
|
|
|
|
def tearDown(self):
|
|
super(RouterDBTestCase, self).tearDown()
|
|
del test_config['config_files']
|
|
|
|
def test_router_remove_router_interface_wrong_subnet_returns_400(self):
|
|
with self.router() as r:
|
|
with self.subnet() as s:
|
|
with self.subnet(cidr='10.0.10.0/24') as s1:
|
|
with self.port(subnet=s1, no_delete=True) as p:
|
|
self._router_interface_action('add',
|
|
r['router']['id'],
|
|
None,
|
|
p['port']['id'])
|
|
self._router_interface_action('remove',
|
|
r['router']['id'],
|
|
s['subnet']['id'],
|
|
p['port']['id'],
|
|
exc.HTTPBadRequest.code)
|
|
#remove properly to clean-up
|
|
self._router_interface_action('remove',
|
|
r['router']['id'],
|
|
None,
|
|
p['port']['id'])
|
|
|
|
def test_router_remove_router_interface_wrong_port_returns_404(self):
|
|
with self.router() as r:
|
|
with self.subnet() as s:
|
|
with self.port(subnet=s, no_delete=True) as p:
|
|
self._router_interface_action('add',
|
|
r['router']['id'],
|
|
None,
|
|
p['port']['id'])
|
|
# create another port for testing failure case
|
|
res = self._create_port('json', p['port']['network_id'])
|
|
p2 = self.deserialize('json', res)
|
|
self._router_interface_action('remove',
|
|
r['router']['id'],
|
|
None,
|
|
p2['port']['id'],
|
|
exc.HTTPNotFound.code)
|
|
# remove correct interface to cleanup
|
|
self._router_interface_action('remove',
|
|
r['router']['id'],
|
|
None,
|
|
p['port']['id'])
|
|
# remove extra port created
|
|
self._delete('ports', p2['port']['id'])
|
|
|
|
def test_multi_tenant_flip_alllocation(self):
|
|
tenant1_id = _uuid()
|
|
tenant2_id = _uuid()
|
|
with contextlib.nested(
|
|
self.network(tenant_id=tenant1_id),
|
|
self.network(tenant_id=tenant2_id)) as (n1, n2):
|
|
with contextlib.nested(
|
|
self.subnet(network=n1, cidr='11.0.0.0/24'),
|
|
self.subnet(network=n2, cidr='12.0.0.0/24'),
|
|
self.subnet(cidr='13.0.0.0/24')) as (s1, s2, psub):
|
|
with contextlib.nested(
|
|
self.router(tenant_id=tenant1_id),
|
|
self.router(tenant_id=tenant2_id),
|
|
self.port(subnet=s1, tenant_id=tenant1_id),
|
|
self.port(subnet=s2, tenant_id=tenant2_id)) as (r1, r2,
|
|
p1, p2):
|
|
self._set_net_external(psub['subnet']['network_id'])
|
|
s1id = p1['port']['fixed_ips'][0]['subnet_id']
|
|
s2id = p2['port']['fixed_ips'][0]['subnet_id']
|
|
s1 = {'subnet': {'id': s1id}}
|
|
s2 = {'subnet': {'id': s2id}}
|
|
self._add_external_gateway_to_router(
|
|
r1['router']['id'],
|
|
psub['subnet']['network_id'])
|
|
self._add_external_gateway_to_router(
|
|
r2['router']['id'],
|
|
psub['subnet']['network_id'])
|
|
self._router_interface_action(
|
|
'add', r1['router']['id'],
|
|
s1['subnet']['id'], None)
|
|
self._router_interface_action(
|
|
'add', r2['router']['id'],
|
|
s2['subnet']['id'], None)
|
|
fl1 = self._make_floatingip_for_tenant_port(
|
|
net_id=psub['subnet']['network_id'],
|
|
port_id=p1['port']['id'],
|
|
tenant_id=tenant1_id)
|
|
multiFloatPatch = patch(
|
|
'httplib.HTTPConnection',
|
|
create=True,
|
|
new=fake_server.VerifyMultiTenantFloatingIP)
|
|
multiFloatPatch.start()
|
|
fl2 = self._make_floatingip_for_tenant_port(
|
|
net_id=psub['subnet']['network_id'],
|
|
port_id=p2['port']['id'],
|
|
tenant_id=tenant2_id)
|
|
multiFloatPatch.stop()
|
|
self._delete('floatingips', fl1['floatingip']['id'])
|
|
self._delete('floatingips', fl2['floatingip']['id'])
|
|
self._router_interface_action(
|
|
'remove', r1['router']['id'],
|
|
s1['subnet']['id'], None)
|
|
self._router_interface_action(
|
|
'remove', r2['router']['id'],
|
|
s2['subnet']['id'], None)
|
|
|
|
def _make_floatingip_for_tenant_port(self, net_id, port_id, tenant_id):
|
|
data = {'floatingip': {'floating_network_id': net_id,
|
|
'tenant_id': tenant_id,
|
|
'port_id': port_id}}
|
|
floatingip_req = self.new_create_request('floatingips', data, self.fmt)
|
|
res = floatingip_req.get_response(self.ext_api)
|
|
return self.deserialize(self.fmt, res)
|
|
|
|
def test_floatingip_with_invalid_create_port(self):
|
|
self._test_floatingip_with_invalid_create_port(
|
|
'neutron.plugins.bigswitch.plugin.NeutronRestProxyV2')
|
|
|
|
def test_create_floatingip_no_ext_gateway_return_404(self):
|
|
with self.subnet(cidr='10.0.10.0/24') as public_sub:
|
|
self._set_net_external(public_sub['subnet']['network_id'])
|
|
with self.port() as private_port:
|
|
with self.router():
|
|
res = self._create_floatingip(
|
|
'json',
|
|
public_sub['subnet']['network_id'],
|
|
port_id=private_port['port']['id'])
|
|
self.assertEqual(res.status_int, exc.HTTPNotFound.code)
|
|
|
|
def test_router_update_gateway(self):
|
|
with self.router() as r:
|
|
with self.subnet() as s1:
|
|
with self.subnet(cidr='10.0.10.0/24') as s2:
|
|
self._set_net_external(s1['subnet']['network_id'])
|
|
self._add_external_gateway_to_router(
|
|
r['router']['id'],
|
|
s1['subnet']['network_id'])
|
|
body = self._show('routers', r['router']['id'])
|
|
net_id = (body['router']
|
|
['external_gateway_info']['network_id'])
|
|
self.assertEqual(net_id, s1['subnet']['network_id'])
|
|
self._set_net_external(s2['subnet']['network_id'])
|
|
self._add_external_gateway_to_router(
|
|
r['router']['id'],
|
|
s2['subnet']['network_id'])
|
|
body = self._show('routers', r['router']['id'])
|
|
net_id = (body['router']
|
|
['external_gateway_info']['network_id'])
|
|
self.assertEqual(net_id, s2['subnet']['network_id'])
|
|
self._remove_external_gateway_from_router(
|
|
r['router']['id'],
|
|
s2['subnet']['network_id'])
|
|
|
|
def test_router_add_interface_overlapped_cidr(self):
|
|
self.skipTest("Plugin does not support")
|
|
|
|
def test_router_add_interface_overlapped_cidr_returns_400(self):
|
|
self.skipTest("Plugin does not support")
|
|
|
|
def test_list_nets_external(self):
|
|
self.skipTest("Plugin does not support")
|
|
|
|
def test_router_update_gateway_with_existed_floatingip(self):
|
|
with self.subnet(cidr='10.0.10.0/24') as subnet:
|
|
self._set_net_external(subnet['subnet']['network_id'])
|
|
with self.floatingip_with_assoc() as fip:
|
|
self._add_external_gateway_to_router(
|
|
fip['floatingip']['router_id'],
|
|
subnet['subnet']['network_id'],
|
|
expected_code=exc.HTTPConflict.code)
|
|
|
|
def test_router_remove_interface_wrong_subnet_returns_400(self):
|
|
with self.router() as r:
|
|
with self.subnet(cidr='10.0.10.0/24') as s:
|
|
with self.port(no_delete=True) as p:
|
|
self._router_interface_action('add',
|
|
r['router']['id'],
|
|
None,
|
|
p['port']['id'])
|
|
self._router_interface_action('remove',
|
|
r['router']['id'],
|
|
s['subnet']['id'],
|
|
p['port']['id'],
|
|
exc.HTTPBadRequest.code)
|
|
#remove properly to clean-up
|
|
self._router_interface_action('remove',
|
|
r['router']['id'],
|
|
None,
|
|
p['port']['id'])
|
|
|
|
def test_router_remove_interface_wrong_port_returns_404(self):
|
|
with self.router() as r:
|
|
with self.subnet(cidr='10.0.10.0/24'):
|
|
with self.port(no_delete=True) as p:
|
|
self._router_interface_action('add',
|
|
r['router']['id'],
|
|
None,
|
|
p['port']['id'])
|
|
# create another port for testing failure case
|
|
res = self._create_port('json', p['port']['network_id'])
|
|
p2 = self.deserialize('json', res)
|
|
self._router_interface_action('remove',
|
|
r['router']['id'],
|
|
None,
|
|
p2['port']['id'],
|
|
exc.HTTPNotFound.code)
|
|
# remove correct interface to cleanup
|
|
self._router_interface_action('remove',
|
|
r['router']['id'],
|
|
None,
|
|
p['port']['id'])
|
|
# remove extra port created
|
|
self._delete('ports', p2['port']['id'])
|
|
|
|
def test_send_data(self):
|
|
fmt = 'json'
|
|
plugin_obj = NeutronManager.get_plugin()
|
|
|
|
with self.router() as r:
|
|
r_id = r['router']['id']
|
|
|
|
with self.subnet(cidr='10.0.10.0/24') as s:
|
|
s_id = s['subnet']['id']
|
|
|
|
with self.router() as r1:
|
|
r1_id = r1['router']['id']
|
|
body = self._router_interface_action('add', r_id, s_id,
|
|
None)
|
|
self.assertIn('port_id', body)
|
|
r_port_id = body['port_id']
|
|
body = self._show('ports', r_port_id)
|
|
self.assertEqual(body['port']['device_id'], r_id)
|
|
|
|
with self.subnet(cidr='10.0.20.0/24') as s1:
|
|
s1_id = s1['subnet']['id']
|
|
body = self._router_interface_action('add', r1_id,
|
|
s1_id, None)
|
|
self.assertIn('port_id', body)
|
|
r1_port_id = body['port_id']
|
|
body = self._show('ports', r1_port_id)
|
|
self.assertEqual(body['port']['device_id'], r1_id)
|
|
|
|
with self.subnet(cidr='11.0.0.0/24') as public_sub:
|
|
public_net_id = public_sub['subnet']['network_id']
|
|
self._set_net_external(public_net_id)
|
|
|
|
with self.port() as prv_port:
|
|
prv_fixed_ip = prv_port['port']['fixed_ips'][0]
|
|
priv_sub_id = prv_fixed_ip['subnet_id']
|
|
self._add_external_gateway_to_router(
|
|
r_id, public_net_id)
|
|
self._router_interface_action('add', r_id,
|
|
priv_sub_id,
|
|
None)
|
|
|
|
priv_port_id = prv_port['port']['id']
|
|
res = self._create_floatingip(
|
|
fmt, public_net_id,
|
|
port_id=priv_port_id)
|
|
self.assertEqual(res.status_int,
|
|
exc.HTTPCreated.code)
|
|
floatingip = self.deserialize(fmt, res)
|
|
|
|
result = plugin_obj._send_all_data()
|
|
self.assertEqual(result[0], 200)
|
|
|
|
self._delete('floatingips',
|
|
floatingip['floatingip']['id'])
|
|
self._remove_external_gateway_from_router(
|
|
r_id, public_net_id)
|
|
self._router_interface_action('remove', r_id,
|
|
priv_sub_id,
|
|
None)
|
|
self._router_interface_action('remove', r_id, s_id,
|
|
None)
|
|
self._show('ports', r_port_id,
|
|
expected_code=exc.HTTPNotFound.code)
|
|
self._router_interface_action('remove', r1_id, s1_id,
|
|
None)
|
|
self._show('ports', r1_port_id,
|
|
expected_code=exc.HTTPNotFound.code)
|
|
|
|
def test_router_rules_update(self):
|
|
with self.router() as r:
|
|
r_id = r['router']['id']
|
|
router_rules = [{'destination': '1.2.3.4/32',
|
|
'source': '4.3.2.1/32',
|
|
'action': 'permit',
|
|
'nexthops': ['4.4.4.4', '4.4.4.5']}]
|
|
body = self._update('routers', r_id,
|
|
{'router': {'router_rules': router_rules}})
|
|
|
|
body = self._show('routers', r['router']['id'])
|
|
self.assertIn('router_rules', body['router'])
|
|
rules = body['router']['router_rules']
|
|
self.assertEqual(_strip_rule_ids(rules), router_rules)
|
|
# Try after adding another rule
|
|
router_rules.append({'source': 'external',
|
|
'destination': '8.8.8.8/32',
|
|
'action': 'permit', 'nexthops': []})
|
|
body = self._update('routers', r['router']['id'],
|
|
{'router': {'router_rules': router_rules}})
|
|
|
|
body = self._show('routers', r['router']['id'])
|
|
self.assertIn('router_rules', body['router'])
|
|
rules = body['router']['router_rules']
|
|
self.assertEqual(_strip_rule_ids(rules), router_rules)
|
|
|
|
def test_router_rules_separation(self):
|
|
with self.router() as r1:
|
|
with self.router() as r2:
|
|
r1_id = r1['router']['id']
|
|
r2_id = r2['router']['id']
|
|
router1_rules = [{'destination': '5.6.7.8/32',
|
|
'source': '8.7.6.5/32',
|
|
'action': 'permit',
|
|
'nexthops': ['8.8.8.8', '9.9.9.9']}]
|
|
router2_rules = [{'destination': '1.2.3.4/32',
|
|
'source': '4.3.2.1/32',
|
|
'action': 'permit',
|
|
'nexthops': ['4.4.4.4', '4.4.4.5']}]
|
|
body1 = self._update('routers', r1_id,
|
|
{'router':
|
|
{'router_rules': router1_rules}})
|
|
body2 = self._update('routers', r2_id,
|
|
{'router':
|
|
{'router_rules': router2_rules}})
|
|
|
|
body1 = self._show('routers', r1_id)
|
|
body2 = self._show('routers', r2_id)
|
|
rules1 = body1['router']['router_rules']
|
|
rules2 = body2['router']['router_rules']
|
|
self.assertEqual(_strip_rule_ids(rules1), router1_rules)
|
|
self.assertEqual(_strip_rule_ids(rules2), router2_rules)
|
|
|
|
def test_router_rules_validation(self):
|
|
with self.router() as r:
|
|
r_id = r['router']['id']
|
|
good_rules = [{'destination': '1.2.3.4/32',
|
|
'source': '4.3.2.1/32',
|
|
'action': 'permit',
|
|
'nexthops': ['4.4.4.4', '4.4.4.5']}]
|
|
|
|
body = self._update('routers', r_id,
|
|
{'router': {'router_rules': good_rules}})
|
|
body = self._show('routers', r_id)
|
|
self.assertIn('router_rules', body['router'])
|
|
self.assertEqual(good_rules,
|
|
_strip_rule_ids(body['router']['router_rules']))
|
|
|
|
# Missing nexthops should be populated with an empty list
|
|
light_rules = copy.deepcopy(good_rules)
|
|
del light_rules[0]['nexthops']
|
|
body = self._update('routers', r_id,
|
|
{'router': {'router_rules': light_rules}})
|
|
body = self._show('routers', r_id)
|
|
self.assertIn('router_rules', body['router'])
|
|
light_rules[0]['nexthops'] = []
|
|
self.assertEqual(light_rules,
|
|
_strip_rule_ids(body['router']['router_rules']))
|
|
# bad CIDR
|
|
bad_rules = copy.deepcopy(good_rules)
|
|
bad_rules[0]['destination'] = '1.1.1.1'
|
|
body = self._update('routers', r_id,
|
|
{'router': {'router_rules': bad_rules}},
|
|
expected_code=exc.HTTPBadRequest.code)
|
|
# bad next hop
|
|
bad_rules = copy.deepcopy(good_rules)
|
|
bad_rules[0]['nexthops'] = ['1.1.1.1', 'f2']
|
|
body = self._update('routers', r_id,
|
|
{'router': {'router_rules': bad_rules}},
|
|
expected_code=exc.HTTPBadRequest.code)
|
|
# bad action
|
|
bad_rules = copy.deepcopy(good_rules)
|
|
bad_rules[0]['action'] = 'dance'
|
|
body = self._update('routers', r_id,
|
|
{'router': {'router_rules': bad_rules}},
|
|
expected_code=exc.HTTPBadRequest.code)
|
|
# duplicate rule with opposite action
|
|
bad_rules = copy.deepcopy(good_rules)
|
|
bad_rules.append(copy.deepcopy(bad_rules[0]))
|
|
bad_rules.append(copy.deepcopy(bad_rules[0]))
|
|
bad_rules[1]['source'] = 'any'
|
|
bad_rules[2]['action'] = 'deny'
|
|
body = self._update('routers', r_id,
|
|
{'router': {'router_rules': bad_rules}},
|
|
expected_code=exc.HTTPBadRequest.code)
|
|
# duplicate nexthop
|
|
bad_rules = copy.deepcopy(good_rules)
|
|
bad_rules[0]['nexthops'] = ['1.1.1.1', '1.1.1.1']
|
|
body = self._update('routers', r_id,
|
|
{'router': {'router_rules': bad_rules}},
|
|
expected_code=exc.HTTPBadRequest.code)
|
|
# make sure light rules persisted during bad updates
|
|
body = self._show('routers', r_id)
|
|
self.assertIn('router_rules', body['router'])
|
|
self.assertEqual(light_rules,
|
|
_strip_rule_ids(body['router']['router_rules']))
|
|
|
|
def test_router_rules_config_change(self):
|
|
cfg.CONF.set_override('tenant_default_router_rule',
|
|
['*:any:any:deny',
|
|
'*:8.8.8.8/32:any:permit:1.2.3.4'],
|
|
'ROUTER')
|
|
with self.router() as r:
|
|
body = self._show('routers', r['router']['id'])
|
|
expected_rules = [{'source': 'any', 'destination': 'any',
|
|
'nexthops': [], 'action': 'deny'},
|
|
{'source': '8.8.8.8/32', 'destination': 'any',
|
|
'nexthops': ['1.2.3.4'], 'action': 'permit'}]
|
|
self.assertEqual(expected_rules,
|
|
_strip_rule_ids(body['router']['router_rules']))
|
|
|
|
def test_rule_exhaustion(self):
|
|
cfg.CONF.set_override('max_router_rules', 10, 'ROUTER')
|
|
with self.router() as r:
|
|
rules = []
|
|
for i in xrange(1, 12):
|
|
rule = {'source': 'any', 'nexthops': [],
|
|
'destination': '1.1.1.' + str(i) + '/32',
|
|
'action': 'permit'}
|
|
rules.append(rule)
|
|
self._update('routers', r['router']['id'],
|
|
{'router': {'router_rules': rules}},
|
|
expected_code=exc.HTTPBadRequest.code)
|
|
|
|
def test_rollback_on_router_create(self):
|
|
tid = test_api_v2._uuid()
|
|
self.errhttpPatch = patch('httplib.HTTPConnection', create=True,
|
|
new=fake_server.HTTPConnectionMock500)
|
|
self.errhttpPatch.start()
|
|
self._create_router('json', tid)
|
|
self.errhttpPatch.stop()
|
|
self.assertTrue(len(self._get_routers(tid)) == 0)
|
|
|
|
def test_rollback_on_router_update(self):
|
|
with self.router() as r:
|
|
data = {'router': {'name': 'aNewName'}}
|
|
self.errhttpPatch = patch('httplib.HTTPConnection', create=True,
|
|
new=fake_server.HTTPConnectionMock500)
|
|
self.errhttpPatch.start()
|
|
self.new_update_request('routers', data,
|
|
r['router']['id']).get_response(self.api)
|
|
self.errhttpPatch.stop()
|
|
updatedr = self._get_routers(r['router']['tenant_id'])[0]
|
|
# name should have stayed the same due to failure
|
|
self.assertEqual(r['router']['name'], updatedr['name'])
|
|
|
|
def test_rollback_on_router_delete(self):
|
|
with self.router() as r:
|
|
self.errhttpPatch = patch('httplib.HTTPConnection', create=True,
|
|
new=fake_server.HTTPConnectionMock500)
|
|
self.errhttpPatch.start()
|
|
self._delete('routers', r['router']['id'],
|
|
expected_code=exc.HTTPInternalServerError.code)
|
|
self.errhttpPatch.stop()
|
|
self.assertEqual(r['router']['id'],
|
|
self._get_routers(r['router']['tenant_id']
|
|
)[0]['id'])
|
|
|
|
def _get_routers(self, tenant_id):
|
|
ctx = context.Context('', tenant_id)
|
|
return self.plugin_obj.get_routers(ctx)
|
|
|
|
|
|
def _strip_rule_ids(rules):
|
|
cleaned = []
|
|
for rule in rules:
|
|
del rule['id']
|
|
cleaned.append(rule)
|
|
return cleaned
|