Statistics
| Branch: | Tag: | Revision:

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

History | View | Annotate | Download (14.5 kB)

1
# Copyright 2012-2014 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 decimal import Decimal
40
from snf_django.utils.testing import BaseAPITest
41
from synnefo.cyclades_settings import cyclades_services
42
from synnefo.lib.services import get_service_path
43
from synnefo.lib import join_urls
44

    
45
PLANKTON_URL = get_service_path(cyclades_services, 'image',
46
                                version='v1.0')
47
IMAGES_URL = join_urls(PLANKTON_URL, "images/")
48

    
49

    
50
def assert_backend_closed(func):
51
    @wraps(func)
52
    def wrapper(self, backend):
53
        result = func(self, backend)
54
        if backend.called is True:
55
            backend.return_value.close.assert_called_once_with()
56
        return result
57
    return wrapper
58

    
59

    
60
@patch("synnefo.plankton.backend.get_pithos_backend")
61
class PlanktonTest(BaseAPITest):
62
    def test_register_image(self, backend):
63
        required = {
64
            "HTTP_X_IMAGE_META_NAME": u"TestImage\u2602",
65
            "HTTP_X_IMAGE_META_LOCATION": "pithos://4321-4321/images/foo"}
66
        # Check valid name
67
        headers = deepcopy(required)
68
        headers.pop("HTTP_X_IMAGE_META_NAME")
69
        response = self.post(IMAGES_URL, **headers)
70
        self.assertBadRequest(response)
71
        self.assertTrue("name" in response.content)
72
        headers["HTTP_X_IMAGE_META_NAME"] = ""
73
        response = self.post(IMAGES_URL, **headers)
74
        self.assertBadRequest(response)
75
        self.assertTrue("name" in response.content)
76
        # Check valid location
77
        headers = deepcopy(required)
78
        headers.pop("HTTP_X_IMAGE_META_LOCATION")
79
        response = self.post(IMAGES_URL, **headers)
80
        self.assertBadRequest(response)
81
        self.assertTrue("location" in response.content)
82
        headers["HTTP_X_IMAGE_META_LOCATION"] = ""
83
        response = self.post(IMAGES_URL, **headers)
84
        self.assertBadRequest(response)
85
        self.assertTrue("location" in response.content)
86
        headers["HTTP_X_IMAGE_META_LOCATION"] = "pitho://4321-4321/images/foo"
87
        response = self.post(IMAGES_URL, **headers)
88
        self.assertBadRequest(response)
89
        self.assertTrue("location" in response.content)
90
        headers["HTTP_X_IMAGE_META_LOCATION"] = "pithos://4321-4321/foo"
91
        response = self.post(IMAGES_URL, **headers)
92
        self.assertBadRequest(response)
93
        self.assertTrue("location" in response.content)
94
        # ID not supported
95
        headers = deepcopy(required)
96
        headers["HTTP_X_IMAGE_META_ID"] = "1234"
97
        response = self.post(IMAGES_URL, **headers)
98
        self.assertBadRequest(response)
99
        headers = deepcopy(required)
100
        # ID not supported
101
        headers = deepcopy(required)
102
        headers["HTTP_X_IMAGE_META_LOLO"] = "1234"
103
        response = self.post(IMAGES_URL, **headers)
104
        self.assertBadRequest(response)
105
        headers = deepcopy(required)
106
        headers["HTTP_X_IMAGE_META_STORE"] = "pitho"
107
        response = self.post(IMAGES_URL, **headers)
108
        self.assertBadRequest(response)
109
        self.assertTrue("store " in response.content)
110
        headers = deepcopy(required)
111
        headers["HTTP_X_IMAGE_META_DISK_FORMAT"] = "diskdumpp"
112
        response = self.post(IMAGES_URL, **headers)
113
        self.assertBadRequest(response)
114
        self.assertTrue("disk format" in response.content)
115
        headers = deepcopy(required)
116
        headers["HTTP_X_IMAGE_META_CONTAINER_FORMAT"] = "baree"
117
        response = self.post(IMAGES_URL, **headers)
118
        self.assertBadRequest(response)
119
        self.assertTrue("container format" in response.content)
120

    
121
        backend().get_object_meta.return_value = {"uuid": "1234-1234-1234",
122
                                                  "bytes": 42,
123
                                                  "hash": "unique_hash"}
124
        headers = deepcopy(required)
125
        headers["HTTP_X_IMAGE_META_SIZE"] = "foo"
126
        response = self.post(IMAGES_URL, **headers)
127
        self.assertBadRequest(response)
128
        self.assertTrue("size" in response.content)
129
        headers["HTTP_X_IMAGE_META_SIZE"] = "43"
130
        response = self.post(IMAGES_URL, **headers)
131
        self.assertBadRequest(response)
132
        self.assertTrue("size" in response.content)
