← Back to branch summary

~ubuntu-branches/ubuntu/vivid/neutron/vivid

« back to all changes in this revision

Viewing changes to neutron/tests/unit/plugins/ml2/drivers/test_type_vlan.py

  • Committer: Package Import Robot
  • Author(s): Chuck Short
  • Date: 2015-04-15 13:59:07 UTC
  • mfrom: (1.1.22)
  • Revision ID: package-import@ubuntu.com-20150415135907-z10fr18evag1ozq3
Tags: 1:2015.1~rc1-0ubuntu1
* New upstream milestone release:
  - debian/control: Update dependencies. 
  - debian/patches/disable-udev-tests.patch: Dropped no longer needed.
  - debian/patches/fixup-driver-test-execution.patch: Dropped no longer needed.
  - debian/patches/skip-iptest.patch: Skip failing test
  - debian/neutron-plugin-openvswitch-agent.install: Added neutron-ovsvapp-agent binary.
  - debian/neutron-plugin-cisco.install: Added neutron-cisco-apic-service-agent and 
    neutron-cisco-apic-host-agent

Show diffs side-by-side

added added

removed removed

Lines of Context:
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})