Revision d6a92fa0 snf-pithos-app/pithos/api/test/containers.py

b/snf-pithos-app/pithos/api/test/containers.py
38 38
    strnextling, pithos_settings, pithos_test_settings
39 39
from pithos.backends.random_word import get_random_word
40 40

  
41
from synnefo.lib import join_urls
42

  
41 43
import django.utils.simplejson as json
42 44
from django.http import urlencode
43 45

  
......
62 64
            objects[name] = data
63 65

  
64 66
        t1 = datetime.datetime.utcnow()
65
        r = self.head('/v1/%s/apples' % self.user)
67
        url = join_urls(self.pithos_path, self.user, 'apples')
68
        r = self.head(url)
66 69
        self.assertEqual(int(r['X-Container-Object-Count']), len(objects))
67 70
        self.assertEqual(int(r['X-Container-Bytes-Used']),
68 71
                         sum([len(i) for i in objects.values()]))
......
103 106
        cname = self.cnames[0]
104 107
        onames = self.objects[cname].keys()
105 108
        oname = onames.pop()
106
        r = self.post('/v1/%s/%s/%s' % (self.user, cname, oname),
107
                      content_type='',
108
                      HTTP_X_OBJECT_SHARING='read=*')
109
        url = join_urls(self.pithos_path, self.user, cname, oname)
110
        r = self.post(url, content_type='', HTTP_X_OBJECT_SHARING='read=*')
109 111
        self.assertEqual(r.status_code, 202)
110 112

  
111 113
        # publish another object
112 114
        other = onames.pop()
113
        r = self.post('/v1/%s/%s/%s' % (self.user, cname, other),
114
                      content_type='',
115
                      HTTP_X_OBJECT_PUBLIC='true')
115
        url = join_urls(self.pithos_path, self.user, cname, other)
116
        r = self.post(url, content_type='', HTTP_X_OBJECT_PUBLIC='true')
116 117
        self.assertEqual(r.status_code, 202)
117 118

  
118 119
        # list shared and assert only the shared object is returned
119
        r = self.get('/v1/%s/%s?shared=' % (self.user, cname))
120
        url = join_urls(self.pithos_path, self.user, cname)
121
        r = self.get('%s?shared=' % url)
120 122
        self.assertEqual(r.status_code, 200)
121 123
        objects = r.content.split('\n')
122 124
        objects.remove('')
123 125
        self.assertEqual([oname], objects)
124 126

  
125 127
        # list detailed shared and assert only the shared object is returned
126
        r = self.get('/v1/%s/%s?shared=&format=json' % (self.user, cname))
128
        url = join_urls(self.pithos_path, self.user, cname)
129
        r = self.get('%s?shared=&format=json' % url)
127 130
        self.assertEqual(r.status_code, 200)
128 131
        try:
129 132
            objects = json.loads(r.content)
......
135 138

  
136 139
        # publish the shared object and assert it is still listed in the
137 140
        # shared objects
138
        r = self.post('/v1/%s/%s/%s' % (self.user, cname, oname),
139
                      content_type='',
140
                      HTTP_X_OBJECT_PUBLIC='true')
141
        url = join_urls(self.pithos_path, self.user, cname, oname)
142
        r = self.post(url, content_type='', HTTP_X_OBJECT_PUBLIC='true')
141 143
        self.assertEqual(r.status_code, 202)
142
        r = self.get('/v1/%s/%s?shared=&format=json' % (self.user, cname))
144
        url = join_urls(self.pithos_path, self.user, cname)
145
        r = self.get('%s?shared=&format=json' % url)
143 146
        self.assertEqual(r.status_code, 200)
144 147
        try:
145 148
            objects = json.loads(r.content)
......
154 157
        descendant = strnextling(oname)
155 158
        self.upload_object(cname, descendant)
156 159
        # request shared and assert child obejct is not listed
157
        r = self.get('/v1/%s/%s?shared=' % (self.user, cname))
160
        url = join_urls(self.pithos_path, self.user, cname)
161
        r = self.get('%s?shared=' % url)
158 162
        self.assertEqual(r.status_code, 200)
159 163
        objects = r.content.split('\n')
160 164
        objects.remove('')
......
167 171
        descendant = '%s/%s' % (oname, get_random_word(8))
168 172
        self.upload_object(cname, descendant)
169 173
        # request shared
170
        r = self.get('/v1/%s/%s?shared=' % (self.user, cname))
174
        url = join_urls(self.pithos_path, self.user, cname)
175
        r = self.get('%s?shared=' % url)
171 176
        self.assertEqual(r.status_code, 200)
