← Back to branch summary

~ubuntu-branches/ubuntu/precise/python-quantumclient/precise

« back to all changes in this revision

Viewing changes to quantum/client/tests/unit/test_cli.py

  • Committer: Package Import Robot
  • Author(s): Ghe Rivero
  • Date: 2012-01-29 12:07:42 UTC
  • Revision ID: package-import@ubuntu.com-20120129120742-ezgggji4gkj0oh27
Tags: upstream-2012.1~e3
Import upstream version 2012.1~e3

Show diffs side-by-side

added added

removed removed

Lines of Context:
quantum/client/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.client import cli_lib as cli
 
31
from quantum.client import Client
 
32
from quantum.db import api as db
 
33
from quantum.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'] = \
 
45
          'quantum.plugins.sample.SamplePlugin.FakePlugin'
 
46
        #TODO: make the version of the API router configurable
 
47
        self.api = server.APIRouterV11(options)
 
48
 
 
49
        self.tenant_id = "test_tenant"
 
50
        self.network_name_1 = "test_network_1"
 
51
        self.network_name_2 = "test_network_2"
 
52
        # Prepare client and plugin manager
 
53
        self.client = Client(tenant=self.tenant_id, format=FORMAT,
 
54
                             testingStub=client_stubs.FakeHTTPConnection)
 
55
        # Redirect stdout
 
56
        self.fake_stdout = client_stubs.FakeStdout()
 
57
        sys.stdout = self.fake_stdout
 
58
 
 
59
    def tearDown(self):
 
60
        """Clear the test environment"""
 
61
        db.clear_db()
 
62
        sys.stdout = sys.__stdout__
 
63
 
 
64
    def _verify_list_networks(self):
 
65
            # Verification - get raw result from db
 
66
            nw_list = db.network_list(self.tenant_id)
 
67
            networks = [dict(id=nw.uuid, name=nw.name) for nw in nw_list]
 
68
            # Fill CLI template
 
69
            output = cli.prepare_output('list_nets', self.tenant_id,
 
70
                                        dict(networks=networks))
 
71
            # Verify!
 
72
            # Must add newline at the end to match effect of print call
 
73
            self.assertEquals(self.fake_stdout.make_string(), output + '\n')
 
74
 
 
75
    def _verify_create_network(self):
 
76
            # Verification - get raw result from db
 
77
            nw_list = db.network_list(self.tenant_id)
 
78
            if len(nw_list) != 1:
 
79
                self.fail("No network created")
 
80
            network_id = nw_list[0].uuid
 
81
            # Fill CLI template
 
82
            output = cli.prepare_output('create_net', self.tenant_id,
 
83
                                        dict(network_id=network_id))
 
84
            # Verify!
 
85
            # Must add newline at the end to match effect of print call
 
86
            self.assertEquals(self.fake_stdout.make_string(), output + '\n')
 
87
 
 
88
    def _verify_delete_network(self, network_id):
 
89
            # Verification - get raw result from db
 
90
            nw_list = db.network_list(self.tenant_id)
 
91
            if len(nw_list) != 0:
 
92
                self.fail("DB should not contain any network")
 
93
            # Fill CLI template
 
94
            output = cli.prepare_output('delete_net', self.tenant_id,
 
95
                                        dict(network_id=network_id))
 
96
            # Verify!
 
97
            # Must add newline at the end to match effect of print call
 
98
            self.assertEquals(self.fake_stdout.make_string(), output + '\n')
 
99
 
 
100
    def _verify_update_network(self):
 
101
            # Verification - get raw result from db
 
102
            nw_list = db.network_list(self.tenant_id)
 
103
            network_data = {'id': nw_list[0].uuid,
 
104
                            'name': nw_list[0].name}
 
105
            # Fill CLI template
 
106
            output = cli.prepare_output('update_net', self.tenant_id,
 
107
                                        dict(network=network_data))
 
108
            # Verify!
 
109
            # Must add newline at the end to match effect of print call
 
110
            self.assertEquals(self.fake_stdout.make_string(), output + '\n')
 
111
 
 
112
    def _verify_show_network(self):
 
113
            # Verification - get raw result from db
 
114
            nw = db.network_list(self.tenant_id)[0]
 
115
            network = dict(id=nw.uuid, name=nw.name)
 
116
            # Fill CLI template
 
117
            output = cli.prepare_output('show_net', self.tenant_id,
 
118
                                        dict(network=network))
 
119
            # Verify!
 
120
            # Must add newline at the end to match effect of print call
 
121
            self.assertEquals(self.fake_stdout.make_string(), output + '\n')
 
122
 
 
123
    def _verify_list_ports(self, network_id):
 
124
            # Verification - get raw result from db
 
