← Back to branch summary

~ubuntu-branches/ubuntu/raring/glance/raring-proposed

« back to all changes in this revision

Viewing changes to glance/tests/unit/v2/test_image_data_resource.py

  • Committer: Package Import Robot
  • Author(s): Chuck Short
  • Date: 2013-03-20 07:42:22 UTC
  • mto: This revision was merged to the branch mainline in revision 68.
  • Revision ID: package-import@ubuntu.com-20130320074222-ty2todj0yxtnq3lb
Tags: upstream-2013.1~rc1
Import upstream version 2013.1~rc1

Show diffs side-by-side

added added

removed removed

Lines of Context:
glance/tests/unit/v2/test_image_data_resource.py
18
18
import webob
19
19
 
20
20
import glance.api.v2.image_data
 
21
from glance.common import exception
21
22
from glance.openstack.common import uuidutils
22
23
from glance.tests.unit import base
23
24
import glance.tests.unit.utils as unit_test_utils
24
25
import glance.tests.utils as test_utils
25
26
 
26
27
 
 
28
class Raise(object):
 
29
    def __init__(self, exc):
 
30
        self.exc = exc
 
31
 
 
32
    def __call__(self, *args, **kwargs):
 
33
        raise self.exc
 
34
 
 
35
 
 
36
class FakeImage(object):
 
37
    def __init__(self, image_id=None, data=None, checksum=None, size=0,
 
38
                 locations=None, container_format='bear', disk_format='rawr',
 
39
                 status=None):
 
40
        self.image_id = image_id
 
41
        self.data = data
 
42
        self.checksum = checksum
 
43
        self.size = size
 
44
        self.locations = locations
 
45
        self.container_format = container_format
 
46
        self.disk_format = disk_format
 
47
        self._status = status
 
48
 
 
49
    @property
 
50
    def status(self):
 
51
        return self._status
 
52
 
 
53
    @status.setter
 
54
    def status(self, value):
 
55
        if isinstance(self._status, BaseException):
 
56
            raise self._status
 
57
        else:
 
58
            self._status = value
 
59
 
 
60
    def get_data(self):
 
61
        return self.data
 
62
 
 
63
    def set_data(self, data, size=None):
 
64
        self.data = ''.join(data)
 
65
        self.size = size
 
66
        self.status = 'modified-by-fake'
 
67
 
 
68
 
 
69
class FakeImageRepo(object):
 
70
    def __init__(self, result=None):
 
71
        self.result = result
 
72
 
 
73
    def get(self, image_id):
 
74
        if isinstance(self.result, BaseException):
 
75
            raise self.result
 
76
        else:
 
77
            return self.result
 
78
 
 
79
    def save(self, image):
 
80
        self.saved_image = image
 
81
 
 
82
 
 
83
class FakeGateway(object):
 
84
    def __init__(self, repo):
 
85
        self.repo = repo
 
86
 
 
87
    def get_repo(self, context):
 
88
        return self.repo
 
89
 
 
90
 
27
91
class TestImagesController(base.StoreClearingUnitTest):
28
92
    def setUp(self):
29
93
        super(TestImagesController, self).setUp()
30
94
 
31
95
        self.config(verbose=True, debug=True)
32
 
        self.notifier = unit_test_utils.FakeNotifier()
 
96
        self.image_repo = FakeImageRepo()
 
97
        self.gateway = FakeGateway(self.image_repo)
33
98
        self.controller = glance.api.v2.image_data.ImageDataController(
34
 
                db_api=unit_test_utils.FakeDB(),
35
 
                store_api=unit_test_utils.FakeStoreAPI(),
36
 
                policy_enforcer=unit_test_utils.FakePolicyEnforcer(),
37
 
                notifier=self.notifier)
 
99
                gateway=self.gateway)
38
100
 
39
101
    def test_download(self):
40
102
        request = unit_test_utils.get_fake_request()
41
 
        output = self.controller.download(request, unit_test_utils.UUID1)
42
 
        self.assertEqual(set(['data', 'meta']), set(output.keys()))
43
 
        self.assertEqual(3, output['meta']['size'])
44
 
        self.assertEqual('XXX', output['data'])
 
103
        image = FakeImage('abcd', locations=['http://example.com/image'])
 
