351 lines
16 KiB
Python
351 lines
16 KiB
Python
# Copyright (c) 2019 Intel Corporation.
|
|
#
|
|
# 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 mock
|
|
from neutron_lib import constants
|
|
from neutron_lib import context
|
|
from oslo_config import cfg
|
|
import webob.exc
|
|
|
|
from neutron.db import db_base_plugin_v2
|
|
from neutron.db import segments_db
|
|
from neutron.extensions import network_segment_range as ext_range
|
|
from neutron.services.network_segment_range import plugin as plugin_range
|
|
from neutron.tests.unit.db import test_db_base_plugin_v2
|
|
|
|
SERVICE_PLUGIN_KLASS = ('neutron.services.network_segment_range.plugin.'
|
|
'NetworkSegmentRangePlugin')
|
|
TEST_PLUGIN_KLASS = (
|
|
'neutron.tests.unit.extensions.test_network_segment_range.'
|
|
'NetworkSegmentRangeTestPlugin')
|
|
TARGET_PLUGIN = 'neutron.plugins.ml2.plugin.Ml2Plugin'
|
|
|
|
|
|
class NetworkSegmentRangeExtensionManager(object):
|
|
|
|
def get_resources(self):
|
|
return ext_range.Network_segment_range.get_resources()
|
|
|
|
def get_actions(self):
|
|
return []
|
|
|
|
def get_request_extensions(self):
|
|
return []
|
|
|
|
|
|
class NetworkSegmentRangeTestBase(test_db_base_plugin_v2.
|
|
NeutronDbPluginV2TestCase):
|
|
|
|
def _create_network_segment_range(self, fmt, expected_res_status=None,
|
|
**kwargs):
|
|
network_segment_range = {'network_segment_range': {}}
|
|
for k, v in kwargs.items():
|
|
network_segment_range['network_segment_range'][k] = str(v)
|
|
|
|
network_segment_range_req = self.new_create_request(
|
|
'network-segment-ranges', network_segment_range, fmt)
|
|
|
|
network_segment_range_res = network_segment_range_req.get_response(
|
|
self.ext_api)
|
|
|
|
if expected_res_status:
|
|
self.assertEqual(expected_res_status,
|
|
network_segment_range_res.status_int)
|
|
return network_segment_range_res
|
|
|
|
def network_segment_range(self, **kwargs):
|
|
res = self._create_network_segment_range(self.fmt, **kwargs)
|
|
if res.status_int >= webob.exc.HTTPClientError.code:
|
|
raise webob.exc.HTTPClientError(code=res.status_int)
|
|
return self.deserialize(self.fmt, res)
|
|
|
|
def _test_create_network_segment_range(self, expected=None, **kwargs):
|
|
network_segment_range = self.network_segment_range(**kwargs)
|
|
self._validate_resource(network_segment_range, kwargs,
|
|
'network_segment_range')
|
|
if expected:
|
|
self._compare_resource(network_segment_range, expected,
|
|
'network_segment_range')
|
|
return network_segment_range
|
|
|
|
def _test_update_network_segment_range(self, range_id,
|
|
data, expected=None):
|
|
update_req = self.new_update_request(
|
|
'network-segment-ranges', data, range_id)
|
|
|
|
update_res = update_req.get_response(self.ext_api)
|
|
if expected:
|
|
network_segment_range = self.deserialize(self.fmt, update_res)
|
|
self._compare_resource(network_segment_range, expected,
|
|
'network_segment_range')
|
|
return network_segment_range
|
|
|
|
return update_res
|
|
|
|
|
|
class NetworkSegmentRangeTestPlugin(db_base_plugin_v2.NeutronDbPluginV2,
|
|
plugin_range.NetworkSegmentRangePlugin):
|
|
"""Test plugin to mixin the network segment range extension."""
|
|
__native_pagination_support = True
|
|
__native_sorting_support = True
|
|
__filter_validation_support = True
|
|
|
|
supported_extension_aliases = ["provider", "network-segment-range"]
|
|
|
|
def __init__(self):
|
|
super(NetworkSegmentRangeTestPlugin, self).__init__()
|
|
self.type_manager = mock.Mock()
|
|
|
|
|
|
class TestNetworkSegmentRange(NetworkSegmentRangeTestBase):
|
|
|
|
def setUp(self, plugin=None):
|
|
if not plugin:
|
|
plugin = TEST_PLUGIN_KLASS
|
|
service_plugins = {'network_segment_range_plugin_name':
|
|
SERVICE_PLUGIN_KLASS}
|
|
cfg.CONF.set_override('service_plugins', [SERVICE_PLUGIN_KLASS])
|
|
ext_mgr = NetworkSegmentRangeExtensionManager()
|
|
super(TestNetworkSegmentRange, self).setUp(
|
|
plugin=plugin, ext_mgr=ext_mgr, service_plugins=service_plugins)
|
|
|
|
def _test_create_network_segment_range(self, expected=None, **kwargs):
|
|
for d in (kwargs, expected):
|
|
if d is None:
|
|
continue
|
|
d.setdefault('name', '')
|
|
d.setdefault('shared', True)
|
|
d.setdefault('project_id', None)
|
|
d.setdefault('network_type', constants.TYPE_VLAN)
|
|
d.setdefault('physical_network', 'phys_net')
|
|
d.setdefault('minimum', 200)
|
|
d.setdefault('maximum', 300)
|
|
return (super(TestNetworkSegmentRange, self).
|
|
_test_create_network_segment_range(expected, **kwargs))
|
|
|
|
def test_create_network_segment_range_empty_name(self):
|
|
expected_range = {'name': '',
|
|
'shared': True,
|
|
'project_id': None,
|
|
'network_type': constants.TYPE_VLAN,
|
|
'physical_network': 'phys_net',
|
|
'minimum': 200,
|
|
'maximum': 300}
|
|
self._test_create_network_segment_range(expected=expected_range)
|
|
|
|
def test_create_network_segment_range_with_name(self):
|
|
expected_range = {'name': 'foo-range-name',
|
|
'shared': True,
|
|
'project_id': None,
|
|
'network_type': constants.TYPE_VLAN,
|
|
'physical_network': 'phys_net',
|
|
'minimum': 200,
|
|
'maximum': 300}
|
|
self._test_create_network_segment_range(
|
|
name='foo-range-name',
|
|
expected=expected_range)
|
|
|
|
def test_create_network_segment_range_unsupported_network_type(self):
|
|
exc = self.assertRaises(webob.exc.HTTPClientError,
|
|
self._test_create_network_segment_range,
|
|
network_type='foo-network-type')
|
|
self.assertEqual(webob.exc.HTTPClientError.code, exc.code)
|
|
self.assertIn('The server could not comply with the request',
|
|
exc.explanation)
|
|
|
|
def test_create_network_segment_range_no_physical_network(self):
|
|
expected_range = {'shared': True,
|
|
'project_id': None,
|
|
'network_type': constants.TYPE_VXLAN,
|
|
'physical_network': None}
|
|
self._test_create_network_segment_range(
|
|
network_type=constants.TYPE_VXLAN,
|
|
physical_network=None,
|
|
expected=expected_range)
|
|
|
|
def test_create_network_segment_range_tenant_specific(self):
|
|
expected_range = {'shared': False,
|
|
'project_id': test_db_base_plugin_v2.TEST_TENANT_ID,
|
|
'network_type': constants.TYPE_VLAN,
|
|
'physical_network': 'phys_net',
|
|
'minimum': 200,
|
|
'maximum': 300}
|
|
self._test_create_network_segment_range(
|
|
shared=False,
|
|
project_id=test_db_base_plugin_v2.TEST_TENANT_ID,
|
|
network_type=constants.TYPE_VLAN,
|
|
physical_network='phys_net',
|
|
expected=expected_range)
|
|
|
|
def test_create_network_segment_ranges_in_certain_order(self):
|
|
ctx = context.get_admin_context()
|
|
range1 = self._test_create_network_segment_range(
|
|
name='foo-range1', physical_network='phys_net1')
|
|
range2 = self._test_create_network_segment_range(
|
|
name='foo-range2', physical_network='phys_net2')
|
|
range3 = self._test_create_network_segment_range(
|
|
name='foo-range3', physical_network='phys_net3')
|
|
network_segment_ranges = (
|
|
NetworkSegmentRangeTestPlugin.get_network_segment_ranges(
|
|
NetworkSegmentRangeTestPlugin(), ctx))
|
|
self.assertEqual(range1['network_segment_range']['id'],
|
|
network_segment_ranges[0]['id'])
|
|
self.assertEqual(range2['network_segment_range']['id'],
|
|
network_segment_ranges[1]['id'])
|
|
self.assertEqual(range3['network_segment_range']['id'],
|
|
network_segment_ranges[2]['id'])
|
|
|
|
def test_create_network_segment_range_failed_with_vlan_minimum_id(self):
|
|
exc = self.assertRaises(webob.exc.HTTPClientError,
|
|
self._test_create_network_segment_range,
|
|
minimum=0)
|
|
self.assertEqual(webob.exc.HTTPClientError.code, exc.code)
|
|
self.assertIn('The server could not comply with the request',
|
|
exc.explanation)
|
|
|
|
def test_create_network_segment_range_failed_with_vlan_maximum_id(self):
|
|
exc = self.assertRaises(webob.exc.HTTPClientError,
|
|
self._test_create_network_segment_range,
|
|
minimum=4095)
|
|
self.assertEqual(webob.exc.HTTPServerError.code, exc.code)
|
|
self.assertIn('The server could not comply with the request',
|
|
exc.explanation)
|
|
|
|
def test_create_network_segment_range_failed_with_tunnel_minimum_id(self):
|
|
tunnel_type = [constants.TYPE_VXLAN,
|
|
constants.TYPE_GRE,
|
|
constants.TYPE_GENEVE]
|
|
for network_type in tunnel_type:
|
|
exc = self.assertRaises(webob.exc.HTTPClientError,
|
|
self._test_create_network_segment_range,
|
|
network_type=network_type,
|
|
physical_network=None,
|
|
minimum=0)
|
|
self.assertEqual(webob.exc.HTTPClientError.code, exc.code)
|
|
self.assertIn('The server could not comply with the request',
|
|
exc.explanation)
|
|
|
|
def test_create_network_segment_range_failed_with_tunnel_maximum_id(self):
|
|
expected_res = [(constants.TYPE_VXLAN, 2 ** 24),
|
|
(constants.TYPE_GRE, 2 ** 32),
|
|
(constants.TYPE_GENEVE, 2 ** 24)]
|
|
for network_type, max_id in expected_res:
|
|
exc = self.assertRaises(webob.exc.HTTPClientError,
|
|
self._test_create_network_segment_range,
|
|
network_type=network_type,
|
|
physical_network=None,
|
|
maximum=max_id)
|
|
if network_type == constants.TYPE_GRE:
|
|
self.assertEqual(webob.exc.HTTPClientError.code, exc.code)
|
|
else:
|
|
self.assertEqual(webob.exc.HTTPServerError.code, exc.code)
|
|
self.assertIn('The server could not comply with the request',
|
|
exc.explanation)
|
|
|
|
def test_update_network_segment_range_set_name(self):
|
|
network_segment_range = self._test_create_network_segment_range()
|
|
with mock.patch.object(segments_db, 'min_max_actual_segments_in_range',
|
|
return_value=(None, None)):
|
|
result = self._update(
|
|
'network-segment-ranges',
|
|
network_segment_range['network_segment_range']['id'],
|
|
{'network_segment_range': {'name': 'foo-name'}},
|
|
expected_code=webob.exc.HTTPOk.code)
|
|
self.assertEqual('foo-name',
|
|
result['network_segment_range']['name'])
|
|
|
|
def test_update_network_segment_range_set_name_to_empty(self):
|
|
network_segment_range = self._test_create_network_segment_range(
|
|
name='foo-range-name')
|
|
with mock.patch.object(segments_db, 'min_max_actual_segments_in_range',
|
|
return_value=(None, None)):
|
|
result = self._update(
|
|
'network-segment-ranges',
|
|
network_segment_range['network_segment_range']['id'],
|
|
{'network_segment_range': {'name': ''}},
|
|
expected_code=webob.exc.HTTPOk.code)
|
|
self.assertEqual('', result['network_segment_range']['name'])
|
|
|
|
def test_update_network_segment_range_min_max(self):
|
|
network_segment_range = self._test_create_network_segment_range()
|
|
with mock.patch.object(segments_db, 'min_max_actual_segments_in_range',
|
|
return_value=(None, None)):
|
|
result = self._update(
|
|
'network-segment-ranges',
|
|
network_segment_range['network_segment_range']['id'],
|
|
{'network_segment_range': {'minimum': 1200, 'maximum': 1300}},
|
|
expected_code=webob.exc.HTTPOk.code)
|
|
self.assertEqual(1200, result['network_segment_range']['minimum'])
|
|
self.assertEqual(1300, result['network_segment_range']['maximum'])
|
|
|
|
def test_get_network_segment_range(self):
|
|
network_segment_range = self._test_create_network_segment_range()
|
|
req = self.new_show_request(
|
|
'network-segment-ranges',
|
|
network_segment_range['network_segment_range']['id'])
|
|
res = self.deserialize(self.fmt, req.get_response(self.ext_api))
|
|
self.assertEqual(
|
|
network_segment_range['network_segment_range']['id'],
|
|
res['network_segment_range']['id'])
|
|
|
|
def test_list_network_segment_ranges(self):
|
|
self._test_create_network_segment_range(name='foo-range1')
|
|
self._test_create_network_segment_range(
|
|
name='foo-range2', minimum=400, maximum=500)
|
|
res = self._list('network-segment-ranges')
|
|
self.assertEqual(2, len(res['network_segment_ranges']))
|
|
|
|
def test_list_network_segment_ranges_with_sort(self):
|
|
range1 = self._test_create_network_segment_range(
|
|
name='foo-range1', physical_network='phys_net1')
|
|
range2 = self._test_create_network_segment_range(
|
|
name='foo-range2', physical_network='phys_net2')
|
|
self._test_list_with_sort('network-segment-range',
|
|
(range2, range1),
|
|
[('name', 'desc')])
|
|
|
|
def test_list_network_segment_ranges_with_pagination(self):
|
|
range1 = self._test_create_network_segment_range(
|
|
name='foo-range1', physical_network='phys_net1')
|
|
range2 = self._test_create_network_segment_range(
|
|
name='foo-range2', physical_network='phys_net2')
|
|
range3 = self._test_create_network_segment_range(
|
|
name='foo-range3', physical_network='phys_net3')
|
|
self._test_list_with_pagination(
|
|
'network-segment-range',
|
|
(range1, range2, range3),
|
|
('name', 'asc'), 2, 2)
|
|
|
|
def test_list_network_segment_ranges_with_pagination_reverse(self):
|
|
range1 = self._test_create_network_segment_range(
|
|
name='foo-range1', physical_network='phys_net1')
|
|
range2 = self._test_create_network_segment_range(
|
|
name='foo-range2', physical_network='phys_net2')
|
|
range3 = self._test_create_network_segment_range(
|
|
name='foo-range3', physical_network='phys_net3')
|
|
self._test_list_with_pagination_reverse(
|
|
'network-segment-range',
|
|
(range1, range2, range3),
|
|
('name', 'asc'), 2, 2)
|
|
|
|
def test_delete_network_segment_range(self):
|
|
network_segment_range = self._test_create_network_segment_range()
|
|
with mock.patch.object(segments_db, 'network_segments_exist_in_range',
|
|
return_value=False):
|
|
self._delete('network-segment-ranges',
|
|
network_segment_range['network_segment_range']['id'])
|
|
self._show('network-segment-ranges',
|
|
network_segment_range['network_segment_range']['id'],
|
|
expected_code=webob.exc.HTTPNotFound.code)
|