172 177
        objects = r.content.split('\n')
173 178
        objects.remove('')
......
179 184
        cname = self.cnames[0]
180 185
        onames = self.objects[cname].keys()
181 186
        oname = onames.pop()
182
        r = self.post('/v1/%s/%s/%s' % (self.user, cname, oname),
183
                      content_type='',
184
                      HTTP_X_OBJECT_PUBLIC='true')
187
        url = join_urls(self.pithos_path, self.user, cname, oname)
188
        r = self.post(url, content_type='', HTTP_X_OBJECT_PUBLIC='true')
185 189
        self.assertEqual(r.status_code, 202)
186 190

  
187 191
        # share another
188 192
        other = onames.pop()
189
        r = self.post('/v1/%s/%s/%s' % (self.user, cname, other),
190
                      content_type='',
191
                      HTTP_X_OBJECT_SHARING='read=alice')
193
        url = join_urls(self.pithos_path, self.user, cname, other)
194
        r = self.post(url, content_type='', HTTP_X_OBJECT_SHARING='read=alice')
192 195
        self.assertEqual(r.status_code, 202)
193 196

  
194 197
        # list public and assert only the public object is returned
195
        r = self.get('/v1/%s/%s?public=' % (self.user, cname))
198
        url = join_urls(self.pithos_path, self.user, cname)
199
        r = self.get('%s?public=' % url)
196 200
        objects = r.content.split('\n')
197 201
        self.assertEqual(r.status_code, 200)
198 202
        self.assertTrue(oname in r.content.split('\n'))
199 203
        (self.assertTrue(o not in objects) for o in o_names[1:])
200 204

  
201 205
        # list detailed public and assert only the public object is returned
202
        r = self.get('/v1/%s/%s?public=&format=json' % (self.user, cname))
206
        url = join_urls(self.pithos_path, self.user, cname)
207
        r = self.get('%s?public=&format=json' % url)
203 208
        self.assertEqual(r.status_code, 200)
204 209
        try:
205 210
            objects = json.loads(r.content)
......
211 216

  
212 217
        # share the public object and assert it is still listed in the
213 218
        # public objects
214
        r = self.post('/v1/%s/%s/%s' % (self.user, cname, oname),
215
                      content_type='',
216
                      HTTP_X_OBJECT_SHARING='read=alice')
219
        url = join_urls(self.pithos_path, self.user, cname, oname)
220
        r = self.post(url, content_type='', HTTP_X_OBJECT_SHARING='read=alice')
217 221
        self.assertEqual(r.status_code, 202)
218
        r = self.get('/v1/%s/%s?public=&format=json' % (self.user, cname))
222
        url = join_urls(self.pithos_path, self.user, cname)
223
        r = self.get('%s?public=&format=json' % url)
219 224
        self.assertEqual(r.status_code, 200)
220 225
        try:
221 226
            objects = json.loads(r.content)
......
225 230
        self.assertTrue('x_object_sharing' in objects[0])
226 231
        self.assertTrue('x_object_public' in objects[0])
227 232

  
233
        url = join_urls(self.pithos_path, self.user, cname)
234

  
228 235
        # Assert listing the container public contents is forbidden to not
229 236
        # shared users
230
        r = self.get('/v1/%s/%s?public=&format=json' % (
231
            self.user, cname), user='bob')
237
        r = self.get('%s?public=&format=json' % url, user='bob')
232 238
        self.assertEqual(r.status_code, 403)
233 239

  
234 240
        # Assert listing the container public contents to shared users
235
        r = self.get('/v1/%s/%s?public=&format=json' % (
236
            self.user, cname), user='alice')
241
        r = self.get('%s?public=&format=json' % url, user='alice')
237 242
        self.assertEqual(r.status_code, 200)
238 243
        try:
239 244
            objects = json.loads(r.content)
......
249 254
        descendant = strnextling(oname)
250 255
        self.upload_object(cname, descendant)
251 256
        # request public and assert child obejct is not listed
252
        r = self.get('/v1/%s/%s?public=' % (self.user, cname))
257
        r = self.get('%s?public=' % url)
253 258
        objects = r.content.split('\n')
254 259
        objects.remove('')
255 260
        self.assertEqual(r.status_code, 200)
......
262 267
        descendant = '%s/%s' % (oname, get_random_word(8))
263 268
        self.upload_object(cname, descendant)
264 269
        # request public
265
        r = self.get('/v1/%s/%s?public=' % (self.user, cname))
270
        r = self.get('%s?public=' % url)
266 271
        self.assertEqual(r.status_code, 200)
267 272
        objects = r.content.split('\n')
