Statistics
| Branch: | Tag: | Revision:

root / snf-cyclades-app / synnefo / plankton / tests.py @ 37f87d98

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 django.test import TestCase
37

    
38
from contextlib import contextmanager
39
from mock import patch
40
from functools import wraps
41
from copy import deepcopy
42

    
43

    
44
FILTERS = ('name', 'container_format', 'disk_format', 'status', 'size_min',
45
           'size_max')
46
PARAMS = ('sort_key', 'sort_dir')
47
SORT_KEY_OPTIONS = ('id', 'name', 'status', 'size', 'disk_format',
48
                    'container_format', 'created_at', 'updated_at')
49
SORT_DIR_OPTIONS = ('asc', 'desc')
50
LIST_FIELDS = ('status', 'name', 'disk_format', 'container_format', 'size',
51
               'id')
52
DETAIL_FIELDS = ('name', 'disk_format', 'container_format', 'size', 'checksum',
53
                 'location', 'created_at', 'updated_at', 'deleted_at',
54
                 'status', 'is_public', 'owner', 'properties', 'id')
55
ADD_FIELDS = ('name', 'id', 'store', 'disk_format', 'container_format', 'size',
56
              'checksum', 'is_public', 'owner', 'properties', 'location')
57
UPDATE_FIELDS = ('name', 'disk_format', 'container_format', 'is_public',
58
                 'owner', 'properties', 'status')
59

    
60

    
61
@contextmanager
62
def astakos_user(user):
63
    """
64
    Context manager to mock astakos response.
65

66
    usage:
67
    with astakos_user("user@user.com"):
68
        .... make api calls ....
69

70
    """
71
    from synnefo.plankton import util
72
    orig_method = util.get_user
73

    
74
    def dummy_get_user(request, *args, **kwargs):
75
        request.user = {'username': user, 'groups': []}
76
        request.user_uniq = user
77
    util.get_user = dummy_get_user
78
    yield
79
    util.get_user = orig_method
80

    
81

    
82
class BaseTestCase(TestCase):
83
    def get(self, url, user='user', *args, **kwargs):
84
        with astakos_user(user):
85
            response = self.client.get(url, *args, **kwargs)
86
        return response
87

    
88
    def delete(self, url, user='user'):
89
        with astakos_user(user):
90
            response = self.client.delete(url)
91
        return response
92

    
93
    def post(self, url, user='user', params={}, ctype='json', *args, **kwargs):
94
        if ctype == 'json':
95
            content_type = 'application/json'
96
        with astakos_user(user):
97
            response = self.client.post(url, params, content_type=content_type,
98
                                        *args, **kwargs)
99
        return response
100

    
101
    def put(self, url, user='user', params={}, ctype='json', *args, **kwargs):
102
        if ctype == 'json':
103
            content_type = 'application/json'
104
        with astakos_user(user):
105
            response = self.client.put(url, params, content_type=content_type,
106
                    *args, **kwargs)
107
        return response
108

    
109
    def assertSuccess(self, response):
110
        self.assertTrue(response.status_code in [200, 203, 204])
111

    
112
    def assertFault(self, response, status_code, name):
113
        self.assertEqual(response.status_code, status_code)
114
        fault = json.loads(response.content)
115
        self.assertEqual(fault.keys(), [name])
116

    
117
    def assertBadRequest(self, response):
118
        self.assertFault(response, 400, 'badRequest')
119

    
120
    def assertItemNotFound(self, response):
121
        self.assertFault(response, 404, 'itemNotFound')
122

    
123

    
124
DummyImages = {
125
 '0786a349-9725-48ec-8b86-8598eefc4043':
126
 {'checksum': u'e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855',
127
  u'container_format': u'bare',
128
  'created_at': '2012-12-04 09:50:20',
129
  'deleted_at': '',
130
  u'disk_format': u'diskdump',
131
  'id': u'0786a349-9725-48ec-8b86-8598eefc4043',
132
  'is_public': True,
133
  'location': u'pithos://foo@example.com/container/foo3',
134
  u'name': u'dummyname',
135
  'owner': u'foo@example.com',
136
  'properties': {},
137
  'size': 500L,
138
  u'status': u'available',
139
  'store': 'pithos',
140
  'updated_at': '2012-12-04 09:50:54'},
141

    
142
 'd8aa85b8-410b-4550-953d-6797572534e6':
143
 {'checksum': u'e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855',
144
  u'container_format': u'bare',
145
  'created_at': '2012-11-26 11:56:42',
146
  'deleted_at': '',
147
  u'disk_format': u'diskdump',
148
  'id': u'd8aa85b8-410b-4550-953d-6797572534e6',
149
  'is_public': False,
150
  'location': u'pithos://foo@example.com/container/private',
151
  u'name': u'dummyname2',
152
  'owner': u'foo@example.com',
153
  'properties': {},
154
  'size': 10000L,
155
  u'status': u'available',
156
  'store': 'pithos',
157
  'updated_at': '2012-11-26 11:57:09'},
158

    
159
 '264fb9ac-2458-421c-b460-6a765a92825c':
160
 {'checksum': u'0c6d0586744781218672fff2d7ed94cc32efb02a6a8eb589a0628f0e22bd5a7f',
161
  u'container_format': u'bare',
162
  'created_at': '2012-11-26 11:52:54',
163
  'deleted_at': '',
164
  u'disk_format': u'diskdump',
165
  'id': u'264fb9ac-2458-421c-b460-6a765a92825c',
166
  'is_public': True,
167
  'location': u'pithos://foo@example.com/container/baz.diskdump',
168
  u'name': u'"dummyname3"',
169
  'owner': u'foo@example.com',
170
  'properties': {u'description': u'Debian Squeeze Base System',
171
                 u'gui': u'No GUI',
172
                 u'kernel': u'2.6.32',
173
                 u'os': u'debian',
174
                 u'osfamily': u'linux',
175
                 u'root_partition': u'1',
176
                 u'size': u'451',
177
                 u'sortorder': u'1',
178
                 u'users': u'root'},
179
  'size': 473772032L,
180
  u'status': u'available',
181
  'store': 'pithos',
182
  'updated_at': '2012-11-26 11:55:40'}}