104
        self.image_repo.result = image
 
105
        image = self.controller.download(request, unit_test_utils.UUID1)
 
106
        self.assertEqual(image.image_id, 'abcd')
45
107
 
46
 
    def test_download_no_data(self):
 
108
    def test_download_no_location(self):
47
109
        request = unit_test_utils.get_fake_request()
 
110
        self.image_repo.result = FakeImage('abcd')
48
111
        self.assertRaises(webob.exc.HTTPNotFound, self.controller.download,
49
112
                          request, unit_test_utils.UUID2)
50
113
 
51
114
    def test_download_non_existent_image(self):
52
115
        request = unit_test_utils.get_fake_request()
 
116
        self.image_repo.result = exception.NotFound()
53
117
        self.assertRaises(webob.exc.HTTPNotFound, self.controller.download,
54
118
                          request, uuidutils.generate_uuid())
55
119
 
56
 
    def test_upload_download(self):
57
 
        request = unit_test_utils.get_fake_request()
 
120
    def test_download_forbidden(self):
 
121
        request = unit_test_utils.get_fake_request()
 
122
        self.image_repo.result = exception.Forbidden()
 
123
        self.assertRaises(webob.exc.HTTPForbidden, self.controller.download,
 
124
                          request, uuidutils.generate_uuid())
 
125
 
 
126
    def test_upload(self):
 
127
        request = unit_test_utils.get_fake_request()
 
128
        image = FakeImage('abcd')
 
129
        self.image_repo.result = image
58
130
        self.controller.upload(request, unit_test_utils.UUID2, 'YYYY', 4)
59
 
        output = self.controller.download(request, unit_test_utils.UUID2)
60
 
        self.assertEqual(set(['data', 'meta']), set(output.keys()))
61
 
        self.assertEqual(4, output['meta']['size'])
62
 
        self.assertEqual('YYYY', output['data'])
63
 
        self.assertEqual(output['meta']['status'], 'active')
64
 
 
65
 
    def test_upload_download_prepare_notification(self):
 
131
        self.assertEqual(image.data, 'YYYY')
 
132
        self.assertEqual(image.size, 4)
 
133
 
 
134
    def test_upload_status(self):
 
135
        request = unit_test_utils.get_fake_request()
 
136
        image = FakeImage('abcd')
 
137
        self.image_repo.result = image
 
138
        insurance = {'called': False}
 
139
 
 
140
        def read_data():
 
141
            insurance['called'] = True
 
142
            self.assertEqual(self.image_repo.saved_image.status, 'saving')
 
143
            yield 'YYYY'
 
144
 
 
145
        self.controller.upload(request, unit_test_utils.UUID2,
 
146
                               read_data(), None)
 
147
        self.assertTrue(insurance['called'])
 
148
        self.assertEqual(self.image_repo.saved_image.status,
 
149
                         'modified-by-fake')
 
150
 
 
151
    def test_upload_no_size(self):
 
152
        request = unit_test_utils.get_fake_request()
 
153
        image = FakeImage('abcd')
 
154
        self.image_repo.result = image
 
155
        self.controller.upload(request, unit_test_utils.UUID2, 'YYYY', None)
 
156
        self.assertEqual(image.data, 'YYYY')
 
157
        self.assertEqual(image.size, None)
 
158
 
 
159
    def test_upload_invalid(self):
 
160
        request = unit_test_utils.get_fake_request()
 
161
        image = FakeImage('abcd')
 
162
        image.status = ValueError()
 
163
        self.image_repo.result = image
 
164
        self.assertRaises(webob.exc.HTTPBadRequest, self.controller.upload,
 
165
                          request, unit_test_utils.UUID1, 'YYYY', 4)
 
166
 
 
167
    def test_upload_non_existent_image(self):
 
168
        request = unit_test_utils.get_fake_request()
 
169
        self.image_repo.result = exception.NotFound()
 
170
        self.assertRaises(webob.exc.HTTPNotFound, self.controller.upload,
 
171
                          request, uuidutils.generate_uuid(), 'ABC', 3)
 
172
 
 
173
    def test_upload_data_exists(self):
 
174
        request = unit_test_utils.get_fake_request()
 
