neutron-14.0.3.tar.gz: .../drivers/test_type_vlan.py

"Fossies" - the Fresh Open Source Software Archive

Member "neutron-14.0.3/neutron/tests/unit/plugins/ml2/drivers/test_type_vlan.py" (22 Oct 2019, 16909 Bytes) of package /linux/misc/openstack/neutron-14.0.3.tar.gz:

As a special service "Fossies" has tried to format the requested source page into HTML format using (guessed) Python source code syntax highlighting (style: standard) with prefixed line numbers. Alternatively you can here view or download the uninterpreted source code file. See also the latest Fossies "Diffs" side-by-side code changes report for "test_type_vlan.py":

14.0.2_vs_14.0.3.

1 # Copyright (c) 2014 Thales Services SAS 2 # All Rights Reserved. 3 # 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 7 # 8 # http://www.apache.org/licenses/LICENSE-2.0 9 # 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 14 # under the License. 15 16 import mock 17 from neutron_lib import constants as p_const 18 from neutron_lib import context 19 from neutron_lib.db import api as db_api 20 from neutron_lib import exceptions as exc 21 from neutron_lib.plugins.ml2 import api 22 from neutron_lib.plugins import utils as plugin_utils 23 from oslo_config import cfg 24 from testtools import matchers 25 26 from neutron.objects import network_segment_range as obj_network_segment_range 27 from neutron.objects.plugins.ml2 import vlanallocation as vlan_alloc_obj 28 from neutron.plugins.ml2.drivers import type_vlan 29 from neutron.tests.unit import testlib_api 30 31 PROVIDER_NET = 'phys_net1' 32 TENANT_NET = 'phys_net2' 33 VLAN_MIN = 200 34 VLAN_MAX = 209 35 NETWORK_VLAN_RANGES = [PROVIDER_NET, "%s:%s:%s" % 36 (TENANT_NET, VLAN_MIN, VLAN_MAX)] 37 UPDATED_VLAN_RANGES = { 38 PROVIDER_NET: [], 39 TENANT_NET: [(VLAN_MIN + 5, VLAN_MAX + 5)], 40 } 41 EMPTY_VLAN_RANGES = { 42 PROVIDER_NET: [] 43 } 44 CORE_PLUGIN = 'ml2' 45 SERVICE_PLUGIN_KLASS = ('neutron.services.network_segment_range.plugin.' 46 'NetworkSegmentRangePlugin') 47 48 49 class VlanTypeTest(testlib_api.SqlTestCase): 50 51 def setUp(self): 52 super(VlanTypeTest, self).setUp() 53 cfg.CONF.set_override('network_vlan_ranges', 54 NETWORK_VLAN_RANGES, 55 group='ml2_type_vlan') 56 self.network_vlan_ranges = plugin_utils.parse_network_vlan_ranges( 57 NETWORK_VLAN_RANGES) 58 self.driver = type_vlan.VlanTypeDriver() 59 self.driver._sync_vlan_allocations() 60 self.context = context.Context() 61 self.driver.physnet_mtus = [] 62 self.setup_coreplugin(CORE_PLUGIN) 63 64 def test_parse_network_exception_handling(self): 65 with mock.patch.object(plugin_utils, 66 'parse_network_vlan_ranges') as parse_ranges: 67 parse_ranges.side_effect = Exception('any exception') 68 self.assertRaises(SystemExit, 69 self.driver._parse_network_vlan_ranges) 70 71 @db_api.CONTEXT_READER 72 def _get_allocation(self, context, segment): 73 return vlan_alloc_obj.VlanAllocation.get_object( 74 context, 75 physical_network=segment[api.PHYSICAL_NETWORK], 76 vlan_id=segment[api.SEGMENTATION_ID]) 77 78 def test_partial_segment_is_partial_segment(self): 79 segment = {api.NETWORK_TYPE: p_const.TYPE_VLAN} 80 self.assertTrue(self.driver.is_partial_segment(segment)) 81 82 def test_specific_segment_is_not_partial_segment(self): 83 segment = {api.NETWORK_TYPE: p_const.TYPE_VLAN, 84 api.PHYSICAL_NETWORK: PROVIDER_NET, 85 api.SEGMENTATION_ID: 1} 86 self.assertFalse(self.driver.is_partial_segment(segment)) 87 88 def test_validate_provider_segment(self): 89 segment = {api.NETWORK_TYPE: p_const.TYPE_VLAN, 90 api.PHYSICAL_NETWORK: PROVIDER_NET, 91 api.SEGMENTATION_ID: 1} 92 self.assertIsNone(self.driver.validate_provider_segment(segment)) 93 94 def test_validate_provider_segment_without_segmentation_id(self): 95 segment = {api.NETWORK_TYPE: p_const.TYPE_VLAN, 96 api.PHYSICAL_NETWORK: TENANT_NET} 97 self.driver.validate_provider_segment(segment) 98 99 def test_validate_provider_segment_without_physical_network(self): 100 segment = {api.NETWORK_TYPE: p_const.TYPE_VLAN} 101 self.driver.validate_provider_segment(segment) 102 103 def test_validate_provider_segment_no_phys_network_seg_id_0(self): 104 segment = {api.NETWORK_TYPE: p_const.TYPE_VLAN, 105 api.SEGMENTATION_ID: 0} 106 self.assertRaises(exc.InvalidInput, 107 self.driver.validate_provider_segment, 108 segment) 109 110 def test_validate_provider_segment_with_missing_physical_network(self): 111 segment = {api.NETWORK_TYPE: p_const.TYPE_VLAN, 112 api.SEGMENTATION_ID: 1} 113 self.assertRaises(exc.InvalidInput, 114 self.driver.validate_provider_segment, 115 segment) 116 117 def test_validate_provider_segment_with_invalid_physical_network(self): 118 segment = {api.NETWORK_TYPE: p_const.TYPE_VLAN, 119 api.PHYSICAL_NETWORK: 'other_phys_net', 120 api.SEGMENTATION_ID: 1} 121 self.assertRaises(exc.InvalidInput, 122 self.driver.validate_provider_segment, 123 segment) 124 125 def test_validate_provider_segment_with_invalid_segmentation_id(self): 126 segment = {api.NETWORK_TYPE: p_const.TYPE_VLAN, 127 api.PHYSICAL_NETWORK: PROVIDER_NET} 128 segmentation_ids = [ 129 p_const.MIN_VLAN_TAG - 1, 130 p_const.MAX_VLAN_TAG + 1] 131 for segmentation_id in segmentation_ids: 132 segment[api.SEGMENTATION_ID] = segmentation_id 133 self.assertRaises(exc.InvalidInput, 134 self.driver.validate_provider_segment, 135 segment) 136 137 def test_validate_provider_segment_with_invalid_input(self): 138 segment = {api.NETWORK_TYPE: p_const.TYPE_VLAN, 139 api.PHYSICAL_NETWORK: PROVIDER_NET, 140 api.SEGMENTATION_ID: 1, 141 'invalid': 1} 142 self.assertRaises(exc.InvalidInput, 143 self.driver.validate_provider_segment, 144 segment) 145 146 def test_validate_provider_segment_with_physical_network_only(self): 147 segment = {api.NETWORK_TYPE: p_const.TYPE_VLAN, 148 api.PHYSICAL_NETWORK: PROVIDER_NET} 149 self.assertRaises(exc.InvalidInput, 150 self.driver.validate_provider_segment, 151 segment) 152 153 def test_sync_vlan_allocations(self): 154 def check_in_ranges(network_vlan_ranges): 155 vlan_min, vlan_max = network_vlan_ranges[TENANT_NET][0] 156 segment = {api.NETWORK_TYPE: p_const.TYPE_VLAN, 157 api.PHYSICAL_NETWORK: TENANT_NET} 158 159 segment[api.SEGMENTATION_ID] = vlan_min - 1 160 self.assertIsNone( 161 self._get_allocation(self.context, segment)) 162 segment[api.SEGMENTATION_ID] = vlan_max + 1 163 self.assertIsNone( 164 self._get_allocation(self.context, segment)) 165 166 segment[api.SEGMENTATION_ID] = vlan_min 167 self.assertFalse( 168 self._get_allocation(self.context, segment).allocated) 169 segment[api.SEGMENTATION_ID] = vlan_max 170 self.assertFalse( 171 self._get_allocation(self.context, segment).allocated) 172 173 check_in_ranges(self.network_vlan_ranges) 174 self.driver.network_vlan_ranges = UPDATED_VLAN_RANGES 175 self.driver._sync_vlan_allocations() 176 check_in_ranges(UPDATED_VLAN_RANGES) 177 178 self.driver.network_vlan_ranges = EMPTY_VLAN_RANGES 179 self.driver._sync_vlan_allocations() 180 181 vlan_min, vlan_max = UPDATED_VLAN_RANGES[TENANT_NET][0] 182 segment = {api.NETWORK_TYPE: p_const.TYPE_VLAN, 183 api.PHYSICAL_NETWORK: TENANT_NET} 184 segment[api.SEGMENTATION_ID] = vlan_min 185 self.assertIsNone( 186 self._get_allocation(self.context, segment)) 187 segment[api.SEGMENTATION_ID] = vlan_max 188 self.assertIsNone( 189 self._get_allocation(self.context, segment)) 190 191 def test_reserve_provider_segment(self): 192 segment = {api.NETWORK_TYPE: p_const.TYPE_VLAN, 193 api.PHYSICAL_NETWORK: PROVIDER_NET, 194 api.SEGMENTATION_ID: 101} 195 alloc = self._get_allocation(self.context, segment) 196 self.assertIsNone(alloc) 197 observed = self.driver.reserve_provider_segment(self.context, segment) 198 alloc = self._get_allocation(self.context, observed) 199 self.assertTrue(alloc.allocated) 200 201 def test_reserve_provider_segment_already_allocated(self): 202 segment = {api.NETWORK_TYPE: p_const.TYPE_VLAN, 203 api.PHYSICAL_NETWORK: PROVIDER_NET, 204 api.SEGMENTATION_ID: 101} 205 observed = self.driver.reserve_provider_segment(self.context, segment) 206 self.assertRaises(exc.VlanIdInUse, 207 self.driver.reserve_provider_segment, 208 self.context, 209 observed) 210 211 def test_reserve_provider_segment_in_tenant_pools(self): 212 segment = {api.NETWORK_TYPE: p_const.TYPE_VLAN, 213 api.PHYSICAL_NETWORK: TENANT_NET, 214 api.SEGMENTATION_ID: VLAN_MIN} 215 alloc = self._get_allocation(self.context, segment) 216 self.assertFalse(alloc.allocated) 217 observed = self.driver.reserve_provider_segment(self.context, segment) 218 alloc = self._get_allocation(self.context, observed) 219 self.assertTrue(alloc.allocated) 220 221 def test_reserve_provider_segment_without_segmentation_id(self): 222 segment = {api.NETWORK_TYPE: p_const.TYPE_VLAN, 223 api.PHYSICAL_NETWORK: TENANT_NET} 224 observed = self.driver.reserve_provider_segment(self.context, segment) 225 alloc = self._get_allocation(self.context, observed) 226 self.assertTrue(alloc.allocated) 227 vlan_id = observed[api.SEGMENTATION_ID] 228 self.assertThat(vlan_id, matchers.GreaterThan(VLAN_MIN - 1)) 229 self.assertThat(vlan_id, matchers.LessThan(VLAN_MAX + 1)) 230 231 def test_reserve_provider_segment_without_physical_network(self): 232 segment = {api.NETWORK_TYPE: p_const.TYPE_VLAN} 233 observed = self.driver.reserve_provider_segment(self.context, segment) 234 alloc = self._get_allocation(self.context, observed) 235 self.assertTrue(alloc.allocated) 236 vlan_id = observed[api.SEGMENTATION_ID] 237 self.assertThat(vlan_id, matchers.GreaterThan(VLAN_MIN - 1)) 238 self.assertThat(vlan_id, matchers.LessThan(VLAN_MAX + 1)) 239 self.assertEqual(TENANT_NET, observed[api.PHYSICAL_NETWORK]) 240 241 def test_reserve_provider_segment_all_allocateds(self): 242 for __ in range(VLAN_MIN, VLAN_MAX + 1): 243 self.driver.allocate_tenant_segment(self.context) 244 segment = {api.NETWORK_TYPE: p_const.TYPE_VLAN} 245 self.assertRaises(exc.NoNetworkAvailable, 246 self.driver.reserve_provider_segment, 247 self.context, 248 segment) 249 250 def test_get_mtu(self): 251 cfg.CONF.set_override('global_physnet_mtu', 1475) 252 cfg.CONF.set_override('path_mtu', 1400, group='ml2') 253 self.driver.physnet_mtus = {'physnet1': 1450, 'physnet2': 1400} 254 self.assertEqual(1450, self.driver.get_mtu('physnet1')) 255 256 cfg.CONF.set_override('global_physnet_mtu', 1375) 257 cfg.CONF.set_override('path_mtu', 1400, group='ml2') 258 self.driver.physnet_mtus = {'physnet1': 1450, 'physnet2': 1400} 259 self.assertEqual(1375, self.driver.get_mtu('physnet1')) 260 261 cfg.CONF.set_override('global_physnet_mtu', 0) 262 cfg.CONF.set_override('path_mtu', 1400, group='ml2') 263 self.driver.physnet_mtus = {'physnet1': 1450, 'physnet2': 1400} 264 self.assertEqual(1450, self.driver.get_mtu('physnet1')) 265 266 cfg.CONF.set_override('global_physnet_mtu', 0) 267 cfg.CONF.set_override('path_mtu', 0, group='ml2') 268 self.driver.physnet_mtus = {} 269 self.assertEqual(0, self.driver.get_mtu('physnet1')) 270 271 def test_allocate_tenant_segment(self): 272 for __ in range(VLAN_MIN, VLAN_MAX + 1): 273 segment = self.driver.allocate_tenant_segment(self.context) 274 alloc = self._get_allocation(self.context, segment) 275 self.assertTrue(alloc.allocated) 276 vlan_id = segment[api.SEGMENTATION_ID] 277 self.assertThat(vlan_id, matchers.GreaterThan(VLAN_MIN - 1)) 278 self.assertThat(vlan_id, matchers.LessThan(VLAN_MAX + 1)) 279 self.assertEqual(TENANT_NET, segment[api.PHYSICAL_NETWORK]) 280 281 def test_allocate_tenant_segment_no_available(self): 282 for __ in range(VLAN_MIN, VLAN_MAX + 1): 283 self.driver.allocate_tenant_segment(self.context) 284 segment = self.driver.allocate_tenant_segment(self.context) 285 self.assertIsNone(segment) 286 287 def test_release_segment(self): 288 segment = self.driver.allocate_tenant_segment(self.context) 289 self.driver.release_segment(self.context, segment) 290 alloc = self._get_allocation(self.context, segment) 291 self.assertFalse(alloc.allocated) 292 293 def test_release_segment_unallocated(self): 294 segment = {api.NETWORK_TYPE: p_const.TYPE_VLAN, 295 api.PHYSICAL_NETWORK: PROVIDER_NET, 296 api.SEGMENTATION_ID: 101} 297 with mock.patch.object(type_vlan.LOG, 'warning') as log_warn: 298 self.driver.release_segment(self.context, segment) 299 log_warn.assert_called_once_with( 300 "No vlan_id %(vlan_id)s found on physical network " 301 "%(physical_network)s", 302 {'vlan_id': 101, 'physical_network': PROVIDER_NET}) 303 304 305 class VlanTypeAllocationTest(testlib_api.SqlTestCase): 306 307 def test_allocate_tenant_segment_in_order_of_config(self): 308 ranges = NETWORK_VLAN_RANGES + ['phys_net3:20:30'] 309 cfg.CONF.set_override('network_vlan_ranges', 310 ranges, 311 group='ml2_type_vlan') 312 driver = type_vlan.VlanTypeDriver() 313 driver.physnet_mtus = [] 314 driver._sync_vlan_allocations() 315 # swap config order from DB order after sync has happened to 316 # ensure config order is followed and not DB order 317 cfg.CONF.set_override('network_vlan_ranges', 318 list(reversed(ranges)), 319 group='ml2_type_vlan') 320 driver._parse_network_vlan_ranges() 321 ctx = context.Context() 322 for vlan in range(11): 323 # all of physnet3 should be exhausted first 324 self.assertEqual( 325 {'network_type': 'vlan', 'physical_network': 'phys_net3', 326 'segmentation_id': mock.ANY, 'mtu': 1500}, 327 driver.allocate_tenant_segment(ctx)) 328 for vlan in range(10): 329 # then physnet2 330 self.assertEqual( 331 {'network_type': 'vlan', 'physical_network': 'phys_net2', 332 'segmentation_id': mock.ANY, 'mtu': 1500}, 333 driver.allocate_tenant_segment(ctx)) 334 # then nothing 335 self.assertFalse(driver.allocate_tenant_segment(ctx)) 336 337 338 class VlanTypeTestWithNetworkSegmentRange(testlib_api.SqlTestCase): 339 340 def setUp(self): 341 super(VlanTypeTestWithNetworkSegmentRange, self).setUp() 342 cfg.CONF.set_override('network_vlan_ranges', 343 NETWORK_VLAN_RANGES, 344 group='ml2_type_vlan') 345 cfg.CONF.set_override('service_plugins', [SERVICE_PLUGIN_KLASS]) 346 self.network_vlan_ranges = plugin_utils.parse_network_vlan_ranges( 347 NETWORK_VLAN_RANGES) 348 self.driver = type_vlan.VlanTypeDriver() 349 self.driver._sync_vlan_allocations() 350 self.context = context.Context() 351 self.setup_coreplugin(CORE_PLUGIN) 352 353 def test__populate_new_default_network_segment_ranges(self): 354 # _populate_new_default_network_segment_ranges will be called when 355 # the type driver initializes with `network_segment_range` loaded as 356 # one of the `service_plugins` 357 ret = obj_network_segment_range.NetworkSegmentRange.get_objects( 358 self.context) 359 self.assertEqual(1, len(ret)) 360 network_segment_range = ret[0] 361 self.assertTrue(network_segment_range.default) 362 self.assertTrue(network_segment_range.shared) 363 self.assertIsNone(network_segment_range.project_id) 364 self.assertEqual(p_const.TYPE_VLAN, network_segment_range.network_type) 365 self.assertEqual(TENANT_NET, network_segment_range.physical_network) 366 self.assertEqual(VLAN_MIN, network_segment_range.minimum) 367 self.assertEqual(VLAN_MAX, network_segment_range.maximum) 368 369 def test__delete_expired_default_network_segment_ranges(self): 370 self.driver._delete_expired_default_network_segment_ranges() 371 ret = obj_network_segment_range.NetworkSegmentRange.get_objects( 372 self.context) 373 self.assertEqual(0, len(ret))