Statistics
| Branch: | Tag: | Revision:

root / snf-cyclades-app / synnefo / plankton / tests.py @ 52194c77

History | View | Annotate | Download (10.9 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
    def dummy_get_user(request, *args, **kwargs):
72
        request.user = {'username': user, 'groups': []}
73
        request.user_uniq = user
74

    
75
    with patch('synnefo.plankton.util.get_user') as m:
76
        m.side_effect = dummy_get_user
77
        yield
78

    
79

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

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

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

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

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

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

    
115
    def assertBadRequest(self, response):
116
        self.assertFault(response, 400, 'badRequest')
117

    
118
    def assertItemNotFound(self, response):
119
        self.assertFault(response, 404, 'itemNotFound')
120

    
121

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

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

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

    
182

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

    
192

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

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

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

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

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

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

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

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