neutron/tests/unit/plugins/ml2/drivers/test_type_vlan.py

   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 testtools import matchers
  18
  19 from neutron.common import exceptions as exc
  20 import neutron.db.api as db
  21 from neutron.plugins.common import constants as p_const
  22 from neutron.plugins.common import utils as plugin_utils
  23 from neutron.plugins.ml2 import config
  24 from neutron.plugins.ml2 import driver_api as api
  25 from neutron.plugins.ml2.drivers import type_vlan
  26 from neutron.tests.unit import testlib_api
  27
  28 PROVIDER_NET = 'phys_net1'
  29 TENANT_NET = 'phys_net2'
  30 VLAN_MIN = 200
  31 VLAN_MAX = 209
  32 NETWORK_VLAN_RANGES = [PROVIDER_NET, "%s:%s:%s" %
  33                        (TENANT_NET, VLAN_MIN, VLAN_MAX)]
  34 UPDATED_VLAN_RANGES = {
  35     PROVIDER_NET: [],
  36     TENANT_NET: [(VLAN_MIN + 5, VLAN_MAX + 5)],
  37 }
  38
  39
  40 class VlanTypeTest(testlib_api.SqlTestCase):
  41
  42     def setUp(self):
  43         super(VlanTypeTest, self).setUp()
  44         config.cfg.CONF.set_override('network_vlan_ranges',
  45                                      NETWORK_VLAN_RANGES,
  46                                      group='ml2_type_vlan')
  47         self.network_vlan_ranges = plugin_utils.parse_network_vlan_ranges(
  48             NETWORK_VLAN_RANGES)
  49         self.driver = type_vlan.VlanTypeDriver()
  50         self.driver._sync_vlan_allocations()
  51         self.session = db.get_session()
  52         self.driver.physnet_mtus = []
  53
  54     def test_parse_network_exception_handling(self):
  55         with mock.patch.object(plugin_utils,
  56                                'parse_network_vlan_ranges') as parse_ranges:
  57             parse_ranges.side_effect = Exception('any exception')
  58             self.assertRaises(SystemExit,
  59                               self.driver._parse_network_vlan_ranges)
  60
  61     def _get_allocation(self, session, segment):
  62         return session.query(type_vlan.VlanAllocation).filter_by(
  63             physical_network=segment[api.PHYSICAL_NETWORK],
  64             vlan_id=segment[api.SEGMENTATION_ID]).first()
  65
  66     def test_partial_segment_is_partial_segment(self):
  67         segment = {api.NETWORK_TYPE: p_const.TYPE_VLAN}
  68         self.assertTrue(self.driver.is_partial_segment(segment))
  69
  70     def test_specific_segment_is_not_partial_segment(self):
  71         segment = {api.NETWORK_TYPE: p_const.TYPE_VLAN,
  72                    api.PHYSICAL_NETWORK: PROVIDER_NET,
  73                    api.SEGMENTATION_ID: 1}
  74         self.assertFalse(self.driver.is_partial_segment(segment))
  75
  76     def test_validate_provider_segment(self):
  77         segment = {api.NETWORK_TYPE: p_const.TYPE_VLAN,
  78                    api.PHYSICAL_NETWORK: PROVIDER_NET,
  79                    api.SEGMENTATION_ID: 1}
  80         self.assertIsNone(self.driver.validate_provider_segment(segment))
  81
  82     def test_validate_provider_segment_without_segmentation_id(self):
  83         segment = {api.NETWORK_TYPE: p_const.TYPE_VLAN,
  84                    api.PHYSICAL_NETWORK: TENANT_NET}
  85         self.driver.validate_provider_segment(segment)
  86
  87     def test_validate_provider_segment_without_physical_network(self):
  88         segment = {api.NETWORK_TYPE: p_const.TYPE_VLAN}
  89         self.driver.validate_provider_segment(segment)
  90
  91     def test_validate_provider_segment_with_missing_physical_network(self):
  92         segment = {api.NETWORK_TYPE: p_const.TYPE_VLAN,
  93                    api.SEGMENTATION_ID: 1}
  94         self.assertRaises(exc.InvalidInput,
  95                           self.driver.validate_provider_segment,
  96                           segment)
  97
  98     def test_validate_provider_segment_with_invalid_physical_network(self):
  99         segment = {api.NETWORK_TYPE: p_const.TYPE_VLAN,
 100                    api.PHYSICAL_NETWORK: 'other_phys_net',
 101                    api.SEGMENTATION_ID: 1}
 102         self.assertRaises(exc.InvalidInput,
 103                           self.driver.validate_provider_segment,
 104                           segment)
 105
 106     def test_validate_provider_segment_with_invalid_segmentation_id(self):
 107         segment = {api.NETWORK_TYPE: p_const.TYPE_VLAN,
 108                    api.PHYSICAL_NETWORK: PROVIDER_NET,
 109                    api.SEGMENTATION_ID: 5000}
 110         self.assertRaises(exc.InvalidInput,
 111                           self.driver.validate_provider_segment,
 112                           segment)
 113
 114     def test_validate_provider_segment_with_invalid_input(self):
 115         segment = {api.NETWORK_TYPE: p_const.TYPE_VLAN,
 116                    api.PHYSICAL_NETWORK: PROVIDER_NET,
 117                    api.SEGMENTATION_ID: 1,
 118                    'invalid': 1}
 119         self.assertRaises(exc.InvalidInput,
 120                           self.driver.validate_provider_segment,
 121                           segment)
 122
 123     def test_sync_vlan_allocations(self):
 124         def check_in_ranges(network_vlan_ranges):
 125             vlan_min, vlan_max = network_vlan_ranges[TENANT_NET][0]
 126             segment = {api.NETWORK_TYPE: p_const.TYPE_VLAN,
 127                        api.PHYSICAL_NETWORK: TENANT_NET}
 128
 129             segment[api.SEGMENTATION_ID] = vlan_min - 1
 130             self.assertIsNone(
 131                 self._get_allocation(self.session, segment))
 132             segment[api.SEGMENTATION_ID] = vlan_max + 1
 133             self.assertIsNone(
 134                 self._get_allocation(self.session, segment))
 135
 136             segment[api.SEGMENTATION_ID] = vlan_min
 137             self.assertFalse(
 138                 self._get_allocation(self.session, segment).allocated)
 139             segment[api.SEGMENTATION_ID] = vlan_max
 140             self.assertFalse(
 141                 self._get_allocation(self.session, segment).allocated)
 142
 143         check_in_ranges(self.network_vlan_ranges)
 144         self.driver.network_vlan_ranges = UPDATED_VLAN_RANGES
 145         self.driver._sync_vlan_allocations()
 146         check_in_ranges(UPDATED_VLAN_RANGES)
 147
 148     def test_reserve_provider_segment(self):
 149         segment = {api.NETWORK_TYPE: p_const.TYPE_VLAN,
 150                    api.PHYSICAL_NETWORK: PROVIDER_NET,
 151                    api.SEGMENTATION_ID: 101}
 152         alloc = self._get_allocation(self.session, segment)
 153         self.assertIsNone(alloc)
 154         observed = self.driver.reserve_provider_segment(self.session, segment)
 155         alloc = self._get_allocation(self.session, observed)
 156         self.assertTrue(alloc.allocated)
 157
 158     def test_reserve_provider_segment_already_allocated(self):
 159         segment = {api.NETWORK_TYPE: p_const.TYPE_VLAN,
 160                    api.PHYSICAL_NETWORK: PROVIDER_NET,
 161                    api.SEGMENTATION_ID: 101}
 162         observed = self.driver.reserve_provider_segment(self.session, segment)
 163         self.assertRaises(exc.VlanIdInUse,
 164                           self.driver.reserve_provider_segment,
 165                           self.session,
 166                           observed)
 167
 168     def test_reserve_provider_segment_in_tenant_pools(self):
 169         segment = {api.NETWORK_TYPE: p_const.TYPE_VLAN,
 170                    api.PHYSICAL_NETWORK: TENANT_NET,
 171                    api.SEGMENTATION_ID: VLAN_MIN}
 172         alloc = self._get_allocation(self.session, segment)
 173         self.assertFalse(alloc.allocated)
 174         observed = self.driver.reserve_provider_segment(self.session, segment)
 175         alloc = self._get_allocation(self.session, observed)
 176         self.assertTrue(alloc.allocated)
 177
 178     def test_reserve_provider_segment_without_segmentation_id(self):
 179         segment = {api.NETWORK_TYPE: p_const.TYPE_VLAN,
 180                    api.PHYSICAL_NETWORK: TENANT_NET}
 181         observed = self.driver.reserve_provider_segment(self.session, segment)
 182         alloc = self._get_allocation(self.session, observed)
 183         self.assertTrue(alloc.allocated)
 184         vlan_id = observed[api.SEGMENTATION_ID]
 185         self.assertThat(vlan_id, matchers.GreaterThan(VLAN_MIN - 1))
 186         self.assertThat(vlan_id, matchers.LessThan(VLAN_MAX + 1))
 187
 188     def test_reserve_provider_segment_without_physical_network(self):
 189         segment = {api.NETWORK_TYPE: p_const.TYPE_VLAN}
 190         observed = self.driver.reserve_provider_segment(self.session, segment)
 191         alloc = self._get_allocation(self.session, observed)
 192         self.assertTrue(alloc.allocated)
 193         vlan_id = observed[api.SEGMENTATION_ID]
 194         self.assertThat(vlan_id, matchers.GreaterThan(VLAN_MIN - 1))
 195         self.assertThat(vlan_id, matchers.LessThan(VLAN_MAX + 1))
 196         self.assertEqual(TENANT_NET, observed[api.PHYSICAL_NETWORK])
 197
 198     def test_reserve_provider_segment_all_allocateds(self):
 199         for __ in range(VLAN_MIN, VLAN_MAX + 1):
 200             self.driver.allocate_tenant_segment(self.session)
 201         segment = {api.NETWORK_TYPE: p_const.TYPE_VLAN}
 202         self.assertRaises(exc.NoNetworkAvailable,
 203                           self.driver.reserve_provider_segment,
 204                           self.session,
 205                           segment)
 206
 207     def test_get_mtu(self):
 208         config.cfg.CONF.set_override('segment_mtu', 1475, group='ml2')
 209         config.cfg.CONF.set_override('path_mtu', 1400, group='ml2')
 210         self.driver.physnet_mtus = {'physnet1': 1450, 'physnet2': 1400}
 211         self.assertEqual(1450, self.driver.get_mtu('physnet1'))
 212
 213         config.cfg.CONF.set_override('segment_mtu', 1375, group='ml2')
 214         config.cfg.CONF.set_override('path_mtu', 1400, group='ml2')
 215         self.driver.physnet_mtus = {'physnet1': 1450, 'physnet2': 1400}
 216         self.assertEqual(1375, self.driver.get_mtu('physnet1'))
 217
 218         config.cfg.CONF.set_override('segment_mtu', 0, group='ml2')
 219         config.cfg.CONF.set_override('path_mtu', 1400, group='ml2')
 220         self.driver.physnet_mtus = {'physnet1': 1450, 'physnet2': 1400}
 221         self.assertEqual(1450, self.driver.get_mtu('physnet1'))
 222
 223         config.cfg.CONF.set_override('segment_mtu', 0, group='ml2')
 224         config.cfg.CONF.set_override('path_mtu', 0, group='ml2')
 225         self.driver.physnet_mtus = {}
 226         self.assertEqual(0, self.driver.get_mtu('physnet1'))
 227
 228     def test_allocate_tenant_segment(self):
 229         for __ in range(VLAN_MIN, VLAN_MAX + 1):
 230             segment = self.driver.allocate_tenant_segment(self.session)
 231             alloc = self._get_allocation(self.session, segment)
 232             self.assertTrue(alloc.allocated)
 233             vlan_id = segment[api.SEGMENTATION_ID]
 234             self.assertThat(vlan_id, matchers.GreaterThan(VLAN_MIN - 1))
 235             self.assertThat(vlan_id, matchers.LessThan(VLAN_MAX + 1))
 236             self.assertEqual(TENANT_NET, segment[api.PHYSICAL_NETWORK])
 237
 238     def test_allocate_tenant_segment_no_available(self):
 239         for __ in range(VLAN_MIN, VLAN_MAX + 1):
 240             self.driver.allocate_tenant_segment(self.session)
 241         segment = self.driver.allocate_tenant_segment(self.session)
 242         self.assertIsNone(segment)
 243
 244     def test_release_segment(self):
 245         segment = self.driver.allocate_tenant_segment(self.session)
 246         self.driver.release_segment(self.session, segment)
 247         alloc = self._get_allocation(self.session, segment)
 248         self.assertFalse(alloc.allocated)
 249
 250     def test_release_segment_unallocated(self):
 251         segment = {api.NETWORK_TYPE: p_const.TYPE_VLAN,
 252                    api.PHYSICAL_NETWORK: PROVIDER_NET,
 253                    api.SEGMENTATION_ID: 101}
 254         with mock.patch.object(type_vlan.LOG, 'warning') as log_warn:
 255             self.driver.release_segment(self.session, segment)
 256             log_warn.assert_called_once_with(
 257                 "No vlan_id %(vlan_id)s found on physical network "
 258                 "%(physical_network)s",
 259                 {'vlan_id': 101, 'physical_network': PROVIDER_NET})