1
# Copyright 2012 OpenStack LLC.
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
8
# http://www.apache.org/licenses/LICENSE-2.0
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
16
from lxml import etree
19
from cinder.api.contrib import extended_snapshot_attributes
20
from cinder import exception
21
from cinder import flags
22
from cinder.openstack.common import jsonutils
23
from cinder import test
24
from cinder.tests.api.openstack import fakes
25
from cinder import volume
31
UUID1 = '00000000-0000-0000-0000-000000000001'
32
UUID2 = '00000000-0000-0000-0000-000000000002'
35
def _get_default_snapshot_param():
39
'status': 'available',
42
'display_name': 'Default name',
43
'display_description': 'Default description',
49
def fake_snapshot_get(self, context, snapshot_id):
50
param = _get_default_snapshot_param()
54
def fake_snapshot_get_all(self, context, search_opts=None):
55
param = _get_default_snapshot_param()
59
class ExtendedSnapshotAttributesTest(test.TestCase):
60
content_type = 'application/json'
61
prefix = 'os-extended-snapshot-attributes:'
64
super(ExtendedSnapshotAttributesTest, self).setUp()
65
self.stubs.Set(volume.api.API, 'get_snapshot', fake_snapshot_get)
66
self.stubs.Set(volume.api.API, 'get_all_snapshots',
67
fake_snapshot_get_all)
69
def _make_request(self, url):
70
req = webob.Request.blank(url)
71
req.headers['Accept'] = self.content_type
72
res = req.get_response(fakes.wsgi_app())
75
def _get_snapshot(self, body):
76
return jsonutils.loads(body).get('snapshot')
78
def _get_snapshots(self, body):
79
return jsonutils.loads(body).get('snapshots')
81
def assertSnapshotAttributes(self, snapshot, project_id, progress):
82
self.assertEqual(snapshot.get('%sproject_id' % self.prefix),
84
self.assertEqual(snapshot.get('%sprogress' % self.prefix), progress)
87
url = '/v1/fake/snapshots/%s' % UUID2
88
res = self._make_request(url)
90
self.assertEqual(res.status_int, 200)
91
self.assertSnapshotAttributes(self._get_snapshot(res.body),
95
def test_detail(self):
96
url = '/v1/fake/snapshots/detail'
97
res = self._make_request(url)
99
self.assertEqual(res.status_int, 200)
100
for i, snapshot in enumerate(self._get_snapshots(res.body)):
101
self.assertSnapshotAttributes(snapshot,
105
def test_no_instance_passthrough_404(self):
107
def fake_snapshot_get(*args, **kwargs):
108
raise exception.InstanceNotFound()
110
self.stubs.Set(volume.api.API, 'get_snapshot', fake_snapshot_get)
111
url = '/v1/fake/snapshots/70f6db34-de8d-4fbd-aafb-4065bdfa6115'
112
res = self._make_request(url)
114
self.assertEqual(res.status_int, 404)
117
class ExtendedSnapshotAttributesXmlTest(ExtendedSnapshotAttributesTest):
118
content_type = 'application/xml'
119
ext = extended_snapshot_attributes
120
prefix = '{%s}' % ext.Extended_snapshot_attributes.namespace
122
def _get_snapshot(self, body):
123
return etree.XML(body)
125
def _get_snapshots(self, body):
126
return etree.XML(body).getchildren()