1
# vim: tabstop=4 shiftwidth=4 softtabstop=4
3
# Copyright 2011,2012 Nicira, Inc.
6
# Licensed under the Apache License, Version 2.0 (the "License"); you may
7
# not use this file except in compliance with the License. You may obtain
8
# a copy of the License at
10
# http://www.apache.org/licenses/LICENSE-2.0
12
# Unless required by applicable law or agreed to in writing, software
13
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
14
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
15
# License for the specific language governing permissions and limitations
20
from nova import context
22
from nova.db.sqlalchemy import models
23
from nova.db.sqlalchemy import session as sql_session
24
from nova import exception
25
from nova import flags
26
from nova.network.quantum import client as quantum_client
27
from nova.network.quantum import fake_client
28
from nova.network.quantum import manager as quantum_manager
29
from nova.network.quantum import melange_connection
30
from nova.network.quantum import melange_ipam_lib
31
from nova.network.quantum import quantum_connection
32
from nova.openstack.common import log as logging
35
from nova import utils
37
LOG = logging.getLogger(__name__)
41
networks = [{'label': 'project1-net1',
44
'cidr': '100.168.0.0/24',
45
'cidr_v6': '100:1db8::/64',
46
'gateway_v6': '100:1db8::1',
48
'netmask': '255.255.255.0',
50
'bridge_interface': None,
51
'gateway': '100.168.0.1',
52
'broadcast': '100.168.0.255',
56
'vpn_public_address': None,
57
'project_id': 'fake_project1',
59
{'label': 'project2-net1',
62
'cidr': '101.168.1.0/24',
63
'cidr_v6': '101:1db9::/64',
64
'gateway_v6': '101:1db9::1',
66
'netmask': '255.255.255.0',
68
'bridge_interface': None,
69
'gateway': '101.168.1.1',
70
'broadcast': '101.168.1.255',
74
'project_id': 'fake_project2',
79
'cidr': '102.0.0.0/24',
80
'cidr_v6': '102:1dba::/64',
81
'gateway_v6': '102:1dba::1',
83
'netmask': '255.255.255.0',
85
'bridge_interface': None,
86
'gateway': '102.0.0.1',
87
'broadcast': '102.0.0.255',
93
{'label': "project2-net2",
96
'cidr': '103.0.0.0/24',
97
'cidr_v6': '103:1dbb::/64',
98
'gateway_v6': '103:1dbb::1',
100
'netmask': '255.255.255.0',
102
'bridge_interface': None,
103
'gateway': '103.0.0.1',
104
'broadcast': '103.0.0.255',
108
'project_id': "fake_project2",
112
class QuantumConnectionTestCase(test.TestCase):
114
def test_connection(self):
115
fc = fake_client.FakeClient(LOG)
116
qc = quantum_connection.QuantumClientConnection(client=fc)
119
net1_uuid = qc.create_network(t, net1_name)
120
self.assertEquals(net1_name, qc.get_network_name(t, net1_uuid))
121
self.assertTrue(qc.network_exists(t, net1_uuid))
122
self.assertFalse(qc.network_exists(t, "fake-uuid"))
123
self.assertFalse(qc.network_exists("fake-tenant", net1_uuid))
125
nets = qc.get_networks(t)['networks']
126
self.assertEquals(len(nets), 1)
127
self.assertEquals(nets[0]['id'], net1_uuid)
130
for i in range(0, num_ports):
131
qc.create_and_attach_port(t, net1_uuid,
132
'iface' + str(i), state='ACTIVE')
134
self.assertEquals(len(qc.get_attached_ports(t, net1_uuid)), num_ports)
136
for i in range(0, num_ports):
137
port_uuid = qc.get_port_by_attachment(t, net1_uuid,
139
self.assertTrue(port_uuid)
140
qc.detach_and_delete_port(t, net1_uuid, port_uuid)
142
self.assertEquals(len(qc.get_attached_ports(t, net1_uuid)), 0)
144
# test port not found
145
qc.create_and_attach_port(t, net1_uuid, 'foo', state='ACTIVE')
146
port_uuid = qc.get_port_by_attachment(t, net1_uuid, 'foo')
147
qc.detach_and_delete_port(t, net1_uuid, port_uuid)
148
self.assertRaises(quantum_client.QuantumNotFoundException,
149
qc.detach_and_delete_port, t,
150
net1_uuid, port_uuid)
152
qc.delete_network(t, net1_uuid)
153
self.assertFalse(qc.network_exists(t, net1_uuid))
154
self.assertEquals(len(qc.get_networks(t)['networks']), 0)
156
self.assertRaises(quantum_client.QuantumNotFoundException,
157
qc.get_network_name, t, net1_uuid)
160
# this is a base class to be used by other QuantumManager Test classes
161
class QuantumNovaTestCase(test.TestCase):
164
super(QuantumNovaTestCase, self).setUp()
166
self.flags(quantum_use_dhcp=True)
167
self.flags(l3_lib="nova.network.l3.LinuxNetL3")
168
linuxdrv = "nova.network.linux_net.LinuxOVSInterfaceDriver"
169
self.flags(linuxnet_interface_driver=linuxdrv)
170
fc = fake_client.FakeClient(LOG)
171
qc = quantum_connection.QuantumClientConnection(client=fc)
173
self.net_man = quantum_manager.QuantumManager(
174
ipam_lib="nova.network.quantum.nova_ipam_lib",
177
def func(arg1, arg2):
180
def func2(arg1, arg2, arg3):
186
self.net_man.driver.update_dhcp_hostfile_with_text = func
187
self.net_man.driver.restart_dhcp = func2
188
self.net_man.driver.kill_dhcp = func1
190
# Tests seem to create some networks by default, which
191
# we don't want. So we delete them.
193
ctx = context.RequestContext('user1', 'fake_project1').elevated()
194
for n in db.network_get_all(ctx):
195
db.network_delete_safe(ctx, n['id'])
197
# Other unit tests (e.g., test_compute.py) have a nasty
198
# habit of of creating fixed IPs and not cleaning up, which
199
# can confuse these tests, so we remove all existing fixed
200
# ips before starting.
201
session = sql_session.get_session()
202
result = session.query(models.FixedIp).all()
203
with session.begin():
204
for fip_ref in result:
205
session.delete(fip_ref)
207
self.net_man.init_host()
209
def _create_network(self, n):
210
ctx = context.RequestContext('user1', n['project_id'])
211
nwks = self.net_man.create_networks(
213
label=n['label'], cidr=n['cidr'],
214
multi_host=n['multi_host'],
215
num_networks=1, network_size=256,
216
cidr_v6=n['cidr_v6'],
217
gateway=n['gateway'],
218
gateway_v6=n['gateway_v6'], bridge=None,
219
bridge_interface=None, dns1=n['dns1'],
220
project_id=n['project_id'],
221
priority=n['priority'])
222
n['uuid'] = nwks[0]['uuid']
225
class QuantumAllocationTestCase(QuantumNovaTestCase):
226
def test_get_network_in_db(self):
227
context = self.mox.CreateMockAnything()
228
context.elevated().AndReturn('elevated')
229
self.mox.StubOutWithMock(db, 'network_get_by_uuid')
230
self.net_man.context = context
231
db.network_get_by_uuid('elevated', 'quantum_net_id').AndReturn(
236
network = self.net_man.get_network(context, ('quantum_net_id',
238
self.assertEquals(network['quantum_net_id'], 'quantum_net_id')
239
self.assertEquals(network['uuid'], 1)
241
def test_get_network_not_in_db(self):
242
context = self.mox.CreateMockAnything()
243
context.elevated().AndReturn('elevated')
244
self.mox.StubOutWithMock(db, 'network_get_by_uuid')
245
self.net_man.context = context
246
db.network_get_by_uuid('elevated', 'quantum_net_id').AndReturn(None)
250
network = self.net_man.get_network(context, ('quantum_net_id',
252
self.assertEquals(network['quantum_net_id'], 'quantum_net_id')
253
self.assertEquals(network['uuid'], 'quantum_net_id')
256
class QuantumDeallocationTestCase(QuantumNovaTestCase):
257
def test_deallocate_port(self):
258
quantum = self.mox.CreateMock(
259
quantum_connection.QuantumClientConnection)
260
quantum.get_port_by_attachment('q_tenant_id', 'net_id',
261
'interface_id').AndReturn('port_id')
262
quantum.detach_and_delete_port('q_tenant_id', 'net_id', 'port_id')
263
self.net_man.q_conn = quantum
267
self.net_man.deallocate_port('interface_id', 'net_id', 'q_tenant_id',
270
def test_deallocate_port_logs_error(self):
271
quantum = self.mox.CreateMock(
272
quantum_connection.QuantumClientConnection)
273
quantum.get_port_by_attachment('q_tenant_id', 'net_id',
274
'interface_id').AndRaise(Exception)
275
self.net_man.q_conn = quantum
277
self.mox.StubOutWithMock(quantum_manager.LOG, 'exception')
278
quantum_manager.LOG.exception(mox.Regex(r'port deallocation failed'))
282
self.net_man.deallocate_port('interface_id', 'net_id', 'q_tenant_id',
285
def test_deallocate_ip_address(self):
286
ipam = self.mox.CreateMock(melange_ipam_lib.QuantumMelangeIPAMLib)
287
ipam.get_tenant_id_by_net_id('context', 'net_id', {'uuid': 1},
288
'project_id').AndReturn('ipam_tenant_id')
289
self.net_man.ipam = ipam
291
self.net_man.deallocate_ip_address('context', 'net_id', 'project_id',
292
{'uuid': 1}, 'instance_id')
294
def test_deallocate_ip_address_2(self):
295
ipam = self.mox.CreateMock(melange_ipam_lib.QuantumMelangeIPAMLib)
296
ipam.get_tenant_id_by_net_id('context', 'net_id', {'uuid': 1},
297
'project_id').AndRaise(Exception())
298
self.net_man.ipam = ipam
300
self.mox.StubOutWithMock(quantum_manager.LOG, 'exception')
301
quantum_manager.LOG.exception(mox.Regex(r'ipam deallocation failed'))
304
self.net_man.deallocate_ip_address('context', 'net_id', 'project_id',
305
{'uuid': 1}, 'instance_id')
308
class QuantumManagerTestCase(QuantumNovaTestCase):
309
def test_create_and_delete_nets(self):
313
def _create_nets(self):
315
self._create_network(n)
317
def _delete_nets(self):
319
ctx = context.RequestContext('user1', n['project_id'])
320
self.net_man.delete_network(ctx, None, n['uuid'])
321
self.assertRaises(exception.NoNetworksFound,
322
db.network_get_all, ctx.elevated())
324
def _validate_nw_info(self, nw_info, expected_net_labels):
326
self.assertEquals(len(nw_info), len(expected_net_labels))
328
ctx = context.RequestContext('user1', 'foo').elevated()
330
for n in db.network_get_all(ctx):
331
all_net_map[n['label']] = n
333
for i in range(0, len(nw_info)):
335
net = all_net_map[expected_net_labels[i]]
337
# simple test assumes that each starting prefix is unique
338
expected_v4_cidr_start = net['cidr'].split(".")[0].lower()
339
expected_v6_cidr_start = net['cidr_v6'].split(":")[0].lower()
341
for subnet in vif['network']['subnets']:
342
addr = subnet['ips'][0]['address']
343
if subnet['version'] == 4:
344
address_start = addr.split(".")[0].lower()
345
self.assertTrue(expected_v4_cidr_start, address_start)
347
address_start = addr.split(":")[0].lower()
348
self.assertTrue(expected_v6_cidr_start, address_start)
350
# confirm that there is a DHCP device on corresponding net
351
for l in expected_net_labels:
353
tenant_id = (n['project_id'] or
354
FLAGS.quantum_default_tenant_id)
355
ports = self.net_man.q_conn.get_attached_ports(
356
tenant_id, n['uuid'])
357
self.assertEquals(len(ports), 2) # gw + instance VIF
359
# make sure we aren't allowed to delete network with
361
self.assertRaises(exception.NetworkBusy,
362
self.net_man.delete_network,
363
ctx, None, n['uuid'])
365
def _check_vifs(self, expect_num_vifs):
366
ctx = context.RequestContext('user1', "").elevated()
367
self.assertEqual(len(db.virtual_interface_get_all(ctx)),
370
def _allocate_and_deallocate_instance(self, project_id, requested_networks,
373
ctx = context.RequestContext('user1', project_id)
376
instance_ref = db.instance_create(ctx,
377
{"project_id": project_id})
379
nw_info = self.net_man.allocate_for_instance(ctx.elevated(),
380
instance_id=instance_ref['id'], host="",
382
project_id=project_id,
383
requested_networks=requested_networks)
385
self._check_vifs(len(nw_info))
387
self._validate_nw_info(nw_info, expected_labels)
389
nw_info = self.net_man.get_instance_nw_info(ctx, instance_ref['id'],
390
instance_ref['uuid'],
391
instance_ref['instance_type_id'], "",
392
project_id=project_id)
394
self._check_vifs(len(nw_info))
395
self._validate_nw_info(nw_info, expected_labels)
399
nid = vif['network']['id']
400
pid = self.net_man.q_conn.get_port_by_attachment(
401
project_id, nid, vif['id'])
403
pid = self.net_man.q_conn.get_port_by_attachment(
404
FLAGS.quantum_default_tenant_id,
406
self.assertTrue(pid is not None)
407
port_net_pairs.append((pid, nid))
409
self.net_man.deallocate_for_instance(ctx,
410
instance_id=instance_ref['id'],
411
project_id=project_id)
413
for pid, nid in port_net_pairs:
414
self.assertRaises(quantum_client.QuantumNotFoundException,
415
self.net_man.q_conn.detach_and_delete_port,
416
project_id, nid, pid)
417
self.assertRaises(quantum_client.QuantumNotFoundException,
418
self.net_man.q_conn.detach_and_delete_port,
419
FLAGS.quantum_default_tenant_id, nid, pid)
423
def test_allocate_and_deallocate_instance_static(self):
425
self._allocate_and_deallocate_instance("fake_project1", None,
426
['public', 'project1-net1'])
429
def test_allocate_and_deallocate_instance_dynamic(self):
432
project_id = "fake_project2"
433
ctx = context.RequestContext('user1', project_id)
434
all_valid_networks = self.net_man.ipam.get_project_and_global_net_ids(
436
requested_networks = [(n[0], None) for n in all_valid_networks]
438
self.net_man.validate_networks(ctx, requested_networks)
441
for n in db.network_get_all(ctx.elevated()):
442
label_map[n['uuid']] = n['label']
443
expected_labels = [label_map[uid] for uid, _i in requested_networks]
445
self._allocate_and_deallocate_instance(project_id, requested_networks,
449
def test_validate_bad_network(self):
450
ctx = context.RequestContext('user1', 'fake_project1')
451
self.assertRaises(exception.NetworkNotFound,
452
self.net_man.validate_networks, ctx, [("", None)])
454
def test_create_net_external_uuid(self):
455
"""Tests use case where network can be created directly via
456
Quantum API, then the UUID is passed in via nova-manage"""
457
project_id = "foo_project"
458
ctx = context.RequestContext('user1', project_id)
459
net_id = self.net_man.q_conn.create_network(project_id, 'net1')
460
self.net_man.create_networks(
471
bridge_interface=None,
473
project_id=project_id,
476
net = db.network_get_by_uuid(ctx.elevated(), net_id)
477
self.assertTrue(net is not None)
478
self.assertEquals(net['uuid'], net_id)
480
def test_create_net_external_uuid_and_host_is_set(self):
481
"""Make sure network['host'] is set when creating a network via the
483
project_id = "foo_project"
484
ctx = context.RequestContext('user1', project_id)
485
net_id = self.net_man.q_conn.create_network(project_id, 'net2')
486
self.net_man.create_networks(
487
ctx, label='achtungbaby2', cidr="9.9.8.0/24", multi_host=False,
488
num_networks=1, network_size=256, cidr_v6=None,
489
gateway="9.9.8.1", gateway_v6=None, bridge=None,
490
bridge_interface=None, dns1="8.8.8.8", project_id=project_id,
491
priority=8, uuid=net_id)
492
net = db.network_get_by_uuid(ctx.elevated(), net_id)
493
self.assertTrue(net is not None)
494
self.assertEquals(net['uuid'], net_id)
495
self.assertTrue(net['host'] != None)
498
class QuantumNovaMACGenerationTestCase(QuantumNovaTestCase):
499
def test_local_mac_address_creation(self):
500
self.flags(use_melange_mac_generation=False)
501
fake_mac = "ab:cd:ef:ab:cd:ef"
502
self.stubs.Set(utils, "generate_mac_address",
504
project_id = "fake_project1"
505
ctx = context.RequestContext('user1', project_id)
506
self._create_network(networks[0])
508
all_valid_networks = self.net_man.ipam.get_project_and_global_net_ids(
510
requested_networks = [(n[0], None) for n in all_valid_networks]
512
instance_ref = db.api.instance_create(ctx,
513
{"project_id": project_id})
514
nw_info = self.net_man.allocate_for_instance(ctx,
515
instance_id=instance_ref['id'], host="",
517
project_id=project_id,
518
requested_networks=requested_networks)
519
self.assertEqual(nw_info[0]['address'], fake_mac)
521
def test_melange_mac_address_creation(self):
522
self.flags(use_melange_mac_generation=True)
523
fake_mac = "ab:cd:ef:ab:cd:ef"
524
self.stubs.Set(melange_connection.MelangeConnection, "create_vif",
525
lambda w, x, y, z: fake_mac)
526
project_id = "fake_project1"
527
ctx = context.RequestContext('user1', project_id)
528
self._create_network(networks[0])
530
all_valid_networks = self.net_man.ipam.get_project_and_global_net_ids(
532
requested_networks = [(n[0], None) for n in all_valid_networks]
534
instance_ref = db.api.instance_create(ctx,
535
{"project_id": project_id})
536
nw_info = self.net_man.allocate_for_instance(ctx,
537
instance_id=instance_ref['id'], host="",
539
project_id=project_id,
540
requested_networks=requested_networks)
541
self.assertEqual(nw_info[0]['address'], fake_mac)
544
class QuantumNovaPortSecurityTestCase(QuantumNovaTestCase):
545
def test_port_securty(self):
546
self.flags(use_melange_mac_generation=True)
547
self.flags(quantum_use_port_security=True)
548
fake_mac = "ab:cd:ef:ab:cd:ef"
549
self.stubs.Set(melange_connection.MelangeConnection, "create_vif",
550
lambda w, x, y, z: fake_mac)
551
project_id = "fake_project1"
552
ctx = context.RequestContext('user1', project_id)
553
self._create_network(networks[0])
555
all_valid_networks = self.net_man.ipam.get_project_and_global_net_ids(
557
requested_networks = [(n[0], None) for n in all_valid_networks]
559
instance_ref = db.api.instance_create(ctx,
560
{"project_id": project_id})
561
oldfunc = self.net_man.q_conn.create_and_attach_port
563
# Make sure we get the appropriate mac set in allowed_address_pairs
564
# if port security is enabled.
565
def _instrumented_create_and_attach_port(tenant_id, net_id,
566
interface_id, **kwargs):
567
self.assertTrue('allowed_address_pairs' in kwargs.keys())
568
pairs = kwargs['allowed_address_pairs']
569
self.assertTrue(pairs[0]['mac_address'] == fake_mac)
570
self.net_man.q_conn.create_and_attach_port = oldfunc
571
return oldfunc(tenant_id, net_id, interface_id, **kwargs)
572
_port_attach = _instrumented_create_and_attach_port
573
self.net_man.q_conn.create_and_attach_port = _port_attach
574
nw_info = self.net_man.allocate_for_instance(ctx,
575
instance_id=instance_ref['id'], host="",
577
project_id=project_id,
578
requested_networks=requested_networks)
579
self.assertEqual(nw_info[0]['address'], fake_mac)
581
def test_port_securty_negative(self):
582
self.flags(use_melange_mac_generation=True)
583
self.flags(quantum_use_port_security=False)
584
fake_mac = "ab:cd:ef:ab:cd:ef"
585
self.stubs.Set(melange_connection.MelangeConnection, "create_vif",
586
lambda w, x, y, z: fake_mac)
587
project_id = "fake_project1"
588
ctx = context.RequestContext('user1', project_id)
589
self._create_network(networks[0])
591
all_valid_networks = self.net_man.ipam.get_project_and_global_net_ids(
593
requested_networks = [(n[0], None) for n in all_valid_networks]
595
instance_ref = db.api.instance_create(ctx,
596
{"project_id": project_id})
597
oldfunc = self.net_man.q_conn.create_and_attach_port
599
# Make sure no pairs are passed in if port security is turned off
600
def _instrumented_create_and_attach_port(tenant_id, net_id,
601
interface_id, **kwargs):
602
self.assertTrue('allowed_address_pairs' in kwargs.keys())
603
pairs = kwargs['allowed_address_pairs']
604
self.assertTrue(len(pairs) == 0)
605
self.net_man.q_conn.create_and_attach_port = oldfunc
606
return oldfunc(tenant_id, net_id, interface_id, **kwargs)
607
_port_attach = _instrumented_create_and_attach_port
608
self.net_man.q_conn.create_and_attach_port = _port_attach
609
nw_info = self.net_man.allocate_for_instance(ctx,
610
instance_id=instance_ref['id'], host="",
612
project_id=project_id,
613
requested_networks=requested_networks)
614
self.assertEqual(nw_info[0]['address'], fake_mac)
617
class QuantumMelangeTestCase(test.TestCase):
619
super(QuantumMelangeTestCase, self).setUp()
621
fc = fake_client.FakeClient(LOG)
622
qc = quantum_connection.QuantumClientConnection(client=fc)
624
self.net_man = quantum_manager.QuantumManager(
625
ipam_lib="nova.network.quantum.nova_ipam_lib",
628
def test_get_instance_uuids_by_ip_filter(self):
629
fake_context = context.RequestContext('user', 'project')
631
filters = {'ip': address}
633
self.net_man.ipam = self.mox.CreateMockAnything()
634
self.net_man.ipam.get_instance_ids_by_ip_address(fake_context,
635
address).AndReturn(['instance_id'])
637
instance = self.mox.CreateMockAnything()
638
instance.uuid = 'instance_uuid'
640
self.mox.StubOutWithMock(db, 'instance_get')
641
db.instance_get(fake_context, 'instance_id').AndReturn(instance)
645
uuids = self.net_man.get_instance_uuids_by_ip_filter(fake_context,
647
self.assertEquals(uuids, [{'instance_uuid':'instance_uuid'}])
added
removed
nova/tests/network/test_quantum.py