268 273
        self.assertTrue(oname in objects)
......
341 346
#
342 347
#
343 348
#        o = self.upload_random_data(self.container[1], 'folder2/object')
344
#        objs = self.client.list_objects(self.container[1], shared=True, public=True)
349
#        objs = self.client.list_objects(
350
#            self.container[1], shared=True, public=True)
345 351
#        self.assertEqual(objs, ['folder1', 'folder1/object', 'folder2'])
346 352
#        objs = cl.list_objects(
347 353
#            self.container[1], shared=True, public=True, account=get_user()
......
350 356
#
351 357
    def test_list_objects(self):
352 358
        cname = self.cnames[0]
353
        r = self.get('/v1/%s/%s' % (self.user, cname))
359
        url = join_urls(self.pithos_path, self.user, cname)
360
        r = self.get(url)
354 361
        self.assertTrue(r.status_code, 200)
355 362
        objects = r.content.split('\n')
356 363
        if '' in objects:
......
361 368
        self.create_container('test')
362 369
        self.upload_object('test', '/objectname')
363 370

  
364
        r = self.get('/v1/%s/test' % self.user)
371
        url = join_urls(self.pithos_path, self.user, 'test')
372

  
373
        r = self.get(url)
365 374
        objects = r.content.split('\n')
366 375
        if '' in objects:
367 376
            objects.remove('')
368 377
        self.assertEqual(objects, ['/objectname'])
369 378

  
370
        r = self.get('/v1/%s/test?format=json' % self.user)
379
        r = self.get('%s?format=json' % url)
371 380
        try:
372 381
            objects = json.loads(r.content)
373 382
        except:
374 383
            self.fail('json format expected')
375 384
        self.assertEqual([o['name'] for o in objects], ['/objectname'])
376 385

  
377
        r = self.get('/v1/%s/test?format=xml' % self.user)
386
        r = self.get('%s?format=xml' % url)
378 387
        try:
379 388
            objects = minidom.parseString(r.content)
380 389
        except:
......
385 394

  
386 395
    def test_list_objects_with_limit_marker(self):
387 396
        cname = self.cnames[0]
388
        r = self.get('/v1/%s/%s?limit=qwert' % (self.user, cname))
397
        url = join_urls(self.pithos_path, self.user, cname)
398
        r = self.get('%s?limit=qwert' % url)
389 399
        self.assertTrue(r.status_code != 500)
390 400

  
391
        r = self.get('/v1/%s/%s?limit=2' % (self.user, cname))
401
        r = self.get('%s?limit=2' % url)
392 402
        self.assertEqual(r.status_code, 200)
393 403
        objects = r.content.split('\n')
394 404
        if '' in objects:
......
401 411
                   'moms_birthday.jpg']
402 412
        limit = 4
403 413
        for m in markers:
404
            r = self.get('/v1/%s/%s?limit=%s&marker=%s' % (
405
                self.user, cname, limit, m))
414
            r = self.get('%s?limit=%s&marker=%s' % (url, limit, m))
406 415
            objects = r.content.split('\n')
407 416
            if '' in objects:
408 417
                objects.remove('')
......
414 423
    @pithos_test_settings(API_LIST_LIMIT=10)
415 424
    def test_list_limit_exceeds(self):
416 425
        self.create_container('container')
426
        url = join_urls(self.pithos_path, self.user, 'container')
417 427

  
418 428
        for _ in range(pithos_settings.API_LIST_LIMIT + 1):
419 429
            self.upload_object('container')
420 430

  
421
        r = self.get('/v1/%s/container?format=json' % self.user)
431
        r = self.get('%s?format=json' % url)
422 432
        try:
423 433
            objects = json.loads(r.content)
424 434
        except:
......
427 437
                         len(objects))
428 438

  
429 439
    def test_list_pseudo_hierarchical_folders(self):
430
        r = self.get('/v1/%s/apples?prefix=photos&delimiter=/' % self.user)
440
        url = join_urls(self.pithos_path, self.user, 'apples')
441
        r = self.get('%s?prefix=photos&delimiter=/' % url)
431 442
        self.assertEqual(r.status_code, 200)
432 443
        objects = r.content.split('\n')
433 444
        if '' in objects:
......
436 447
            ['photos/animals/', 'photos/me.jpg', 'photos/plants/'],
437 448
            objects)
438 449

  
439
        r = self.get(
440
            '/v1/%s/apples?prefix=photos/animals&delimiter=/' % self.user)
450
        r = self.get('%s?prefix=photos/animals&delimiter=/' % url)
441 451
        objects = r.content.split('\n')
442 452
        if '' in objects:
