"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))