156 |
156 |
def _get_path(self, parsed):
|
157 |
157 |
# If there are parameters, add them
|
158 |
158 |
if parsed[3]:
|
159 |
|
return unquote(parsed[2] + ";" + parsed[3])
|
|
159 |
return unquote(parsed[2] + ";" + parsed[3])
|
160 |
160 |
else:
|
161 |
|
return unquote(parsed[2])
|
|
161 |
return unquote(parsed[2])
|
162 |
162 |
|
163 |
163 |
def copy(self, path, data={}, content_type=MULTIPART_CONTENT,
|
164 |
164 |
follow=False, **extra):
|
... | ... | |
296 |
296 |
**extra)
|
297 |
297 |
return response
|
298 |
298 |
|
299 |
|
def update_account_meta(self, meta, user=None):
|
|
299 |
def update_account_meta(self, meta, user=None, verify_status=True):
|
300 |
300 |
user = user or self.user
|
301 |
301 |
kwargs = dict(
|
302 |
302 |
('HTTP_X_ACCOUNT_META_%s' % k, str(v)) for k, v in meta.items())
|
303 |
303 |
url = join_urls(self.pithos_path, user)
|
304 |
304 |
r = self.post('%s?update=' % url, user=user, **kwargs)
|
305 |
|
self.assertEqual(r.status_code, 202)
|
|
305 |
if verify_status:
|
|
306 |
self.assertEqual(r.status_code, 202)
|
306 |
307 |
account_meta = self.get_account_meta(user=user)
|
307 |
308 |
(self.assertTrue('X-Account-Meta-%s' % k in account_meta) for
|
308 |
309 |
k in meta.keys())
|
309 |
310 |
(self.assertEqual(account_meta['X-Account-Meta-%s' % k], v) for
|
310 |
311 |
k, v in meta.items())
|
311 |
312 |
|
312 |
|
def reset_account_meta(self, meta, user=None):
|
|
313 |
def reset_account_meta(self, meta, user=None, verify_status=True):
|
313 |
314 |
user = user or self.user
|
314 |
315 |
kwargs = dict(
|
315 |
316 |
('HTTP_X_ACCOUNT_META_%s' % k, str(v)) for k, v in meta.items())
|
316 |
317 |
url = join_urls(self.pithos_path, user)
|
317 |
318 |
r = self.post(url, user=user, **kwargs)
|
318 |
|
self.assertEqual(r.status_code, 202)
|
|
319 |
if verify_status:
|
|
320 |
self.assertEqual(r.status_code, 202)
|
319 |
321 |
account_meta = self.get_account_meta(user=user)
|
320 |
322 |
(self.assertTrue('X-Account-Meta-%s' % k in account_meta) for
|
321 |
323 |
k in meta.keys())
|
322 |
324 |
(self.assertEqual(account_meta['X-Account-Meta-%s' % k], v) for
|
323 |
325 |
k, v in meta.items())
|
324 |
326 |
|
325 |
|
def delete_account_meta(self, meta, user=None):
|
|
327 |
def delete_account_meta(self, meta, user=None, verify_status=True):
|
326 |
328 |
user = user or self.user
|
327 |
329 |
transform = lambda k: 'HTTP_%s' % k.replace('-', '_').upper()
|
328 |
330 |
kwargs = dict((transform(k), '') for k, v in meta.items())
|
329 |
331 |
url = join_urls(self.pithos_path, user)
|
330 |
332 |
r = self.post('%s?update=' % url, user=user, **kwargs)
|
331 |
|
self.assertEqual(r.status_code, 202)
|
|
333 |
if verify_status:
|
|
334 |
self.assertEqual(r.status_code, 202)
|
332 |
335 |
account_meta = self.get_account_meta(user=user)
|
333 |
336 |
(self.assertTrue('X-Account-Meta-%s' % k not in account_meta) for
|
334 |
337 |
k in meta.keys())
|
335 |
338 |
return r
|
336 |
339 |
|
337 |
|
def delete_account_groups(self, groups, user=None):
|
|
340 |
def delete_account_groups(self, groups, user=None, verify_status=True):
|
338 |
341 |
user = user or self.user
|
339 |
342 |
url = join_urls(self.pithos_path, user)
|
340 |
343 |
r = self.post('%s?update=' % url, user=user, **groups)
|
341 |
|
self.assertEqual(r.status_code, 202)
|
|
344 |
if verify_status:
|
|
345 |
self.assertEqual(r.status_code, 202)
|
342 |
346 |
account_groups = self.get_account_groups()
|
343 |
347 |
(self.assertTrue(k not in account_groups) for k in groups.keys())
|
344 |
348 |
return r
|
345 |
349 |
|
346 |
|
def get_account_info(self, until=None, user=None):
|
|
350 |
def get_account_info(self, until=None, user=None, verify_status=True):
|
347 |
351 |
user = user or self.user
|
348 |
352 |
url = join_urls(self.pithos_path, user)
|
349 |
353 |
if until is not None:
|
... | ... | |
353 |
357 |
})
|
354 |
358 |
url = urlunsplit(parts)
|
355 |
359 |
r = self.head(url, user=user)
|
356 |
|
self.assertEqual(r.status_code, 204)
|
|
360 |
if verify_status:
|
|
361 |
self.assertEqual(r.status_code, 204)
|
357 |
362 |
return r
|
358 |
363 |
|
359 |
364 |
def get_account_meta(self, until=None, user=None):
|
... | ... | |
368 |
373 |
headers = dict(r._headers.values())
|
369 |
374 |
return filter_headers(headers, prefix)
|
370 |
375 |
|
371 |
|
def get_container_info(self, container, until=None, user=None):
|
|
376 |
def get_container_info(self, container, until=None, user=None,
|
|
377 |
verify_status=True):
|
372 |
378 |
user = user or self.user
|
373 |
379 |
url = join_urls(self.pithos_path, user, container)
|
374 |
380 |
if until is not None:
|
... | ... | |
378 |
384 |
})
|
379 |
385 |
url = urlunsplit(parts)
|
380 |
386 |
r = self.head(url, user=user)
|
381 |
|
self.assertEqual(r.status_code, 204)
|
|
387 |
if verify_status:
|
|
388 |
self.assertEqual(r.status_code, 204)
|
382 |
389 |
return r
|
383 |
390 |
|
384 |
391 |
def get_container_meta(self, container, until=None, user=None):
|
... | ... | |
387 |
394 |
headers = dict(r._headers.values())
|
388 |
395 |
return filter_headers(headers, prefix)
|
389 |
396 |
|
390 |
|
def update_container_meta(self, container, meta, user=None):
|
|
397 |
def update_container_meta(self, container, meta=None, user=None,
|
|
398 |
verify_status=True):
|
391 |
399 |
user = user or self.user
|
|
400 |
meta = meta or {get_random_name(): get_random_name()}
|
392 |
401 |
kwargs = dict(
|
393 |
402 |
('HTTP_X_CONTAINER_META_%s' % k, str(v)) for k, v in meta.items())
|
394 |
403 |
url = join_urls(self.pithos_path, user, container)
|
395 |
404 |
r = self.post('%s?update=' % url, user=user, **kwargs)
|
396 |
|
self.assertEqual(r.status_code, 202)
|
|
405 |
if verify_status:
|
|
406 |
self.assertEqual(r.status_code, 202)
|
397 |
407 |
container_meta = self.get_container_meta(container, user=user)
|
398 |
408 |
(self.assertTrue('X-Container-Meta-%s' % k in container_meta) for
|
399 |
409 |
k in meta.keys())
|
400 |
410 |
(self.assertEqual(container_meta['X-Container-Meta-%s' % k], v) for
|
401 |
411 |
k, v in meta.items())
|
|
412 |
return r
|
402 |
413 |
|
403 |
414 |
def list_containers(self, format='json', headers={}, user=None, **params):
|
404 |
415 |
user = user or self.user
|
... | ... | |
425 |
436 |
elif format == 'xml':
|
426 |
437 |
return minidom.parseString(r.content)
|
427 |
438 |
|
428 |
|
def delete_container_content(self, cname, user=None):
|
|
439 |
def delete_container_content(self, cname, user=None, verify_status=True):
|
429 |
440 |
user = user or self.user
|
430 |
441 |
url = join_urls(self.pithos_path, user, cname)
|
431 |
442 |
r = self.delete('%s?delimiter=/' % url, user=user)
|
432 |
|
self.assertEqual(r.status_code, 204)
|
|
443 |
if verify_status:
|
|
444 |
self.assertEqual(r.status_code, 204)
|
433 |
445 |
return r
|
434 |
446 |
|
435 |
|
def delete_container(self, cname, user=None):
|
|
447 |
def delete_container(self, cname, user=None, verify_status=True):
|
436 |
448 |
user = user or self.user
|
437 |
449 |
url = join_urls(self.pithos_path, user, cname)
|
438 |
450 |
r = self.delete(url, user=user)
|
439 |
|
self.assertEqual(r.status_code, 204)
|
|
451 |
if verify_status:
|
|
452 |
self.assertEqual(r.status_code, 204)
|
|
453 |
return r
|
|
454 |
|
|
455 |
def delete_object(self, cname, oname, user=None, verify_status=True):
|
|
456 |
user = user or self.user
|
|
457 |
url = join_urls(self.pithos_path, user, cname, oname)
|
|
458 |
r = self.delete(url, user=user)
|
|
459 |
if verify_status:
|
|
460 |
self.assertEqual(r.status_code, 204)
|
440 |
461 |
return r
|
441 |
462 |
|
442 |
|
def create_container(self, cname=None, user=None):
|
|
463 |
def create_container(self, cname=None, user=None, verify_status=True):
|
443 |
464 |
cname = cname or get_random_name()
|
444 |
465 |
user = user or self.user
|
445 |
466 |
url = join_urls(self.pithos_path, user, cname)
|
446 |
467 |
r = self.put(url, user=user, data='')
|
447 |
|
self.assertTrue(r.status_code in (202, 201))
|
|
468 |
if verify_status:
|
|
469 |
self.assertTrue(r.status_code in (202, 201))
|
448 |
470 |
return cname, r
|
449 |
471 |
|
450 |
|
def upload_object(self, cname, oname=None, length=None, verify=True,
|
|
472 |
def upload_object(self, cname, oname=None, length=None, verify_status=True,
|
451 |
473 |
user=None, **meta):
|
452 |
474 |
oname = oname or get_random_name()
|
453 |
475 |
length = length or random.randint(TEST_BLOCK_SIZE, 2 * TEST_BLOCK_SIZE)
|
... | ... | |
457 |
479 |
for k, v in meta.iteritems())
|
458 |
480 |
url = join_urls(self.pithos_path, user, cname, oname)
|
459 |
481 |
r = self.put(url, user=user, data=data, **headers)
|
460 |
|
if verify:
|
|
482 |
if verify_status:
|
461 |
483 |
self.assertEqual(r.status_code, 201)
|
462 |
484 |
return oname, data, r
|
463 |
485 |
|
464 |
486 |
def update_object_data(self, cname, oname=None, length=None,
|
465 |
487 |
content_type=None, content_range=None,
|
466 |
|
verify=True, user=None, **meta):
|
|
488 |
verify_status=True, user=None, **meta):
|
467 |
489 |
oname = oname or get_random_name()
|
468 |
490 |
length = length or random.randint(TEST_BLOCK_SIZE, 2 * TEST_BLOCK_SIZE)
|
469 |
491 |
content_type = content_type or 'application/octet-stream'
|
... | ... | |
476 |
498 |
url = join_urls(self.pithos_path, user, cname, oname)
|
477 |
499 |
r = self.post(url, user=user, data=data, content_type=content_type,
|
478 |
500 |
**headers)
|
479 |
|
if verify:
|
|
501 |
if verify_status:
|
480 |
502 |
self.assertEqual(r.status_code, 204)
|
481 |
503 |
return oname, data, r
|
482 |
504 |
|
... | ... | |
488 |
510 |
content_range='bytes */*',
|
489 |
511 |
user=user)
|
490 |
512 |
|
491 |
|
def create_folder(self, cname, oname=None, user=None, **headers):
|
|
513 |
def create_folder(self, cname, oname=None, user=None, verify_status=True,
|
|
514 |
**headers):
|
492 |
515 |
user = user or self.user
|
493 |
516 |
oname = oname or get_random_name()
|
494 |
517 |
url = join_urls(self.pithos_path, user, cname, oname)
|
495 |
518 |
r = self.put(url, user=user, data='',
|
496 |
519 |
content_type='application/directory', **headers)
|
497 |
|
self.assertEqual(r.status_code, 201)
|
|
520 |
if verify_status:
|
|
521 |
self.assertEqual(r.status_code, 201)
|
498 |
522 |
return oname, r
|
499 |
523 |
|
500 |
|
def list_objects(self, cname, prefix=None, user=None):
|
|
524 |
def list_objects(self, cname, prefix=None, user=None, verify_status=True):
|
501 |
525 |
user = user or self.user
|
502 |
526 |
url = join_urls(self.pithos_path, user, cname)
|
503 |
527 |
path = '%s?format=json' % url
|
504 |
528 |
if prefix is not None:
|
505 |
529 |
path = '%s&prefix=%s' % (path, prefix)
|
506 |
530 |
r = self.get(path, user=user)
|
507 |
|
self.assertTrue(r.status_code in (200, 204))
|
|
531 |
if verify_status:
|
|
532 |
self.assertTrue(r.status_code in (200, 204))
|
508 |
533 |
try:
|
509 |
534 |
objects = json.loads(r.content)
|
510 |
535 |
except:
|
... | ... | |
512 |
537 |
return objects
|
513 |
538 |
|
514 |
539 |
def get_object_info(self, container, object, version=None, until=None,
|
515 |
|
user=None):
|
|
540 |
user=None, verify_status=True):
|
516 |
541 |
user = user or self.user
|
517 |
542 |
url = join_urls(self.pithos_path, user, container, object)
|
518 |
543 |
if until is not None:
|
... | ... | |
524 |
549 |
if version:
|
525 |
550 |
url = '%s?version=%s' % (url, version)
|
526 |
551 |
r = self.head(url, user=user)
|
527 |
|
self.assertEqual(r.status_code, 200)
|
|
552 |
if verify_status:
|
|
553 |
self.assertEqual(r.status_code, 200)
|
528 |
554 |
return r
|
529 |
555 |
|
530 |
556 |
def get_object_meta(self, container, object, version=None, until=None,
|
... | ... | |
536 |
562 |
headers = dict(r._headers.values())
|
537 |
563 |
return filter_headers(headers, prefix)
|
538 |
564 |
|
539 |
|
def update_object_meta(self, container, object, meta, user=None):
|
|
565 |
def update_object_meta(self, container, object, meta=None, user=None,
|
|
566 |
verify_status=True):
|
540 |
567 |
user = user or self.user
|
|
568 |
meta = meta or {get_random_name(): get_random_name()}
|
541 |
569 |
kwargs = dict(
|
542 |
570 |
('HTTP_X_OBJECT_META_%s' % k, str(v)) for k, v in meta.items())
|
543 |
571 |
url = join_urls(self.pithos_path, user, container, object)
|
544 |
572 |
r = self.post('%s?update=' % url, user=user, content_type='', **kwargs)
|
545 |
|
self.assertEqual(r.status_code, 202)
|
|
573 |
if verify_status:
|
|
574 |
self.assertEqual(r.status_code, 202)
|
546 |
575 |
object_meta = self.get_object_meta(container, object, user=user)
|
547 |
576 |
(self.assertTrue('X-Objecr-Meta-%s' % k in object_meta) for
|
548 |
577 |
k in meta.keys())
|
549 |
578 |
(self.assertEqual(object_meta['X-Object-Meta-%s' % k], v) for
|
550 |
579 |
k, v in meta.items())
|
|
580 |
return r
|
551 |
581 |
|
552 |
582 |
def assert_extended(self, data, format, type, size=10000):
|
553 |
583 |
if format == 'xml':
|