175
        image = FakeImage()
 
176
        image.set_data = Raise(exception.Duplicate)
 
177
        self.image_repo.result = image
 
178
        self.assertRaises(webob.exc.HTTPConflict, self.controller.upload,
 
179
                          request, unit_test_utils.UUID1, 'YYYY', 4)
 
180
 
 
181
    def test_upload_storage_full(self):
 
182
        request = unit_test_utils.get_fake_request()
 
183
        image = FakeImage()
 
184
        image.set_data = Raise(exception.StorageFull)
 
185
        self.image_repo.result = image
 
186
        self.assertRaises(webob.exc.HTTPRequestEntityTooLarge,
 
187
                          self.controller.upload,
 
188
                          request, unit_test_utils.UUID2, 'YYYYYYY', 7)
 
189
 
 
190
    def test_upload_storage_forbidden(self):
 
191
        request = unit_test_utils.get_fake_request(user=unit_test_utils.USER2)
 
192
        image = FakeImage()
 
193
        image.set_data = Raise(exception.Forbidden)
 
194
        self.image_repo.result = image
 
195
        self.assertRaises(webob.exc.HTTPForbidden, self.controller.upload,
 
196
                          request, unit_test_utils.UUID2, 'YY', 2)
 
197
 
 
198
    def test_upload_storage_write_denied(self):
 
199
        request = unit_test_utils.get_fake_request(user=unit_test_utils.USER3)
 
200
        image = FakeImage()
 
201
        image.set_data = Raise(exception.StorageWriteDenied)
 
202
        self.image_repo.result = image
 
203
        self.assertRaises(webob.exc.HTTPServiceUnavailable,
 
204
                          self.controller.upload,
 
205
                          request, unit_test_utils.UUID2, 'YY', 2)
 
206
 
 
207
    def _test_upload_download_prepare_notification(self):
66
208
        request = unit_test_utils.get_fake_request()
67
209
        self.controller.upload(request, unit_test_utils.UUID2, 'YYYY', 4)
68
210
        output = self.controller.download(request, unit_test_utils.UUID2)
84
226
        self.assertTrue(prepare_updated_at <= output['meta']['updated_at'])
85
227
        self.assertEqual(output_log[0], prepare_log)
86
228
 
87
 
    def test_upload_download_upload_notification(self):
 
229
    def _test_upload_download_upload_notification(self):
88
230
        request = unit_test_utils.get_fake_request()
89
231
        self.controller.upload(request, unit_test_utils.UUID2, 'YYYY', 4)
90
232
        output = self.controller.download(request, unit_test_utils.UUID2)
98
240
        self.assertEqual(len(output_log), 3)
99
241
        self.assertEqual(output_log[1], upload_log)
100
242
 
101
 
    def test_upload_download_activate_notification(self):
 
243
    def _test_upload_download_activate_notification(self):
102
244
        request = unit_test_utils.get_fake_request()
103
245
        self.controller.upload(request, unit_test_utils.UUID2, 'YYYY', 4)
104
246
        output = self.controller.download(request, unit_test_utils.UUID2)
112
254
        self.assertEqual(len(output_log), 3)
113
255
        self.assertEqual(output_log[2], activate_log)
114
256
 
115
 
    def test_upload_non_existent_image(self):
116
 
        request = unit_test_utils.get_fake_request()
117
 
        self.assertRaises(webob.exc.HTTPNotFound, self.controller.upload,
118
 
                          request, uuidutils.generate_uuid(), 'YYYY', 4)
119
 
 
120
 
    def test_upload_data_exists(self):
121
 
        request = unit_test_utils.get_fake_request()
122
 
        self.assertRaises(webob.exc.HTTPConflict, self.controller.upload,
123
 
                          request, unit_test_utils.UUID1, 'YYYY', 4)
124
 
 
125
 
    def test_upload_storage_full(self):
126
 
        request = unit_test_utils.get_fake_request()
127
 
        self.assertRaises(webob.exc.HTTPRequestEntityTooLarge,
128
 
                          self.controller.upload,
129
 
                          request, unit_test_utils.UUID2, 'YYYYYYY', 7)
130
 
 
131
 
    def test_upload_storage_forbidden(self):