125
            port_list = db.port_list(network_id)
 
126
            ports = [dict(id=port.uuid, state=port.state)
 
127
                     for port in port_list]
 
128
            # Fill CLI template
 
129
            output = cli.prepare_output('list_ports', self.tenant_id,
 
130
                                        dict(network_id=network_id,
 
131
                                             ports=ports))
 
132
            # Verify!
 
133
            # Must add newline at the end to match effect of print call
 
134
            self.assertEquals(self.fake_stdout.make_string(), output + '\n')
 
135
 
 
136
    def _verify_create_port(self, network_id):
 
137
            # Verification - get raw result from db
 
138
            port_list = db.port_list(network_id)
 
139
            if len(port_list) != 1:
 
140
                self.fail("No port created")
 
141
            port_id = port_list[0].uuid
 
142
            # Fill CLI template
 
143
            output = cli.prepare_output('create_port', self.tenant_id,
 
144
                                        dict(network_id=network_id,
 
145
                                             port_id=port_id))
 
146
            # Verify!
 
147
            # Must add newline at the end to match effect of print call
 
148
            self.assertEquals(self.fake_stdout.make_string(), output + '\n')
 
149
 
 
150
    def _verify_delete_port(self, network_id, port_id):
 
151
            # Verification - get raw result from db
 
152
            port_list = db.port_list(network_id)
 
153
            if len(port_list) != 0:
 
154
                self.fail("DB should not contain any port")
 
155
            # Fill CLI template
 
156
            output = cli.prepare_output('delete_port', self.tenant_id,
 
157
                                        dict(network_id=network_id,
 
158
                                             port_id=port_id))
 
159
            # Verify!
 
160
            # Must add newline at the end to match effect of print call
 
161
            self.assertEquals(self.fake_stdout.make_string(), output + '\n')
 
162
 
 
163
    def _verify_update_port(self, network_id, port_id):
 
164
            # Verification - get raw result from db
 
165
            port = db.port_get(port_id, network_id)
 
166
            port_data = {'id': port.uuid, 'state': port.state}
 
167
            # Fill CLI template
 
168
            output = cli.prepare_output('update_port', self.tenant_id,
 
169
                                        dict(network_id=network_id,
 
170
                                             port=port_data))
 
171
            # Verify!
 
172
            # Must add newline at the end to match effect of print call
 
173
            self.assertEquals(self.fake_stdout.make_string(), output + '\n')
 
174
 
 
175
    def _verify_show_port(self, network_id, port_id):
 
176
            # Verification - get raw result from db
 
177
            # TODO(salvatore-orlando): Must resolve this issue with
 
178
            # attachment in separate bug fix.
 
179
            port = db.port_get(port_id, network_id)
 
180
            port_data = {'id': port.uuid, 'state': port.state,
 
181
                         'attachment': "<none>"}
 
182
            if port.interface_id is not None:
 
183
                port_data['attachment'] = port.interface_id
 
184
 
 
185
            # Fill CLI template
 
186
            output = cli.prepare_output('show_port', self.tenant_id,
 
187
                                        dict(network_id=network_id,
 
188
                                             port=port_data))
 
189
            # Verify!
 
190
            # Must add newline at the end to match effect of print call
 
191
            self.assertEquals(self.fake_stdout.make_string(), output + '\n')
 
192
 
 
193
    def _verify_plug_iface(self, network_id, port_id):
 
194
            # Verification - get raw result from db
 
195
            port = db.port_get(port_id, network_id)
 
196
            # Fill CLI template
 
197
            output = cli.prepare_output("plug_iface", self.tenant_id,
 
198
                                        dict(network_id=network_id,
 
199
                                             port_id=port['uuid'],
 
200
                                             attachment=port['interface_id']))
 
201
            # Verify!
 
202
            # Must add newline at the end to match effect of print call
 
203
            self.assertEquals(self.fake_stdout.make_string(), output + '\n')
 
204
 
 
205
    def _verify_unplug_iface(self, network_id, port_id):
 
206
            # Verification - get raw result from db
 
207
            port = db.port_get(port_id, network_id)
 
208
            # Fill CLI template
 
209
            output = cli.prepare_output("unplug_iface", self.tenant_id,
 
210
                                        dict(network_id=network_id,
 
211
                                             port_id=port['uuid']))
 
212
            # Verify!
 
213
            # Must add newline at the end to match effect of print call
 
214
            self.assertEquals(self.fake_stdout.make_string(), output + '\n')
 
215
 
 
216
    def test_list_networks(self):
 
217
        try:
 
218
            # Pre-populate data for testing using db api
 
219
            db.network_create(self.tenant_id, self.network_name_1)
 
220
            db.network_create(self.tenant_id, self.network_name_2)
 
