1
# vim: tabstop=4 shiftwidth=4 softtabstop=4
3
# Copyright 2011 Rackspace
4
# Copyright (c) 2011 X.commerce, a business unit of eBay Inc.
7
# Licensed under the Apache License, Version 2.0 (the "License"); you may
8
# not use this file except in compliance with the License. You may obtain
9
# a copy of the License at
11
# http://www.apache.org/licenses/LICENSE-2.0
13
# Unless required by applicable law or agreed to in writing, software
14
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
15
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
16
# License for the specific language governing permissions and limitations
21
from nova import context
23
from nova import exception
24
from nova import flags
25
from nova import log as logging
29
from nova import utils
30
from nova.network import manager as network_manager
31
from nova.tests import fake_network
34
LOG = logging.getLogger(__name__)
41
'uuid': "aaaaaaaa-aaaa-aaaa-aaaa-aaaaaaaaaaaa",
45
'cidr': '192.168.0.0/24',
46
'cidr_v6': '2001:db8::/64',
47
'gateway_v6': '2001:db8::1',
49
'netmask': '255.255.255.0',
51
'bridge_interface': 'fake_fa0',
52
'gateway': '192.168.0.1',
53
'broadcast': '192.168.0.255',
54
'dns1': '192.168.0.1',
55
'dns2': '192.168.0.2',
58
'project_id': 'fake_project',
59
'vpn_public_address': '192.168.0.2'},
61
'uuid': "bbbbbbbb-bbbb-bbbb-bbbb-bbbbbbbbbbbb",
65
'cidr': '192.168.1.0/24',
66
'cidr_v6': '2001:db9::/64',
67
'gateway_v6': '2001:db9::1',
69
'netmask': '255.255.255.0',
71
'bridge_interface': 'fake_fa1',
72
'gateway': '192.168.1.1',
73
'broadcast': '192.168.1.255',
74
'dns1': '192.168.0.1',
75
'dns2': '192.168.0.2',
78
'project_id': 'fake_project',
79
'vpn_public_address': '192.168.1.2'}]
81
fixed_ips = [{'id': 0,
83
'address': '192.168.0.100',
86
'virtual_interface_id': 0,
90
'address': '192.168.1.100',
93
'virtual_interface_id': 0,
101
floating_ip_fields = {'id': 0,
102
'address': '192.168.10.100',
107
'auto_assigned': False}
110
'address': 'DE:AD:BE:EF:00:00',
111
'uuid': '00000000-0000-0000-0000-0000000000000000',
115
'address': 'DE:AD:BE:EF:00:01',
116
'uuid': '00000000-0000-0000-0000-0000000000000001',
120
'address': 'DE:AD:BE:EF:00:02',
121
'uuid': '00000000-0000-0000-0000-0000000000000002',
126
class FlatNetworkTestCase(test.TestCase):
128
super(FlatNetworkTestCase, self).setUp()
129
self.network = network_manager.FlatManager(host=HOST)
130
temp = utils.import_object('nova.network.minidns.MiniDNS')
131
self.network.instance_dns_manager = temp
132
self.network.instance_dns_domain = ''
134
self.context = context.RequestContext('testuser', 'testproject',
138
super(FlatNetworkTestCase, self).tearDown()
139
self.network.instance_dns_manager.delete_dns_file()
141
def test_get_instance_nw_info(self):
142
fake_get_instance_nw_info = fake_network.fake_get_instance_nw_info
144
nw_info = fake_get_instance_nw_info(self.stubs, 0, 2)
145
self.assertFalse(nw_info)
147
nw_info = fake_get_instance_nw_info(self.stubs, 1, 2)
149
for i, (nw, info) in enumerate(nw_info):
151
check = {'bridge': 'fake_br%d' % nid,
152
'cidr': '192.168.%s.0/24' % nid,
153
'cidr_v6': '2001:db8:0:%x::/64' % nid,
154
'id': '00000000-0000-0000-0000-00000000000000%02d' % nid,
157
'bridge_interface': None,
160
self.assertDictMatch(nw, check)
162
check = {'broadcast': '192.168.%d.255' % nid,
163
'dhcp_server': '192.168.%d.1' % nid,
164
'dns': ['192.168.%d.3' % nid, '192.168.%d.4' % nid],
165
'gateway': '192.168.%d.1' % nid,
166
'gateway_v6': 'fe80::def',
169
'label': 'test%d' % nid,
170
'mac': 'DE:AD:BE:EF:00:%02x' % nid,
173
'00000000-0000-0000-0000-00000000000000%02d' % nid,
174
'should_create_vlan': False,
175
'should_create_bridge': False}
176
self.assertDictMatch(info, check)
178
check = [{'enabled': 'DONTCARE',
179
'ip': '2001:db8:0:1::%x' % nid,
181
'gateway': 'fe80::def'}]
182
self.assertDictListMatch(info['ip6s'], check)
184
num_fixed_ips = len(info['ips'])
185
check = [{'enabled': 'DONTCARE',
186
'ip': '192.168.%d.%03d' % (nid, ip_num + 99),
187
'netmask': '255.255.255.0',
188
'gateway': '192.168.%d.1' % nid}
189
for ip_num in xrange(1, num_fixed_ips + 1)]
190
self.assertDictListMatch(info['ips'], check)
192
def test_validate_networks(self):
193
self.mox.StubOutWithMock(db, 'network_get')
194
self.mox.StubOutWithMock(db, 'network_get_all_by_uuids')
195
self.mox.StubOutWithMock(db, "fixed_ip_get_by_address")
197
requested_networks = [("bbbbbbbb-bbbb-bbbb-bbbb-bbbbbbbbbbbb",
199
db.network_get_all_by_uuids(mox.IgnoreArg(),
200
mox.IgnoreArg()).AndReturn(networks)
201
db.network_get(mox.IgnoreArg(),
202
mox.IgnoreArg()).AndReturn(networks[1])
204
ip = fixed_ips[1].copy()
205
ip['instance_id'] = None
206
db.fixed_ip_get_by_address(mox.IgnoreArg(),
207
mox.IgnoreArg()).AndReturn(ip)
210
self.network.validate_networks(self.context, requested_networks)
212
def test_validate_reserved(self):
213
context_admin = context.RequestContext('testuser', 'testproject',
215
nets = self.network.create_networks(context_admin, 'fake',
216
'192.168.0.0/24', False, 1,
217
256, None, None, None, None, None)
218
self.assertEqual(1, len(nets))
220
self.assertEqual(3, db.network_count_reserved_ips(context_admin,
223
def test_validate_networks_none_requested_networks(self):
224
self.network.validate_networks(self.context, None)
226
def test_validate_networks_empty_requested_networks(self):
227
requested_networks = []
230
self.network.validate_networks(self.context, requested_networks)
232
def test_validate_networks_invalid_fixed_ip(self):
233
self.mox.StubOutWithMock(db, 'network_get_all_by_uuids')
234
requested_networks = [(1, "192.168.0.100.1")]
235
db.network_get_all_by_uuids(mox.IgnoreArg(),
236
mox.IgnoreArg()).AndReturn(networks)
239
self.assertRaises(exception.FixedIpInvalid,
240
self.network.validate_networks, self.context,
243
def test_validate_networks_empty_fixed_ip(self):
244
self.mox.StubOutWithMock(db, 'network_get_all_by_uuids')
246
requested_networks = [(1, "")]
247
db.network_get_all_by_uuids(mox.IgnoreArg(),
248
mox.IgnoreArg()).AndReturn(networks)
251
self.assertRaises(exception.FixedIpInvalid,
252
self.network.validate_networks,
253
self.context, requested_networks)
255
def test_validate_networks_none_fixed_ip(self):
256
self.mox.StubOutWithMock(db, 'network_get_all_by_uuids')
258
requested_networks = [(1, None)]
259
db.network_get_all_by_uuids(mox.IgnoreArg(),
260
mox.IgnoreArg()).AndReturn(networks)
263
self.network.validate_networks(self.context, requested_networks)
265
def test_add_fixed_ip_instance_without_vpn_requested_networks(self):
266
self.mox.StubOutWithMock(db, 'network_get')
267
self.mox.StubOutWithMock(db, 'network_update')
268
self.mox.StubOutWithMock(db, 'fixed_ip_associate_pool')
269
self.mox.StubOutWithMock(db, 'instance_get')
270
self.mox.StubOutWithMock(db,
271
'virtual_interface_get_by_instance_and_network')
272
self.mox.StubOutWithMock(db, 'fixed_ip_update')
274
db.fixed_ip_update(mox.IgnoreArg(),
277
db.virtual_interface_get_by_instance_and_network(mox.IgnoreArg(),
278
mox.IgnoreArg(), mox.IgnoreArg()).AndReturn({'id': 0})
280
db.instance_get(mox.IgnoreArg(),
281
mox.IgnoreArg()).AndReturn({'security_groups':
283
db.instance_get(self.context,
284
1).AndReturn({'display_name': HOST,
285
'uuid': 'test-00001'})
286
db.instance_get(mox.IgnoreArg(),
287
mox.IgnoreArg()).AndReturn({'availability_zone': ''})
288
db.fixed_ip_associate_pool(mox.IgnoreArg(),
290
mox.IgnoreArg()).AndReturn('192.168.0.101')
291
db.network_get(mox.IgnoreArg(),
292
mox.IgnoreArg()).AndReturn(networks[0])
293
db.network_update(mox.IgnoreArg(), mox.IgnoreArg(), mox.IgnoreArg())
295
self.network.add_fixed_ip_to_instance(self.context, 1, HOST,
298
def test_mini_dns_driver(self):
299
zone1 = "example.org"
300
zone2 = "example.com"
301
driver = self.network.instance_dns_manager
302
driver.create_entry("hostone", "10.0.0.1", "A", zone1)
303
driver.create_entry("hosttwo", "10.0.0.2", "A", zone1)
304
driver.create_entry("hostthree", "10.0.0.3", "A", zone1)
305
driver.create_entry("hostfour", "10.0.0.4", "A", zone1)
306
driver.create_entry("hostfive", "10.0.0.5", "A", zone2)
308
driver.delete_entry("hostone", zone1)
309
driver.modify_address("hostfour", "10.0.0.1", zone1)
310
driver.modify_address("hostthree", "10.0.0.1", zone1)
311
names = driver.get_entries_by_address("10.0.0.1", zone1)
312
self.assertEqual(len(names), 2)
313
self.assertIn('hostthree', names)
314
self.assertIn('hostfour', names)
316
names = driver.get_entries_by_address("10.0.0.5", zone2)
317
self.assertEqual(len(names), 1)
318
self.assertIn('hostfive', names)
320
addresses = driver.get_entries_by_name("hosttwo", zone1)
321
self.assertEqual(len(addresses), 1)
322
self.assertIn('10.0.0.2', addresses)
324
self.assertRaises(exception.InvalidInput,
331
def test_instance_dns(self):
332
fixedip = '192.168.0.101'
333
self.mox.StubOutWithMock(db, 'network_get')
334
self.mox.StubOutWithMock(db, 'network_update')
335
self.mox.StubOutWithMock(db, 'fixed_ip_associate_pool')
336
self.mox.StubOutWithMock(db, 'instance_get')
337
self.mox.StubOutWithMock(db,
338
'virtual_interface_get_by_instance_and_network')
339
self.mox.StubOutWithMock(db, 'fixed_ip_update')
341
db.fixed_ip_update(mox.IgnoreArg(),
344
db.virtual_interface_get_by_instance_and_network(mox.IgnoreArg(),
345
mox.IgnoreArg(), mox.IgnoreArg()).AndReturn({'id': 0})
347
db.instance_get(mox.IgnoreArg(),
348
mox.IgnoreArg()).AndReturn({'security_groups':
351
db.instance_get(self.context,
352
1).AndReturn({'display_name': HOST,
353
'uuid': 'test-00001'})
354
db.instance_get(mox.IgnoreArg(),
355
mox.IgnoreArg()).AndReturn({'availability_zone': ''})
356
db.fixed_ip_associate_pool(mox.IgnoreArg(),
358
mox.IgnoreArg()).AndReturn(fixedip)
359
db.network_get(mox.IgnoreArg(),
360
mox.IgnoreArg()).AndReturn(networks[0])
361
db.network_update(mox.IgnoreArg(), mox.IgnoreArg(), mox.IgnoreArg())
364
self.network.add_fixed_ip_to_instance(self.context, 1, HOST,
366
instance_manager = self.network.instance_dns_manager
367
addresses = instance_manager.get_entries_by_name(HOST,
368
self.network.instance_dns_domain)
369
self.assertEqual(len(addresses), 1)
370
self.assertEqual(addresses[0], fixedip)
371
addresses = instance_manager.get_entries_by_name('test-00001',
372
self.network.instance_dns_domain)
373
self.assertEqual(len(addresses), 1)
374
self.assertEqual(addresses[0], fixedip)
377
class VlanNetworkTestCase(test.TestCase):
379
super(VlanNetworkTestCase, self).setUp()
380
self.network = network_manager.VlanManager(host=HOST)
382
self.context = context.RequestContext('testuser', 'testproject',
385
def test_vpn_allocate_fixed_ip(self):
386
self.mox.StubOutWithMock(db, 'fixed_ip_associate')
387
self.mox.StubOutWithMock(db, 'fixed_ip_update')
388
self.mox.StubOutWithMock(db,
389
'virtual_interface_get_by_instance_and_network')
391
db.fixed_ip_associate(mox.IgnoreArg(),
395
reserved=True).AndReturn('192.168.0.1')
396
db.fixed_ip_update(mox.IgnoreArg(),
399
db.virtual_interface_get_by_instance_and_network(mox.IgnoreArg(),
400
mox.IgnoreArg(), mox.IgnoreArg()).AndReturn({'id': 0})
403
network = dict(networks[0])
404
network['vpn_private_address'] = '192.168.0.2'
405
self.network.allocate_fixed_ip(None, 0, network, vpn=True)
407
def test_vpn_allocate_fixed_ip_no_network_id(self):
408
network = dict(networks[0])
409
network['vpn_private_address'] = '192.168.0.2'
411
context_admin = context.RequestContext('testuser', 'testproject',
413
self.assertRaises(exception.FixedIpNotFoundForNetwork,
414
self.network.allocate_fixed_ip,
420
def test_allocate_fixed_ip(self):
421
self.mox.StubOutWithMock(db, 'fixed_ip_associate_pool')
422
self.mox.StubOutWithMock(db, 'fixed_ip_update')
423
self.mox.StubOutWithMock(db,
424
'virtual_interface_get_by_instance_and_network')
425
self.mox.StubOutWithMock(db, 'instance_get')
427
db.instance_get(mox.IgnoreArg(),
428
mox.IgnoreArg()).AndReturn({'security_groups':
430
db.fixed_ip_associate_pool(mox.IgnoreArg(),
432
mox.IgnoreArg()).AndReturn('192.168.0.1')
433
db.fixed_ip_update(mox.IgnoreArg(),
436
db.virtual_interface_get_by_instance_and_network(mox.IgnoreArg(),
437
mox.IgnoreArg(), mox.IgnoreArg()).AndReturn({'id': 0})
440
network = dict(networks[0])
441
network['vpn_private_address'] = '192.168.0.2'
442
self.network.allocate_fixed_ip(self.context, 0, network)
444
def test_create_networks_too_big(self):
445
self.assertRaises(ValueError, self.network.create_networks, None,
446
num_networks=4094, vlan_start=1)
448
def test_create_networks_too_many(self):
449
self.assertRaises(ValueError, self.network.create_networks, None,
450
num_networks=100, vlan_start=1,
451
cidr='192.168.0.1/24', network_size=100)
453
def test_validate_networks(self):
454
def network_get(_context, network_id):
455
return networks[network_id]
457
self.stubs.Set(db, 'network_get', network_get)
458
self.mox.StubOutWithMock(db, 'network_get_all_by_uuids')
459
self.mox.StubOutWithMock(db, "fixed_ip_get_by_address")
461
requested_networks = [("bbbbbbbb-bbbb-bbbb-bbbb-bbbbbbbbbbbb",
463
db.network_get_all_by_uuids(mox.IgnoreArg(),
465
mox.IgnoreArg()).AndReturn(networks)
467
fixed_ips[1]['network_id'] = networks[1]['id']
468
fixed_ips[1]['instance_id'] = None
469
db.fixed_ip_get_by_address(mox.IgnoreArg(),
470
mox.IgnoreArg()).AndReturn(fixed_ips[1])
473
self.network.validate_networks(self.context, requested_networks)
475
def test_validate_networks_none_requested_networks(self):
476
self.network.validate_networks(self.context, None)
478
def test_validate_networks_empty_requested_networks(self):
479
requested_networks = []
482
self.network.validate_networks(self.context, requested_networks)
484
def test_validate_networks_invalid_fixed_ip(self):
485
self.mox.StubOutWithMock(db, 'network_get_all_by_uuids')
486
requested_networks = [(1, "192.168.0.100.1")]
487
db.network_get_all_by_uuids(mox.IgnoreArg(),
489
mox.IgnoreArg()).AndReturn(networks)
492
self.assertRaises(exception.FixedIpInvalid,
493
self.network.validate_networks, self.context,
496
def test_validate_networks_empty_fixed_ip(self):
497
self.mox.StubOutWithMock(db, 'network_get_all_by_uuids')
499
requested_networks = [(1, "")]
500
db.network_get_all_by_uuids(mox.IgnoreArg(),
502
mox.IgnoreArg()).AndReturn(networks)
505
self.assertRaises(exception.FixedIpInvalid,
506
self.network.validate_networks,
507
self.context, requested_networks)
509
def test_validate_networks_none_fixed_ip(self):
510
self.mox.StubOutWithMock(db, 'network_get_all_by_uuids')
512
requested_networks = [(1, None)]
513
db.network_get_all_by_uuids(mox.IgnoreArg(),
515
mox.IgnoreArg()).AndReturn(networks)
517
self.network.validate_networks(self.context, requested_networks)
519
def test_floating_ip_owned_by_project(self):
520
ctxt = context.RequestContext('testuser', 'testproject',
523
# raises because floating_ip project_id is None
524
floating_ip = {'address': '10.0.0.1',
526
self.assertRaises(exception.NotAuthorized,
527
self.network._floating_ip_owned_by_project,
531
# raises because floating_ip project_id is not equal to ctxt project_id
532
floating_ip = {'address': '10.0.0.1',
533
'project_id': ctxt.project_id + '1'}
534
self.assertRaises(exception.NotAuthorized,
535
self.network._floating_ip_owned_by_project,
539
# does not raise (floating ip is owned by ctxt project)
540
floating_ip = {'address': '10.0.0.1',
541
'project_id': ctxt.project_id}
542
self.network._floating_ip_owned_by_project(ctxt, floating_ip)
544
def test_allocate_floating_ip(self):
545
ctxt = context.RequestContext('testuser', 'testproject',
548
def fake1(*args, **kwargs):
549
return {'address': '10.0.0.1'}
551
def fake2(*args, **kwargs):
554
def fake3(*args, **kwargs):
557
self.stubs.Set(self.network.db, 'floating_ip_allocate_address', fake1)
559
# this time should raise
560
self.stubs.Set(self.network.db, 'floating_ip_count_by_project', fake2)
561
self.assertRaises(exception.QuotaError,
562
self.network.allocate_floating_ip,
566
# this time should not
567
self.stubs.Set(self.network.db, 'floating_ip_count_by_project', fake3)
568
self.network.allocate_floating_ip(ctxt, ctxt.project_id)
570
def test_deallocate_floating_ip(self):
571
ctxt = context.RequestContext('testuser', 'testproject',
574
def fake1(*args, **kwargs):
577
def fake2(*args, **kwargs):
578
return {'address': '10.0.0.1', 'fixed_ip_id': 1}
580
def fake3(*args, **kwargs):
581
return {'address': '10.0.0.1', 'fixed_ip_id': None}
583
self.stubs.Set(self.network.db, 'floating_ip_deallocate', fake1)
584
self.stubs.Set(self.network, '_floating_ip_owned_by_project', fake1)
586
# this time should raise because floating ip is associated to fixed_ip
587
self.stubs.Set(self.network.db, 'floating_ip_get_by_address', fake2)
588
self.assertRaises(exception.FloatingIpAssociated,
589
self.network.deallocate_floating_ip,
593
# this time should not raise
594
self.stubs.Set(self.network.db, 'floating_ip_get_by_address', fake3)
595
self.network.deallocate_floating_ip(ctxt, ctxt.project_id)
597
def test_associate_floating_ip(self):
598
ctxt = context.RequestContext('testuser', 'testproject',
601
def fake1(*args, **kwargs):
604
# floating ip that's already associated
605
def fake2(*args, **kwargs):
606
return {'address': '10.0.0.1',
611
# floating ip that isn't associated
612
def fake3(*args, **kwargs):
613
return {'address': '10.0.0.1',
618
# fixed ip with remote host
619
def fake4(*args, **kwargs):
620
return {'address': '10.0.0.1',
623
'network_id': 'blah'}
625
def fake4_network(*args, **kwargs):
626
return {'multi_host': False, 'host': 'jibberjabber'}
628
# fixed ip with local host
629
def fake5(*args, **kwargs):
630
return {'address': '10.0.0.1',
633
'network_id': 'blahblah'}
635
def fake5_network(*args, **kwargs):
636
return {'multi_host': False, 'host': 'testhost'}
638
def fake6(*args, **kwargs):
641
def fake7(*args, **kwargs):
644
def fake8(*args, **kwargs):
645
raise exception.ProcessExecutionError('',
646
'Cannot find device "em0"\n')
648
# raises because interface doesn't exist
649
self.stubs.Set(self.network.db,
650
'floating_ip_fixed_ip_associate',
652
self.stubs.Set(self.network.db, 'floating_ip_disassociate', fake1)
653
self.stubs.Set(self.network.driver, 'bind_floating_ip', fake8)
654
self.assertRaises(exception.NoFloatingIpInterface,
655
self.network._associate_floating_ip,
661
self.stubs.Set(self.network, '_floating_ip_owned_by_project', fake1)
663
# raises because floating_ip is already associated to a fixed_ip
664
self.stubs.Set(self.network.db, 'floating_ip_get_by_address', fake2)
665
self.assertRaises(exception.FloatingIpAssociated,
666
self.network.associate_floating_ip,
671
self.stubs.Set(self.network.db, 'floating_ip_get_by_address', fake3)
673
# does not raise and makes call remotely
675
self.stubs.Set(self.network.db, 'fixed_ip_get_by_address', fake4)
676
self.stubs.Set(self.network.db, 'network_get', fake4_network)
677
self.stubs.Set(rpc, 'cast', fake6)
678
self.network.associate_floating_ip(ctxt, mox.IgnoreArg(),
680
self.assertFalse(self.local)
682
# does not raise and makes call locally
684
self.stubs.Set(self.network.db, 'fixed_ip_get_by_address', fake5)
685
self.stubs.Set(self.network.db, 'network_get', fake5_network)
686
self.stubs.Set(self.network, '_associate_floating_ip', fake7)
687
self.network.associate_floating_ip(ctxt, mox.IgnoreArg(),
689
self.assertTrue(self.local)
691
def test_disassociate_floating_ip(self):
692
ctxt = context.RequestContext('testuser', 'testproject',
695
def fake1(*args, **kwargs):
698
# floating ip that isn't associated
699
def fake2(*args, **kwargs):
700
return {'address': '10.0.0.1',
705
# floating ip that is associated
706
def fake3(*args, **kwargs):
707
return {'address': '10.0.0.1',
712
# fixed ip with remote host
713
def fake4(*args, **kwargs):
714
return {'address': '10.0.0.1',
717
'network_id': 'blah'}
719
def fake4_network(*args, **kwargs):
720
return {'multi_host': False,
721
'host': 'jibberjabber'}
723
# fixed ip with local host
724
def fake5(*args, **kwargs):
725
return {'address': '10.0.0.1',
728
'network_id': 'blahblah'}
730
def fake5_network(*args, **kwargs):
731
return {'multi_host': False, 'host': 'testhost'}
733
def fake6(*args, **kwargs):
736
def fake7(*args, **kwargs):
739
self.stubs.Set(self.network, '_floating_ip_owned_by_project', fake1)
741
# raises because floating_ip is not associated to a fixed_ip
742
self.stubs.Set(self.network.db, 'floating_ip_get_by_address', fake2)
743
self.assertRaises(exception.FloatingIpNotAssociated,
744
self.network.disassociate_floating_ip,
748
self.stubs.Set(self.network.db, 'floating_ip_get_by_address', fake3)
750
# does not raise and makes call remotely
752
self.stubs.Set(self.network.db, 'fixed_ip_get', fake4)
753
self.stubs.Set(self.network.db, 'network_get', fake4_network)
754
self.stubs.Set(rpc, 'cast', fake6)
755
self.network.disassociate_floating_ip(ctxt, mox.IgnoreArg())
756
self.assertFalse(self.local)
758
# does not raise and makes call locally
760
self.stubs.Set(self.network.db, 'fixed_ip_get', fake5)
761
self.stubs.Set(self.network.db, 'network_get', fake5_network)
762
self.stubs.Set(self.network, '_disassociate_floating_ip', fake7)
763
self.network.disassociate_floating_ip(ctxt, mox.IgnoreArg())
764
self.assertTrue(self.local)
766
def test_add_fixed_ip_instance_without_vpn_requested_networks(self):
767
self.mox.StubOutWithMock(db, 'network_get')
768
self.mox.StubOutWithMock(db, 'fixed_ip_associate_pool')
769
self.mox.StubOutWithMock(db, 'instance_get')
770
self.mox.StubOutWithMock(db,
771
'virtual_interface_get_by_instance_and_network')
772
self.mox.StubOutWithMock(db, 'fixed_ip_update')
774
db.fixed_ip_update(mox.IgnoreArg(),
777
db.virtual_interface_get_by_instance_and_network(mox.IgnoreArg(),
778
mox.IgnoreArg(), mox.IgnoreArg()).AndReturn({'id': 0})
780
db.instance_get(mox.IgnoreArg(),
781
mox.IgnoreArg()).AndReturn({'security_groups':
783
'availability_zone': ''})
784
db.fixed_ip_associate_pool(mox.IgnoreArg(),
786
mox.IgnoreArg()).AndReturn('192.168.0.101')
787
db.network_get(mox.IgnoreArg(),
788
mox.IgnoreArg()).AndReturn(networks[0])
790
self.network.add_fixed_ip_to_instance(self.context, 1, HOST,
793
def test_ip_association_and_allocation_of_other_project(self):
794
"""Makes sure that we cannot deallocaate or disassociate
795
a public ip of other project"""
797
def network_get(_context, network_id):
798
return networks[network_id]
800
self.stubs.Set(db, 'network_get', network_get)
802
context1 = context.RequestContext('user', 'project1')
803
context2 = context.RequestContext('user', 'project2')
806
float_addr = db.floating_ip_create(context1.elevated(),
808
'project_id': context1.project_id})
810
instance = db.instance_create(context1,
811
{'project_id': 'project1'})
813
fix_addr = db.fixed_ip_associate_pool(context1.elevated(),
816
# Associate the IP with non-admin user context
817
self.assertRaises(exception.NotAuthorized,
818
self.network.associate_floating_ip,
823
# Deallocate address from other project
824
self.assertRaises(exception.NotAuthorized,
825
self.network.deallocate_floating_ip,
829
# Now Associates the address to the actual project
830
self.network.associate_floating_ip(context1, float_addr, fix_addr)
832
# Now try dis-associating from other project
833
self.assertRaises(exception.NotAuthorized,
834
self.network.disassociate_floating_ip,
838
# Clean up the ip addresses
839
self.network.disassociate_floating_ip(context1, float_addr)
840
self.network.deallocate_floating_ip(context1, float_addr)
841
self.network.deallocate_fixed_ip(context1, fix_addr)
842
db.floating_ip_destroy(context1.elevated(), float_addr)
843
db.fixed_ip_disassociate(context1.elevated(), fix_addr)
846
class CommonNetworkTestCase(test.TestCase):
849
super(CommonNetworkTestCase, self).setUp()
850
self.context = context.RequestContext('fake', 'fake')
852
def fake_create_fixed_ips(self, context, network_id, fixed_cidr=None):
855
def test_remove_fixed_ip_from_instance(self):
856
manager = fake_network.FakeNetworkManager()
857
manager.remove_fixed_ip_from_instance(self.context, 99, '10.0.0.1')
859
self.assertEquals(manager.deallocate_called, '10.0.0.1')
861
def test_remove_fixed_ip_from_instance_bad_input(self):
862
manager = fake_network.FakeNetworkManager()
863
self.assertRaises(exception.FixedIpNotFoundForSpecificInstance,
864
manager.remove_fixed_ip_from_instance,
865
self.context, 99, 'bad input')
867
def test_validate_cidrs(self):
868
manager = fake_network.FakeNetworkManager()
869
nets = manager.create_networks(None, 'fake', '192.168.0.0/24',
870
False, 1, 256, None, None, None,
872
self.assertEqual(1, len(nets))
873
cidrs = [str(net['cidr']) for net in nets]
874
self.assertTrue('192.168.0.0/24' in cidrs)
876
def test_validate_cidrs_split_exact_in_half(self):
877
manager = fake_network.FakeNetworkManager()
878
nets = manager.create_networks(None, 'fake', '192.168.0.0/24',
879
False, 2, 128, None, None, None,
881
self.assertEqual(2, len(nets))
882
cidrs = [str(net['cidr']) for net in nets]
883
self.assertTrue('192.168.0.0/25' in cidrs)
884
self.assertTrue('192.168.0.128/25' in cidrs)
886
def test_validate_cidrs_split_cidr_in_use_middle_of_range(self):
887
manager = fake_network.FakeNetworkManager()
888
self.mox.StubOutWithMock(manager.db, 'network_get_all')
889
ctxt = mox.IgnoreArg()
890
manager.db.network_get_all(ctxt).AndReturn([{'id': 1,
891
'cidr': '192.168.2.0/24'}])
893
nets = manager.create_networks(None, 'fake', '192.168.0.0/16',
894
False, 4, 256, None, None, None,
896
self.assertEqual(4, len(nets))
897
cidrs = [str(net['cidr']) for net in nets]
898
exp_cidrs = ['192.168.0.0/24', '192.168.1.0/24', '192.168.3.0/24',
900
for exp_cidr in exp_cidrs:
901
self.assertTrue(exp_cidr in cidrs)
902
self.assertFalse('192.168.2.0/24' in cidrs)
904
def test_validate_cidrs_smaller_subnet_in_use(self):
905
manager = fake_network.FakeNetworkManager()
906
self.mox.StubOutWithMock(manager.db, 'network_get_all')
907
ctxt = mox.IgnoreArg()
908
manager.db.network_get_all(ctxt).AndReturn([{'id': 1,
909
'cidr': '192.168.2.9/25'}])
911
# ValueError: requested cidr (192.168.2.0/24) conflicts with
912
# existing smaller cidr
913
args = (None, 'fake', '192.168.2.0/24', False, 1, 256, None, None,
915
self.assertRaises(ValueError, manager.create_networks, *args)
917
def test_validate_cidrs_split_smaller_cidr_in_use(self):
918
manager = fake_network.FakeNetworkManager()
919
self.mox.StubOutWithMock(manager.db, 'network_get_all')
920
ctxt = mox.IgnoreArg()
921
manager.db.network_get_all(ctxt).AndReturn([{'id': 1,
922
'cidr': '192.168.2.0/25'}])
924
nets = manager.create_networks(None, 'fake', '192.168.0.0/16',
925
False, 4, 256, None, None, None, None,
927
self.assertEqual(4, len(nets))
928
cidrs = [str(net['cidr']) for net in nets]
929
exp_cidrs = ['192.168.0.0/24', '192.168.1.0/24', '192.168.3.0/24',
931
for exp_cidr in exp_cidrs:
932
self.assertTrue(exp_cidr in cidrs)
933
self.assertFalse('192.168.2.0/24' in cidrs)
935
def test_validate_cidrs_split_smaller_cidr_in_use2(self):
936
manager = fake_network.FakeNetworkManager()
937
self.mox.StubOutWithMock(manager.db, 'network_get_all')
938
ctxt = mox.IgnoreArg()
939
manager.db.network_get_all(ctxt).AndReturn([{'id': 1,
940
'cidr': '192.168.2.9/29'}])
942
nets = manager.create_networks(None, 'fake', '192.168.2.0/24',
943
False, 3, 32, None, None, None, None,
945
self.assertEqual(3, len(nets))
946
cidrs = [str(net['cidr']) for net in nets]
947
exp_cidrs = ['192.168.2.32/27', '192.168.2.64/27', '192.168.2.96/27']
948
for exp_cidr in exp_cidrs:
949
self.assertTrue(exp_cidr in cidrs)
950
self.assertFalse('192.168.2.0/27' in cidrs)
952
def test_validate_cidrs_split_all_in_use(self):
953
manager = fake_network.FakeNetworkManager()
954
self.mox.StubOutWithMock(manager.db, 'network_get_all')
955
ctxt = mox.IgnoreArg()
956
in_use = [{'id': 1, 'cidr': '192.168.2.9/29'},
957
{'id': 2, 'cidr': '192.168.2.64/26'},
958
{'id': 3, 'cidr': '192.168.2.128/26'}]
959
manager.db.network_get_all(ctxt).AndReturn(in_use)
961
args = (None, 'fake', '192.168.2.0/24', False, 3, 64, None, None,
963
# ValueError: Not enough subnets avail to satisfy requested num_
964
# networks - some subnets in requested range already
966
self.assertRaises(ValueError, manager.create_networks, *args)
968
def test_validate_cidrs_one_in_use(self):
969
manager = fake_network.FakeNetworkManager()
970
args = (None, 'fake', '192.168.0.0/24', False, 2, 256, None, None,
972
# ValueError: network_size * num_networks exceeds cidr size
973
self.assertRaises(ValueError, manager.create_networks, *args)
975
def test_validate_cidrs_already_used(self):
976
manager = fake_network.FakeNetworkManager()
977
self.mox.StubOutWithMock(manager.db, 'network_get_all')
978
ctxt = mox.IgnoreArg()
979
manager.db.network_get_all(ctxt).AndReturn([{'id': 1,
980
'cidr': '192.168.0.0/24'}])
982
# ValueError: cidr already in use
983
args = (None, 'fake', '192.168.0.0/24', False, 1, 256, None, None,
985
self.assertRaises(ValueError, manager.create_networks, *args)
987
def test_validate_cidrs_too_many(self):
988
manager = fake_network.FakeNetworkManager()
989
args = (None, 'fake', '192.168.0.0/24', False, 200, 256, None, None,
991
# ValueError: Not enough subnets avail to satisfy requested
993
self.assertRaises(ValueError, manager.create_networks, *args)
995
def test_validate_cidrs_split_partial(self):
996
manager = fake_network.FakeNetworkManager()
997
nets = manager.create_networks(None, 'fake', '192.168.0.0/16',
998
False, 2, 256, None, None, None, None,
1000
returned_cidrs = [str(net['cidr']) for net in nets]
1001
self.assertTrue('192.168.0.0/24' in returned_cidrs)
1002
self.assertTrue('192.168.1.0/24' in returned_cidrs)
1004
def test_validate_cidrs_conflict_existing_supernet(self):
1005
manager = fake_network.FakeNetworkManager()
1006
self.mox.StubOutWithMock(manager.db, 'network_get_all')
1007
ctxt = mox.IgnoreArg()
1008
fakecidr = [{'id': 1, 'cidr': '192.168.0.0/8'}]
1009
manager.db.network_get_all(ctxt).AndReturn(fakecidr)
1010
self.mox.ReplayAll()
1011
args = (None, 'fake', '192.168.0.0/24', False, 1, 256, None, None,
1013
# ValueError: requested cidr (192.168.0.0/24) conflicts
1014
# with existing supernet
1015
self.assertRaises(ValueError, manager.create_networks, *args)
1017
def test_create_networks(self):
1018
cidr = '192.168.0.0/24'
1019
manager = fake_network.FakeNetworkManager()
1020
self.stubs.Set(manager, '_create_fixed_ips',
1021
self.fake_create_fixed_ips)
1022
args = [None, 'foo', cidr, None, 1, 256, 'fd00::/48', None, None,
1024
self.assertTrue(manager.create_networks(*args))
1026
def test_create_networks_cidr_already_used(self):
1027
manager = fake_network.FakeNetworkManager()
1028
self.mox.StubOutWithMock(manager.db, 'network_get_all')
1029
ctxt = mox.IgnoreArg()
1030
fakecidr = [{'id': 1, 'cidr': '192.168.0.0/24'}]
1031
manager.db.network_get_all(ctxt).AndReturn(fakecidr)
1032
self.mox.ReplayAll()
1033
args = [None, 'foo', '192.168.0.0/24', None, 1, 256,
1034
'fd00::/48', None, None, None, None, None]
1035
self.assertRaises(ValueError, manager.create_networks, *args)
1037
def test_create_networks_many(self):
1038
cidr = '192.168.0.0/16'
1039
manager = fake_network.FakeNetworkManager()
1040
self.stubs.Set(manager, '_create_fixed_ips',
1041
self.fake_create_fixed_ips)
1042
args = [None, 'foo', cidr, None, 10, 256, 'fd00::/48', None, None,
1044
self.assertTrue(manager.create_networks(*args))
1046
def test_get_instance_uuids_by_ip_regex(self):
1047
manager = fake_network.FakeNetworkManager()
1048
_vifs = manager.db.virtual_interface_get_all(None)
1049
fake_context = context.RequestContext('user', 'project')
1051
# Greedy get eveything
1052
res = manager.get_instance_uuids_by_ip_filter(fake_context,
1054
self.assertEqual(len(res), len(_vifs))
1057
res = manager.get_instance_uuids_by_ip_filter(fake_context,
1059
self.assertFalse(res)
1062
res = manager.get_instance_uuids_by_ip_filter(fake_context,
1063
{'ip': '172.16.0.2'})
1064
self.assertTrue(res)
1065
self.assertEqual(len(res), 1)
1066
self.assertEqual(res[0]['instance_id'], _vifs[1]['instance_id'])
1069
res = manager.get_instance_uuids_by_ip_filter(fake_context,
1070
{'ip': '173.16.0.2'})
1071
self.assertTrue(res)
1072
self.assertEqual(len(res), 1)
1073
self.assertEqual(res[0]['instance_id'], _vifs[2]['instance_id'])
1075
# Get instance 0 and 1
1076
res = manager.get_instance_uuids_by_ip_filter(fake_context,
1077
{'ip': '172.16.0.*'})
1078
self.assertTrue(res)
1079
self.assertEqual(len(res), 2)
1080
self.assertEqual(res[0]['instance_id'], _vifs[0]['instance_id'])
1081
self.assertEqual(res[1]['instance_id'], _vifs[1]['instance_id'])
1083
# Get instance 1 and 2
1084
res = manager.get_instance_uuids_by_ip_filter(fake_context,
1085
{'ip': '17..16.0.2'})
1086
self.assertTrue(res)
1087
self.assertEqual(len(res), 2)
1088
self.assertEqual(res[0]['instance_id'], _vifs[1]['instance_id'])
1089
self.assertEqual(res[1]['instance_id'], _vifs[2]['instance_id'])
1091
def test_get_instance_uuids_by_ipv6_regex(self):
1092
manager = fake_network.FakeNetworkManager()
1093
_vifs = manager.db.virtual_interface_get_all(None)
1094
fake_context = context.RequestContext('user', 'project')
1096
# Greedy get eveything
1097
res = manager.get_instance_uuids_by_ip_filter(fake_context,
1099
self.assertEqual(len(res), len(_vifs))
1102
res = manager.get_instance_uuids_by_ip_filter(fake_context,
1103
{'ip6': '.*1034.*'})
1104
self.assertFalse(res)
1107
res = manager.get_instance_uuids_by_ip_filter(fake_context,
1108
{'ip6': '2001:.*2'})
1109
self.assertTrue(res)
1110
self.assertEqual(len(res), 1)
1111
self.assertEqual(res[0]['instance_id'], _vifs[1]['instance_id'])
1114
ip6 = '2001:db8:69:1f:dead:beff:feff:ef03'
1115
res = manager.get_instance_uuids_by_ip_filter(fake_context,
1117
self.assertTrue(res)
1118
self.assertEqual(len(res), 1)
1119
self.assertEqual(res[0]['instance_id'], _vifs[2]['instance_id'])
1121
# Get instance 0 and 1
1122
res = manager.get_instance_uuids_by_ip_filter(fake_context,
1123
{'ip6': '.*ef0[1,2]'})
1124
self.assertTrue(res)
1125
self.assertEqual(len(res), 2)
1126
self.assertEqual(res[0]['instance_id'], _vifs[0]['instance_id'])
1127
self.assertEqual(res[1]['instance_id'], _vifs[1]['instance_id'])
1129
# Get instance 1 and 2
1130
ip6 = '2001:db8:69:1.:dead:beff:feff:ef0.'
1131
res = manager.get_instance_uuids_by_ip_filter(fake_context,
1133
self.assertTrue(res)
1134
self.assertEqual(len(res), 2)
1135
self.assertEqual(res[0]['instance_id'], _vifs[1]['instance_id'])
1136
self.assertEqual(res[1]['instance_id'], _vifs[2]['instance_id'])
1138
def test_get_instance_uuids_by_ip(self):
1139
manager = fake_network.FakeNetworkManager()
1140
_vifs = manager.db.virtual_interface_get_all(None)
1141
fake_context = context.RequestContext('user', 'project')
1144
res = manager.get_instance_uuids_by_ip_filter(fake_context,
1146
self.assertFalse(res)
1150
res = manager.get_instance_uuids_by_ip_filter(fake_context,
1152
self.assertFalse(res)
1156
res = manager.get_instance_uuids_by_ip_filter(fake_context,
1158
self.assertTrue(res)
1159
self.assertEqual(len(res), 1)
1160
self.assertEqual(res[0]['instance_id'], _vifs[1]['instance_id'])
1164
res = manager.get_instance_uuids_by_ip_filter(fake_context,
1166
self.assertTrue(res)
1167
self.assertEqual(len(res), 1)
1168
self.assertEqual(res[0]['instance_id'], _vifs[2]['instance_id'])
1170
def test_get_network(self):
1171
manager = fake_network.FakeNetworkManager()
1172
fake_context = context.RequestContext('user', 'project')
1173
self.mox.StubOutWithMock(manager.db, 'network_get_all_by_uuids')
1174
manager.db.network_get_all_by_uuids(
1176
mox.IgnoreArg()).AndReturn(networks)
1177
self.mox.ReplayAll()
1178
uuid = 'aaaaaaaa-aaaa-aaaa-aaaa-aaaaaaaaaaaa'
1179
network = manager.get_network(fake_context, uuid)
1180
self.assertEqual(network['uuid'], uuid)
1182
def test_get_network_not_found(self):
1183
manager = fake_network.FakeNetworkManager()
1184
fake_context = context.RequestContext('user', 'project')
1185
self.mox.StubOutWithMock(manager.db, 'network_get_all_by_uuids')
1186
manager.db.network_get_all_by_uuids(mox.IgnoreArg(),
1187
mox.IgnoreArg()).AndReturn([])
1188
self.mox.ReplayAll()
1189
uuid = 'eeeeeeee-eeee-eeee-eeee-eeeeeeeeeeee'
1190
self.assertRaises(exception.NetworkNotFound,
1191
manager.get_network, fake_context, uuid)
1193
def test_get_all_networks(self):
1194
manager = fake_network.FakeNetworkManager()
1195
fake_context = context.RequestContext('user', 'project')
1196
self.mox.StubOutWithMock(manager.db, 'network_get_all')
1197
manager.db.network_get_all(mox.IgnoreArg()).AndReturn(networks)
1198
self.mox.ReplayAll()
1199
output = manager.get_all_networks(fake_context)
1200
self.assertEqual(len(networks), 2)
1201
self.assertEqual(output[0]['uuid'],
1202
'aaaaaaaa-aaaa-aaaa-aaaa-aaaaaaaaaaaa')
1203
self.assertEqual(output[1]['uuid'],
1204
'bbbbbbbb-bbbb-bbbb-bbbb-bbbbbbbbbbbb')
1206
def test_disassociate_network(self):
1207
manager = fake_network.FakeNetworkManager()
1208
fake_context = context.RequestContext('user', 'project')
1209
self.mox.StubOutWithMock(manager.db, 'network_get_all_by_uuids')
1210
manager.db.network_get_all_by_uuids(
1212
mox.IgnoreArg()).AndReturn(networks)
1213
self.mox.ReplayAll()
1214
uuid = 'aaaaaaaa-aaaa-aaaa-aaaa-aaaaaaaaaaaa'
1215
manager.disassociate_network(fake_context, uuid)
1217
def test_disassociate_network_not_found(self):
1218
manager = fake_network.FakeNetworkManager()
1219
fake_context = context.RequestContext('user', 'project')
1220
self.mox.StubOutWithMock(manager.db, 'network_get_all_by_uuids')
1221
manager.db.network_get_all_by_uuids(mox.IgnoreArg(),
1222
mox.IgnoreArg()).AndReturn([])
1223
self.mox.ReplayAll()
1224
uuid = 'eeeeeeee-eeee-eeee-eeee-eeeeeeeeeeee'
1225
self.assertRaises(exception.NetworkNotFound,
1226
manager.disassociate_network, fake_context, uuid)
1229
class TestRPCFixedManager(network_manager.RPCAllocateFixedIP,
1230
network_manager.NetworkManager):
1231
"""Dummy manager that implements RPCAllocateFixedIP"""
1234
class RPCAllocateTestCase(test.TestCase):
1235
"""Tests nova.network.manager.RPCAllocateFixedIP"""
1237
super(RPCAllocateTestCase, self).setUp()
1238
self.rpc_fixed = TestRPCFixedManager()
1239
self.context = context.RequestContext('fake', 'fake')
1241
def test_rpc_allocate(self):
1242
"""Test to verify bug 855030 doesn't resurface.
1244
Mekes sure _rpc_allocate_fixed_ip returns a value so the call
1245
returns properly and the greenpool completes."""
1246
address = '10.10.10.10'
1248
def fake_allocate(*args, **kwargs):
1251
def fake_network_get(*args, **kwargs):
1254
self.stubs.Set(self.rpc_fixed, 'allocate_fixed_ip', fake_allocate)
1255
self.stubs.Set(self.rpc_fixed.db, 'network_get', fake_network_get)
1256
rval = self.rpc_fixed._rpc_allocate_fixed_ip(self.context,
1259
self.assertEqual(rval, address)
1262
class TestFloatingIPManager(network_manager.FloatingIP,
1263
network_manager.NetworkManager):
1264
"""Dummy manager that implements FloatingIP"""
1267
class FloatingIPTestCase(test.TestCase):
1268
"""Tests nova.network.manager.FloatingIP"""
1270
super(FloatingIPTestCase, self).setUp()
1271
self.network = TestFloatingIPManager()
1272
temp = utils.import_object('nova.network.minidns.MiniDNS')
1273
self.network.floating_dns_manager = temp
1274
self.network.db = db
1275
self.project_id = 'testproject'
1276
self.context = context.RequestContext('testuser', self.project_id,
1280
super(FloatingIPTestCase, self).tearDown()
1281
self.network.floating_dns_manager.delete_dns_file()
1283
def test_double_deallocation(self):
1284
instance_ref = db.api.instance_create(self.context,
1285
{"project_id": self.project_id})
1286
# Run it twice to make it fault if it does not handle
1287
# instances without fixed networks
1288
# If this fails in either, it does not handle having no addresses
1289
self.network.deallocate_for_instance(self.context,
1290
instance_id=instance_ref['id'])
1291
self.network.deallocate_for_instance(self.context,
1292
instance_id=instance_ref['id'])
1294
def test_deallocation_deleted_instance(self):
1295
instance_ref = db.api.instance_create(self.context,
1296
{"project_id": self.project_id, "deleted": True})
1297
self.network.deallocate_for_instance(self.context,
1298
instance_id=instance_ref['id'])
1300
def test_floating_dns_create_conflict(self):
1301
zone = "example.org"
1302
address1 = "10.10.10.11"
1306
self.network.add_dns_entry(self.context, address1, name1, "A", zone)
1308
self.assertRaises(exception.FloatingIpDNSExists,
1309
self.network.add_dns_entry, self.context,
1310
address1, name1, "A", zone)
1312
def test_floating_create_and_get(self):
1313
zone = "example.org"
1314
address1 = "10.10.10.11"
1317
entries = self.network.get_dns_entries_by_address(self.context,
1319
self.assertFalse(entries)
1321
self.network.add_dns_entry(self.context, address1, name1, "A", zone)
1322
self.network.add_dns_entry(self.context, address1, name2, "A", zone)
1323
entries = self.network.get_dns_entries_by_address(self.context,
1325
self.assertEquals(len(entries), 2)
1326
self.assertEquals(entries[0], name1)
1327
self.assertEquals(entries[1], name2)
1329
entries = self.network.get_dns_entries_by_name(self.context,
1331
self.assertEquals(len(entries), 1)
1332
self.assertEquals(entries[0], address1)
1334
def test_floating_dns_delete(self):
1335
zone = "example.org"
1336
address1 = "10.10.10.11"
1340
self.network.add_dns_entry(self.context, address1, name1, "A", zone)
1341
self.network.add_dns_entry(self.context, address1, name2, "A", zone)
1342
self.network.delete_dns_entry(self.context, name1, zone)
1344
entries = self.network.get_dns_entries_by_address(self.context,
1346
self.assertEquals(len(entries), 1)
1347
self.assertEquals(entries[0], name2)
1349
self.assertRaises(exception.NotFound,
1350
self.network.delete_dns_entry, self.context,
1353
def test_floating_dns_domains_public(self):
1355
domain1 = "example.org"
1356
domain2 = "example.com"
1357
address1 = '10.10.10.10'
1358
entryname = 'testentry'
1360
context_admin = context.RequestContext('testuser', 'testproject',
1363
self.assertRaises(exception.AdminRequired,
1364
self.network.create_public_dns_domain, self.context,
1366
self.network.create_public_dns_domain(context_admin, domain1,
1368
self.network.create_public_dns_domain(context_admin, domain2,
1371
domains = self.network.get_dns_domains(self.context)
1372
self.assertEquals(len(domains), 2)
1373
self.assertEquals(domains[0]['domain'], domain1)
1374
self.assertEquals(domains[1]['domain'], domain2)
1375
self.assertEquals(domains[0]['project'], 'testproject')
1376
self.assertEquals(domains[1]['project'], 'fakeproject')
1378
self.network.add_dns_entry(self.context, address1, entryname,
1380
entries = self.network.get_dns_entries_by_name(self.context,
1382
self.assertEquals(len(entries), 1)
1383
self.assertEquals(entries[0], address1)
1385
self.assertRaises(exception.AdminRequired,
1386
self.network.delete_dns_domain, self.context,
1388
self.network.delete_dns_domain(context_admin, domain1)
1389
self.network.delete_dns_domain(context_admin, domain2)
1391
# Verify that deleting the domain deleted the associated entry
1392
entries = self.network.get_dns_entries_by_name(self.context,
1394
self.assertFalse(entries)
1396
def test_delete_all_by_ip(self):
1397
domain1 = "example.org"
1398
domain2 = "example.com"
1399
address = "10.10.10.10"
1403
def fake_domains(context):
1404
return [{'domain': 'example.org', 'scope': 'public'},
1405
{'domain': 'example.com', 'scope': 'public'},
1406
{'domain': 'test.example.org', 'scope': 'public'}]
1408
self.stubs.Set(self.network, 'get_dns_domains', fake_domains)
1410
context_admin = context.RequestContext('testuser', 'testproject',
1413
self.network.create_public_dns_domain(context_admin, domain1,
1415
self.network.create_public_dns_domain(context_admin, domain2,
1418
domains = self.network.get_dns_domains(self.context)
1419
for domain in domains:
1420
self.network.add_dns_entry(self.context, address,
1421
name1, "A", domain['domain'])
1422
self.network.add_dns_entry(self.context, address,
1423
name2, "A", domain['domain'])
1424
entries = self.network.get_dns_entries_by_address(self.context,
1427
self.assertEquals(len(entries), 2)
1429
self.network._delete_all_entries_for_ip(self.context, address)
1431
for domain in domains:
1432
entries = self.network.get_dns_entries_by_address(self.context,
1435
self.assertFalse(entries)
1437
self.network.delete_dns_domain(context_admin, domain1)
1438
self.network.delete_dns_domain(context_admin, domain2)
1441
class NetworkPolicyTestCase(test.TestCase):
1443
super(NetworkPolicyTestCase, self).setUp()
1448
self.context = context.get_admin_context()
1451
super(NetworkPolicyTestCase, self).tearDown()
1454
def _set_rules(self, rules):
1455
nova.common.policy.set_brain(nova.common.policy.HttpBrain(rules))
1457
def test_check_policy(self):
1458
self.mox.StubOutWithMock(nova.policy, 'enforce')
1460
'project_id': self.context.project_id,
1461
'user_id': self.context.user_id,
1463
nova.policy.enforce(self.context, 'network:get_all', target)
1464
self.mox.ReplayAll()
1465
network_manager.check_policy(self.context, 'get_all')
1466
self.mox.UnsetStubs()
1467
self.mox.VerifyAll()
1470
class InstanceDNSTestCase(test.TestCase):
1471
"""Tests nova.network.manager instance DNS"""
1473
super(InstanceDNSTestCase, self).setUp()
1474
self.network = TestFloatingIPManager()
1475
temp = utils.import_object('nova.network.minidns.MiniDNS')
1476
self.network.instance_dns_manager = temp
1477
temp = utils.import_object('nova.network.dns_driver.DNSDriver')
1478
self.network.floating_dns_manager = temp
1479
self.network.db = db
1480
self.project_id = 'testproject'
1481
self.context = context.RequestContext('testuser', self.project_id,
1485
super(InstanceDNSTestCase, self).tearDown()
1486
self.network.instance_dns_manager.delete_dns_file()
1488
def test_dns_domains_private(self):
1490
domain1 = 'example.org'
1492
context_admin = context.RequestContext('testuser', 'testproject',
1495
self.assertRaises(exception.AdminRequired,
1496
self.network.create_private_dns_domain, self.context,
1499
self.network.create_private_dns_domain(context_admin, domain1, zone1)
1500
domains = self.network.get_dns_domains(self.context)
1501
self.assertEquals(len(domains), 1)
1502
self.assertEquals(domains[0]['domain'], domain1)
1503
self.assertEquals(domains[0]['availability_zone'], zone1)
1505
self.assertRaises(exception.AdminRequired,
1506
self.network.delete_dns_domain, self.context,
1508
self.network.delete_dns_domain(context_admin, domain1)
1511
domain1 = "example.org"
1512
domain2 = "example.com"
1515
class LdapDNSTestCase(test.TestCase):
1516
"""Tests nova.network.ldapdns.LdapDNS"""
1518
super(LdapDNSTestCase, self).setUp()
1520
self.saved_ldap = sys.modules.get('ldap')
1521
import nova.auth.fakeldap
1522
sys.modules['ldap'] = nova.auth.fakeldap
1524
temp = utils.import_object('nova.network.ldapdns.FakeLdapDNS')
1526
self.driver.create_domain(domain1)
1527
self.driver.create_domain(domain2)
1530
super(LdapDNSTestCase, self).tearDown()
1531
self.driver.delete_domain(domain1)
1532
self.driver.delete_domain(domain2)
1533
sys.modules['ldap'] = self.saved_ldap
1535
def test_ldap_dns_domains(self):
1536
domains = self.driver.get_domains()
1537
self.assertEqual(len(domains), 2)
1538
self.assertIn(domain1, domains)
1539
self.assertIn(domain2, domains)
1541
def test_ldap_dns_create_conflict(self):
1542
address1 = "10.10.10.11"
1546
self.driver.create_entry(name1, address1, "A", domain1)
1548
self.assertRaises(exception.FloatingIpDNSExists,
1549
self.driver.create_entry,
1550
name1, address1, "A", domain1)
1552
def test_ldap_dns_create_and_get(self):
1553
address1 = "10.10.10.11"
1556
entries = self.driver.get_entries_by_address(address1, domain1)
1557
self.assertFalse(entries)
1559
self.driver.create_entry(name1, address1, "A", domain1)
1560
self.driver.create_entry(name2, address1, "A", domain1)
1561
entries = self.driver.get_entries_by_address(address1, domain1)
1562
self.assertEquals(len(entries), 2)
1563
self.assertEquals(entries[0], name1)
1564
self.assertEquals(entries[1], name2)
1566
entries = self.driver.get_entries_by_name(name1, domain1)
1567
self.assertEquals(len(entries), 1)
1568
self.assertEquals(entries[0], address1)
1570
def test_ldap_dns_delete(self):
1571
address1 = "10.10.10.11"
1575
self.driver.create_entry(name1, address1, "A", domain1)
1576
self.driver.create_entry(name2, address1, "A", domain1)
1577
entries = self.driver.get_entries_by_address(address1, domain1)
1578
self.assertEquals(len(entries), 2)
1580
self.driver.delete_entry(name1, domain1)
1581
entries = self.driver.get_entries_by_address(address1, domain1)
1582
LOG.debug("entries: %s" % entries)
1583
self.assertEquals(len(entries), 1)
1584
self.assertEquals(entries[0], name2)
1586
self.assertRaises(exception.NotFound,
1587
self.driver.delete_entry,
added
removed
nova/tests/test_network.py
