Statistics
| Branch: | Tag: | Revision:

root / snf-cyclades-app / synnefo / plankton / tests.py @ 4e571185

History | View | Annotate | Download (11.4 kB)

1
# Copyright 2012 GRNET S.A. All rights reserved.
2
#
3
# Redistribution and use in source and binary forms, with or
4
# without modification, are permitted provided that the following
5
# conditions are met:
6
#
7
#   1. Redistributions of source code must retain the above
8
#      copyright notice, this list of conditions and the following
9
#      disclaimer.
10
#
11
#   2. Redistributions in binary form must reproduce the above
12
#      copyright notice, this list of conditions and the following
13
#      disclaimer in the documentation and/or other materials
14
#      provided with the distribution.
15
#
16
# THIS SOFTWARE IS PROVIDED BY GRNET S.A. ``AS IS'' AND ANY EXPRESS
17
# OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED
18
# WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR
19
# PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL GRNET S.A OR
20
# CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
21
# SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
22
# LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF
23
# USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED
24
# AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT
25
# LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN
26
# ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
27
# POSSIBILITY OF SUCH DAMAGE.
28
#
29
# The views and conclusions contained in the software and
30
# documentation are those of the authors and should not be
31
# interpreted as representing official policies, either expressed
32
# or implied, of GRNET S.A.
33

    
34
import json
35

    
36
from mock import patch
37
from functools import wraps
38
from copy import deepcopy
39
from snf_django.utils.testing import BaseAPITest
40
from synnefo.cyclades_settings import cyclades_services
41
from synnefo.lib.services import get_service_path
42
from synnefo.lib import join_urls
43

    
44

    
45
class PlanktonAPITest(BaseAPITest):
46
    def setUp(self, *args, **kwargs):
47
        super(PlanktonAPITest, self).setUp(*args, **kwargs)
48
        self.api_path = get_service_path(cyclades_services, 'image',
49
                                             version='v1.0')
50
    def myget(self, path, *args, **kwargs):
51
        path = join_urls(self.api_path, path)
52
        return self.get(path, *args, **kwargs)
53

    
54
    def myput(self, path, *args, **kwargs):
55
        path = join_urls(self.api_path, path)
56
        return self.put(path, *args, **kwargs)
57

    
58
    def mypost(self, path, *args, **kwargs):
59
        path = join_urls(self.api_path, path)
60
        return self.post(path, *args, **kwargs)
61

    
62
    def mydelete(self, path, *args, **kwargs):
63
        path = join_urls(self.api_path, path)
64
        return self.delete(path, *args, **kwargs)
65

    
66

    
67
FILTERS = ('name', 'container_format', 'disk_format', 'status', 'size_min',
68
           'size_max')
69
PARAMS = ('sort_key', 'sort_dir')
70
SORT_KEY_OPTIONS = ('id', 'name', 'status', 'size', 'disk_format',
71
                    'container_format', 'created_at', 'updated_at')
72
SORT_DIR_OPTIONS = ('asc', 'desc')
73
LIST_FIELDS = ('status', 'name', 'disk_format', 'container_format', 'size',
74
               'id')
75
DETAIL_FIELDS = ('name', 'disk_format', 'container_format', 'size', 'checksum',
76
                 'location', 'created_at', 'updated_at', 'deleted_at',
77
                 'status', 'is_public', 'owner', 'properties', 'id')
78
ADD_FIELDS = ('name', 'id', 'store', 'disk_format', 'container_format', 'size',
79
              'checksum', 'is_public', 'owner', 'properties', 'location')
80
UPDATE_FIELDS = ('name', 'disk_format', 'container_format', 'is_public',
81
                 'owner', 'properties', 'status')
