Coverage for manila/tests/network/test_standalone_network_plugin.py: 100%
228 statements
« prev ^ index » next coverage.py v7.11.0, created at 2026-02-18 22:19 +0000
1# Copyright 2015 Mirantis, Inc.
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.
16from unittest import mock
18import ddt
19import netaddr
20from oslo_config import cfg
22from manila.common import constants
23from manila import context
24from manila import exception
25from manila.network import standalone_network_plugin as plugin
26from manila import test
27from manila.tests import utils as test_utils
29CONF = cfg.CONF
31fake_context = context.RequestContext(
32 user_id='fake user', project_id='fake project', is_admin=False)
33fake_share_server = dict(id='fake_share_server_id')
34fake_share_network = dict(id='fake_share_network_id')
35fake_share_network_subnet = dict(id='fake_share_network_subnet_id')
38@ddt.ddt
39class StandaloneNetworkPluginTest(test.TestCase):
41 @ddt.data('custom_config_group_name', 'DEFAULT')
42 def test_init_only_with_required_data_v4(self, group_name):
43 data = {
44 group_name: {
45 'standalone_network_plugin_gateway': '10.0.0.1',
46 'standalone_network_plugin_mask': '24',
47 },
48 }
49 with test_utils.create_temp_config_with_opts(data):
50 instance = plugin.StandaloneNetworkPlugin(
51 config_group_name=group_name)
53 self.assertEqual('10.0.0.1', instance.gateway)
54 self.assertEqual('24', instance.mask)
55 self.assertIsNone(instance.segmentation_id)
56 self.assertIsNone(instance.allowed_ip_ranges)
57 self.assertEqual(4, instance.ip_version)
58 self.assertEqual(netaddr.IPNetwork('10.0.0.1/24'), instance.net)
59 self.assertEqual(['10.0.0.1/24'], instance.allowed_cidrs)
60 self.assertEqual(
61 ('10.0.0.0', '10.0.0.1', '10.0.0.255'),
62 instance.reserved_addresses)
64 @ddt.data('custom_config_group_name', 'DEFAULT')
65 def test_init_with_all_data_v4(self, group_name):
66 data = {
67 group_name: {
68 'standalone_network_plugin_gateway': '10.0.0.1',
69 'standalone_network_plugin_mask': '255.255.0.0',
70 'standalone_network_plugin_network_type': 'vlan',
71 'standalone_network_plugin_segmentation_id': 1001,
72 'standalone_network_plugin_allowed_ip_ranges': (
73 '10.0.0.3-10.0.0.7,10.0.0.69-10.0.0.157,10.0.0.213'),
74 'network_plugin_ipv4_enabled': True,
75 },
76 }
77 allowed_cidrs = [
78 '10.0.0.3/32', '10.0.0.4/30', '10.0.0.69/32', '10.0.0.70/31',
79 '10.0.0.72/29', '10.0.0.80/28', '10.0.0.96/27', '10.0.0.128/28',
80 '10.0.0.144/29', '10.0.0.152/30', '10.0.0.156/31', '10.0.0.213/32',
81 ]
82 with test_utils.create_temp_config_with_opts(data):
83 instance = plugin.StandaloneNetworkPlugin(
84 config_group_name=group_name)
86 self.assertEqual(4, instance.ip_version)
87 self.assertEqual('10.0.0.1', instance.gateway)
88 self.assertEqual('255.255.0.0', instance.mask)
89 self.assertEqual('vlan', instance.network_type)
90 self.assertEqual(1001, instance.segmentation_id)
91 self.assertEqual(allowed_cidrs, instance.allowed_cidrs)
92 self.assertEqual(
93 ['10.0.0.3-10.0.0.7', '10.0.0.69-10.0.0.157', '10.0.0.213'],
94 instance.allowed_ip_ranges)
95 self.assertEqual(
96 netaddr.IPNetwork('10.0.0.1/255.255.0.0'), instance.net)
97 self.assertEqual(
98 ('10.0.0.0', '10.0.0.1', '10.0.255.255'),
99 instance.reserved_addresses)
101 @ddt.data('custom_config_group_name', 'DEFAULT')
102 def test_init_only_with_required_data_v6(self, group_name):
103 data = {
104 group_name: {
105 'standalone_network_plugin_gateway': (
106 '2001:cdba::3257:9652'),
107 'standalone_network_plugin_mask': '48',
108 'network_plugin_ipv6_enabled': True,
109 },
110 }
111 with test_utils.create_temp_config_with_opts(data):
112 instance = plugin.StandaloneNetworkPlugin(
113 config_group_name=group_name)
115 self.assertEqual(
116 '2001:cdba::3257:9652', instance.gateway)
117 self.assertEqual('48', instance.mask)
118 self.assertIsNone(instance.segmentation_id)
119 self.assertIsNone(instance.allowed_ip_ranges)
120 self.assertEqual(6, instance.ip_version)
121 self.assertEqual(
122 netaddr.IPNetwork('2001:cdba::3257:9652/48'),
123 instance.net)
124 self.assertEqual(
125 ['2001:cdba::3257:9652/48'], instance.allowed_cidrs)
126 self.assertEqual(
127 ('2001:cdba::', '2001:cdba::3257:9652',
128 netaddr.IPAddress('2001:cdba:0:ffff:ffff:ffff:ffff:ffff').format()
129 ),
130 instance.reserved_addresses)
132 @ddt.data('custom_config_group_name', 'DEFAULT')
133 def test_init_with_all_data_v6(self, group_name):
134 data = {
135 group_name: {
136 'standalone_network_plugin_gateway': '2001:db8::0001',
137 'standalone_network_plugin_mask': '88',
138 'standalone_network_plugin_network_type': 'vlan',
139 'standalone_network_plugin_segmentation_id': 3999,
140 'standalone_network_plugin_allowed_ip_ranges': (
141 '2001:db8::-2001:db8:0000:0000:0000:007f:ffff:ffff'),
142 'network_plugin_ipv6_enabled': True,
143 },
144 }
145 with test_utils.create_temp_config_with_opts(data):
146 instance = plugin.StandaloneNetworkPlugin(
147 config_group_name=group_name)
149 self.assertEqual(6, instance.ip_version)
150 self.assertEqual('2001:db8::0001', instance.gateway)
151 self.assertEqual('88', instance.mask)
152 self.assertEqual('vlan', instance.network_type)
153 self.assertEqual(3999, instance.segmentation_id)
154 self.assertEqual(['2001:db8::/89'], instance.allowed_cidrs)
155 self.assertEqual(
156 ['2001:db8::-2001:db8:0000:0000:0000:007f:ffff:ffff'],
157 instance.allowed_ip_ranges)
158 self.assertEqual(
159 netaddr.IPNetwork('2001:db8::0001/88'), instance.net)
160 self.assertEqual(
161 ('2001:db8::', '2001:db8::0001', '2001:db8::ff:ffff:ffff'),
162 instance.reserved_addresses)
164 @ddt.data('flat', 'vlan', 'vxlan', 'gre')
165 def test_init_with_valid_network_types_v4(self, network_type):
166 data = {
167 'DEFAULT': {
168 'standalone_network_plugin_gateway': '10.0.0.1',
169 'standalone_network_plugin_mask': '255.255.0.0',
170 'standalone_network_plugin_network_type': network_type,
171 'standalone_network_plugin_segmentation_id': 1001,
172 'network_plugin_ipv4_enabled': True,
173 },
174 }
175 with test_utils.create_temp_config_with_opts(data):
176 instance = plugin.StandaloneNetworkPlugin(
177 config_group_name='DEFAULT')
179 self.assertEqual(instance.network_type, network_type)
181 @ddt.data(
182 'foo', 'foovlan', 'vlanfoo', 'foovlanbar', 'None', 'Vlan', 'vlaN')
183 def test_init_with_fake_network_types_v4(self, fake_network_type):
184 data = {
185 'DEFAULT': {
186 'standalone_network_plugin_gateway': '10.0.0.1',
187 'standalone_network_plugin_mask': '255.255.0.0',
188 'standalone_network_plugin_network_type': fake_network_type,
189 'standalone_network_plugin_segmentation_id': 1001,
190 'network_plugin_ipv4_enabled': True,
191 },
192 }
193 with test_utils.create_temp_config_with_opts(data):
194 self.assertRaises(
195 cfg.ConfigFileValueError,
196 plugin.StandaloneNetworkPlugin,
197 config_group_name='DEFAULT',
198 )
200 @ddt.data('custom_config_group_name', 'DEFAULT')
201 def test_invalid_init_without_any_config_definitions(self, group_name):
202 self.assertRaises(
203 exception.NetworkBadConfigurationException,
204 plugin.StandaloneNetworkPlugin,
205 config_group_name=group_name)
207 @ddt.data(
208 {},
209 {'gateway': '20.0.0.1'},
210 {'mask': '8'},
211 {'gateway': '20.0.0.1', 'mask': '33'},
212 {'gateway': '20.0.0.256', 'mask': '16'})
213 def test_invalid_init_required_data_improper(self, data):
214 group_name = 'custom_group_name'
215 if 'gateway' in data:
216 data['standalone_network_plugin_gateway'] = data.pop('gateway')
217 if 'mask' in data:
218 data['standalone_network_plugin_mask'] = data.pop('mask')
219 data = {group_name: data}
220 with test_utils.create_temp_config_with_opts(data):
221 self.assertRaises(
222 exception.NetworkBadConfigurationException,
223 plugin.StandaloneNetworkPlugin,
224 config_group_name=group_name)
226 @ddt.data(
227 'fake',
228 '11.0.0.0-11.0.0.5-11.0.0.11',
229 '11.0.0.0-11.0.0.5',
230 '10.0.10.0-10.0.10.5',
231 '10.0.0.0-10.0.0.5,fake',
232 '10.0.10.0-10.0.10.5,10.0.0.0-10.0.0.5',
233 '10.0.10.0-10.0.10.5,10.0.0.10-10.0.10.5',
234 '10.0.0.0-10.0.0.5,10.0.10.0-10.0.10.5')
235 def test_invalid_init_incorrect_allowed_ip_ranges_v4(self, ip_range):
236 group_name = 'DEFAULT'
237 data = {
238 group_name: {
239 'standalone_network_plugin_gateway': '10.0.0.1',
240 'standalone_network_plugin_mask': '255.255.255.0',
241 'standalone_network_plugin_allowed_ip_ranges': ip_range,
242 },
243 }
244 with test_utils.create_temp_config_with_opts(data):
245 self.assertRaises(
246 exception.NetworkBadConfigurationException,
247 plugin.StandaloneNetworkPlugin,
248 config_group_name=group_name)
250 @ddt.data(
251 {'gateway': '2001:db8::0001', 'vers': 4},
252 {'gateway': '10.0.0.1', 'vers': 6})
253 @ddt.unpack
254 def test_invalid_init_mismatch_of_versions(self, gateway, vers):
255 group_name = 'DEFAULT'
256 data = {
257 group_name: {
258 'standalone_network_plugin_gateway': gateway,
259 'standalone_network_plugin_mask': '25',
260 },
261 }
262 if vers == 4:
263 data[group_name]['network_plugin_ipv4_enabled'] = True
264 if vers == 6:
265 data[group_name]['network_plugin_ipv4_enabled'] = False
266 data[group_name]['network_plugin_ipv6_enabled'] = True
268 with test_utils.create_temp_config_with_opts(data):
269 self.assertRaises(
270 exception.NetworkBadConfigurationException,
271 plugin.StandaloneNetworkPlugin,
272 config_group_name=group_name)
274 def test_deallocate_network(self):
275 share_server_id = 'fake_share_server_id'
276 data = {
277 'DEFAULT': {
278 'standalone_network_plugin_gateway': '10.0.0.1',
279 'standalone_network_plugin_mask': '24',
280 },
281 }
282 fake_allocations = [{'id': 'fake1'}, {'id': 'fake2'}]
283 with test_utils.create_temp_config_with_opts(data):
284 instance = plugin.StandaloneNetworkPlugin()
285 self.mock_object(
286 instance.db, 'network_allocations_get_for_share_server',
287 mock.Mock(return_value=fake_allocations))
288 self.mock_object(instance.db, 'network_allocation_delete')
290 instance.deallocate_network(fake_context, share_server_id)
292 (instance.db.network_allocations_get_for_share_server.
293 assert_called_once_with(fake_context, share_server_id))
294 (instance.db.network_allocation_delete.
295 assert_has_calls([
296 mock.call(fake_context, 'fake1'),
297 mock.call(fake_context, 'fake2'),
298 ]))
300 def test_allocate_network_zero_addresses_ipv4(self):
301 data = {
302 'DEFAULT': {
303 'standalone_network_plugin_gateway': '10.0.0.1',
304 'standalone_network_plugin_mask': '24',
305 },
306 }
307 with test_utils.create_temp_config_with_opts(data):
308 instance = plugin.StandaloneNetworkPlugin()
309 self.mock_object(instance.db, 'share_network_subnet_update')
311 allocations = instance.allocate_network(
312 fake_context, fake_share_server, fake_share_network,
313 fake_share_network_subnet, count=0)
315 self.assertEqual([], allocations)
316 instance.db.share_network_subnet_update.assert_called_once_with(
317 fake_context, fake_share_network_subnet['id'],
318 dict(network_type=None, segmentation_id=None,
319 cidr=str(instance.net.cidr),
320 gateway=str(instance.gateway),
321 ip_version=4,
322 mtu=1500))
324 def test_allocate_network_zero_addresses_ipv6(self):
325 data = {
326 'DEFAULT': {
327 'standalone_network_plugin_gateway': '2001:db8::0001',
328 'standalone_network_plugin_mask': '64',
329 'network_plugin_ipv6_enabled': True,
330 },
331 }
332 with test_utils.create_temp_config_with_opts(data):
333 instance = plugin.StandaloneNetworkPlugin()
334 self.mock_object(instance.db, 'share_network_subnet_update')
336 allocations = instance.allocate_network(
337 fake_context, fake_share_server, fake_share_network,
338 fake_share_network_subnet, count=0)
340 self.assertEqual([], allocations)
341 instance.db.share_network_subnet_update.assert_called_once_with(
342 fake_context, fake_share_network_subnet['id'],
343 dict(network_type=None, segmentation_id=None,
344 cidr=str(instance.net.cidr),
345 gateway=str(instance.gateway),
346 ip_version=6,
347 mtu=1500))
349 @ddt.data('admin', 'user')
350 def test_allocate_network_one_ip_address_ipv4_no_usages_exist(self, label):
351 data = {
352 'DEFAULT': {
353 'standalone_network_plugin_network_type': 'vlan',
354 'standalone_network_plugin_segmentation_id': 1003,
355 'standalone_network_plugin_gateway': '10.0.0.1',
356 'standalone_network_plugin_mask': '24',
357 },
358 }
359 with test_utils.create_temp_config_with_opts(data):
360 instance = plugin.StandaloneNetworkPlugin(label=label)
361 if label != 'admin':
362 self.mock_object(instance.db, 'share_network_subnet_update')
363 self.mock_object(instance.db, 'network_allocation_create')
364 self.mock_object(
365 instance.db, 'network_allocations_get_by_ip_address',
366 mock.Mock(return_value=[]))
368 allocations = instance.allocate_network(
369 fake_context, fake_share_server, fake_share_network,
370 fake_share_network_subnet)
372 self.assertEqual(1, len(allocations))
373 na_data = {
374 'network_type': 'vlan',
375 'segmentation_id': 1003,
376 'cidr': '10.0.0.0/24',
377 'gateway': '10.0.0.1',
378 'ip_version': 4,
379 'mtu': 1500,
380 }
381 if label != 'admin':
382 instance.db.share_network_subnet_update.assert_called_once_with(
383 fake_context, fake_share_network_subnet['id'], na_data)
384 na_data['share_network_subnet_id'] = \
385 fake_share_network_subnet['id']
386 instance.db.network_allocations_get_by_ip_address.assert_has_calls(
387 [mock.call(fake_context, '10.0.0.2')])
388 instance.db.network_allocation_create.assert_called_once_with(
389 fake_context,
390 dict(share_server_id=fake_share_server['id'],
391 ip_address='10.0.0.2', status=constants.STATUS_ACTIVE,
392 label=label, **na_data))
394 def test_allocate_network_two_ip_addresses_ipv4_two_usages_exist(self):
395 ctxt = type('FakeCtxt', (object,), {'fake': ['10.0.0.2', '10.0.0.4']})
397 def fake_get_allocations_by_ip_address(context, ip_address):
398 if ip_address not in context.fake:
399 context.fake.append(ip_address)
400 return []
401 else:
402 return context.fake
404 data = {
405 'DEFAULT': {
406 'standalone_network_plugin_gateway': '10.0.0.1',
407 'standalone_network_plugin_mask': '24',
408 },
409 }
410 with test_utils.create_temp_config_with_opts(data):
411 instance = plugin.StandaloneNetworkPlugin()
412 self.mock_object(instance.db, 'share_network_subnet_update')
413 self.mock_object(instance.db, 'network_allocation_create')
414 self.mock_object(
415 instance.db, 'network_allocations_get_by_ip_address',
416 mock.Mock(side_effect=fake_get_allocations_by_ip_address))
418 allocations = instance.allocate_network(
419 ctxt, fake_share_server, fake_share_network,
420 fake_share_network_subnet, count=2)
422 self.assertEqual(2, len(allocations))
423 na_data = {
424 'network_type': None,
425 'segmentation_id': None,
426 'cidr': str(instance.net.cidr),
427 'gateway': str(instance.gateway),
428 'ip_version': 4,
429 'mtu': 1500,
430 }
431 instance.db.share_network_subnet_update.assert_called_once_with(
432 ctxt, fake_share_network_subnet['id'], dict(**na_data))
433 instance.db.network_allocations_get_by_ip_address.assert_has_calls(
434 [mock.call(ctxt, '10.0.0.2'), mock.call(ctxt, '10.0.0.3'),
435 mock.call(ctxt, '10.0.0.4'), mock.call(ctxt, '10.0.0.5')])
436 na_data['share_network_subnet_id'] = fake_share_network_subnet['id']
437 instance.db.network_allocation_create.assert_has_calls([
438 mock.call(
439 ctxt,
440 dict(share_server_id=fake_share_server['id'],
441 ip_address='10.0.0.3', status=constants.STATUS_ACTIVE,
442 label='user', **na_data)),
443 mock.call(
444 ctxt,
445 dict(share_server_id=fake_share_server['id'],
446 ip_address='10.0.0.5', status=constants.STATUS_ACTIVE,
447 label='user', **na_data)),
448 ])
450 def test_allocate_network_no_available_ipv4_addresses(self):
451 data = {
452 'DEFAULT': {
453 'standalone_network_plugin_gateway': '10.0.0.1',
454 'standalone_network_plugin_mask': '30',
455 },
456 }
457 with test_utils.create_temp_config_with_opts(data):
458 instance = plugin.StandaloneNetworkPlugin()
459 self.mock_object(instance.db, 'share_network_subnet_update')
460 self.mock_object(instance.db, 'network_allocation_create')
461 self.mock_object(
462 instance.db, 'network_allocations_get_by_ip_address',
463 mock.Mock(return_value=['not empty list']))
465 self.assertRaises(
466 exception.NetworkBadConfigurationException,
467 instance.allocate_network,
468 fake_context, fake_share_server, fake_share_network,
469 fake_share_network_subnet)
471 instance.db.share_network_subnet_update.assert_called_once_with(
472 fake_context, fake_share_network_subnet['id'],
473 dict(network_type=None, segmentation_id=None,
474 cidr=str(instance.net.cidr),
475 gateway=str(instance.gateway),
476 ip_version=4,
477 mtu=1500))
478 instance.db.network_allocations_get_by_ip_address.assert_has_calls(
479 [mock.call(fake_context, '10.0.0.2')])
481 def _setup_manage_network_allocations(self, label=None):
482 data = {
483 'DEFAULT': {
484 'standalone_network_plugin_gateway': '192.168.0.1',
485 'standalone_network_plugin_mask': '24',
486 },
487 }
488 with test_utils.create_temp_config_with_opts(data):
489 instance = plugin.StandaloneNetworkPlugin(label=label)
491 return instance
493 @ddt.data('admin', None)
494 def test_manage_network_allocations(self, label):
495 allocations = ['192.168.0.11', '192.168.0.12', 'fd12::2000']
497 instance = self._setup_manage_network_allocations(label=label)
498 if not label:
499 self.mock_object(instance, '_verify_share_network_subnet')
500 self.mock_object(instance.db, 'share_network_subnet_update')
501 self.mock_object(instance.db, 'network_allocation_create')
503 result = instance.manage_network_allocations(
504 fake_context, allocations, fake_share_server,
505 fake_share_network, fake_share_network_subnet)
507 self.assertEqual(['fd12::2000'], result)
509 network_data = {
510 'network_type': instance.network_type,
511 'segmentation_id': instance.segmentation_id,
512 'cidr': str(instance.net.cidr),
513 'gateway': str(instance.gateway),
514 'ip_version': instance.ip_version,
515 'mtu': instance.mtu,
516 }
518 data_list = [{
519 'share_server_id': fake_share_server['id'],
520 'ip_address': x,
521 'status': constants.STATUS_ACTIVE,
522 'label': instance.label,
523 } for x in ['192.168.0.11', '192.168.0.12']]
525 data_list[0].update(network_data)
526 data_list[1].update(network_data)
528 if not label:
529 instance.db.share_network_subnet_update.assert_called_once_with(
530 fake_context, fake_share_network_subnet['id'], network_data)
531 data_list[0]['share_network_subnet_id'] = (
532 fake_share_network_subnet['id'])
533 data_list[1]['share_network_subnet_id'] = (
534 fake_share_network_subnet['id'])
535 instance._verify_share_network_subnet.assert_called_once_with(
536 fake_share_server['id'], fake_share_network_subnet)
538 instance.db.network_allocation_create.assert_has_calls([
539 mock.call(fake_context, data_list[0]),
540 mock.call(fake_context, data_list[1])
541 ])
543 def test_unmanage_network_allocations(self):
544 instance = self._setup_manage_network_allocations()
545 self.mock_object(instance, 'deallocate_network')
546 instance.unmanage_network_allocations('context', 'server_id')
547 instance.deallocate_network.assert_called_once_with(
548 'context', 'server_id')
550 def _setup_include_network_info(self):
551 data = {
552 'DEFAULT': {
553 'standalone_network_plugin_gateway': '192.168.0.1',
554 'standalone_network_plugin_mask': '24',
555 },
556 }
557 with test_utils.create_temp_config_with_opts(data):
558 instance = plugin.StandaloneNetworkPlugin()
560 return instance
562 def test_include_network_info(self):
563 instance = self._setup_include_network_info()
564 self.mock_object(instance, '_save_network_info')
565 instance.include_network_info(fake_share_network)
566 instance._save_network_info.assert_called_once_with(
567 None, fake_share_network, save_db=False)