221
 
 
222
            cli.list_nets(self.client, self.tenant_id)
 
223
        except:
 
224
            LOG.exception("Exception caught: %s", sys.exc_info())
 
225
            self.fail("test_list_networks failed due to an exception")
 
226
 
 
227
        LOG.debug("Operation completed. Verifying result")
 
228
        LOG.debug(self.fake_stdout.content)
 
229
        self._verify_list_networks()
 
230
 
 
231
    def test_create_network(self):
 
232
        try:
 
233
            cli.create_net(self.client, self.tenant_id, "test")
 
234
        except:
 
235
            LOG.exception("Exception caught: %s", sys.exc_info())
 
236
            self.fail("test_create_network failed due to an exception")
 
237
 
 
238
        LOG.debug("Operation completed. Verifying result")
 
239
        LOG.debug(self.fake_stdout.content)
 
240
        self._verify_create_network()
 
241
 
 
242
    def test_delete_network(self):
 
243
        try:
 
244
            db.network_create(self.tenant_id, self.network_name_1)
 
245
            network_id = db.network_list(self.tenant_id)[0]['uuid']
 
246
            cli.delete_net(self.client, self.tenant_id, network_id)
 
247
        except:
 
248
            LOG.exception("Exception caught: %s", sys.exc_info())
 
249
            self.fail("test_delete_network failed due to an exception")
 
250
 
 
251
        LOG.debug("Operation completed. Verifying result")
 
252
        LOG.debug(self.fake_stdout.content)
 
253
        self._verify_delete_network(network_id)
 
254
 
 
255
    def test_show_network(self):
 
256
        try:
 
257
            # Load some data into the datbase
 
258
            net = db.network_create(self.tenant_id, self.network_name_1)
 
259
            cli.show_net(self.client, self.tenant_id, net['uuid'])
 
260
        except:
 
261
            LOG.exception("Exception caught: %s", sys.exc_info())
 
262
            self.fail("test_detail_network failed due to an exception")
 
263
 
 
264
        LOG.debug("Operation completed. Verifying result")
 
265
        LOG.debug(self.fake_stdout.content)
 
266
        self._verify_show_network()
 
267
 
 
268
    def test_update_network(self):
 
269
        try:
 
270
            net = db.network_create(self.tenant_id, self.network_name_1)
 
271
            network_id = net['uuid']
 
272
            cli.update_net(self.client, self.tenant_id,
 
273
                           network_id, 'name=%s' % self.network_name_2)
 
274
        except:
 
275
            LOG.exception("Exception caught: %s", sys.exc_info())
 
276
            self.fail("test_update_network failed due to an exception")
 
277
 
 
278
        LOG.debug("Operation completed. Verifying result")
 
279
        LOG.debug(self.fake_stdout.content)
 
280
        self._verify_update_network()
 
281
 
 
282
    def test_list_ports(self):
 
283
        try:
 
284
            # Pre-populate data for testing using db api
 
285
            net = db.network_create(self.tenant_id, self.network_name_1)
 
286
            network_id = net['uuid']
 
287
            db.port_create(network_id)
 
288
            db.port_create(network_id)
 
289
            cli.list_ports(self.client, self.tenant_id, network_id)
 
290
        except:
 
291
            LOG.exception("Exception caught: %s", sys.exc_info())
 
292
            self.fail("test_list_ports failed due to an exception")
 
293
 
 
294
        LOG.debug("Operation completed. Verifying result")
 
295
        LOG.debug(self.fake_stdout.content)
 
296
        self._verify_list_ports(network_id)
 
297
 
 
298
    def test_create_port(self):
 
299
        network_id = None
 
300
        try:
 
301
            # Pre-populate data for testing using db api
 
302
            net = db.network_create(self.tenant_id, self.network_name_1)
 
303
            network_id = net['uuid']
 
304
            cli.create_port(self.client, self.tenant_id, network_id)
 
305
        except:
 
306
            LOG.exception("Exception caught: %s", sys.exc_info())
 
307
            self.fail("test_create_port failed due to an exception")
 
308
 
 
309
        LOG.debug("Operation completed. Verifying result")
 
310
        LOG.debug(self.fake_stdout.content)
 
311
        self._verify_create_port(network_id)
 
312
 
 
313
    def test_delete_port(self):
 
314
        network_id = None
 
315
        port_id = None
 
316
        try:
 
317
            # Pre-populate data for testing using db api
 
318
            net = db.network_create(self.tenant_id, self.network_name_1)
 
319
            network_id = net['uuid']
 
320
            port = db.port_create(network_id)
 
321
            port_id = port['uuid']
 
322
            cli.delete_port(self.client, self.tenant_id, network_id, port_id)
 
323
        except:
 
