Revision fdb15edb
| b/snf-cyclades-app/synnefo/plankton/tests.py | ||
|---|---|---|
| 37 | 37 |
from functools import wraps |
| 38 | 38 |
from copy import deepcopy |
| 39 | 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.compute_path = get_service_path(cyclades_services, 'image', |
|
| 49 |
version='v1.0') |
|
| 50 |
def myget(self, path, *args, **kwargs): |
|
| 51 |
path = join_urls(self.compute_path, path) |
|
| 52 |
return self.get(path, *args, **kwargs) |
|
| 53 |
|
|
| 54 |
def myput(self, path, *args, **kwargs): |
|
| 55 |
path = join_urls(self.compute_path, path) |
|
| 56 |
return self.put(path, *args, **kwargs) |
|
| 57 |
|
|
| 58 |
def mypost(self, path, *args, **kwargs): |
|
| 59 |
path = join_urls(self.compute_path, path) |
|
| 60 |
return self.post(path, *args, **kwargs) |
|
| 61 |
|
|
| 62 |
def mydelete(self, path, *args, **kwargs): |
|
| 63 |
path = join_urls(self.compute_path, path) |
|
| 64 |
return self.delete(path, *args, **kwargs) |
|
| 40 | 65 |
|
| 41 | 66 |
|
| 42 | 67 |
FILTERS = ('name', 'container_format', 'disk_format', 'status', 'size_min',
|
| ... | ... | |
| 128 | 153 |
|
| 129 | 154 |
|
| 130 | 155 |
@patch("synnefo.plankton.backend.ImageBackend")
|
| 131 |
class PlanktonTest(BaseAPITest):
|
|
| 156 |
class PlanktonTest(PlanktonAPITest):
|
|
| 132 | 157 |
@assert_backend_closed |
| 133 | 158 |
def test_list_images(self, backend): |
| 134 | 159 |
backend.return_value.list_images.return_value =\ |
| 135 | 160 |
deepcopy(DummyImages).values() |
| 136 |
response = self.get("/plankton/images/")
|
|
| 161 |
response = self.myget("images/")
|
|
| 137 | 162 |
self.assertSuccess(response) |
| 138 | 163 |
images = json.loads(response.content) |
| 139 | 164 |
for api_image in images: |
| ... | ... | |
| 150 | 175 |
def test_list_images_detail(self, backend): |
| 151 | 176 |
backend.return_value.list_images.return_value =\ |
| 152 | 177 |
deepcopy(DummyImages).values() |
| 153 |
response = self.get("/plankton/images/detail")
|
|
| 178 |
response = self.myget("images/detail")
|
|
| 154 | 179 |
self.assertSuccess(response) |
| 155 | 180 |
images = json.loads(response.content) |
| 156 | 181 |
for api_image in images: |
| ... | ... | |
| 167 | 192 |
def test_list_images_filters(self, backend): |
| 168 | 193 |
backend.return_value.list_images.return_value =\ |
| 169 | 194 |
deepcopy(DummyImages).values() |
| 170 |
response = self.get("/plankton/images/?size_max=1000")
|
|
| 195 |
response = self.myget("images/?size_max=1000")
|
|
| 171 | 196 |
self.assertSuccess(response) |
| 172 | 197 |
backend.return_value\ |
| 173 | 198 |
.list_images.assert_called_once_with({'size_max': 1000},
|
| ... | ... | |
| 176 | 201 |
|
| 177 | 202 |
@assert_backend_closed |
| 178 | 203 |
def test_list_images_filters_error_1(self, backend): |
| 179 |
response = self.get("/plankton/images/?size_max=")
|
|
| 204 |
response = self.myget("images/?size_max=")
|
|
| 180 | 205 |
self.assertBadRequest(response) |
| 181 | 206 |
|
| 182 | 207 |
@assert_backend_closed |
| 183 | 208 |
def test_list_images_filters_error_2(self, backend): |
| 184 |
response = self.get("/plankton/images/?size_min=foo")
|
|
| 209 |
response = self.myget("images/?size_min=foo")
|
|
| 185 | 210 |
self.assertBadRequest(response) |
| 186 | 211 |
|
| 187 | 212 |
@assert_backend_closed |
| 188 | 213 |
def test_update_image(self, backend): |
| 189 | 214 |
db_image = DummyImages.values()[0] |
| 190 |
response = self.put("/plankton/images/%s" % db_image['id'],
|
|
| 191 |
json.dumps({}),
|
|
| 192 |
'json', HTTP_X_IMAGE_META_OWNER='user2') |
|
| 215 |
response = self.myput("images/%s" % db_image['id'],
|
|
| 216 |
json.dumps({}),
|
|
| 217 |
'json', HTTP_X_IMAGE_META_OWNER='user2')
|
|
| 193 | 218 |
self.assertSuccess(response) |
| 194 | 219 |
backend.return_value.update_metadata.assert_called_once_with(db_image['id'], |
| 195 | 220 |
{"owner": "user2"})
|
| ... | ... | |
| 197 | 222 |
@assert_backend_closed |
| 198 | 223 |
def test_add_image_member(self, backend): |
| 199 | 224 |
image_id = DummyImages.values()[0]['id'] |
| 200 |
response = self.put("/plankton/images/%s/members/user3" % image_id,
|
|
| 201 |
json.dumps({}), 'json')
|
|
| 225 |
response = self.myput("images/%s/members/user3" % image_id,
|
|
| 226 |
json.dumps({}), 'json')
|
|
| 202 | 227 |
self.assertSuccess(response) |
| 203 | 228 |
backend.return_value.add_user.assert_called_once_with(image_id, |
| 204 | 229 |
'user3') |
| ... | ... | |
| 206 | 231 |
@assert_backend_closed |
| 207 | 232 |
def test_remove_image_member(self, backend): |
| 208 | 233 |
image_id = DummyImages.values()[0]['id'] |
| 209 |
response = self.delete("/plankton/images/%s/members/user3" % image_id)
|
|
| 234 |
response = self.mydelete("images/%s/members/user3" % image_id)
|
|
| 210 | 235 |
self.assertSuccess(response) |
| 211 | 236 |
backend.return_value.remove_user.assert_called_once_with(image_id, |
| 212 | 237 |
'user3') |
| ... | ... | |
| 214 | 239 |
@assert_backend_closed |
| 215 | 240 |
def test_add_image(self, backend): |
| 216 | 241 |
location = "pithos://uuid/container/name/" |
| 217 |
response = self.post("/plankton/images/",
|
|
| 218 |
json.dumps({}),
|
|
| 219 |
'json', |
|
| 220 |
HTTP_X_IMAGE_META_NAME='dummy_name', |
|
| 221 |
HTTP_X_IMAGE_META_OWNER='dummy_owner', |
|
| 222 |
HTTP_X_IMAGE_META_LOCATION=location) |
|
| 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)
|
|
| 223 | 248 |
self.assertSuccess(response) |
| 224 | 249 |
backend.return_value.register.assert_called_once_with('dummy_name',
|
| 225 | 250 |
location, |
| ... | ... | |
| 227 | 252 |
|
| 228 | 253 |
@assert_backend_closed |
| 229 | 254 |
def test_get_image(self, backend): |
| 230 |
response = self.get("/plankton/images/123")
|
|
| 255 |
response = self.myget("images/123")
|
|
| 231 | 256 |
self.assertEqual(response.status_code, 501) |
| 232 | 257 |
|
| 233 | 258 |
@assert_backend_closed |
| 234 | 259 |
def test_delete_image(self, backend): |
| 235 |
response = self.delete("/plankton/images/123")
|
|
| 260 |
response = self.mydelete("images/123")
|
|
| 236 | 261 |
self.assertEqual(response.status_code, 204) |
| 237 | 262 |
backend.return_value.unregister.assert_called_once_with('123')
|
| 238 | 263 |
backend.return_value._delete.assert_not_called() |
Also available in: Unified diff