← Back to branch summary

~zulcss/cinder/cinder-ca-g2

« back to all changes in this revision

Viewing changes to cinder/tests/api/openstack/volume/test_snapshots.py

  • Committer: Package Import Robot
  • Author(s): Chuck Short
  • Date: 2012-10-09 08:26:21 UTC
  • mfrom: (3.1.10 quantal)
  • Revision ID: package-import@ubuntu.com-20121009082621-stc86vcuyzyrp7ow
Tags: 2012.2-0ubuntu2
https://launchpad.net/bugs/1064366
* debian/cinder_tgt.conf: Add missing configuration file. (LP: #1064366) 
* debian/README.Debian: Added note about migration from nova-volume
  to cinder-volume.

Show diffs side-by-side

added added

removed removed

Lines of Context:
cinder/tests/api/openstack/volume/test_snapshots.py
 
1
# Copyright 2011 Denali Systems, Inc.
 
2
# All Rights Reserved.
 
3
#
 
4
#    Licensed under the Apache License, Version 2.0 (the "License"); you may
 
5
#    not use this file except in compliance with the License. You may obtain
 
6
#    a copy of the License at
 
7
#
 
8
#         http://www.apache.org/licenses/LICENSE-2.0
 
9
#
 
10
#    Unless required by applicable law or agreed to in writing, software
 
11
#    distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
 
12
#    WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
 
13
#    License for the specific language governing permissions and limitations
 
14
#    under the License.
 
15
 
 
16
import datetime
 
17
 
 
18
from lxml import etree
 
19
import webob
 
20
 
 
21
from cinder.api.openstack.volume import snapshots
 
22
from cinder import db
 
23
from cinder import exception
 
24
from cinder import flags
 
25
from cinder.openstack.common import log as logging
 
26
from cinder import test
 
27
from cinder import volume
 
28
from cinder.tests.api.openstack import fakes
 
29
 
 
30
 
 
31
FLAGS = flags.FLAGS
 
32
LOG = logging.getLogger(__name__)
 
33
 
 
34
UUID = '00000000-0000-0000-0000-000000000001'
 
35
INVALID_UUID = '00000000-0000-0000-0000-000000000002'
 
36
 
 
37
 
 
38
def _get_default_snapshot_param():
 
39
    return {
 
40
        'id': UUID,
 
41
        'volume_id': 12,
 
42
        'status': 'available',
 
43
        'volume_size': 100,
 
44
        'created_at': None,
 
45
        'display_name': 'Default name',
 
46
        'display_description': 'Default description',
 
47
        }
 
48
 
 
49
 
 
50
def stub_snapshot_create(self, context, volume_id, name, description):
 
51
    snapshot = _get_default_snapshot_param()
 
52
    snapshot['volume_id'] = volume_id
 
53
    snapshot['display_name'] = name
 
54
    snapshot['display_description'] = description
 
55
    return snapshot
 
56
 
 
57
 
 
58
def stub_snapshot_delete(self, context, snapshot):
 
59
    if snapshot['id'] != UUID:
 
60
        raise exception.NotFound
 
61
 
 
62
 
 
63
def stub_snapshot_get(self, context, snapshot_id):
 
64
    if snapshot_id != UUID:
 
65
        raise exception.NotFound
 
66
 
 
67
    param = _get_default_snapshot_param()
 
68
    return param
 
69
 
 
70
 
 
71
def stub_snapshot_get_all(self, context, search_opts=None):
 
72
    param = _get_default_snapshot_param()
 
73
    return [param]
 
74
 
 
75
 
 
76
class SnapshotApiTest(test.TestCase):
 
77
    def setUp(self):
 
78
        super(SnapshotApiTest, self).setUp()
 
79
        self.controller = snapshots.SnapshotsController()
 
80
 
 
81
        self.stubs.Set(db, 'snapshot_get_all_by_project',
 
82
                       fakes.stub_snapshot_get_all_by_project)
 
83
        self.stubs.Set(db, 'snapshot_get_all',
 
84
                      fakes.stub_snapshot_get_all)
 
85
 
 
86
    def test_snapshot_create(self):
 
87
        self.stubs.Set(volume.api.API, "create_snapshot", stub_snapshot_create)
 
88
        self.stubs.Set(volume.api.API, 'get', fakes.stub_volume_get)
 
89
        snapshot = {"volume_id": '12',
 
90
                "force": False,
 
91
                "display_name": "Snapshot Test Name",
 
92
                "display_description": "Snapshot Test Desc"}
 
93
        body = dict(snapshot=snapshot)
 
94
        req = fakes.HTTPRequest.blank('/v1/snapshots')
 
95
        resp_dict = self.controller.create(req, body)
 
96
 
 
97
        self.assertTrue('snapshot' in resp_dict)
 
98
        self.assertEqual(resp_dict['snapshot']['display_name'],
 
99
                        snapshot['display_name'])
 
100
        self.assertEqual(resp_dict['snapshot']['display_description'],
 
101
                        snapshot['display_description'])
 
102
 
 
103
    def test_snapshot_create_force(self):
 
104
        self.stubs.Set(volume.api.API, "create_snapshot_force",
 
105
            stub_snapshot_create)
 
106
        self.stubs.Set(volume.api.API, 'get', fakes.stub_volume_get)
 
107
        snapshot = {"volume_id": '12',
 
108
                "force": True,
 
109
                "display_name": "Snapshot Test Name",
 
110
                "display_description": "Snapshot Test Desc"}
 
111
        body = dict(snapshot=snapshot)
 
112
        req = fakes.HTTPRequest.blank('/v1/snapshots')
 
113
        resp_dict = self.controller.create(req, body)
 
114
 
 
115
        self.assertTrue('snapshot' in resp_dict)
 
116
        self.assertEqual(resp_dict['snapshot']['display_name'],
 
117
                        snapshot['display_name'])
 
118
        self.assertEqual(resp_dict['snapshot']['display_description'],
 
119
                        snapshot['display_description'])
 
120
 
 
121
        snapshot = {"volume_id": "12",
 
122
                "force": "**&&^^%%$$##@@",
 
123
                "display_name": "Snapshot Test Name",
 
124
                "display_description": "Snapshot Test Desc"}
 
125
        body = dict(snapshot=snapshot)
 
126
        req = fakes.HTTPRequest.blank('/v1/snapshots')
 
127
        self.assertRaises(exception.InvalidParameterValue,
 
128
                          self.controller.create,
 
129
                          req,
 
130
                          body)
 
131
 
 
132
    def test_snapshot_delete(self):
 
133
        self.stubs.Set(volume.api.API, "get_snapshot", stub_snapshot_get)
 
134
        self.stubs.Set(volume.api.API, "delete_snapshot", stub_snapshot_delete)
 
135
 
 
136
        snapshot_id = UUID
 
137
        req = fakes.HTTPRequest.blank('/v1/snapshots/%s' % snapshot_id)
 
138
        resp = self.controller.delete(req, snapshot_id)
 
139
        self.assertEqual(resp.status_int, 202)
 
140
 
 
141
    def test_snapshot_delete_invalid_id(self):
 
142
        self.stubs.Set(volume.api.API, "delete_snapshot", stub_snapshot_delete)
 
143
        snapshot_id = INVALID_UUID
 
144
        req = fakes.HTTPRequest.blank('/v1/snapshots/%s' % snapshot_id)
 
145
        self.assertRaises(webob.exc.HTTPNotFound,
 
146
                          self.controller.delete,
 
147
                          req,
 
148
                          snapshot_id)
 
149
 
 
150
    def test_snapshot_show(self):
 
151
        self.stubs.Set(volume.api.API, "get_snapshot", stub_snapshot_get)
 
152
        req = fakes.HTTPRequest.blank('/v1/snapshots/%s' % UUID)
 
153
        resp_dict = self.controller.show(req, UUID)
 
154
 
 
155
        self.assertTrue('snapshot' in resp_dict)
 
156
        self.assertEqual(resp_dict['snapshot']['id'], UUID)
 
157
 
 
158
    def test_snapshot_show_invalid_id(self):
 
159
        snapshot_id = INVALID_UUID
 
160
        req = fakes.HTTPRequest.blank('/v1/snapshots/%s' % snapshot_id)
 
161
        self.assertRaises(webob.exc.HTTPNotFound,
 
162
                          self.controller.show,
 
163
                          req,
 
164
                          snapshot_id)
 
165
 
 
166
    def test_snapshot_detail(self):
 
167
        self.stubs.Set(volume.api.API, "get_all_snapshots",
 
168
            stub_snapshot_get_all)
 
169
        req = fakes.HTTPRequest.blank('/v1/snapshots/detail')
 
170
        resp_dict = self.controller.detail(req)
 
171
 
 
172
        self.assertTrue('snapshots' in resp_dict)
 
173
        resp_snapshots = resp_dict['snapshots']
 
174
        self.assertEqual(len(resp_snapshots), 1)
 
175
 
 
176
        resp_snapshot = resp_snapshots.pop()
 
177
        self.assertEqual(resp_snapshot['id'], UUID)
 
178
 
 
179
    def test_snapshot_list_by_status(self):
 
180
        def stub_snapshot_get_all_by_project(context, project_id):
 
181
            return [
 
182
                fakes.stub_snapshot(1, display_name='backup1',
 
183
                                    status='available'),
 
184
                fakes.stub_snapshot(2, display_name='backup2',
 
185
                                    status='available'),
 
186
                fakes.stub_snapshot(3, display_name='backup3',
 
187
                                    status='creating'),
 
188
            ]
 
189
        self.stubs.Set(db, 'snapshot_get_all_by_project',
 
190
                       stub_snapshot_get_all_by_project)
 
191
 
 
192
        # no status filter
 
193
        req = fakes.HTTPRequest.blank('/v1/snapshots')
 
194
        resp = self.controller.index(req)
 
195
        self.assertEqual(len(resp['snapshots']), 3)
 
196
        # single match
 
197
        req = fakes.HTTPRequest.blank('/v1/snapshots?status=creating')
 
198
        resp = self.controller.index(req)
 
199
        self.assertEqual(len(resp['snapshots']), 1)
 
200
        self.assertEqual(resp['snapshots'][0]['status'], 'creating')
 
201
        # multiple match
 
202
        req = fakes.HTTPRequest.blank('/v1/snapshots?status=available')
 
203
        resp = self.controller.index(req)
 
204
        self.assertEqual(len(resp['snapshots']), 2)
 
205
        for snapshot in resp['snapshots']:
 
206
            self.assertEquals(snapshot['status'], 'available')
 
207
        # no match
 
208
        req = fakes.HTTPRequest.blank('/v1/snapshots?status=error')
 
209
        resp = self.controller.index(req)
 
210
        self.assertEqual(len(resp['snapshots']), 0)
 
211
 
 
212
    def test_snapshot_list_by_volume(self):
 
213
        def stub_snapshot_get_all_by_project(context, project_id):
 
214
            return [
 
215
                fakes.stub_snapshot(1, volume_id='vol1', status='creating'),
 
216
                fakes.stub_snapshot(2, volume_id='vol1', status='available'),
 
217
                fakes.stub_snapshot(3, volume_id='vol2', status='available'),
 
218
            ]
 
219
        self.stubs.Set(db, 'snapshot_get_all_by_project',
 
220
                       stub_snapshot_get_all_by_project)
 
221
 
 
222
        # single match
 
223
        req = fakes.HTTPRequest.blank('/v1/snapshots?volume_id=vol2')
 
224
        resp = self.controller.index(req)
 
225
        self.assertEqual(len(resp['snapshots']), 1)
 
226
        self.assertEqual(resp['snapshots'][0]['volume_id'], 'vol2')
 
227
        # multiple match
 
228
        req = fakes.HTTPRequest.blank('/v1/snapshots?volume_id=vol1')
 
229
        resp = self.controller.index(req)
 
230
        self.assertEqual(len(resp['snapshots']), 2)
 
231
        for snapshot in resp['snapshots']:
 
232
            self.assertEqual(snapshot['volume_id'], 'vol1')
 
233
        # multiple filters
 
234
        req = fakes.HTTPRequest.blank('/v1/snapshots?volume_id=vol1'
 
235
                                      '&status=available')
 
236
        resp = self.controller.index(req)
 
237
        self.assertEqual(len(resp['snapshots']), 1)
 
238
        self.assertEqual(resp['snapshots'][0]['volume_id'], 'vol1')
 
239
        self.assertEqual(resp['snapshots'][0]['status'], 'available')
 
240
 
 
241
    def test_snapshot_list_by_name(self):
 
242
        def stub_snapshot_get_all_by_project(context, project_id):
 
243
            return [
 
244
                fakes.stub_snapshot(1, display_name='backup1'),
 
245
                fakes.stub_snapshot(2, display_name='backup2'),
 
246
                fakes.stub_snapshot(3, display_name='backup3'),
 
247
            ]
 
248
        self.stubs.Set(db, 'snapshot_get_all_by_project',
 
249
                       stub_snapshot_get_all_by_project)
 
250
 
 
251
        # no display_name filter
 
252
        req = fakes.HTTPRequest.blank('/v1/snapshots')
 
253
        resp = self.controller.index(req)
 
254
        self.assertEqual(len(resp['snapshots']), 3)
 
255
        # filter by one name
 
256
        req = fakes.HTTPRequest.blank('/v1/snapshots?display_name=backup2')
 
257
        resp = self.controller.index(req)
 
258
        self.assertEqual(len(resp['snapshots']), 1)
 
259
        self.assertEquals(resp['snapshots'][0]['display_name'], 'backup2')
 
260
        # filter no match
 
261
        req = fakes.HTTPRequest.blank('/v1/snapshots?display_name=backup4')
 
262
        resp = self.controller.index(req)
 
263
        self.assertEqual(len(resp['snapshots']), 0)
 
264
 
 
265
    def test_admin_list_snapshots_limited_to_project(self):
 
266
        req = fakes.HTTPRequest.blank('/v1/fake/snapshots',
 
267
                                      use_admin_context=True)
 
268
        res = self.controller.index(req)
 
269
 
 
270
        self.assertTrue('snapshots' in res)
 
271
        self.assertEqual(1, len(res['snapshots']))
 
272
 
 
273
    def test_admin_list_snapshots_all_tenants(self):
 
274
        req = fakes.HTTPRequest.blank('/v1/fake/snapshots?all_tenants=1',
 
275
                                      use_admin_context=True)
 
276
        res = self.controller.index(req)
 
277
        self.assertTrue('snapshots' in res)
 
278
        self.assertEqual(3, len(res['snapshots']))
 
279
 
 
280
    def test_all_tenants_non_admin_gets_all_tenants(self):
 
281
        req = fakes.HTTPRequest.blank('/v1/fake/snapshots?all_tenants=1')
 
282
        res = self.controller.index(req)
 
283
        self.assertTrue('snapshots' in res)
 
284
        self.assertEqual(1, len(res['snapshots']))
 
285
 
 
286
    def test_non_admin_get_by_project(self):
 
287
        req = fakes.HTTPRequest.blank('/v1/fake/snapshots')
 
288
        res = self.controller.index(req)
 
289
        self.assertTrue('snapshots' in res)
 
290
        self.assertEqual(1, len(res['snapshots']))
 
291
 
 
292
 
 
293
class SnapshotSerializerTest(test.TestCase):
 
294
    def _verify_snapshot(self, snap, tree):
 
295
        self.assertEqual(tree.tag, 'snapshot')
 
296
 
 
297
        for attr in ('id', 'status', 'size', 'created_at',
 
298
                     'display_name', 'display_description', 'volume_id'):
 
299
            self.assertEqual(str(snap[attr]), tree.get(attr))
 
300
 
 
301
    def test_snapshot_show_create_serializer(self):
 
302
        serializer = snapshots.SnapshotTemplate()
 
303
        raw_snapshot = dict(
 
304
            id='snap_id',
 
305
            status='snap_status',
 
306
            size=1024,
 
307
            created_at=datetime.datetime.now(),
 
308
            display_name='snap_name',
 
309
            display_description='snap_desc',
 
310
            volume_id='vol_id',
 
311
            )
 
312
        text = serializer.serialize(dict(snapshot=raw_snapshot))
 
313
 
 
314
        print text
 
315
        tree = etree.fromstring(text)
 
316
 
 
317
        self._verify_snapshot(raw_snapshot, tree)
 
318
 
 
319
    def test_snapshot_index_detail_serializer(self):
 
320
        serializer = snapshots.SnapshotsTemplate()
 
321
        raw_snapshots = [dict(
 
322
                id='snap1_id',
 
323
                status='snap1_status',
 
324
                size=1024,
 
325
                created_at=datetime.datetime.now(),
 
326
                display_name='snap1_name',
 
327
                display_description='snap1_desc',
 
328
                volume_id='vol1_id',
 
329
                ),
 
330
                       dict(
 
331
                id='snap2_id',
 
332
                status='snap2_status',
 
333
                size=1024,
 
334
                created_at=datetime.datetime.now(),
 
335
                display_name='snap2_name',
 
336
                display_description='snap2_desc',
 
337
                volume_id='vol2_id',
 
338
                )]
 
339
        text = serializer.serialize(dict(snapshots=raw_snapshots))
 
340
 
 
341
        print text
 
342
        tree = etree.fromstring(text)
 
343
 
 
344
        self.assertEqual('snapshots', tree.tag)
 
345
        self.assertEqual(len(raw_snapshots), len(tree))
 
346
        for idx, child in enumerate(tree):
 
347
            self._verify_snapshot(raw_snapshots[idx], child)
 
348
 
 
349
 
 
350
class SnapshotsUnprocessableEntityTestCase(test.TestCase):
 
351
 
 
352
    """
 
353
    Tests of places we throw 422 Unprocessable Entity from
 
354
    """
 
355
 
 
356
    def setUp(self):
 
357
        super(SnapshotsUnprocessableEntityTestCase, self).setUp()
 
358
        self.controller = snapshots.SnapshotsController()
 
359
 
 
360
    def _unprocessable_snapshot_create(self, body):
 
361
        req = fakes.HTTPRequest.blank('/v2/fake/snapshots')
 
362
        req.method = 'POST'
 
363
 
 
364
        self.assertRaises(webob.exc.HTTPUnprocessableEntity,
 
365
                          self.controller.create, req, body)
 
366
 
 
367
    def test_create_no_body(self):
 
368
        self._unprocessable_snapshot_create(body=None)
 
369
 
 
370
    def test_create_missing_snapshot(self):
 
371
        body = {'foo': {'a': 'b'}}
 
372
        self._unprocessable_snapshot_create(body=body)
 
373
 
 
374
    def test_create_malformed_entity(self):
 
375
        body = {'snapshot': 'string'}
 
376
        self._unprocessable_snapshot_create(body=body)