1
# Copyright 2013, Nachi Ueno, NTT MCL, Inc.
4
# Licensed under the Apache License, Version 2.0 (the "License"); you may
5
# not use this file except in compliance with the License. You may obtain
6
# a copy of the License at
8
# http://www.apache.org/licenses/LICENSE-2.0
10
# Unless required by applicable law or agreed to in writing, software
11
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
12
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
13
# License for the specific language governing permissions and limitations
18
from oslo_config import cfg
19
from oslo_log import log as logging
22
from neutron.common import constants
23
from neutron.db import extraroute_db
24
from neutron.extensions import extraroute
25
from neutron.extensions import l3
26
from neutron.openstack.common import uuidutils
27
from neutron.tests.unit import test_api_v2
28
from neutron.tests.unit import test_l3_plugin as test_l3
31
LOG = logging.getLogger(__name__)
33
_uuid = uuidutils.generate_uuid
34
_get_path = test_api_v2._get_path
37
class ExtraRouteTestExtensionManager(object):
39
def get_resources(self):
40
l3.RESOURCE_ATTRIBUTE_MAP['routers'].update(
41
extraroute.EXTENDED_ATTRIBUTES_2_0['routers'])
42
return l3.L3.get_resources()
44
def get_actions(self):
47
def get_request_extensions(self):
51
# This plugin class is for tests with plugin that integrates L3.
52
class TestExtraRouteIntPlugin(test_l3.TestL3NatIntPlugin,
53
extraroute_db.ExtraRoute_db_mixin):
54
supported_extension_aliases = ["external-net", "router", "extraroute"]
57
# A fake l3 service plugin class with extra route capability for
58
# plugins that delegate away L3 routing functionality
59
class TestExtraRouteL3NatServicePlugin(test_l3.TestL3NatServicePlugin,
60
extraroute_db.ExtraRoute_db_mixin):
61
supported_extension_aliases = ["router", "extraroute"]
64
class ExtraRouteDBTestCaseBase(object):
65
def _routes_update_prepare(self, router_id, subnet_id,
66
port_id, routes, skip_add=False):
68
self._router_interface_action('add', router_id, subnet_id, port_id)
69
self._update('routers', router_id, {'router': {'routes': routes}})
70
return self._show('routers', router_id)
72
def _routes_update_cleanup(self, port_id, subnet_id, router_id, routes):
73
self._update('routers', router_id, {'router': {'routes': routes}})
74
self._router_interface_action('remove', router_id, subnet_id, port_id)
76
def test_route_update_with_one_route(self):
77
routes = [{'destination': '135.207.0.0/16', 'nexthop': '10.0.1.3'}]
78
with self.router() as r:
79
with self.subnet(cidr='10.0.1.0/24') as s:
80
with self.port(subnet=s) as p:
81
body = self._routes_update_prepare(r['router']['id'],
82
None, p['port']['id'],
84
self.assertEqual(body['router']['routes'], routes)
85
self._routes_update_cleanup(p['port']['id'],
86
None, r['router']['id'], [])
88
def test_route_clear_routes_with_None(self):
89
routes = [{'destination': '135.207.0.0/16',
90
'nexthop': '10.0.1.3'},
91
{'destination': '12.0.0.0/8',
92
'nexthop': '10.0.1.4'},
93
{'destination': '141.212.0.0/16',
94
'nexthop': '10.0.1.5'}]
95
with self.router() as r:
96
with self.subnet(cidr='10.0.1.0/24') as s:
97
with self.port(subnet=s) as p:
98
self._routes_update_prepare(r['router']['id'],
99
None, p['port']['id'], routes)
100
body = self._update('routers', r['router']['id'],
101
{'router': {'routes': None}})
102
self.assertEqual(body['router']['routes'], [])
103
self._routes_update_cleanup(p['port']['id'],
104
None, r['router']['id'], [])
106
def test_router_interface_in_use_by_route(self):
107
routes = [{'destination': '135.207.0.0/16',
108
'nexthop': '10.0.1.3'}]
109
with self.router() as r:
110
with self.subnet(cidr='10.0.1.0/24') as s:
111
with self.port(subnet=s) as p:
112
body = self._routes_update_prepare(r['router']['id'],
113
None, p['port']['id'],
115
self.assertEqual(body['router']['routes'], routes)
116
self._router_interface_action(
121
expected_code=exc.HTTPConflict.code)
123
self._routes_update_cleanup(p['port']['id'],
124
None, r['router']['id'], [])
126
def test_route_update_with_multi_routes(self):
127
routes = [{'destination': '135.207.0.0/16',
128
'nexthop': '10.0.1.3'},
129
{'destination': '12.0.0.0/8',
130
'nexthop': '10.0.1.4'},
131
{'destination': '141.212.0.0/16',
132
'nexthop': '10.0.1.5'}]
133
with self.router() as r:
134
with self.subnet(cidr='10.0.1.0/24') as s:
135
with self.port(subnet=s) as p:
136
body = self._routes_update_prepare(r['router']['id'],
137
None, p['port']['id'],
139
self.assertEqual(sorted(body['router']['routes']),
141
self._routes_update_cleanup(p['port']['id'],
142
None, r['router']['id'], [])
144
def test_routes_update_for_multiple_routers(self):
145
routes1 = [{'destination': '135.207.0.0/16',
146
'nexthop': '10.0.0.3'}]
147
routes2 = [{'destination': '12.0.0.0/8',
148
'nexthop': '10.0.0.4'}]
149
with contextlib.nested(
152
self.subnet(cidr='10.0.0.0/24')) as (r1, r2, s):
153
with contextlib.nested(
155
self.port(subnet=s)) as (p1, p2):
156
body = self._routes_update_prepare(r1['router']['id'],
157
None, p1['port']['id'],
159
self.assertEqual(body['router']['routes'], routes1)
161
body = self._routes_update_prepare(r2['router']['id'],
162
None, p2['port']['id'],
164
self.assertEqual(body['router']['routes'], routes2)
166
self._routes_update_cleanup(p1['port']['id'],
167
None, r1['router']['id'], [])
168
self._routes_update_cleanup(p2['port']['id'],
169
None, r2['router']['id'], [])
171
def test_router_update_delete_routes(self):
172
routes_orig = [{'destination': '135.207.0.0/16',
173
'nexthop': '10.0.1.3'},
174
{'destination': '12.0.0.0/8',
175
'nexthop': '10.0.1.4'},
176
{'destination': '141.212.0.0/16',
177
'nexthop': '10.0.1.5'}]
178
routes_left = [{'destination': '135.207.0.0/16',
179
'nexthop': '10.0.1.3'},
180
{'destination': '141.212.0.0/16',
181
'nexthop': '10.0.1.5'}]
182
with self.router() as r:
183
with self.subnet(cidr='10.0.1.0/24') as s:
184
with self.port(subnet=s) as p:
185
body = self._routes_update_prepare(r['router']['id'],
186
None, p['port']['id'],
188
self.assertEqual(sorted(body['router']['routes']),
190
body = self._routes_update_prepare(r['router']['id'],
191
None, p['port']['id'],
194
self.assertEqual(sorted(body['router']['routes']),
196
self._routes_update_cleanup(p['port']['id'],
197
None, r['router']['id'], [])
199
def _test_malformed_route(self, routes):
200
with self.router() as r:
201
with self.subnet(cidr='10.0.1.0/24') as s:
202
with self.port(subnet=s) as p:
203
self._router_interface_action('add',
208
self._update('routers', r['router']['id'],
209
{'router': {'routes': routes}},
210
expected_code=exc.HTTPBadRequest.code)
212
self._router_interface_action('remove',
217
def test_no_destination_route(self):
218
self._test_malformed_route([{'nexthop': '10.0.1.6'}])
220
def test_no_nexthop_route(self):
221
self._test_malformed_route({'destination': '135.207.0.0/16'})
223
def test_none_destination(self):
224
self._test_malformed_route([{'destination': None,
225
'nexthop': '10.0.1.3'}])
227
def test_none_nexthop(self):
228
self._test_malformed_route([{'destination': '135.207.0.0/16',
231
def test_nexthop_is_port_ip(self):
232
with self.router() as r:
233
with self.subnet(cidr='10.0.1.0/24') as s:
234
with self.port(subnet=s) as p:
235
self._router_interface_action('add',
239
port_ip = p['port']['fixed_ips'][0]['ip_address']
240
routes = [{'destination': '135.207.0.0/16',
243
self._update('routers', r['router']['id'],
244
{'router': {'routes':
246
expected_code=exc.HTTPBadRequest.code)
248
self._router_interface_action('remove',
253
def test_router_update_with_too_many_routes(self):
254
with self.router() as r:
255
with self.subnet(cidr='10.0.1.0/24') as s:
256
with self.port(subnet=s) as p:
257
self._router_interface_action('add',
262
routes = [{'destination': '135.207.0.0/16',
263
'nexthop': '10.0.1.3'},
264
{'destination': '12.0.0.0/8',
265
'nexthop': '10.0.1.4'},
266
{'destination': '141.212.0.0/16',
267
'nexthop': '10.0.1.5'},
268
{'destination': '192.168.0.0/16',
269
'nexthop': '10.0.1.6'}]
271
self._update('routers', r['router']['id'],
272
{'router': {'routes':
274
expected_code=exc.HTTPBadRequest.code)
277
self._router_interface_action('remove',
282
def test_router_update_with_dup_address(self):
283
with self.router() as r:
284
with self.subnet(cidr='10.0.1.0/24') as s:
285
with self.port(subnet=s) as p:
286
self._router_interface_action('add',
291
routes = [{'destination': '135.207.0.0/16',
292
'nexthop': '10.0.1.3'},
293
{'destination': '135.207.0.0/16',
294
'nexthop': '10.0.1.3'}]
296
self._update('routers', r['router']['id'],
297
{'router': {'routes':
299
expected_code=exc.HTTPBadRequest.code)
302
self._router_interface_action('remove',
307
def test_router_update_with_invalid_ip_address(self):
308
with self.router() as r:
309
with self.subnet(cidr='10.0.1.0/24') as s:
310
with self.port(subnet=s) as p:
311
self._router_interface_action('add',
316
routes = [{'destination': '512.207.0.0/16',
317
'nexthop': '10.0.1.3'}]
319
self._update('routers', r['router']['id'],
320
{'router': {'routes':
322
expected_code=exc.HTTPBadRequest.code)
324
routes = [{'destination': '127.207.0.0/48',
325
'nexthop': '10.0.1.3'}]
327
self._update('routers', r['router']['id'],
328
{'router': {'routes':
330
expected_code=exc.HTTPBadRequest.code)
332
routes = [{'destination': 'invalid_ip_address',
333
'nexthop': '10.0.1.3'}]
335
self._update('routers', r['router']['id'],
336
{'router': {'routes':
338
expected_code=exc.HTTPBadRequest.code)
341
self._router_interface_action('remove',
346
def test_router_update_with_invalid_nexthop_ip(self):
347
with self.router() as r:
348
with self.subnet(cidr='10.0.1.0/24') as s:
349
with self.port(subnet=s) as p:
350
self._router_interface_action('add',
355
routes = [{'destination': '127.207.0.0/16',
356
'nexthop': ' 300.10.10.4'}]
358
self._update('routers', r['router']['id'],
359
{'router': {'routes':
361
expected_code=exc.HTTPBadRequest.code)
364
self._router_interface_action('remove',
369
def test_router_update_with_nexthop_is_outside_port_subnet(self):
370
with self.router() as r:
371
with self.subnet(cidr='10.0.1.0/24') as s:
372
with self.port(subnet=s) as p:
373
self._router_interface_action('add',
378
routes = [{'destination': '127.207.0.0/16',
379
'nexthop': ' 20.10.10.4'}]
381
self._update('routers', r['router']['id'],
382
{'router': {'routes':
384
expected_code=exc.HTTPBadRequest.code)
387
self._router_interface_action('remove',
392
def test_router_update_on_external_port(self):
393
with self.router() as r:
394
with self.subnet(cidr='10.0.1.0/24') as s:
395
self._set_net_external(s['subnet']['network_id'])
396
self._add_external_gateway_to_router(
398
s['subnet']['network_id'])
399
body = self._show('routers', r['router']['id'])
400
net_id = body['router']['external_gateway_info']['network_id']
401
self.assertEqual(net_id, s['subnet']['network_id'])
402
port_res = self._list_ports(
405
s['subnet']['network_id'],
406
tenant_id=r['router']['tenant_id'],
407
device_owner=constants.DEVICE_OWNER_ROUTER_GW)
408
port_list = self.deserialize('json', port_res)
409
self.assertEqual(len(port_list['ports']), 1)
411
routes = [{'destination': '135.207.0.0/16',
412
'nexthop': '10.0.1.3'}]
414
body = self._update('routers', r['router']['id'],
415
{'router': {'routes':
418
body = self._show('routers', r['router']['id'])
419
self.assertEqual(body['router']['routes'],
422
self._remove_external_gateway_from_router(
424
s['subnet']['network_id'])
425
body = self._show('routers', r['router']['id'])
426
gw_info = body['router']['external_gateway_info']
427
self.assertIsNone(gw_info)
429
def test_router_list_with_sort(self):
430
with contextlib.nested(self.router(name='router1'),
431
self.router(name='router2'),
432
self.router(name='router3')
433
) as (router1, router2, router3):
434
self._test_list_with_sort('router', (router3, router2, router1),
437
def test_router_list_with_pagination(self):
438
with contextlib.nested(self.router(name='router1'),
439
self.router(name='router2'),
440
self.router(name='router3')
441
) as (router1, router2, router3):
442
self._test_list_with_pagination('router',
443
(router1, router2, router3),
444
('name', 'asc'), 2, 2)
446
def test_router_list_with_pagination_reverse(self):
447
with contextlib.nested(self.router(name='router1'),
448
self.router(name='router2'),
449
self.router(name='router3')
450
) as (router1, router2, router3):
451
self._test_list_with_pagination_reverse('router',
454
('name', 'asc'), 2, 2)
457
class ExtraRouteDBIntTestCase(test_l3.L3NatDBIntTestCase,
458
ExtraRouteDBTestCaseBase):
460
def setUp(self, plugin=None, ext_mgr=None):
462
plugin = ('neutron.tests.unit.test_extension_extraroute.'
463
'TestExtraRouteIntPlugin')
464
# for these tests we need to enable overlapping ips
465
cfg.CONF.set_default('allow_overlapping_ips', True)
466
cfg.CONF.set_default('max_routes', 3)
467
ext_mgr = ExtraRouteTestExtensionManager()
468
super(test_l3.L3BaseForIntTests, self).setUp(plugin=plugin,
470
self.setup_notification_driver()
473
class ExtraRouteDBSepTestCase(test_l3.L3NatDBSepTestCase,
474
ExtraRouteDBTestCaseBase):
476
# the plugin without L3 support
477
plugin = 'neutron.tests.unit.test_l3_plugin.TestNoL3NatPlugin'
478
# the L3 service plugin
479
l3_plugin = ('neutron.tests.unit.test_extension_extraroute.'
480
'TestExtraRouteL3NatServicePlugin')
481
service_plugins = {'l3_plugin_name': l3_plugin}
483
# for these tests we need to enable overlapping ips
484
cfg.CONF.set_default('allow_overlapping_ips', True)
485
cfg.CONF.set_default('max_routes', 3)
486
ext_mgr = ExtraRouteTestExtensionManager()
487
super(test_l3.L3BaseForSepTests, self).setUp(
488
plugin=plugin, ext_mgr=ext_mgr,
489
service_plugins=service_plugins)
491
self.setup_notification_driver()
added
removed
neutron/tests/unit/test_extension_extraroute.py
