← Back to branch summary

~ubuntu-branches/ubuntu/raring/nova/raring-proposed

« back to all changes in this revision

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

  • Committer: Package Import Robot
  • Author(s): Chuck Short, Adam Gandelman, Chuck Short
  • Date: 2012-11-23 09:04:58 UTC
  • mfrom: (1.1.66)
  • Revision ID: package-import@ubuntu.com-20121123090458-91565o7aev1i1h71
Tags: 2013.1~g1-0ubuntu1
https://launchpad.net/bugs/1073289
https://launchpad.net/bugs/1076442
[ Adam Gandelman ]
* debian/control: Ensure novaclient is upgraded with nova,
  require python-keystoneclient >= 1:2.9.0. (LP: #1073289)
* debian/patches/{ubuntu/*, rbd-security.patch}: Dropped, applied
  upstream.
* debian/control: Add python-testtools to Build-Depends.

[ Chuck Short ]
* New upstream version.
* Refreshed debian/patches/avoid_setuptools_git_dependency.patch.
* debian/rules: FTBFS if missing binaries.
* debian/nova-scheudler.install: Add missing rabbit-queues and
  nova-rpc-zmq-receiver.
* Remove nova-volume since it doesnt exist anymore, transition to cinder-*.
* debian/rules: install apport hook in the right place.
* debian/patches/ubuntu-show-tests.patch: Display test failures.
* debian/control: Add depends on genisoimage
* debian/control: Suggest guestmount.
* debian/control: Suggest websockify. (LP: #1076442)
* debian/nova.conf: Disable nova-volume service.
* debian/control: Depend on xen-system-* rather than the hypervisor.
* debian/control, debian/mans/nova-conductor.8, debian/nova-conductor.init,
  debian/nova-conductor.install, debian/nova-conductor.logrotate
  debian/nova-conductor.manpages, debian/nova-conductor.postrm
  debian/nova-conductor.upstart.in: Add nova-conductor service.
* debian/control: Add python-fixtures as a build deps.

Show diffs side-by-side

added added

removed removed

Lines of Context:
nova/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
 
from lxml import etree
17
 
import webob
18
 
 
19
 
from nova.api.openstack.volume import snapshots
20
 
from nova import db
21
 
from nova import exception
22
 
from nova import flags
23
 
from nova.openstack.common import log as logging
24
 
from nova.openstack.common import timeutils
25
 
from nova import test
26
 
from nova.tests.api.openstack import fakes
27
 
from nova import volume
28
 
 
29
 
FLAGS = flags.FLAGS
30
 
 
31
 
LOG = logging.getLogger(__name__)
32
 
 
33
 
 
34
 
def _get_default_snapshot_param():
35
 
    return {
36
 
        'id': 123,
37
 
        'volume_id': 12,
38
 
        'status': 'available',
39
 
        'volume_size': 100,
40
 
        'created_at': None,
41
 
        'display_name': 'Default name',
42
 
        'display_description': 'Default description',
43
 
        }
44
 
 
45
 
 
46
 
def stub_snapshot_create(self, context, volume_id, name, description):
47
 
    snapshot = _get_default_snapshot_param()
48
 
    snapshot['volume_id'] = volume_id
49
 
    snapshot['display_name'] = name
50
 
    snapshot['display_description'] = description
51
 
    return snapshot
52
 
 
53
 
 
54
 
def stub_snapshot_delete(self, context, snapshot):
55
 
    if snapshot['id'] != 123:
56
 
        raise exception.NotFound
57
 
 
58
 
 
59
 
def stub_snapshot_get(self, context, snapshot_id):
60
 
    if snapshot_id != 123:
61
 
        raise exception.NotFound
62
 
 
63
 
    param = _get_default_snapshot_param()
64
 
    return param
65
 
 
66
 
 
67
 
def stub_snapshot_get_all(self, context, search_opts=None):
68
 
    param = _get_default_snapshot_param()
69
 
    return [param]
70
 
 
71
 
 
72
 
class SnapshotApiTest(test.TestCase):
73
 
    def setUp(self):
74
 
        super(SnapshotApiTest, self).setUp()
75
 
        self.controller = snapshots.SnapshotsController()
76
 
 
77
 
        self.stubs.Set(db, 'snapshot_get_all_by_project',
78
 
                       fakes.stub_snapshot_get_all_by_project)
79
 
        self.stubs.Set(db, 'snapshot_get_all',
80
 
                      fakes.stub_snapshot_get_all)
81
 
 
82
 
    def test_snapshot_create(self):
83
 
        self.stubs.Set(volume.api.API, "create_snapshot", stub_snapshot_create)
84
 
        self.stubs.Set(volume.api.API, 'get', fakes.stub_volume_get)
85
 
        snapshot = {"volume_id": '12',
86
 
                "force": False,
87
 
                "display_name": "Snapshot Test Name",
88
 
                "display_description": "Snapshot Test Desc"}
89
 
        body = dict(snapshot=snapshot)
90
 
        req = fakes.HTTPRequest.blank('/v1/snapshots')
91
 
        resp_dict = self.controller.create(req, body)
92
 
 
93
 
        self.assertTrue('snapshot' in resp_dict)
94
 
        self.assertEqual(resp_dict['snapshot']['display_name'],
95
 
                        snapshot['display_name'])
96
 
        self.assertEqual(resp_dict['snapshot']['display_description'],
97
 
                        snapshot['display_description'])
98
 
 
99
 
    def test_snapshot_create_force(self):
100
 
        self.stubs.Set(volume.api.API, "create_snapshot_force",
101
 
            stub_snapshot_create)
102
 
        self.stubs.Set(volume.api.API, 'get', fakes.stub_volume_get)
103
 
        snapshot = {"volume_id": '12',
104
 
                "force": True,
105
 
                "display_name": "Snapshot Test Name",
106
 
                "display_description": "Snapshot Test Desc"}
107
 
        body = dict(snapshot=snapshot)
108
 
        req = fakes.HTTPRequest.blank('/v1/snapshots')
109
 
        resp_dict = self.controller.create(req, body)
110
 
 
111
 
        self.assertTrue('snapshot' in resp_dict)
112
 
        self.assertEqual(resp_dict['snapshot']['display_name'],
113
 
                        snapshot['display_name'])
114
 
        self.assertEqual(resp_dict['snapshot']['display_description'],
115
 
                        snapshot['display_description'])
116
 
 
117
 
        # Test invalid force paramter
118
 
        snapshot = {"volume_id": 12,
119
 
                "force": '**&&^^%%$$##@@'}
120
 
        body = dict(snapshot=snapshot)
121
 
        req = fakes.HTTPRequest.blank('/v1/snapshots')
122
 
        self.assertRaises(exception.InvalidParameterValue,
123
 
                          self.controller.create,
124
 
                          req,
125
 
                          body)
126
 
 
127
 
    def test_snapshot_delete(self):
128
 
        self.stubs.Set(volume.api.API, "get_snapshot", stub_snapshot_get)
129
 
        self.stubs.Set(volume.api.API, "delete_snapshot", stub_snapshot_delete)
130
 
 
131
 
        snapshot_id = 123
132
 
        req = fakes.HTTPRequest.blank('/v1/snapshots/%d' % snapshot_id)
133
 
        resp = self.controller.delete(req, snapshot_id)
134
 
        self.assertEqual(resp.status_int, 202)
135
 
 
136
 
    def test_snapshot_delete_invalid_id(self):
137
 
        self.stubs.Set(volume.api.API, "delete_snapshot", stub_snapshot_delete)
138
 
        snapshot_id = 234
139
 
        req = fakes.HTTPRequest.blank('/v1/snapshots/%d' % snapshot_id)
140
 
        self.assertRaises(webob.exc.HTTPNotFound,
141
 
                          self.controller.delete,
142
 
                          req,
143
 
                          snapshot_id)
144
 
 
145
 
    def test_snapshot_show(self):
146
 
        self.stubs.Set(volume.api.API, "get_snapshot", stub_snapshot_get)
147
 
        req = fakes.HTTPRequest.blank('/v1/snapshots/123')
148
 
        resp_dict = self.controller.show(req, 123)
149
 
 
150
 
        self.assertTrue('snapshot' in resp_dict)
151
 
        self.assertEqual(resp_dict['snapshot']['id'], '123')
152
 
 
153
 
    def test_snapshot_show_invalid_id(self):
154
 
        snapshot_id = 234
155
 
        req = fakes.HTTPRequest.blank('/v1/snapshots/%d' % snapshot_id)
156
 
        self.assertRaises(webob.exc.HTTPNotFound,
157
 
                          self.controller.show,
158
 
                          req,
159
 
                          snapshot_id)
160
 
 
161
 
    def test_snapshot_detail(self):
162
 
        self.stubs.Set(volume.api.API, "get_all_snapshots",
163
 
                      stub_snapshot_get_all)
164
 
        req = fakes.HTTPRequest.blank('/v1/snapshots/detail')
165
 
        resp_dict = self.controller.detail(req)
166
 
 
167
 
        self.assertTrue('snapshots' in resp_dict)
168
 
        resp_snapshots = resp_dict['snapshots']
169
 
        self.assertEqual(len(resp_snapshots), 1)
170
 
 
171
 
        resp_snapshot = resp_snapshots.pop()
172
 
        self.assertEqual(resp_snapshot['id'], '123')
173
 
 
174
 
    def test_admin_list_snapshots_limited_to_project(self):
175
 
        req = fakes.HTTPRequest.blank('/v1/fake/snapshots',
176
 
                                      use_admin_context=True)
177
 
        res = self.controller.index(req)
178
 
 
179
 
        self.assertTrue('snapshots' in res)
180
 
        self.assertEqual(1, len(res['snapshots']))
181
 
 
182
 
    def test_admin_list_snapshots_all_tenants(self):
183
 
        req = fakes.HTTPRequest.blank('/v1/fake/snapshots?all_tenants=1',
184
 
                                      use_admin_context=True)
185
 
        res = self.controller.index(req)
186
 
        self.assertTrue('snapshots' in res)
187
 
        self.assertEqual(3, len(res['snapshots']))
188
 
 
189
 
    def test_all_tenants_non_admin_gets_all_tenants(self):
190
 
        req = fakes.HTTPRequest.blank('/v1/fake/snapshots?all_tenants=1')
191
 
        res = self.controller.index(req)
192
 
        self.assertTrue('snapshots' in res)
193
 
        self.assertEqual(1, len(res['snapshots']))
194
 
 
195
 
    def test_non_admin_get_by_project(self):
196
 
        req = fakes.HTTPRequest.blank('/v1/fake/snapshots')
197
 
        res = self.controller.index(req)
198
 
        self.assertTrue('snapshots' in res)
199
 
        self.assertEqual(1, len(res['snapshots']))
200
 
 
201
 
 
202
 
class SnapshotSerializerTest(test.TestCase):
203
 
    def _verify_snapshot(self, snap, tree):
204
 
        self.assertEqual(tree.tag, 'snapshot')
205
 
 
206
 
        for attr in ('id', 'status', 'size', 'created_at',
207
 
                     'display_name', 'display_description', 'volume_id'):
208
 
            self.assertEqual(str(snap[attr]), tree.get(attr))
209
 
 
210
 
    def test_snapshot_show_create_serializer(self):
211
 
        serializer = snapshots.SnapshotTemplate()
212
 
        raw_snapshot = dict(
213
 
            id='snap_id',
214
 
            status='snap_status',
215
 
            size=1024,
216
 
            created_at=timeutils.utcnow(),
217
 
            display_name='snap_name',
218
 
            display_description='snap_desc',
219
 
            volume_id='vol_id',
220
 
            )
221
 
        text = serializer.serialize(dict(snapshot=raw_snapshot))
222
 
 
223
 
        print text
224
 
        tree = etree.fromstring(text)
225
 
 
226
 
        self._verify_snapshot(raw_snapshot, tree)
227
 
 
228
 
    def test_snapshot_index_detail_serializer(self):
229
 
        serializer = snapshots.SnapshotsTemplate()
230
 
        raw_snapshots = [dict(
231
 
                id='snap1_id',
232
 
                status='snap1_status',
233
 
                size=1024,
234
 
                created_at=timeutils.utcnow(),
235
 
                display_name='snap1_name',
236
 
                display_description='snap1_desc',
237
 
                volume_id='vol1_id',
238
 
                ),
239
 
                       dict(
240
 
                id='snap2_id',
241
 
                status='snap2_status',
242
 
                size=1024,
243
 
                created_at=timeutils.utcnow(),
244
 
                display_name='snap2_name',
245
 
                display_description='snap2_desc',
246
 
                volume_id='vol2_id',
247
 
                )]
248
 
        text = serializer.serialize(dict(snapshots=raw_snapshots))
249
 
 
250
 
        print text
251
 
        tree = etree.fromstring(text)
252
 
 
253
 
        self.assertEqual('snapshots', tree.tag)
254
 
        self.assertEqual(len(raw_snapshots), len(tree))
255
 
        for idx, child in enumerate(tree):
256
 
            self._verify_snapshot(raw_snapshots[idx], child)
257
 
 
258
 
 
259
 
class SnapshotsUnprocessableEntityTestCase(test.TestCase):
260
 
 
261
 
    """
262
 
    Tests of places we throw 422 Unprocessable Entity from
263
 
    """
264
 
 
265
 
    def setUp(self):
266
 
        super(SnapshotsUnprocessableEntityTestCase, self).setUp()
267
 
        self.controller = snapshots.SnapshotsController()
268
 
 
269
 
    def _unprocessable_snapshot_create(self, body):
270
 
        req = fakes.HTTPRequest.blank('/v2/fake/snapshots')
271
 
        req.method = 'POST'
272
 
 
273
 
        self.assertRaises(webob.exc.HTTPUnprocessableEntity,
274
 
                          self.controller.create, req, body)
275
 
 
276
 
    def test_create_no_body(self):
277
 
        self._unprocessable_snapshot_create(body=None)
278
 
 
279
 
    def test_create_missing_snapshot(self):
280
 
        body = {'foo': {'a': 'b'}}
281
 
        self._unprocessable_snapshot_create(body=body)
282
 
 
283
 
    def test_create_malformed_entity(self):
284
 
        body = {'snapshot': 'string'}
285
 
        self._unprocessable_snapshot_create(body=body)