183

    
184

    
185
def assert_backend_closed(func):
186
    @wraps(func)
187
    def wrapper(self, backend):
188
        result = func(self, backend)
189
        if backend.called is True:
190
            backend.return_value.close.assert_called_once_with()
191
        return result
192
    return wrapper
193

    
194

    
195
@patch("synnefo.plankton.util.ImageBackend")
196
class PlanktonTest(BaseTestCase):
197
    @assert_backend_closed
198
    def test_list_images(self, backend):
199
        backend.return_value.list.return_value =\
200
                deepcopy(DummyImages).values()
201
        response = self.get("/plankton/images/")
202
        self.assertSuccess(response)
203
        images = json.loads(response.content)
204
        for api_image in images:
205
            id = api_image['id']
206
            pithos_image = dict([(key, val)\
207
                                for key, val in DummyImages[id].items()\
208
                                if key in LIST_FIELDS])
209
            self.assertEqual(api_image, pithos_image)
210
        backend.return_value\
211
                .list.assert_called_once_with({}, {'sort_key': 'created_at',
212
                                                   'sort_dir': 'desc'})
213

    
214
    @assert_backend_closed
215
    def test_list_images_detail(self, backend):
216
        backend.return_value.list.return_value =\
217
                deepcopy(DummyImages).values()
218
        response = self.get("/plankton/images/detail")
219
        self.assertSuccess(response)
220
        images = json.loads(response.content)
221
        for api_image in images:
222
            id = api_image['id']
223
            pithos_image = dict([(key, val)\
224
                                for key, val in DummyImages[id].items()\
225
                                if key in DETAIL_FIELDS])
226
            self.assertEqual(api_image, pithos_image)
227
        backend.return_value\
228
                .list.assert_called_once_with({}, {'sort_key': 'created_at',
229
                                                   'sort_dir': 'desc'})
230

    
231
    @assert_backend_closed
232
    def test_list_images_filters(self, backend):
233
        backend.return_value.list.return_value =\
234
                deepcopy(DummyImages).values()
235
        response = self.get("/plankton/images/?size_max=1000")
236
        self.assertSuccess(response)
237
        backend.return_value\
238
                .list.assert_called_once_with({'size_max': '1000'},
239
                                              {'sort_key': 'created_at',
240
                                               'sort_dir': 'desc'})
241

    
242
    @assert_backend_closed
243
    def test_update_image(self, backend):
244
        db_image = DummyImages.values()[0]
245
        response = self.put("/plankton/images/%s" % db_image['id'],
246
                            json.dumps({}),
247
                            'json', HTTP_X_IMAGE_META_OWNER='user2')
248
        self.assertSuccess(response)
249
        backend.return_value.update.assert_called_once_with(db_image['id'],
250
                                                            {"owner": "user2"})
251

    
252
    @assert_backend_closed
253
    def test_add_image_member(self, backend):
254
        image_id = DummyImages.values()[0]['id']
255
        response = self.put("/plankton/images/%s/members/user3" % image_id,
256
                            json.dumps({}), 'json')
257
        self.assertSuccess(response)
258
        backend.return_value.add_user.assert_called_once_with(image_id,
259
                                                             'user3')
260

    
261
    @assert_backend_closed
262
    def test_remove_image_member(self, backend):
263
        image_id = DummyImages.values()[0]['id']
264
        response = self.delete("/plankton/images/%s/members/user3" % image_id)
265
        self.assertSuccess(response)
266
        backend.return_value.remove_user.assert_called_once_with(image_id,
267
                                                                'user3')
268

    
269
    @assert_backend_closed
270
    def test_add_image(self, backend):
271
        response = self.post("/plankton/images/",
272
                             json.dumps({}),
273
                             'json',
274
                             HTTP_X_IMAGE_META_NAME='dummy_name',
275
                             HTTP_X_IMAGE_META_OWNER='dummy_owner',
276
                             HTTP_X_IMAGE_META_LOCATION='dummy_location')
277
        self.assertSuccess(response)
278
        backend.return_value.register.assert_called_once_with('dummy_name',
279
                                                              'dummy_location',
280
                                                              {'owner': 'dummy_owner'})
281

    
282
    @assert_backend_closed
283
    def test_get_image(self, backend):
284
        response = self.get("/plankton/images/123")
285
        self.assertEqual(response.status_code, 501)