Revision f53483d8 snf-pithos-app/pithos/api/test/__init__.py
| b/snf-pithos-app/pithos/api/test/__init__.py | ||
|---|---|---|
| 34 | 34 |
# interpreted as representing official policies, either expressed |
| 35 | 35 |
# or implied, of GRNET S.A. |
| 36 | 36 |
|
| 37 |
from urlparse import urlunsplit, urlsplit |
|
| 37 |
from urlparse import urlunsplit, urlsplit, urlparse
|
|
| 38 | 38 |
from xml.dom import minidom |
| 39 | 39 |
from urllib import quote, unquote |
| 40 | 40 |
|
| ... | ... | |
| 48 | 48 |
from synnefo.lib import join_urls |
| 49 | 49 |
|
| 50 | 50 |
from django.test import TestCase |
| 51 |
from django.test.client import Client, MULTIPART_CONTENT, FakePayload |
|
| 51 | 52 |
from django.test.simple import DjangoTestSuiteRunner |
| 52 | 53 |
from django.conf import settings |
| 53 | 54 |
from django.utils.http import urlencode |
| ... | ... | |
| 102 | 103 |
'django.db.backends.oracle': 'oracle'} |
| 103 | 104 |
|
| 104 | 105 |
|
| 105 |
def prepate_db_connection():
|
|
| 106 |
def prepare_db_connection():
|
|
| 106 | 107 |
"""Build pithos backend connection string from django default database""" |
| 107 | 108 |
|
| 108 | 109 |
db = settings.DATABASES['default'] |
| ... | ... | |
| 141 | 142 |
def setup_databases(self, **kwargs): |
| 142 | 143 |
old_names, mirrors = super(PithosTestSuiteRunner, |
| 143 | 144 |
self).setup_databases(**kwargs) |
| 144 |
prepate_db_connection()
|
|
| 145 |
prepare_db_connection()
|
|
| 145 | 146 |
return old_names, mirrors |
| 146 | 147 |
|
| 147 | 148 |
def teardown_databases(self, old_config, **kwargs): |
| ... | ... | |
| 151 | 152 |
**kwargs) |
| 152 | 153 |
|
| 153 | 154 |
|
| 155 |
class PithosTestClient(Client): |
|
| 156 |
def copy(self, path, data={}, content_type=MULTIPART_CONTENT,
|
|
| 157 |
follow=False, **extra): |
|
| 158 |
""" |
|
| 159 |
Send a resource to the server using COPY. |
|
| 160 |
""" |
|
| 161 |
parsed = urlparse(path) |
|
| 162 |
r = {
|
|
| 163 |
'CONTENT_TYPE': 'text/html; charset=utf-8', |
|
| 164 |
'PATH_INFO': self._get_path(parsed), |
|
| 165 |
'QUERY_STRING': urlencode(data, doseq=True) or parsed[4], |
|
| 166 |
'REQUEST_METHOD': 'COPY', |
|
| 167 |
'wsgi.input': FakePayload('')
|
|
| 168 |
} |
|
| 169 |
r.update(extra) |
|
| 170 |
|
|
| 171 |
response = self.request(**r) |
|
| 172 |
if follow: |
|
| 173 |
response = self._handle_redirects(response, **extra) |
|
| 174 |
return response |
|
| 175 |
|
|
| 176 |
def move(self, path, data={}, content_type=MULTIPART_CONTENT,
|
|
| 177 |
follow=False, **extra): |
|
| 178 |
""" |
|
| 179 |
Send a resource to the server using MOVE. |
|
| 180 |
""" |
|
| 181 |
parsed = urlparse(path) |
|
| 182 |
r = {
|
|
| 183 |
'CONTENT_TYPE': 'text/html; charset=utf-8', |
|
| 184 |
'PATH_INFO': self._get_path(parsed), |
|
| 185 |
'QUERY_STRING': urlencode(data, doseq=True) or parsed[4], |
|
| 186 |
'REQUEST_METHOD': 'MOVE', |
|
| 187 |
'wsgi.input': FakePayload('')
|
|
| 188 |
} |
|
| 189 |
r.update(extra) |
|
| 190 |
|
|
| 191 |
response = self.request(**r) |
|
| 192 |
if follow: |
|
| 193 |
response = self._handle_redirects(response, **extra) |
|
| 194 |
return response |
|
| 195 |
|
|
| 196 |
|
|
| 154 | 197 |
class PithosAPITest(TestCase): |
| 155 | 198 |
def setUp(self): |
| 199 |
self.client = PithosTestClient() |
|
| 200 |
|
|
| 156 | 201 |
# Override default block size to spead up tests |
| 157 | 202 |
pithos_settings.BACKEND_BLOCK_SIZE = TEST_BLOCK_SIZE |
| 158 | 203 |
pithos_settings.BACKEND_HASH_ALGORITHM = TEST_HASH_ALGORITHM |
| ... | ... | |
| 177 | 222 |
self.delete_container_content(c['name']) |
| 178 | 223 |
self.delete_container(c['name']) |
| 179 | 224 |
|
| 180 |
def head(self, url, user='user', data={}, follow=False, **extra):
|
|
| 225 |
def head(self, url, user='user', token='DummyToken', data={}, follow=False,
|
|
| 226 |
**extra): |
|
| 181 | 227 |
with astakos_user(user): |
| 182 | 228 |
extra = dict((quote(k), quote(v)) for k, v in extra.items()) |
| 183 |
extra.setdefault('HTTP_X_AUTH_TOKEN', 'token')
|
|
| 229 |
if token: |
|
| 230 |
extra['HTTP_X_AUTH_TOKEN'] = token |
|
| 184 | 231 |
response = self.client.head(url, data, follow, **extra) |
| 185 | 232 |
return response |
| 186 | 233 |
|
| 187 |
def get(self, url, user='user', data={}, follow=False, **extra):
|
|
| 234 |
def get(self, url, user='user', token='DummyToken', data={}, follow=False,
|
|
| 235 |
**extra): |
|
| 188 | 236 |
with astakos_user(user): |
| 189 | 237 |
extra = dict((quote(k), quote(v)) for k, v in extra.items()) |
| 190 |
extra.setdefault('HTTP_X_AUTH_TOKEN', 'token')
|
|
| 238 |
if token: |
|
| 239 |
extra['HTTP_X_AUTH_TOKEN'] = token |
|
| 191 | 240 |
response = self.client.get(url, data, follow, **extra) |
| 192 | 241 |
return response |
| 193 | 242 |
|
| 194 |
def delete(self, url, user='user', data={}, follow=False, **extra):
|
|
| 243 |
def delete(self, url, user='user', token='DummyToken', data={},
|
|
| 244 |
follow=False, **extra): |
|
| 195 | 245 |
with astakos_user(user): |
| 196 | 246 |
extra = dict((quote(k), quote(v)) for k, v in extra.items()) |
| 197 |
extra.setdefault('HTTP_X_AUTH_TOKEN', 'token')
|
|
| 247 |
if token: |
|
| 248 |
extra['HTTP_X_AUTH_TOKEN'] = token |
|
| 198 | 249 |
response = self.client.delete(url, data, follow, **extra) |
| 199 | 250 |
return response |
| 200 | 251 |
|
| 201 |
def post(self, url, user='user', data={},
|
|
| 252 |
def post(self, url, user='user', token='DummyToken', data={},
|
|
| 202 | 253 |
content_type='application/octet-stream', follow=False, **extra): |
| 203 | 254 |
with astakos_user(user): |
| 204 | 255 |
extra = dict((quote(k), quote(v)) for k, v in extra.items()) |
| 205 |
extra.setdefault('HTTP_X_AUTH_TOKEN', 'token')
|
|
| 256 |
if token: |
|
| 257 |
extra['HTTP_X_AUTH_TOKEN'] = token |
|
| 206 | 258 |
response = self.client.post(url, data, content_type, follow, |
| 207 | 259 |
**extra) |
| 208 | 260 |
return response |
| 209 | 261 |
|
| 210 |
def put(self, url, user='user', data={},
|
|
| 262 |
def put(self, url, user='user', token='DummyToken', data={},
|
|
| 211 | 263 |
content_type='application/octet-stream', follow=False, **extra): |
| 212 | 264 |
with astakos_user(user): |
| 213 | 265 |
extra = dict((quote(k), quote(v)) for k, v in extra.items()) |
| 214 |
extra.setdefault('HTTP_X_AUTH_TOKEN', 'token')
|
|
| 266 |
if token: |
|
| 267 |
extra['HTTP_X_AUTH_TOKEN'] = token |
|
| 215 | 268 |
response = self.client.put(url, data, content_type, follow, |
| 216 | 269 |
**extra) |
| 217 | 270 |
return response |
| 218 | 271 |
|
| 272 |
def copy(self, url, user='user', token='DummyToken', data={},
|
|
| 273 |
content_type='application/octet-stream', follow=False, **extra): |
|
| 274 |
with astakos_user(user): |
|
| 275 |
extra = dict((quote(k), quote(v)) for k, v in extra.items()) |
|
| 276 |
if token: |
|
| 277 |
extra['HTTP_X_AUTH_TOKEN'] = token |
|
| 278 |
response = self.client.copy(url, data, content_type, follow, |
|
| 279 |
**extra) |
|
| 280 |
return response |
|
| 281 |
|
|
| 282 |
def move(self, url, user='user', token='DummyToken', data={},
|
|
| 283 |
content_type='application/octet-stream', follow=False, **extra): |
|
| 284 |
with astakos_user(user): |
|
| 285 |
extra = dict((quote(k), quote(v)) for k, v in extra.items()) |
|
| 286 |
if token: |
|
| 287 |
extra['HTTP_X_AUTH_TOKEN'] = token |
|
| 288 |
response = self.client.move(url, data, content_type, follow, |
|
| 289 |
**extra) |
|
| 290 |
return response |
|
| 291 |
|
|
| 219 | 292 |
def update_account_meta(self, meta, user=None): |
| 220 | 293 |
user = user or self.user |
| 221 | 294 |
kwargs = dict( |
| ... | ... | |
| 223 | 296 |
url = join_urls(self.pithos_path, user) |
| 224 | 297 |
r = self.post('%s?update=' % url, user=user, **kwargs)
|
| 225 | 298 |
self.assertEqual(r.status_code, 202) |
| 226 |
account_meta = self.get_account_meta() |
|
| 227 |
(self.assertTrue(k in account_meta) for k in meta.keys()) |
|
| 228 |
(self.assertEqual(account_meta[k], v) for k, v in meta.items()) |
|
| 299 |
account_meta = self.get_account_meta(user=user) |
|
| 300 |
(self.assertTrue('X-Account-Meta-%s' % k in account_meta) for
|
|
| 301 |
k in meta.keys()) |
|
| 302 |
(self.assertEqual(account_meta['X-Account-Meta-%s' % k], v) for |
|
| 303 |
k, v in meta.items()) |
|
| 304 |
|
|
| 305 |
def reset_account_meta(self, meta, user=None): |
|
| 306 |
user = user or self.user |
|
| 307 |
kwargs = dict( |
|
| 308 |
('HTTP_X_ACCOUNT_META_%s' % k, str(v)) for k, v in meta.items())
|
|
| 309 |
url = join_urls(self.pithos_path, user) |
|
| 310 |
r = self.post(url, user=user, **kwargs) |
|
| 311 |
self.assertEqual(r.status_code, 202) |
|
| 312 |
account_meta = self.get_account_meta(user=user) |
|
| 313 |
(self.assertTrue('X-Account-Meta-%s' % k in account_meta) for
|
|
| 314 |
k in meta.keys()) |
|
| 315 |
(self.assertEqual(account_meta['X-Account-Meta-%s' % k], v) for |
|
| 316 |
k, v in meta.items()) |
|
| 229 | 317 |
|
| 230 | 318 |
def delete_account_meta(self, meta, user=None): |
| 231 | 319 |
user = user or self.user |
| 232 |
transform = (lambda k: 'HTTP_X_ACCOUNT_META_%s' % |
|
| 233 |
k.replace('-', '_').upper())
|
|
| 320 |
transform = lambda k: 'HTTP_%s' % k.replace('-', '_').upper()
|
|
| 234 | 321 |
kwargs = dict((transform(k), '') for k, v in meta.items()) |
| 235 | 322 |
url = join_urls(self.pithos_path, user) |
| 236 | 323 |
r = self.post('%s?update=' % url, user=user, **kwargs)
|
| 237 | 324 |
self.assertEqual(r.status_code, 202) |
| 238 |
account_meta = self.get_account_meta() |
|
| 239 |
(self.assertTrue(k not in account_meta) for k in meta.keys()) |
|
| 325 |
account_meta = self.get_account_meta(user=user) |
|
| 326 |
(self.assertTrue('X-Account-Meta-%s' % k not in account_meta) for
|
|
| 327 |
k in meta.keys()) |
|
| 240 | 328 |
return r |
| 241 | 329 |
|
| 242 | 330 |
def delete_account_groups(self, groups, user=None): |
| 243 | 331 |
user = user or self.user |
| 244 | 332 |
url = join_urls(self.pithos_path, user) |
| 245 |
transform = (lambda k: 'HTTP_X_ACCOUNT_GROUP_%s' % |
|
| 246 |
k.replace('-', '_').upper())
|
|
| 247 |
kwargs = dict((transform(k), '') for k, v in groups.items()) |
|
| 248 |
r = self.post('%s?update=' % url, user=user, **kwargs)
|
|
| 333 |
r = self.post('%s?update=' % url, user=user, **groups)
|
|
| 249 | 334 |
self.assertEqual(r.status_code, 202) |
| 250 | 335 |
account_groups = self.get_account_groups() |
| 251 | 336 |
(self.assertTrue(k not in account_groups) for k in groups.keys()) |
| ... | ... | |
| 438 | 523 |
def get_object_meta(self, container, object, version=None, until=None, |
| 439 | 524 |
user=None): |
| 440 | 525 |
prefix = 'X-Object-Meta-' |
| 526 |
user = user or self.user |
|
| 441 | 527 |
r = self.get_object_info(container, object, version, until=until, |
| 442 | 528 |
user=user) |
| 443 | 529 |
headers = dict(r._headers.values()) |
Also available in: Unified diff