¡@
OpenStack Study: test_router_db.py**** CubicPower OpenStack Study ****
# 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
**** CubicPower OpenStack Study **** class RouterRulesTestExtensionManager(object):
**** CubicPower OpenStack Study **** def get_resources(self): l3.RESOURCE_ATTRIBUTE_MAP['routers'].update( routerrule.EXTENDED_ATTRIBUTES_2_0['routers']) return l3.L3.get_resources()
**** CubicPower OpenStack Study **** def get_actions(self): return []
**** CubicPower OpenStack Study **** def get_request_extensions(self): return []
**** CubicPower OpenStack Study **** class DHCPOptsTestCase(test_base.BigSwitchTestBase,
test_extradhcp.TestExtraDhcpOpt):
**** CubicPower OpenStack Study **** def setUp(self, plugin=None): self.setup_patches() self.setup_config_files() super(test_extradhcp.ExtraDhcpOptDBTestCase, self).setUp(plugin=self._plugin_name)
**** CubicPower OpenStack Study **** class RouterDBTestCase(test_base.BigSwitchTestBase,
test_l3_plugin.L3NatDBIntTestCase):
**** CubicPower OpenStack Study **** 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()
**** CubicPower OpenStack Study **** def tearDown(self): super(RouterDBTestCase, self).tearDown() del test_config['config_files']
**** CubicPower OpenStack Study **** 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'])
**** CubicPower OpenStack Study **** 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'])
**** CubicPower OpenStack Study **** 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)
**** CubicPower OpenStack Study **** 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)
**** CubicPower OpenStack Study **** def test_floatingip_with_invalid_create_port(self): self._test_floatingip_with_invalid_create_port( 'neutron.plugins.bigswitch.plugin.NeutronRestProxyV2')
**** CubicPower OpenStack Study **** 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)
**** CubicPower OpenStack Study **** 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'])
**** CubicPower OpenStack Study **** def test_router_add_interface_overlapped_cidr(self): self.skipTest("Plugin does not support")
**** CubicPower OpenStack Study **** def test_router_add_interface_overlapped_cidr_returns_400(self): self.skipTest("Plugin does not support")
**** CubicPower OpenStack Study **** def test_list_nets_external(self): self.skipTest("Plugin does not support")
**** CubicPower OpenStack Study **** 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)
**** CubicPower OpenStack Study **** 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'])
**** CubicPower OpenStack Study **** 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'])
**** CubicPower OpenStack Study **** 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)
**** CubicPower OpenStack Study **** 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)
**** CubicPower OpenStack Study **** 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)
**** CubicPower OpenStack Study **** 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']))
**** CubicPower OpenStack Study **** 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']))
**** CubicPower OpenStack Study **** 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)
**** CubicPower OpenStack Study **** 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)
**** CubicPower OpenStack Study **** 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'])
**** CubicPower OpenStack Study **** 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'])
**** CubicPower OpenStack Study **** 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
**** CubicPower OpenStack Study **** def _strip_rule_ids(rules): cleaned = [] for rule in rules: del rule['id'] cleaned.append(rule) return cleaned
|