Revision c977b0b6 snf-pithos-app/pithos/api/test/__init__.py
| b/snf-pithos-app/pithos/api/test/__init__.py | ||
|---|---|---|
| 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': |
Also available in: Unified diff