443 453
            objects.remove('')
444 454
        self.assertEquals(
445 455
            ['photos/animals/cats/', 'photos/animals/dogs/'], objects)
446 456

  
447
        r = self.get('/v1/%s/apples?path=photos' % self.user)
457
        r = self.get('%s?path=photos' % url)
448 458
        objects = r.content.split('\n')
449 459
        if '' in objects:
450 460
            objects.remove('')
451 461
        self.assertEquals(['photos/me.jpg'], objects)
452 462

  
453 463
    def test_extended_list_json(self):
464
        url = join_urls(self.pithos_path, self.user, 'apples')
454 465
        params = {'format': 'json', 'limit': 2, 'prefix': 'photos/animals',
455 466
                  'delimiter': '/'}
456
        r = self.get('/v1/%s/apples?%s' % (self.user, urlencode(params)))
467
        r = self.get('%s?%s' % (url, urlencode(params)))
457 468
        self.assertEqual(r.status_code, 200)
458 469
        try:
459 470
            objects = json.loads(r.content)
......
463 474
        self.assertEqual(objects[1]['subdir'], 'photos/animals/dogs/')
464 475

  
465 476
    def test_extended_list_xml(self):
477
        url = join_urls(self.pithos_path, self.user, 'apples')
466 478
        params = {'format': 'xml', 'limit': 4, 'prefix': 'photos',
467 479
                  'delimiter': '/'}
468
        r = self.get('/v1/%s/apples?%s' % (self.user, urlencode(params)))
480
        r = self.get('%s?%s' % (url, urlencode(params)))
469 481
        self.assertEqual(r.status_code, 200)
470 482
        try:
471 483
            xml = minidom.parseString(r.content)
......
484 496
    def test_list_meta_double_matching(self):
485 497
        # update object meta
486 498
        cname = 'apples'
499
        container_url = join_urls(self.pithos_path, self.user, cname)
487 500
        oname = self.objects[cname].keys().pop()
488 501
        meta = {'quality': 'aaa', 'stock': 'true'}
489 502
        headers = dict(('HTTP_X_OBJECT_META_%s' % k.upper(), v)
490 503
                       for k, v in meta.iteritems())
491
        self.post('/v1/%s/%s/%s' % (self.user, cname, oname),
492
                  content_type='', **headers)
504
        object_url = join_urls(container_url, oname)
505
        self.post(object_url, content_type='', **headers)
493 506

  
494 507
        # list objects that satisfy the criteria
495
        r = self.get('/v1/%s/%s?meta=Quality,Stock' % (self.user, cname))
508
        r = self.get('%s?meta=Quality,Stock' % container_url)
496 509
        self.assertEqual(r.status_code, 200)
497 510
        objects = r.content.split('\n')
498 511
        if '' in objects:
499 512
            objects.remove('')
500 513
        self.assertEqual(objects, [oname])
501 514

  
502
    def tearDown(self):
503
        pass
504

  
505 515
    def test_list_using_meta(self):
506 516
        # update object meta
507 517
        cname = 'apples'
518
        container_url = join_urls(self.pithos_path, self.user, cname)
519

  
508 520
        oname1 = self.objects[cname].keys().pop()
509
        self.post('/v1/%s/%s/%s' % (self.user, cname, oname1),
510
                  content_type='', HTTP_X_OBJECT_META_QUALITY='aaa')
521
        url = join_urls(container_url, oname1)
522
        self.post(url, content_type='', HTTP_X_OBJECT_META_QUALITY='aaa')
523

  
511 524
        oname2 = self.objects[cname].keys().pop()
512
        self.post('/v1/%s/%s/%s' % (self.user, cname, oname2),
513
                  content_type='', HTTP_X_OBJECT_META_QUALITY='ab')
525
        url = join_urls(container_url, cname, oname2)
526
        self.post(url, content_type='', HTTP_X_OBJECT_META_QUALITY='ab')
514 527

  
515 528
        oname3 = self.objects[cname].keys().pop()
516
        self.post('/v1/%s/%s/%s' % (self.user, cname, oname3),
517
                  content_type='', HTTP_X_OBJECT_META_STOCK='100')
529
        url = join_urls(container_url, oname3)
530
        self.post(url, content_type='', HTTP_X_OBJECT_META_STOCK='100')
531

  
518 532
        oname4 = self.objects[cname].keys().pop()
519
        self.post('/v1/%s/%s/%s' % (self.user, cname, oname4),
520
                  content_type='', HTTP_X_OBJECT_META_STOCK='200')
533
        url = join_urls(container_url, oname4)
