« back to all changes in this revision
Viewing changes to tests/unit/test_cli.py
- browse files at revision 49
- compare with another revision
- download diff
- download tarball
- view history from revision 49
- Committer: Salvatore Orlando
- Date: 2011-08-30 16:06:19 UTC
- mfrom: (37.1.22 quantum-835216)
- Revision ID: salvatore.orlando@eu.citrix.com-20110830160619-ay4jfkz9lsy56t0h
merge trunk
- files added:
- bin/cli
- tests/unit/client_tools
- files removed:
- quantum/cli.py
- files modified:
- quantum/api/attachments.py
added
removed
tests/unit/test_cli.py
1
# vim: tabstop=4 shiftwidth=4 softtabstop=4
2
3
# Copyright 2010-2011 ????
4
# All Rights Reserved.
5
#
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
9
#
10
# http://www.apache.org/licenses/LICENSE-2.0
11
#
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
16
# under the License.
17
# @author: Salvatore Orlando, Citrix Systems
18
19
""" Module containing unit tests for Quantum
20
command line interface
21
22
"""
23
24
25
import logging
26
import sys
27
import unittest
28
29
from quantum import api as server
30
from quantum import cli_lib as cli
31
from quantum.client import Client
32
from quantum.db import api as db
33
from tests.unit.client_tools import stubs as client_stubs
34
35
LOG = logging.getLogger('quantum.tests.test_cli')
36
FORMAT = 'json'
37
38
39
class CLITest(unittest.TestCase):
40
41
def setUp(self):
42
"""Prepare the test environment"""
43
options = {}
44
options['plugin_provider'] = 'quantum.plugins.SamplePlugin.FakePlugin'
45
self.api = server.APIRouterV01(options)
46
47
self.tenant_id = "test_tenant"
48
self.network_name_1 = "test_network_1"
49
self.network_name_2 = "test_network_2"
50
# Prepare client and plugin manager
51
self.client = Client(tenant=self.tenant_id, format=FORMAT,
52
testingStub=client_stubs.FakeHTTPConnection)
53
# Redirect stdout
54
self.fake_stdout = client_stubs.FakeStdout()
55
sys.stdout = self.fake_stdout
56
57
def tearDown(self):
58
"""Clear the test environment"""
59
db.clear_db()
60
sys.stdout = sys.__stdout__
61
62
def _verify_list_networks(self):
63
# Verification - get raw result from db
64
nw_list = db.network_list(self.tenant_id)
65
networks = [dict(id=nw.uuid, name=nw.name) for nw in nw_list]
66
# Fill CLI template
67
output = cli.prepare_output('list_nets', self.tenant_id,
68
dict(networks=networks))
69
# Verify!
70
# Must add newline at the end to match effect of print call
71
self.assertEquals(self.fake_stdout.make_string(), output + '\n')
72
73
def _verify_create_network(self):
74
# Verification - get raw result from db
75
nw_list = db.network_list(self.tenant_id)
76
if len(nw_list) != 1:
77
self.fail("No network created")
78
network_id = nw_list[0].uuid
79
# Fill CLI template
80
output = cli.prepare_output('create_net', self.tenant_id,
81
dict(network_id=network_id))
82
# Verify!
83
# Must add newline at the end to match effect of print call
84
self.assertEquals(self.fake_stdout.make_string(), output + '\n')
85
86
def _verify_delete_network(self, network_id):
87
# Verification - get raw result from db
88
nw_list = db.network_list(self.tenant_id)
89
if len(nw_list) != 0:
90
self.fail("DB should not contain any network")
91
# Fill CLI template
92
output = cli.prepare_output('delete_net', self.tenant_id,
93
dict(network_id=network_id))
94
# Verify!
95
# Must add newline at the end to match effect of print call
96
self.assertEquals(self.fake_stdout.make_string(), output + '\n')
97
98
def _verify_rename_network(self):
99
# Verification - get raw result from db
100
nw_list = db.network_list(self.tenant_id)
101
network_data = {'id': nw_list[0].uuid,
102
'name': nw_list[0].name}
103
# Fill CLI template
104
output = cli.prepare_output('rename_net', self.tenant_id,
105
dict(network=network_data))
106
# Verify!
107
# Must add newline at the end to match effect of print call
108
self.assertEquals(self.fake_stdout.make_string(), output + '\n')
109
110
def _verify_show_network(self):
111
# Verification - get raw result from db
112
nw = db.network_list(self.tenant_id)[0]
113
network = dict(id=nw.uuid, name=nw.name)
114
# Fill CLI template
115
output = cli.prepare_output('show_net', self.tenant_id,
116
dict(network=network))
117
# Verify!
118
# Must add newline at the end to match effect of print call
119
self.assertEquals(self.fake_stdout.make_string(), output + '\n')
120
121
def _verify_list_ports(self, network_id):
122
# Verification - get raw result from db
123
port_list = db.port_list(network_id)
124
ports = [dict(id=port.uuid, state=port.state)
125
for port in port_list]
126
# Fill CLI template
127
output = cli.prepare_output('list_ports', self.tenant_id,
128
dict(network_id=network_id,
129
ports=ports))
130
# Verify!
131
# Must add newline at the end to match effect of print call
132
self.assertEquals(self.fake_stdout.make_string(), output + '\n')
133
134
def _verify_create_port(self, network_id):
135
# Verification - get raw result from db
136
port_list = db.port_list(network_id)
137
if len(port_list) != 1:
138
self.fail("No port created")
139
port_id = port_list[0].uuid
140
# Fill CLI template
141
output = cli.prepare_output('create_port', self.tenant_id,
142
dict(network_id=network_id,
143
port_id=port_id))
144
# Verify!
145
# Must add newline at the end to match effect of print call
146
self.assertEquals(self.fake_stdout.make_string(), output + '\n')
147
148
def _verify_delete_port(self, network_id, port_id):
149
# Verification - get raw result from db
150
port_list = db.port_list(network_id)
151
if len(port_list) != 0:
152
self.fail("DB should not contain any port")
153
# Fill CLI template
154
output = cli.prepare_output('delete_port', self.tenant_id,
155
dict(network_id=network_id,
156
port_id=port_id))
157
# Verify!
158
# Must add newline at the end to match effect of print call
159
self.assertEquals(self.fake_stdout.make_string(), output + '\n')
160
161
def _verify_set_port_state(self, network_id, port_id):
162
# Verification - get raw result from db
163
port = db.port_get(port_id, network_id)
164
port_data = {'id': port.uuid, 'state': port.state}
165
# Fill CLI template
166
output = cli.prepare_output('set_port_state', self.tenant_id,
167
dict(network_id=network_id,
168
port=port_data))
169
# Verify!
170
# Must add newline at the end to match effect of print call
171
self.assertEquals(self.fake_stdout.make_string(), output + '\n')
172
173
def _verify_show_port(self, network_id, port_id):
174
# Verification - get raw result from db
175
# TODO(salvatore-orlando): Must resolve this issue with
176
# attachment in separate bug fix.
177
port = db.port_get(port_id, network_id)
178
port_data = {'id': port.uuid, 'state': port.state,
179
'attachment': "<none>"}
180
if port.interface_id is not None:
181
port_data['attachment'] = port.interface_id
182
183
# Fill CLI template
184
output = cli.prepare_output('show_port', self.tenant_id,
185
dict(network_id=network_id,
186
port=port_data))
187
# Verify!
188
# Must add newline at the end to match effect of print call
189
self.assertEquals(self.fake_stdout.make_string(), output + '\n')
190
191
def _verify_plug_iface(self, network_id, port_id):
192
# Verification - get raw result from db
193
port = db.port_get(port_id, network_id)
194
# Fill CLI template
195
output = cli.prepare_output("plug_iface", self.tenant_id,
196
dict(network_id=network_id,
197
port_id=port['uuid'],
198
attachment=port['interface_id']))
199
# Verify!
200
# Must add newline at the end to match effect of print call
201
self.assertEquals(self.fake_stdout.make_string(), output + '\n')
202
203
def _verify_unplug_iface(self, network_id, port_id):
204
# Verification - get raw result from db
205
port = db.port_get(port_id, network_id)
206
# Fill CLI template
207
output = cli.prepare_output("unplug_iface", self.tenant_id,
208
dict(network_id=network_id,
209
port_id=port['uuid']))
210
# Verify!
211
# Must add newline at the end to match effect of print call
212
self.assertEquals(self.fake_stdout.make_string(), output + '\n')
213
214
def test_list_networks(self):
215
try:
216
# Pre-populate data for testing using db api
217
db.network_create(self.tenant_id, self.network_name_1)
218
db.network_create(self.tenant_id, self.network_name_2)
219
220
cli.list_nets(self.client, self.tenant_id)
221
except:
222
LOG.exception("Exception caught: %s", sys.exc_info())
223
self.fail("test_list_networks failed due to an exception")
224
225
LOG.debug("Operation completed. Verifying result")
226
LOG.debug(self.fake_stdout.content)
227
self._verify_list_networks()
228
229
def test_create_network(self):
230
try:
231
cli.create_net(self.client, self.tenant_id, "test")
232
except:
233
LOG.exception("Exception caught: %s", sys.exc_info())
234
self.fail("test_create_network failed due to an exception")
235
236
LOG.debug("Operation completed. Verifying result")
237
LOG.debug(self.fake_stdout.content)
238
self._verify_create_network()
239
240
def test_delete_network(self):
241
try:
242
db.network_create(self.tenant_id, self.network_name_1)
243
network_id = db.network_list(self.tenant_id)[0]['uuid']
244
cli.delete_net(self.client, self.tenant_id, network_id)
245
except:
246
LOG.exception("Exception caught: %s", sys.exc_info())
247
self.fail("test_delete_network failed due to an exception")
248
249
LOG.debug("Operation completed. Verifying result")
250
LOG.debug(self.fake_stdout.content)
251
self._verify_delete_network(network_id)
252
253
def test_show_network(self):
254
try:
255
# Load some data into the datbase
256
net = db.network_create(self.tenant_id, self.network_name_1)
257
cli.show_net(self.client, self.tenant_id, net['uuid'])
258
except:
259
LOG.exception("Exception caught: %s", sys.exc_info())
260
self.fail("test_detail_network failed due to an exception")
261
262
LOG.debug("Operation completed. Verifying result")
263
LOG.debug(self.fake_stdout.content)
264
self._verify_show_network()
265
266
def test_rename_network(self):
267
try:
268
net = db.network_create(self.tenant_id, self.network_name_1)
269
network_id = net['uuid']
270
cli.rename_net(self.client, self.tenant_id,
271
network_id, self.network_name_2)
272
except:
273
LOG.exception("Exception caught: %s", sys.exc_info())
274
self.fail("test_rename_network failed due to an exception")
275
276
LOG.debug("Operation completed. Verifying result")
277
LOG.debug(self.fake_stdout.content)
278
self._verify_rename_network()
279
280
def test_list_ports(self):
281
try:
282
# Pre-populate data for testing using db api
283
net = db.network_create(self.tenant_id, self.network_name_1)
284
network_id = net['uuid']
285
db.port_create(network_id)
286
db.port_create(network_id)
287
cli.list_ports(self.client, self.tenant_id, network_id)
288
except:
289
LOG.exception("Exception caught: %s", sys.exc_info())
290
self.fail("test_list_ports failed due to an exception")
291
292
LOG.debug("Operation completed. Verifying result")
293
LOG.debug(self.fake_stdout.content)
294
self._verify_list_ports(network_id)
295
296
def test_create_port(self):
297
network_id = None
298
try:
299
# Pre-populate data for testing using db api
300
net = db.network_create(self.tenant_id, self.network_name_1)
301
network_id = net['uuid']
302
cli.create_port(self.client, self.tenant_id, network_id)
303
except:
304
LOG.exception("Exception caught: %s", sys.exc_info())
305
self.fail("test_create_port failed due to an exception")
306
307
LOG.debug("Operation completed. Verifying result")
308
LOG.debug(self.fake_stdout.content)
309
self._verify_create_port(network_id)
310
311
def test_delete_port(self):
312
network_id = None
313
port_id = None
314
try:
315
# Pre-populate data for testing using db api
316
net = db.network_create(self.tenant_id, self.network_name_1)
317
network_id = net['uuid']
318
port = db.port_create(network_id)
319
port_id = port['uuid']
320
cli.delete_port(self.client, self.tenant_id, network_id, port_id)
321
except:
322
LOG.exception("Exception caught: %s", sys.exc_info())
323
self.fail("test_delete_port failed due to an exception")
324
325
LOG.debug("Operation completed. Verifying result")
326
LOG.debug(self.fake_stdout.content)
327
self._verify_delete_port(network_id, port_id)
328
329
def test_set_port_state(self):
330
try:
331
net = db.network_create(self.tenant_id, self.network_name_1)
332
network_id = net['uuid']
333
port = db.port_create(network_id)
334
port_id = port['uuid']
335
# Default state is DOWN - change to ACTIVE.
336
cli.set_port_state(self.client, self.tenant_id, network_id,
337
port_id, 'ACTIVE')
338
except:
339
LOG.exception("Exception caught: %s", sys.exc_info())
340
self.fail("test_set_port_state failed due to an exception")
341
342
LOG.debug("Operation completed. Verifying result")
343
LOG.debug(self.fake_stdout.content)
344
self._verify_set_port_state(network_id, port_id)
345
346
def test_show_port_no_attach(self):
347
network_id = None
348
port_id = None
349
try:
350
# Pre-populate data for testing using db api
351
net = db.network_create(self.tenant_id, self.network_name_1)
352
network_id = net['uuid']
353
port = db.port_create(network_id)
354
port_id = port['uuid']
355
cli.show_port(self.client, self.tenant_id, network_id, port_id)
356
except:
357
LOG.exception("Exception caught: %s", sys.exc_info())
358
self.fail("test_show_port_no_attach failed due to an exception")
359
360
LOG.debug("Operation completed. Verifying result")
361
LOG.debug(self.fake_stdout.content)
362
self._verify_show_port(network_id, port_id)
363
364
def test_show_port_with_attach(self):
365
network_id = None
366
port_id = None
367
iface_id = "flavor crystals"
368
try:
369
# Pre-populate data for testing using db api
370
net = db.network_create(self.tenant_id, self.network_name_1)
371
network_id = net['uuid']
372
port = db.port_create(network_id)
373
port_id = port['uuid']
374
db.port_set_attachment(port_id, network_id, iface_id)
375
cli.show_port(self.client, self.tenant_id, network_id, port_id)
376
except:
377
LOG.exception("Exception caught: %s", sys.exc_info())
378
self.fail("test_show_port_with_attach failed due to an exception")
379
380
LOG.debug("Operation completed. Verifying result")
381
LOG.debug(self.fake_stdout.content)
382
self._verify_show_port(network_id, port_id)
383
384
def test_plug_iface(self):
385
network_id = None
386
port_id = None
387
try:
388
# Load some data into the datbase
389
net = db.network_create(self.tenant_id, self.network_name_1)
390
network_id = net['uuid']
391
port = db.port_create(net['uuid'])
392
port_id = port['uuid']
393
cli.plug_iface(self.client, self.tenant_id, network_id,
394
port_id, "test_iface_id")
395
except:
396
LOG.exception("Exception caught: %s", sys.exc_info())
397
self.fail("test_plug_iface failed due to an exception")
398
399
LOG.debug("Operation completed. Verifying result")
400
LOG.debug(self.fake_stdout.content)
401
self._verify_plug_iface(network_id, port_id)
402
403
def test_unplug_iface(self):
404
network_id = None
405
port_id = None
406
try:
407
# Load some data into the datbase
408
net = db.network_create(self.tenant_id, self.network_name_1)
409
network_id = net['uuid']
410
port = db.port_create(net['uuid'])
411
port_id = port['uuid']
412
db.port_set_attachment(port_id, network_id, "test_iface_id")
413
cli.unplug_iface(self.client, self.tenant_id, network_id, port_id)
414
except:
415
LOG.exception("Exception caught: %s", sys.exc_info())
416
self.fail("test_plug_iface failed due to an exception")
417
418
LOG.debug("Operation completed. Verifying result")
419
LOG.debug(self.fake_stdout.content)
420
self._verify_unplug_iface(network_id, port_id)
Loggerhead is a web-based interface for Breezy
Version: 2.0.1