324
            LOG.exception("Exception caught: %s", sys.exc_info())
 
325
            self.fail("test_delete_port failed due to an exception")
 
326
 
 
327
        LOG.debug("Operation completed. Verifying result")
 
328
        LOG.debug(self.fake_stdout.content)
 
329
        self._verify_delete_port(network_id, port_id)
 
330
 
 
331
    def test_update_port(self):
 
332
        try:
 
333
            net = db.network_create(self.tenant_id, self.network_name_1)
 
334
            network_id = net['uuid']
 
335
            port = db.port_create(network_id)
 
336
            port_id = port['uuid']
 
337
            # Default state is DOWN - change to ACTIVE.
 
338
            cli.update_port(self.client, self.tenant_id, network_id,
 
339
                               port_id, 'state=ACTIVE')
 
340
        except:
 
341
            LOG.exception("Exception caught: %s", sys.exc_info())
 
342
            self.fail("test_update_port failed due to an exception")
 
343
 
 
344
        LOG.debug("Operation completed. Verifying result")
 
345
        LOG.debug(self.fake_stdout.content)
 
346
        self._verify_update_port(network_id, port_id)
 
347
 
 
348
    def test_show_port_no_attach(self):
 
349
        network_id = None
 
350
        port_id = None
 
351
        try:
 
352
            # Pre-populate data for testing using db api
 
353
            net = db.network_create(self.tenant_id, self.network_name_1)
 
354
            network_id = net['uuid']
 
355
            port = db.port_create(network_id)
 
356
            port_id = port['uuid']
 
357
            cli.show_port(self.client, self.tenant_id, network_id, port_id)
 
358
        except:
 
359
            LOG.exception("Exception caught: %s", sys.exc_info())
 
360
            self.fail("test_show_port_no_attach failed due to an exception")
 
361
 
 
362
        LOG.debug("Operation completed. Verifying result")
 
363
        LOG.debug(self.fake_stdout.content)
 
364
        self._verify_show_port(network_id, port_id)
 
365
 
 
366
    def test_show_port_with_attach(self):
 
367
        network_id = None
 
368
        port_id = None
 
369
        iface_id = "flavor crystals"
 
370
        try:
 
371
            # Pre-populate data for testing using db api
 
372
            net = db.network_create(self.tenant_id, self.network_name_1)
 
373
            network_id = net['uuid']
 
374
            port = db.port_create(network_id)
 
375
            port_id = port['uuid']
 
376
            db.port_set_attachment(port_id, network_id, iface_id)
 
377
            cli.show_port(self.client, self.tenant_id, network_id, port_id)
 
378
        except:
 
379
            LOG.exception("Exception caught: %s", sys.exc_info())
 
380
            self.fail("test_show_port_with_attach failed due to an exception")
 
381
 
 
382
        LOG.debug("Operation completed. Verifying result")
 
383
        LOG.debug(self.fake_stdout.content)
 
384
        self._verify_show_port(network_id, port_id)
 
385
 
 
386
    def test_plug_iface(self):
 
387
        network_id = None
 
388
        port_id = None
 
389
        try:
 
390
            # Load some data into the datbase
 
391
            net = db.network_create(self.tenant_id, self.network_name_1)
 
392
            network_id = net['uuid']
 
393
            port = db.port_create(net['uuid'])
 
394
            port_id = port['uuid']
 
395
            cli.plug_iface(self.client, self.tenant_id, network_id,
 
396
                           port_id, "test_iface_id")
 
397
        except:
 
398
            LOG.exception("Exception caught: %s", sys.exc_info())
 
399
            self.fail("test_plug_iface failed due to an exception")
 
400
 
 
401
        LOG.debug("Operation completed. Verifying result")
 
402
        LOG.debug(self.fake_stdout.content)
 
403
        self._verify_plug_iface(network_id, port_id)
 
404
 
 
405
    def test_unplug_iface(self):
 
406
        network_id = None
 
407
        port_id = None
 
408
        try:
 
409
            # Load some data into the datbase
 
410
            net = db.network_create(self.tenant_id, self.network_name_1)
 
411
            network_id = net['uuid']
 
412
            port = db.port_create(net['uuid'])
 
413
            port_id = port['uuid']
 
414
            db.port_set_attachment(port_id, network_id, "test_iface_id")
 
415
            cli.unplug_iface(self.client, self.tenant_id, network_id, port_id)
 
416
        except:
 
417
            LOG.exception("Exception caught: %s", sys.exc_info())
 
418
            self.fail("test_plug_iface failed due to an exception")
 
419
 
 
420
        LOG.debug("Operation completed. Verifying result")
 
421
        LOG.debug(self.fake_stdout.content)
 
422
        self._verify_unplug_iface(network_id, port_id)