534
        self.post(url, content_type='', HTTP_X_OBJECT_META_STOCK='200')
521 535

  
522 536
        # test multiple existence criteria matches
523
        r = self.get('/v1/%s/%s?meta=Quality,Stock' % (self.user, cname))
537
        r = self.get('%s?meta=Quality,Stock' % container_url)
524 538
        self.assertEqual(r.status_code, 200)
525 539
        objects = r.content.split('\n')
526 540
        if '' in objects:
......
528 542
        self.assertTrue(objects, sorted([oname1, oname2, oname3, oname4]))
529 543

  
530 544
        # list objects that satisfy the existence criteria
531
        r = self.get('/v1/%s/%s?meta=Stock' % (self.user, cname))
545
        r = self.get('%s?meta=Stock' % container_url)
532 546
        self.assertEqual(r.status_code, 200)
533 547
        objects = r.content.split('\n')
534 548
        if '' in objects:
......
536 550
        self.assertTrue(objects, sorted([oname3, oname4]))
537 551

  
538 552
        # test case insensitive existence criteria matching
539
        r = self.get('/v1/%s/%s?meta=quality' % (self.user, cname))
553
        r = self.get('%s?meta=quality' % container_url)
540 554
        self.assertEqual(r.status_code, 200)
541 555
        objects = r.content.split('\n')
542 556
        if '' in objects:
......
544 558
        self.assertTrue(objects, sorted([oname1, oname2]))
545 559

  
546 560
        # test do not all existencecriteria match
547
        r = self.get('/v1/%s/%s?meta=Quality,Foo' % (self.user, cname))
561
        r = self.get('%s?meta=Quality,Foo' % container_url)
548 562
        self.assertEqual(r.status_code, 200)
549 563
        objects = r.content.split('\n')
550 564
        if '' in objects:
......
552 566
        self.assertTrue(objects, sorted([oname1, oname2]))
553 567

  
554 568
        # test equals criteria
555
        r = self.get('/v1/%s/%s?meta=%s' % (self.user, cname,
556
                     quote('Quality=aaa')))
569
        r = self.get('%s?meta=%s' % (container_url, quote('Quality=aaa')))
557 570
        self.assertEqual(r.status_code, 200)
558 571
        objects = r.content.split('\n')
559 572
        if '' in objects:
......
561 574
        self.assertTrue(objects, [oname1])
562 575

  
563 576
        # test not equals criteria
564
        r = self.get('/v1/%s/%s?meta=%s' % (self.user, cname,
565
                     urlencode('Quality!=aaa')))
577
        r = self.get('%s?meta=%s' % (container_url, urlencode('Quality!=aaa')))
566 578
        self.assertEqual(r.status_code, 200)
567 579
        objects = r.content.split('\n')
568 580
        if '' in objects:
......
570 582
        self.assertTrue(objects, [oname2])
571 583

  
572 584
        # test lte criteria
573
        r = self.get('/v1/%s/%s?meta=%s' % (self.user, cname,
574
                     urlencode('Stock<=120')))
585
        r = self.get('%s?meta=%s' % (container_url, urlencode('Stock<=120')))
575 586
        self.assertEqual(r.status_code, 200)
576 587
        objects = r.content.split('\n')
577 588
        if '' in objects:
......
579 590
        self.assertTrue(objects, [oname3])
580 591

  
581 592
        # test gte criteria
582
        r = self.get('/v1/%s/%s?meta=%s' % (self.user, cname,
583
                     urlencode('Stock>=200')))
593
        r = self.get('%s?meta=%s' % (container_url, urlencode('Stock>=200')))
584 594
        self.assertEqual(r.status_code, 200)
585 595
        objects = r.content.split('\n')
586 596
        if '' in objects:
......
626 636
#        since = now + datetime.timedelta(1)
627 637
#
628 638
#        for f in DATE_FORMATS:
629
#            obj = self.client.list_objects(self.container[0],
630
#                                           if_unmodified_since=since.strftime(f))
639
#            obj = self.client.list_objects(
640
#                self.container[0], if_unmodified_since=since.strftime(f))
631 641
#
632 642
#            #assert unmodified
633 643
#            self.assertEqual(obj, self.client.list_objects(self.container[0]))
......
676 686
#
677 687
#        args = [self.containers[0], 'o1']
678 688
#        kwargs = {'length':101}
679
#        self.assert_raises_fault(413, self.upload_random_data, *args, **kwargs)
689
#        self.assert_raises_fault(
690
#            413, self.upload_random_data, *args, **kwargs)
680 691
#
681 692
#        #reset quota
682 693
#        policy = {'quota':0}

Also available in: Unified diff