132
 
        request = unit_test_utils.get_fake_request(user=unit_test_utils.USER2)
133
 
        self.assertRaises(webob.exc.HTTPForbidden, self.controller.upload,
134
 
                          request, unit_test_utils.UUID2, 'YY', 2)
135
 
 
136
 
    def test_upload_storage_write_denied(self):
137
 
        request = unit_test_utils.get_fake_request(user=unit_test_utils.USER3)
138
 
        self.assertRaises(webob.exc.HTTPServiceUnavailable,
139
 
                          self.controller.upload,
140
 
                          request, unit_test_utils.UUID2, 'YY', 2)
141
 
 
142
 
    def test_upload_download_no_size(self):
143
 
        request = unit_test_utils.get_fake_request()
144
 
        self.controller.upload(request, unit_test_utils.UUID2, 'YYYY', None)
145
 
        output = self.controller.download(request, unit_test_utils.UUID2)
146
 
        self.assertEqual(set(['data', 'meta']), set(output.keys()))
147
 
        self.assertEqual(4, output['meta']['size'])
148
 
        self.assertEqual('YYYY', output['data'])
149
 
 
150
 
 
151
 
class TestImageDataControllerPolicies(base.IsolatedUnitTest):
152
 
 
153
 
    def setUp(self):
154
 
        super(TestImageDataControllerPolicies, self).setUp()
155
 
        self.db = unit_test_utils.FakeDB()
156
 
        self.policy = unit_test_utils.FakePolicyEnforcer()
157
 
        self.controller = glance.api.v2.image_data.ImageDataController(
158
 
                                                self.db,
159
 
                                                policy_enforcer=self.policy)
160
 
 
161
 
    def test_download_unauthorized(self):
162
 
        rules = {"download_image": False}
163
 
        self.policy.set_rules(rules)
164
 
        request = unit_test_utils.get_fake_request()
165
 
        self.assertRaises(webob.exc.HTTPForbidden, self.controller.download,
166
 
                          request, image_id=unit_test_utils.UUID2)
167
 
 
168
257
 
169
258
class TestImageDataDeserializer(test_utils.BaseTestCase):
170
259
 
234
323
 
235
324
    def setUp(self):
236
325
        super(TestImageDataSerializer, self).setUp()
237
 
        self.serializer = glance.api.v2.image_data.ResponseSerializer(
238
 
            notifier=unit_test_utils.FakeNotifier())
 
326
        self.serializer = glance.api.v2.image_data.ResponseSerializer()
239
327
 
240
328
    def test_download(self):
241
329
        request = webob.Request.blank('/')
242
330
        request.environ = {}
243
331
        response = webob.Response()
244
332
        response.request = request
245
 
        fixture = {
246
 
            'data': 'ZZZ',
247
 
            'meta': {'size': 3, 'id': 'asdf', 'checksum': None}
248
 
        }
249
 
        self.serializer.download(response, fixture)
 
333
        image = FakeImage(size=3, data=iter('ZZZ'))
 
334
        self.serializer.download(response, image)
250
335
        self.assertEqual('ZZZ', response.body)
251
336
        self.assertEqual('3', response.headers['Content-Length'])
252
337
        self.assertFalse('Content-MD5' in response.headers)
259
344
        response = webob.Response()
260
345
        response.request = request
261
346
        checksum = '0745064918b49693cca64d6b6a13d28a'
262
 
        fixture = {
263
 
            'data': 'ZZZ',
264
 
            'meta': {'size': 3, 'id': 'asdf', 'checksum': checksum}
265
 
        }
266
 
        self.serializer.download(response, fixture)
 
347
        image = FakeImage(size=3, checksum=checksum, data=iter('ZZZ'))
 
348
        self.serializer.download(response, image)
267
349
        self.assertEqual('ZZZ', response.body)
268
350
        self.assertEqual('3', response.headers['Content-Length'])
269
351
        self.assertEqual(checksum, response.headers['Content-MD5'])
276
358
        response = webob.Response()
277
359
        response.request = request
278
360
        self.serializer.upload(response, {})
279
 
        self.assertEqual(201, response.status_int)
 
361
        self.assertEqual(204, response.status_int)
280
362
        self.assertEqual('0', response.headers['Content-Length'])