82

    
83

    
84
DummyImages = {
85
 '0786a349-9725-48ec-8b86-8598eefc4043':
86
 {'checksum': u'e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855',
87
  u'container_format': u'bare',
88
  'created_at': '2012-12-04 09:50:20',
89
  'deleted_at': '',
90
  u'disk_format': u'diskdump',
91
  'id': u'0786a349-9725-48ec-8b86-8598eefc4043',
92
  'is_public': True,
93
  'location': u'pithos://foo@example.com/container/foo3',
94
  u'name': u'dummyname',
95
  'owner': u'foo@example.com',
96
  'properties': {},
97
  'size': 500L,
98
  u'status': u'available',
99
  'store': 'pithos',
100
  'updated_at': '2012-12-04 09:50:54'},
101

    
102
 'd8aa85b8-410b-4550-953d-6797572534e6':
103
 {'checksum': u'e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855',
104
  u'container_format': u'bare',
105
  'created_at': '2012-11-26 11:56:42',
106
  'deleted_at': '',
107
  u'disk_format': u'diskdump',
108
  'id': u'd8aa85b8-410b-4550-953d-6797572534e6',
109
  'is_public': False,
110
  'location': u'pithos://foo@example.com/container/private',
111
  u'name': u'dummyname2',
112
  'owner': u'foo@example.com',
113
  'properties': {},
114
  'size': 10000L,
115
  u'status': u'available',
116
  'store': 'pithos',
117
  'updated_at': '2012-11-26 11:57:09'},
118

    
119
 '264fb9ac-2458-421c-b460-6a765a92825c':
120
 {'checksum': u'0c6d0586744781218672fff2d7ed94cc32efb02a6a8eb589a0628f0e22bd5a7f',
121
  u'container_format': u'bare',
122
  'created_at': '2012-11-26 11:52:54',
123
  'deleted_at': '',
124
  u'disk_format': u'diskdump',
125
  'id': u'264fb9ac-2458-421c-b460-6a765a92825c',
126
  'is_public': True,
127
  'location': u'pithos://foo@example.com/container/baz.diskdump',
128
  u'name': u'"dummyname3"',
129
  'owner': u'foo@example.com',
130
  'properties': {u'description': u'Debian Squeeze Base System',
131
                 u'gui': u'No GUI',
132
                 u'kernel': u'2.6.32',
133
                 u'os': u'debian',
134
                 u'osfamily': u'linux',
135
                 u'root_partition': u'1',
136
                 u'size': u'451',
137
                 u'sortorder': u'1',
138
                 u'users': u'root'},
139
  'size': 473772032L,
140
  u'status': u'available',
141
  'store': 'pithos',
142
  'updated_at': '2012-11-26 11:55:40'}}
143

    
144

    
145
def assert_backend_closed(func):
146
    @wraps(func)
147
    def wrapper(self, backend):
148
        result = func(self, backend)
149
        if backend.called is True:
150
            backend.return_value.close.assert_called_once_with()
151
        return result
152
    return wrapper
153

    
154

    
155
@patch("synnefo.plankton.backend.ImageBackend")
156
class PlanktonTest(PlanktonAPITest):
157
    @assert_backend_closed
158
    def test_list_images(self, backend):
159
        backend.return_value.list_images.return_value =\
160
                deepcopy(DummyImages).values()
161
        response = self.myget("images/")
162
        self.assertSuccess(response)
163
        images = json.loads(response.content)
164
        for api_image in images:
165
            id = api_image['id']
166
            pithos_image = dict([(key, val)\
167
                                for key, val in DummyImages[id].items()\
168
                                if key in LIST_FIELDS])
169
            self.assertEqual(api_image, pithos_image)
170
        backend.return_value\
171
                .list_images.assert_called_once_with({}, {'sort_key': 'created_at',
172
                                                   'sort_dir': 'desc'})
173

    
174
    @assert_backend_closed
175
    def test_list_images_detail(self, backend):
176
        backend.return_value.list_images.return_value =\
177
                deepcopy(DummyImages).values()
178
        response = self.myget("images/detail")
179
        self.assertSuccess(response)
180
        images = json.loads(response.content)
181
        for api_image in images:
182
            id = api_image['id']
183
            pithos_image = dict([(key, val)\
184
                                for key, val in DummyImages[id].items()\
185
                                if key in DETAIL_FIELDS])
186
            self.assertEqual(api_image, pithos_image)
187
        backend.return_value\