133

    
134
        headers["HTTP_X_IMAGE_META_SIZE"] = 42
135
        headers["HTTP_X_IMAGE_META_CHECKSUM"] = "wrong_checksum"
136
        response = self.post(IMAGES_URL, **headers)
137
        self.assertBadRequest(response)
138

    
139
        backend().get_uuid.return_value =\
140
            ("4321-4321", "images", "foo")
141
        backend().get_object_permissions.return_value = \
142
            ("foo", "foo", {"read": []})
143
        backend().get_object_meta.side_effect = \
144
            [{"uuid": "1234-1234-1234",
145
              "bytes": 42,
146
              "hash": "unique_hash"},
147
             {"uuid": "1234-1234-1234",
148
              "bytes": 42,
149
              "hash": "unique_hash",
150
              'version_timestamp': Decimal('1392487853.863673'),
151
              "plankton:name": u"TestImage\u2602",
152
              "plankton:container_format": "bare",
153
              "plankton:disk_format": "diskdump",
154
              "plankton:status": u"AVAILABLE"}]
155
        headers = deepcopy(required)
156
        response = self.post(IMAGES_URL, **headers)
157
        self.assertSuccess(response)
158
        self.assertEqual(response["x-image-meta-location"],
159
                         "pithos://4321-4321/images/foo")
160
        self.assertEqual(response["x-image-meta-id"], "1234-1234-1234")
161
        self.assertEqual(response["x-image-meta-status"], "AVAILABLE")
162
        self.assertEqual(response["x-image-meta-deleted-at"], "")
163
        self.assertEqual(response["x-image-meta-is-public"], "False")
164
        self.assertEqual(response["x-image-meta-owner"], "4321-4321")
165
        self.assertEqual(response["x-image-meta-size"], "42")
166
        self.assertEqual(response["x-image-meta-checksum"], "unique_hash")
167
        self.assertEqual(response["x-image-meta-name"],
168
                         u"TestImage\u2602".encode("utf-8"))
169
        self.assertEqual(response["x-image-meta-container-format"], "bare")
170
        self.assertEqual(response["x-image-meta-disk-format"], "diskdump")
171
        self.assertEqual(response["x-image-meta-created-at"],
172
                         "2014-02-15 18:10:53")
173
        self.assertEqual(response["x-image-meta-updated-at"],
174
                         "2014-02-15 18:10:53")
175

    
176
        # Extra headers,properties
177
        backend().get_object_meta.side_effect = \
178
            [{"uuid": "1234-1234-1234",
179
              "bytes": 42,
180
              "hash": "unique_hash"},
181
             {"uuid": "1234-1234-1234",
182
              "bytes": 42,
183
              "hash": "unique_hash",
184
              'version_timestamp': Decimal('1392487853.863673'),
185
              "plankton:name": u"TestImage\u2602",
186
              "plankton:container_format": "bare",
187
              "plankton:disk_format": "diskdump",
188
              "plankton:status": u"AVAILABLE"}]
189
        headers = deepcopy(required)
190
        headers["HTTP_X_IMAGE_META_IS_PUBLIC"] = True
191
        headers["HTTP_X_IMAGE_META_PROPERTY_KEY1"] = "val1"
192
        headers["HTTP_X_IMAGE_META_PROPERTY_KEY2"] = u"\u2601"
193
        response = self.post(IMAGES_URL, **headers)
194
        name, args, kwargs = backend().update_object_meta.mock_calls[-1]
195
        metadata = args[5]
196
        self.assertEqual(metadata["plankton:property:key1"], "val1")
197
        self.assertEqual(metadata["plankton:property:key2"],
198
                         u"\u2601".encode("utf-8"))
199
        self.assertSuccess(response)
200

    
201
    def test_unregister_image(self, backend):
202
        backend().get_uuid.return_value = ("img_owner", "images", "foo")
203
        backend().get_object_meta.return_value = {"uuid": "img_uuid",
204
                                                  "bytes": 42,
205
                                                  "plankton:name": "test"}
206
        response = self.delete(join_urls(IMAGES_URL, "img_uuid"))
207
        self.assertEqual(response.status_code, 204)
208
        backend().update_object_meta.assert_called_once_with(
209
            "user", "img_owner", "images", "foo", "plankton", {}, True)
210

    
211
    def test_users(self, backend):
212
        """Test adding/removing and replacing image members"""
213
        # Add user
214
        backend.reset_mock()
215
        backend().get_uuid.return_value = ("img_owner", "images", "foo")
216
        backend().get_object_permissions.return_value = \
217
            ("foo", "foo", {"read": []})
218
        backend().get_object_meta.return_value = {"uuid": "img_uuid",
219
                                                  "bytes": 42,
220
                                                  "plankton:name": "test"}
221
        response = self.put(join_urls(IMAGES_URL, "img_uuid/members/user1"),
222
                            user="user1")
