1
# vim: tabstop=4 shiftwidth=4 softtabstop=4
3
# Copyright 2010-2011 ????
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
17
# @author: Salvatore Orlando, Citrix Systems
19
""" Module containing unit tests for Quantum
20
command line interface
28
from quantum import api as server
29
from quantumclient import ClientV11
30
from quantumclient import cli_lib as cli
31
from quantum.db import api as db
32
from quantum.openstack.common import cfg
33
from quantumclient.tests.unit import stubs as client_stubs
35
LOG = logging.getLogger('quantumclient.tests.test_cli')
40
class CLITest(unittest.TestCase):
43
"""Prepare the test environment"""
44
#TODO: make the version of the API router configurable
45
cfg.CONF.core_plugin = 'quantum.plugins.sample.SamplePlugin.FakePlugin'
46
self.api = server.APIRouterV11()
48
self.tenant_id = "test_tenant"
49
self.network_name_1 = "test_network_1"
50
self.network_name_2 = "test_network_2"
51
self.version = API_VERSION
52
# Prepare client and plugin manager
53
self.client = ClientV11(tenant=self.tenant_id,
55
testingStub=client_stubs.FakeHTTPConnection,
58
self.fake_stdout = client_stubs.FakeStdout()
59
sys.stdout = self.fake_stdout
62
"""Clear the test environment"""
64
sys.stdout = sys.__stdout__
66
def _verify_list_networks(self):
67
# Verification - get raw result from db
68
nw_list = db.network_list(self.tenant_id)
69
networks = [{'id': nw.uuid, 'name': nw.name}
72
output = cli.prepare_output('list_nets',
74
dict(networks=networks),
77
# Must add newline at the end to match effect of print call
78
self.assertEquals(self.fake_stdout.make_string(), output + '\n')
80
def _verify_list_networks_details(self):
81
# Verification - get raw result from db
82
nw_list = db.network_list(self.tenant_id)
83
networks = [dict(id=nw.uuid, name=nw.name) for nw in nw_list]
85
output = cli.prepare_output('list_nets_detail',
87
dict(networks=networks),
90
# Must add newline at the end to match effect of print call
91
self.assertEquals(self.fake_stdout.make_string(), output + '\n')
93
def _verify_list_networks_details_name_filter(self, name):
94
# Verification - get raw result from db
95
nw_list = db.network_list(self.tenant_id)
99
nw_filtered.append(nw)
100
networks = [dict(id=nw.uuid, name=nw.name) for nw in nw_filtered]
102
output = cli.prepare_output('list_nets_detail',
104
dict(networks=networks),
107
# Must add newline at the end to match effect of print call
108
self.assertEquals(self.fake_stdout.make_string(), output + '\n')
110
def _verify_create_network(self):
111
# Verification - get raw result from db
112
nw_list = db.network_list(self.tenant_id)
113
if len(nw_list) != 1:
114
self.fail("No network created")
115
network_id = nw_list[0].uuid
117
output = cli.prepare_output('create_net',
119
dict(network_id=network_id),
122
# Must add newline at the end to match effect of print call
123
self.assertEquals(self.fake_stdout.make_string(), output + '\n')
125
def _verify_delete_network(self, network_id):
126
# Verification - get raw result from db
127
nw_list = db.network_list(self.tenant_id)
128
if len(nw_list) != 0:
129
self.fail("DB should not contain any network")
131
output = cli.prepare_output('delete_net',
133
dict(network_id=network_id),
136
# Must add newline at the end to match effect of print call
137
self.assertEquals(self.fake_stdout.make_string(), output + '\n')
139
def _verify_update_network(self):
140
# Verification - get raw result from db
141
nw_list = db.network_list(self.tenant_id)
142
network_data = {'id': nw_list[0].uuid,
143
'name': nw_list[0].name,
144
'op-status': nw_list[0].op_status}
146
output = cli.prepare_output('update_net',
148
dict(network=network_data),
151
# Must add newline at the end to match effect of print call
152
self.assertEquals(self.fake_stdout.make_string(), output + '\n')
154
def _verify_show_network(self):
155
# Verification - get raw result from db
156
nw = db.network_list(self.tenant_id)[0]
157
network = {'id': nw.uuid,
159
'op-status': nw.op_status}
161
output = cli.prepare_output('show_net',
163
dict(network=network),
166
# Must add newline at the end to match effect of print call
167
self.assertEquals(self.fake_stdout.make_string(), output + '\n')
169
def _verify_show_network_details(self):
170
# Verification - get raw result from db
171
nw = db.network_list(self.tenant_id)[0]
172
network = {'id': nw.uuid,
174
'op-status': nw.op_status}
175
port_list = db.port_list(nw.uuid)
181
'id': '<none>', }, }, ]
183
network['ports'] = []
184
for port in port_list:
185
network['ports'].append({
189
'id': port.interface_id or '<none>', }, })
192
output = cli.prepare_output('show_net_detail',
194
dict(network=network),
197
# Must add newline at the end to match effect of print call
198
self.assertEquals(self.fake_stdout.make_string(), output + '\n')
200
def _verify_list_ports(self, network_id):
201
# Verification - get raw result from db
202
port_list = db.port_list(network_id)
203
ports = [dict(id=port.uuid, state=port.state)
204
for port in port_list]
206
output = cli.prepare_output('list_ports',
208
dict(network_id=network_id,
212
# Must add newline at the end to match effect of print call
213
self.assertEquals(self.fake_stdout.make_string(), output + '\n')
215
def _verify_list_ports_details(self, network_id):
216
# Verification - get raw result from db
217
port_list = db.port_list(network_id)
218
ports = [dict(id=port.uuid, state=port.state)
219
for port in port_list]
221
output = cli.prepare_output('list_ports_detail',
223
dict(network_id=network_id,
227
# Must add newline at the end to match effect of print call
228
self.assertEquals(self.fake_stdout.make_string(), output + '\n')
230
def _verify_create_port(self, network_id):
231
# Verification - get raw result from db
232
port_list = db.port_list(network_id)
233
if len(port_list) != 1:
234
self.fail("No port created")
235
port_id = port_list[0].uuid
237
output = cli.prepare_output('create_port',
239
dict(network_id=network_id,
243
# Must add newline at the end to match effect of print call
244
self.assertEquals(self.fake_stdout.make_string(), output + '\n')
246
def _verify_delete_port(self, network_id, port_id):
247
# Verification - get raw result from db
248
port_list = db.port_list(network_id)
249
if len(port_list) != 0:
250
self.fail("DB should not contain any port")
252
output = cli.prepare_output('delete_port',
254
dict(network_id=network_id,
258
# Must add newline at the end to match effect of print call
259
self.assertEquals(self.fake_stdout.make_string(), output + '\n')
261
def _verify_update_port(self, network_id, port_id):
262
# Verification - get raw result from db
263
port = db.port_get(port_id, network_id)
264
port_data = {'id': port.uuid,
266
'op-status': port.op_status}
268
output = cli.prepare_output('update_port',
270
dict(network_id=network_id,
274
# Must add newline at the end to match effect of print call
275
self.assertEquals(self.fake_stdout.make_string(), output + '\n')
277
def _verify_show_port(self, network_id, port_id):
278
# Verification - get raw result from db
279
port = db.port_get(port_id, network_id)
280
port_data = {'id': port.uuid,
282
'attachment': {'id': port.interface_id or '<none>'},
283
'op-status': port.op_status}
286
output = cli.prepare_output('show_port',
288
dict(network_id=network_id,
292
# Must add newline at the end to match effect of print call
293
self.assertEquals(self.fake_stdout.make_string(), output + '\n')
295
def _verify_show_port_details(self, network_id, port_id):
296
# Verification - get raw result from db
297
# TODO(salvatore-orlando): Must resolve this issue with
298
# attachment in separate bug fix.
299
port = db.port_get(port_id, network_id)
300
port_data = {'id': port.uuid,
302
'attachment': {'id': port.interface_id or '<none>'},
303
'op-status': port.op_status}
306
output = cli.prepare_output('show_port_detail',
308
dict(network_id=network_id,
312
# Must add newline at the end to match effect of print call
313
self.assertEquals(self.fake_stdout.make_string(), output + '\n')
315
def _verify_plug_iface(self, network_id, port_id):
316
# Verification - get raw result from db
317
port = db.port_get(port_id, network_id)
319
output = cli.prepare_output("plug_iface",
321
dict(network_id=network_id,
322
port_id=port['uuid'],
323
attachment=port['interface_id']),
326
# Must add newline at the end to match effect of print call
327
self.assertEquals(self.fake_stdout.make_string(), output + '\n')
329
def _verify_unplug_iface(self, network_id, port_id):
330
# Verification - get raw result from db
331
port = db.port_get(port_id, network_id)
333
output = cli.prepare_output("unplug_iface",
335
dict(network_id=network_id,
336
port_id=port['uuid']),
339
# Must add newline at the end to match effect of print call
340
self.assertEquals(self.fake_stdout.make_string(), output + '\n')
342
def _verify_show_iface(self, network_id, port_id):
343
# Verification - get raw result from db
344
port = db.port_get(port_id, network_id)
345
iface = {'id': port.interface_id or '<none>'}
348
output = cli.prepare_output('show_iface',
350
dict(network_id=network_id,
355
# Must add newline at the end to match effect of print call
356
self.assertEquals(self.fake_stdout.make_string(), output + '\n')
358
def test_list_networks_v10(self):
360
# Pre-populate data for testing using db api
361
db.network_create(self.tenant_id, self.network_name_1)
362
db.network_create(self.tenant_id, self.network_name_2)
364
cli.list_nets(self.client,
368
LOG.exception("Exception caught: %s", sys.exc_info())
369
self.fail("test_list_networks_v10 failed due to an exception")
371
LOG.debug("Operation completed. Verifying result")
372
LOG.debug(self.fake_stdout.content)
373
self._verify_list_networks()
375
def test_list_networks_details_v10(self):
377
# Pre-populate data for testing using db api
378
db.network_create(self.tenant_id, self.network_name_1)
379
db.network_create(self.tenant_id, self.network_name_2)
381
cli.list_nets_detail(self.client,
385
LOG.exception("Exception caught: %s", sys.exc_info())
386
self.fail("test_list_networks_details_v10 failed due to" +
389
LOG.debug("Operation completed. Verifying result")
390
LOG.debug(self.fake_stdout.content)
391
self._verify_list_networks_details()
393
def test_list_networks_v11(self):
395
# Pre-populate data for testing using db api
396
db.network_create(self.tenant_id, self.network_name_1)
397
db.network_create(self.tenant_id, self.network_name_2)
399
cli.list_nets_v11(self.client,
403
LOG.exception("Exception caught: %s", sys.exc_info())
404
self.fail("test_list_networks_v11 failed due to an exception")
406
LOG.debug("Operation completed. Verifying result")
407
LOG.debug(self.fake_stdout.content)
408
self._verify_list_networks()
410
def test_list_networks_details_v11(self):
412
# Pre-populate data for testing using db api
413
db.network_create(self.tenant_id, self.network_name_1)
414
db.network_create(self.tenant_id, self.network_name_2)
416
cli.list_nets_detail_v11(self.client,
420
LOG.exception("Exception caught: %s", sys.exc_info())
421
self.fail("test_list_networks_details_v11 failed due to " +
424
LOG.debug("Operation completed. Verifying result")
425
LOG.debug(self.fake_stdout.content)
426
self._verify_list_networks_details()
428
def test_list_networks_details_v11_name_filter(self):
430
# Pre-populate data for testing using db api
431
db.network_create(self.tenant_id, self.network_name_1)
432
db.network_create(self.tenant_id, self.network_name_2)
434
cli.list_nets_detail_v11(self.client,
437
{'name': self.network_name_1, })
439
LOG.exception("Exception caught: %s", sys.exc_info())
440
self.fail("test_list_networks_details_v11 failed due to " +
443
LOG.debug("Operation completed. Verifying result")
444
LOG.debug(self.fake_stdout.content)
445
self._verify_list_networks_details_name_filter(self.network_name_1)
447
def test_create_network(self):
449
cli.create_net(self.client,
454
LOG.exception("Exception caught: %s", sys.exc_info())
455
self.fail("test_create_network failed due to an exception")
457
LOG.debug("Operation completed. Verifying result")
458
LOG.debug(self.fake_stdout.content)
459
self._verify_create_network()
461
def test_delete_network(self):
463
db.network_create(self.tenant_id, self.network_name_1)
464
network_id = db.network_list(self.tenant_id)[0]['uuid']
465
cli.delete_net(self.client,
470
LOG.exception("Exception caught: %s", sys.exc_info())
471
self.fail("test_delete_network failed due to an exception")
473
LOG.debug("Operation completed. Verifying result")
474
LOG.debug(self.fake_stdout.content)
475
self._verify_delete_network(network_id)
477
def test_show_network(self):
479
# Load some data into the datbase
480
net = db.network_create(self.tenant_id, self.network_name_1)
481
cli.show_net(self.client,
486
LOG.exception("Exception caught: %s", sys.exc_info())
487
self.fail("test_detail_network failed due to an exception")
489
LOG.debug("Operation completed. Verifying result")
490
LOG.debug(self.fake_stdout.content)
491
self._verify_show_network()
493
def test_show_network_details_no_ports(self):
495
# Load some data into the datbase
496
net = db.network_create(self.tenant_id, self.network_name_1)
497
network_id = net['uuid']
498
cli.show_net_detail(self.client,
503
LOG.exception("Exception caught: %s", sys.exc_info())
504
self.fail("test_show_network_details_no_ports failed due to" +
507
LOG.debug("Operation completed. Verifying result")
508
LOG.debug(self.fake_stdout.content)
509
self._verify_show_network_details()
511
def test_show_network_details(self):
512
iface_id = "flavor crystals"
514
# Load some data into the datbase
515
net = db.network_create(self.tenant_id, self.network_name_1)
516
network_id = net['uuid']
517
port = db.port_create(network_id)
518
port_id = port['uuid']
519
db.port_set_attachment(port_id, network_id, iface_id)
520
port = db.port_create(network_id)
521
cli.show_net_detail(self.client,
526
LOG.exception("Exception caught: %s", sys.exc_info())
527
self.fail("test_show_network_details failed due to an exception")
529
LOG.debug("Operation completed. Verifying result")
530
LOG.debug(self.fake_stdout.content)
531
self._verify_show_network_details()
533
def test_update_network(self):
535
net = db.network_create(self.tenant_id, self.network_name_1)
536
network_id = net['uuid']
537
cli.update_net(self.client,
540
'name=%s' % self.network_name_2,
543
LOG.exception("Exception caught: %s", sys.exc_info())
544
self.fail("test_update_network failed due to an exception")
546
LOG.debug("Operation completed. Verifying result")
547
LOG.debug(self.fake_stdout.content)
548
self._verify_update_network()
550
def test_list_ports_v10(self):
552
# Pre-populate data for testing using db api
553
net = db.network_create(self.tenant_id, self.network_name_1)
554
network_id = net['uuid']
555
db.port_create(network_id)
556
db.port_create(network_id)
557
cli.list_ports(self.client,
562
LOG.exception("Exception caught: %s", sys.exc_info())
563
self.fail("test_list_ports failed due to an exception")
565
LOG.debug("Operation completed. Verifying result")
566
LOG.debug(self.fake_stdout.content)
567
self._verify_list_ports(network_id)
569
def test_list_ports_details_v10(self):
571
# Pre-populate data for testing using db api
572
net = db.network_create(self.tenant_id, self.network_name_1)
573
network_id = net['uuid']
574
db.port_create(network_id)
575
db.port_create(network_id)
576
cli.list_ports_detail(self.client,
581
LOG.exception("Exception caught: %s", sys.exc_info())
582
self.fail("test_list_ports_details_v10 failed due to" +
585
LOG.debug("Operation completed. Verifying result")
586
LOG.debug(self.fake_stdout.content)
587
self._verify_list_ports_details(network_id)
589
def test_list_ports_v11(self):
591
# Pre-populate data for testing using db api
592
net = db.network_create(self.tenant_id, self.network_name_1)
593
network_id = net['uuid']
594
db.port_create(network_id)
595
db.port_create(network_id)
597
cli.list_ports_v11(self.client,
602
LOG.exception("Exception caught: %s", sys.exc_info())
603
self.fail("test_list_ports_v11 failed due to an exception")
605
LOG.debug("Operation completed. Verifying result")
606
LOG.debug(self.fake_stdout.content)
607
self._verify_list_ports(network_id)
609
def test_list_ports_details_v11(self):
611
# Pre-populate data for testing using db api
612
net = db.network_create(self.tenant_id, self.network_name_1)
613
network_id = net['uuid']
614
db.port_create(network_id)
615
db.port_create(network_id)
617
cli.list_ports_detail_v11(self.client,
622
LOG.exception("Exception caught: %s", sys.exc_info())
623
self.fail("test_list_ports_details_v11 failed due to " +
626
LOG.debug("Operation completed. Verifying result")
627
LOG.debug(self.fake_stdout.content)
628
self._verify_list_ports_details(network_id)
630
def test_create_port(self):
633
# Pre-populate data for testing using db api
634
net = db.network_create(self.tenant_id, self.network_name_1)
635
network_id = net['uuid']
636
cli.create_port(self.client,
641
LOG.exception("Exception caught: %s", sys.exc_info())
642
self.fail("test_create_port failed due to an exception")
644
LOG.debug("Operation completed. Verifying result")
645
LOG.debug(self.fake_stdout.content)
646
self._verify_create_port(network_id)
648
def test_delete_port(self):
652
# Pre-populate data for testing using db api
653
net = db.network_create(self.tenant_id, self.network_name_1)
654
network_id = net['uuid']
655
port = db.port_create(network_id)
656
port_id = port['uuid']
657
cli.delete_port(self.client,
663
LOG.exception("Exception caught: %s", sys.exc_info())
664
self.fail("test_delete_port failed due to an exception")
666
LOG.debug("Operation completed. Verifying result")
667
LOG.debug(self.fake_stdout.content)
668
self._verify_delete_port(network_id, port_id)
670
def test_update_port(self):
672
net = db.network_create(self.tenant_id, self.network_name_1)
673
network_id = net['uuid']
674
port = db.port_create(network_id)
675
port_id = port['uuid']
676
# Default state is DOWN - change to ACTIVE.
677
cli.update_port(self.client,
684
LOG.exception("Exception caught: %s", sys.exc_info())
685
self.fail("test_update_port failed due to an exception")
687
LOG.debug("Operation completed. Verifying result")
688
LOG.debug(self.fake_stdout.content)
689
self._verify_update_port(network_id, port_id)
691
def test_show_port(self):
695
# Pre-populate data for testing using db api
696
net = db.network_create(self.tenant_id, self.network_name_1)
697
network_id = net['uuid']
698
port = db.port_create(network_id)
699
port_id = port['uuid']
700
cli.show_port(self.client,
706
LOG.exception("Exception caught: %s", sys.exc_info())
707
self.fail("test_show_port failed due to an exception")
709
LOG.debug("Operation completed. Verifying result")
710
LOG.debug(self.fake_stdout.content)
711
self._verify_show_port(network_id, port_id)
713
def test_show_port_details_no_attach(self):
717
# Pre-populate data for testing using db api
718
net = db.network_create(self.tenant_id, self.network_name_1)
719
network_id = net['uuid']
720
port = db.port_create(network_id)
721
port_id = port['uuid']
722
cli.show_port_detail(self.client,
728
LOG.exception("Exception caught: %s", sys.exc_info())
729
self.fail("test_show_port_details_no_attach failed due to" +
732
LOG.debug("Operation completed. Verifying result")
733
LOG.debug(self.fake_stdout.content)
734
self._verify_show_port_details(network_id, port_id)
736
def test_show_port_details_with_attach(self):
739
iface_id = "flavor crystals"
741
# Pre-populate data for testing using db api
742
net = db.network_create(self.tenant_id, self.network_name_1)
743
network_id = net['uuid']
744
port = db.port_create(network_id)
745
port_id = port['uuid']
746
db.port_set_attachment(port_id, network_id, iface_id)
747
cli.show_port_detail(self.client,
753
LOG.exception("Exception caught: %s", sys.exc_info())
754
self.fail("test_show_port_details_with_attach failed due" +
757
LOG.debug("Operation completed. Verifying result")
758
LOG.debug(self.fake_stdout.content)
759
self._verify_show_port_details(network_id, port_id)
761
def test_plug_iface(self):
765
# Load some data into the datbase
766
net = db.network_create(self.tenant_id, self.network_name_1)
767
network_id = net['uuid']
768
port = db.port_create(net['uuid'])
769
port_id = port['uuid']
770
cli.plug_iface(self.client,
777
LOG.exception("Exception caught: %s", sys.exc_info())
778
self.fail("test_plug_iface failed due to an exception")
780
LOG.debug("Operation completed. Verifying result")
781
LOG.debug(self.fake_stdout.content)
782
self._verify_plug_iface(network_id, port_id)
784
def test_unplug_iface(self):
788
# Load some data into the datbase
789
net = db.network_create(self.tenant_id, self.network_name_1)
790
network_id = net['uuid']
791
port = db.port_create(net['uuid'])
792
port_id = port['uuid']
793
db.port_set_attachment(port_id, network_id, "test_iface_id")
794
cli.unplug_iface(self.client,
800
LOG.exception("Exception caught: %s", sys.exc_info())
801
self.fail("test_plug_iface failed due to an exception")
803
LOG.debug("Operation completed. Verifying result")
804
LOG.debug(self.fake_stdout.content)
805
self._verify_unplug_iface(network_id, port_id)
807
def test_show_iface_no_attach(self):
811
# Pre-populate data for testing using db api
812
net = db.network_create(self.tenant_id, self.network_name_1)
813
network_id = net['uuid']
814
port = db.port_create(network_id)
815
port_id = port['uuid']
816
cli.show_iface(self.client,
822
LOG.exception("Exception caught: %s", sys.exc_info())
823
self.fail("test_show_iface_no_attach failed due to" +
826
LOG.debug("Operation completed. Verifying result")
827
LOG.debug(self.fake_stdout.content)
828
self._verify_show_iface(network_id, port_id)
830
def test_show_iface_with_attach(self):
833
iface_id = "flavor crystals"
835
# Pre-populate data for testing using db api
836
net = db.network_create(self.tenant_id, self.network_name_1)
837
network_id = net['uuid']
838
port = db.port_create(network_id)
839
port_id = port['uuid']
840
db.port_set_attachment(port_id, network_id, iface_id)
841
cli.show_iface(self.client,
847
LOG.exception("Exception caught: %s", sys.exc_info())
848
self.fail("test_show_iface_with_attach failed due" +
851
LOG.debug("Operation completed. Verifying result")
852
LOG.debug(self.fake_stdout.content)
853
self._verify_show_iface(network_id, port_id)