188
                .list_images.assert_called_once_with({}, {'sort_key': 'created_at',
189
                                                   'sort_dir': 'desc'})
190

    
191
    @assert_backend_closed
192
    def test_list_images_filters(self, backend):
193
        backend.return_value.list_images.return_value =\
194
                deepcopy(DummyImages).values()
195
        response = self.myget("images/?size_max=1000")
196
        self.assertSuccess(response)
197
        backend.return_value\
198
                .list_images.assert_called_once_with({'size_max': 1000},
199
                                                     {'sort_key': 'created_at',
200
                                                     'sort_dir': 'desc'})
201

    
202
    @assert_backend_closed
203
    def test_list_images_filters_error_1(self, backend):
204
        response = self.myget("images/?size_max=")
205
        self.assertBadRequest(response)
206

    
207
    @assert_backend_closed
208
    def test_list_images_filters_error_2(self, backend):
209
        response = self.myget("images/?size_min=foo")
210
        self.assertBadRequest(response)
211

    
212
    @assert_backend_closed
213
    def test_update_image(self, backend):
214
        db_image = DummyImages.values()[0]
215
        response = self.myput("images/%s" % db_image['id'],
216
                              json.dumps({}),
217
                              'json', HTTP_X_IMAGE_META_OWNER='user2')
218
        self.assertSuccess(response)
219
        backend.return_value.update_metadata.assert_called_once_with(db_image['id'],
220
                                                                     {"owner": "user2"})
221

    
222
    @assert_backend_closed
223
    def test_add_image_member(self, backend):
224
        image_id = DummyImages.values()[0]['id']
225
        response = self.myput("images/%s/members/user3" % image_id,
226
                              json.dumps({}), 'json')
227
        self.assertSuccess(response)
228
        backend.return_value.add_user.assert_called_once_with(image_id,
229
                                                             'user3')
230

    
231
    @assert_backend_closed
232
    def test_remove_image_member(self, backend):
233
        image_id = DummyImages.values()[0]['id']
234
        response = self.mydelete("images/%s/members/user3" % image_id)
235
        self.assertSuccess(response)
236
        backend.return_value.remove_user.assert_called_once_with(image_id,
237
                                                                'user3')
238

    
239
    @assert_backend_closed
240
    def test_add_image(self, backend):
241
        location = "pithos://uuid/container/name/"
242
        response = self.mypost("images/",
243
                               json.dumps({}),
244
                               'json',
245
                               HTTP_X_IMAGE_META_NAME='dummy_name',
246
                               HTTP_X_IMAGE_META_OWNER='dummy_owner',
247
                               HTTP_X_IMAGE_META_LOCATION=location)
248
        self.assertSuccess(response)
249
        backend.return_value.register.assert_called_once_with('dummy_name',
250
                                                              location,
251
                                                      {'owner': 'dummy_owner'})
252

    
253
    @assert_backend_closed
254
    def test_add_image_invalid_name(self, backend):
255
        location = "pithos://uuid/container/name/"
256
        response = self.mypost("images/",
257
                               json.dumps({}),
258
                               'json',
259
                               HTTP_X_IMAGE_META_NAME="\xc2",
260
                               HTTP_X_IMAGE_META_OWNER='dummy_owner',
261
                               HTTP_X_IMAGE_META_LOCATION=location)
262
        self.assertBadRequest(response)
263

    
264
    @assert_backend_closed
265
    def test_get_image(self, backend):
266
        response = self.myget("images/123")
267
        self.assertEqual(response.status_code, 501)
268

    
269
    @assert_backend_closed
270
    def test_delete_image(self, backend):
271
        response = self.mydelete("images/123")
272
        self.assertEqual(response.status_code, 204)
273
        backend.return_value.unregister.assert_called_once_with('123')
274
        backend.return_value._delete.assert_not_called()
275

    
276
    @assert_backend_closed
277
    def test_catch_wrong_api_paths(self, *args):
278
        response = self.myget('nonexistent')
279
        self.assertEqual(response.status_code, 400)
280
        try:
281
            error = json.loads(response.content)
282
        except ValueError:
283
            self.assertTrue(False)