223
        self.assertSuccess(response)
224
        backend().update_object_permissions.assert_called_once_with(
225
            "user1", "img_owner", "images", "foo", {"read": ["user1"]})
226

    
227
        # Remove user
228
        backend().update_object_permissions.reset_mock()
229
        backend().get_object_permissions.return_value = \
230
            ("foo", "foo", {"read": ["user1"]})
231
        response = self.delete(join_urls(IMAGES_URL, "img_uuid/members/user1"),
232
                               user="user1")
233
        self.assertSuccess(response)
234
        backend().update_object_permissions.assert_called_once_with(
235
            "user1", "img_owner", "images", "foo", {"read": []})
236

    
237
        # Update users
238
        backend().get_object_permissions.return_value = \
239
            ("foo", "foo", {"read": ["user1", "user2", "user3"]})
240
        backend().update_object_permissions.reset_mock()
241
        response = self.put(join_urls(IMAGES_URL, "img_uuid/members"),
242
                            params=json.dumps({"memberships":
243
                                               [{"member_id": "foo1"},
244
                                                {"member_id": "foo2"}]}),
245
                            ctype="json",
246
                            user="user1")
247
        self.assertSuccess(response)
248
        backend().update_object_permissions.assert_called_once_with(
249
            "user1", "img_owner", "images", "foo", {"read": ["foo1", "foo2"]})
250

    
251
        # List users
252
        backend().get_object_permissions.return_value = \
253
            ("foo", "foo", {"read": ["user1", "user2", "user3"]})
254
        response = self.get(join_urls(IMAGES_URL, "img_uuid/members"))
255
        self.assertSuccess(response)
256
        res_members = [{"member_id": m, "can_share": False}
257
                       for m in ["user1", "user2", "user3"]]
258
        self.assertEqual(json.loads(response.content)["members"], res_members)
259

    
260
    def test_metadata(self, backend):
261
        backend().get_uuid.return_value = ("img_owner", "images", "foo")
262
        backend().get_object_meta.return_value = \
263
            {"uuid": "img_uuid",
264
             "bytes": 42,
265
             "hash": "unique_hash",
266
             'version_timestamp': Decimal('1392487853.863673'),
267
             "plankton:name": u"TestImage\u2602",
268
             "plankton:container_format": "bare",
269
             "plankton:disk_format": "diskdump",
270
             "plankton:status": u"AVAILABLE"}
271
        backend().get_object_permissions.return_value = \
272
            ("foo", "foo", {"read": ["*", "user1"]})
273
        response = self.head(join_urls(IMAGES_URL, "img_uuid2"))
274
        self.assertSuccess(response)
275
        self.assertEqual(response["x-image-meta-location"],
276
                         "pithos://img_owner/images/foo")
277
        self.assertEqual(response["x-image-meta-id"], "img_uuid")
278
        self.assertEqual(response["x-image-meta-status"], "AVAILABLE")
279
        self.assertEqual(response["x-image-meta-deleted-at"], "")
280
        self.assertEqual(response["x-image-meta-is-public"], "True")
281
        self.assertEqual(response["x-image-meta-owner"], "img_owner")
282
        self.assertEqual(response["x-image-meta-size"], "42")
283
        self.assertEqual(response["x-image-meta-checksum"], "unique_hash")
284
        self.assertEqual(response["x-image-meta-name"],
285
                         u"TestImage\u2602".encode("utf-8"))
286
        self.assertEqual(response["x-image-meta-container-format"], "bare")
287
        self.assertEqual(response["x-image-meta-disk-format"], "diskdump")
288
        self.assertEqual(response["x-image-meta-created-at"],
289
                         "2014-02-15 18:10:53")
290
        self.assertEqual(response["x-image-meta-updated-at"],
291
                         "2014-02-15 18:10:53")
292
        response = self.head(join_urls(IMAGES_URL, "img_uuid2"))
293

    
294
        headers = {"HTTP_X_IMAGE_META_IS_PUBLIC": False,
295
                   "HTTP_X_IMAGE_META_PROPERTY_KEY1": "Val1"}
296
        response = self.put(join_urls(IMAGES_URL, "img_uuid"), **headers)
297
        self.assertSuccess(response)
298
        backend().update_object_permissions.assert_called_once_with(
299
            "user", "img_owner", "images", "foo", {"read": ["user1"]})
300

    
301
    def test_catch_wrong_api_paths(self, *args):
302
        response = self.get(join_urls(PLANKTON_URL, 'nonexistent'))
303
        self.assertEqual(response.status_code, 400)
304
        try:
305
            json.loads(response.content)
306
        except ValueError:
307
            self.assertTrue(False)
308

    
309
    def test_list_images_filters_error_1(self, backend):
310
        response = self.get(join_urls(IMAGES_URL, "?size_max="))
311
        self.assertBadRequest(response)