Statistics
| Branch: | Tag: | Revision:

root / snf-cyclades-app / synnefo / plankton / tests.py @ 028f4731

History | View | Annotate | Download (11 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
  'size': 500L,
97
  u'status': u'available',
98
  'store': 'pithos',
99
  'updated_at': '2012-12-04 09:50:54'},
100

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

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

    
142

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

    
152

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

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

    
192
        backend.return_value\
193
                .list_images.assert_called_once_with({}, {'sort_key': 'created_at',
194
                                                   'sort_dir': 'desc'})
195

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

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

    
212
    @assert_backend_closed
213
    def test_list_images_filters_error_2(self, backend):
214
        response = self.myget("images/?size_min=foo")
215
        self.assertBadRequest(response)
216

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

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

    
236
    @assert_backend_closed
237
    def test_remove_image_member(self, backend):
238
        image_id = DummyImages.values()[0]['id']
239
        response = self.mydelete("images/%s/members/user3" % image_id)
240
        self.assertSuccess(response)
241
        backend.return_value.remove_user.assert_called_once_with(image_id,
242
                                                                'user3')
243

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

    
258
    @assert_backend_closed
259
    def test_get_image(self, backend):
260
        response = self.myget("images/123")
261
        self.assertEqual(response.status_code, 501)
262

    
263
    @assert_backend_closed
264
    def test_delete_image(self, backend):
265
        response = self.mydelete("images/123")
266
        self.assertEqual(response.status_code, 204)
267
        backend.return_value.unregister.assert_called_once_with('123')
268
        backend.return_value._delete.assert_not_called()
269

    
270
    @assert_backend_closed
271
    def test_catch_wrong_api_paths(self, *args):
272
        response = self.myget('nonexistent')
273
        self.assertEqual(response.status_code, 400)
274
        try:
275
            error = json.loads(response.content)
276
        except ValueError:
277
            self.assertTrue(False)