Statistics
| Branch: | Tag: | Revision:

root / snf-pithos-app / pithos / api / test / containers.py @ d6a92fa0

History | View | Annotate | Download (26.9 kB)

1
#!/usr/bin/env python
2
#coding=utf8
3

    
4
# Copyright 2011-2013 GRNET S.A. All rights reserved.
5
#
6
# Redistribution and use in source and binary forms, with or
7
# without modification, are permitted provided that the following
8
# conditions are met:
9
#
10
#   1. Redistributions of source code must retain the above
11
#      copyright notice, this list of conditions and the following
12
#      disclaimer.
13
#
14
#   2. Redistributions in binary form must reproduce the above
15
#      copyright notice, this list of conditions and the following
16
#      disclaimer in the documentation and/or other materials
17
#      provided with the distribution.
18
#
19
# THIS SOFTWARE IS PROVIDED BY GRNET S.A. ``AS IS'' AND ANY EXPRESS
20
# OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED
21
# WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR
22
# PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL GRNET S.A OR
23
# CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
24
# SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
25
# LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF
26
# USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED
27
# AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT
28
# LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN
29
# ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
30
# POSSIBILITY OF SUCH DAMAGE.
31
#
32
# The views and conclusions contained in the software and
33
# documentation are those of the authors and should not be
34
# interpreted as representing official policies, either expressed
35
# or implied, of GRNET S.A.
36

    
37
from pithos.api.test import PithosAPITest, DATE_FORMATS, o_names,\
38
    strnextling, pithos_settings, pithos_test_settings
39
from pithos.backends.random_word import get_random_word
40

    
41
from synnefo.lib import join_urls
42

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

    
46
from xml.dom import minidom
47
from urllib import quote
48

    
49
import random
50
import datetime
51

    
52

    
53
class ContainerHead(PithosAPITest):
54
    def test_get_meta(self):
55
        self.create_container('apples')
56

    
57
        # populate with objects
58
        objects = {}
59
        for i in range(random.randint(1, 100)):
60

    
61
            # upload object
62
            meta = {'foo%s' % i: 'bar'}
63
            name, data, resp = self.upload_object('apples', **meta)
64
            objects[name] = data
65

    
66
        t1 = datetime.datetime.utcnow()
67
        url = join_urls(self.pithos_path, self.user, 'apples')
68
        r = self.head(url)
69
        self.assertEqual(int(r['X-Container-Object-Count']), len(objects))
70
        self.assertEqual(int(r['X-Container-Bytes-Used']),
71
                         sum([len(i) for i in objects.values()]))
72
        self.assertTrue('X-Container-Block-Size' in r)
73
        self.assertTrue('X-Container-Block-Hash' in r)
74
        self.assertTrue('X-Container-Until-Timestamp' not in r)
75
        self.assertEqual(r['X-Container-Policy-Versioning'], 'auto')
76
        self.assertEqual(int(r['X-Container-Policy-Quota']), 0)
77
        t2 = datetime.datetime.strptime(r['Last-Modified'], DATE_FORMATS[2])
78
        delta = (t2 - t1)
79
        threashold = datetime.timedelta(seconds=1)
80
        self.assertTrue(delta < threashold)
81
        self.assertTrue(r['X-Container-Object-Meta'])
82
        (self.assertTrue('foo%s' % i in r['X-Container-Object-Meta'])
83
            for i in range(len(objects)))
84

    
85

    
86
class ContainerGet(PithosAPITest):
87
    def setUp(self):
88
        PithosAPITest.setUp(self)
89

    
90
        self.cnames = ['pears', 'apples']
91
        self.objects = {}
92
        for c in self.cnames:
93
            self.create_container(c)
94

    
95
        self.objects['pears'] = {}
96
        for o in o_names[:8]:
97
            name, data, resp = self.upload_object('pears', o)
98
            self.objects['pears'][name] = data
99
        self.objects['apples'] = {}
100
        for o in o_names[8:]:
101
            name, data, resp = self.upload_object('apples', o)
102
            self.objects['apples'][name] = data
103

    
104
    def test_list_shared(self):
105
        # share an object
106
        cname = self.cnames[0]
107
        onames = self.objects[cname].keys()
108
        oname = onames.pop()
109
        url = join_urls(self.pithos_path, self.user, cname, oname)
110
        r = self.post(url, content_type='', HTTP_X_OBJECT_SHARING='read=*')
111
        self.assertEqual(r.status_code, 202)
112

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

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

    
127
        # list detailed shared and assert only the shared object is returned
128
        url = join_urls(self.pithos_path, self.user, cname)
129
        r = self.get('%s?shared=&format=json' % url)
130
        self.assertEqual(r.status_code, 200)
131
        try:
132
            objects = json.loads(r.content)
133
        except:
134
            self.fail('json format expected')
135
        self.assertEqual([oname], [o['name'] for o in objects])
136
        self.assertTrue('x_object_sharing' in objects[0])
137
        self.assertTrue('x_object_public' not in objects[0])
138

    
139
        # publish the shared object and assert it is still listed in the
140
        # shared objects
141
        url = join_urls(self.pithos_path, self.user, cname, oname)
142
        r = self.post(url, content_type='', HTTP_X_OBJECT_PUBLIC='true')
143
        self.assertEqual(r.status_code, 202)
144
        url = join_urls(self.pithos_path, self.user, cname)
145
        r = self.get('%s?shared=&format=json' % url)
146
        self.assertEqual(r.status_code, 200)
147
        try:
148
            objects = json.loads(r.content)
149
        except:
150
            self.fail('json format expected')
151
        self.assertEqual([oname], [o['name'] for o in objects])
152
        self.assertTrue('x_object_sharing' in objects[0])
153
        # TODO
154
        #self.assertTrue('x_object_public' in objects[0])
155

    
156
        # create child object
157
        descendant = strnextling(oname)
158
        self.upload_object(cname, descendant)
159
        # request shared and assert child obejct is not listed
160
        url = join_urls(self.pithos_path, self.user, cname)
161
        r = self.get('%s?shared=' % url)
162
        self.assertEqual(r.status_code, 200)
163
        objects = r.content.split('\n')
164
        objects.remove('')
165
        self.assertTrue(oname in objects)
166
        self.assertTrue(descendant not in objects)
167

    
168
        # check folder inheritance
169
        oname, _ = self.create_folder(cname, HTTP_X_OBJECT_SHARING='read=*')
170
        # create child object
171
        descendant = '%s/%s' % (oname, get_random_word(8))
172
        self.upload_object(cname, descendant)
173
        # request shared
174
        url = join_urls(self.pithos_path, self.user, cname)
175
        r = self.get('%s?shared=' % url)
176
        self.assertEqual(r.status_code, 200)
177
        objects = r.content.split('\n')
178
        objects.remove('')
179
        self.assertTrue(oname in objects)
180
        self.assertTrue(descendant in objects)
181

    
182
    def test_list_public(self):
183
        # publish an object
184
        cname = self.cnames[0]
185
        onames = self.objects[cname].keys()
186
        oname = onames.pop()
187
        url = join_urls(self.pithos_path, self.user, cname, oname)
188
        r = self.post(url, content_type='', HTTP_X_OBJECT_PUBLIC='true')
189
        self.assertEqual(r.status_code, 202)
190

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

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

    
205
        # list detailed public and assert only the public object is returned
206
        url = join_urls(self.pithos_path, self.user, cname)
207
        r = self.get('%s?public=&format=json' % url)
208
        self.assertEqual(r.status_code, 200)
209
        try:
210
            objects = json.loads(r.content)
211
        except:
212
            self.fail('json format expected')
213
        self.assertEqual([oname], [o['name'] for o in objects])
214
        self.assertTrue('x_object_sharing' not in objects[0])
215
        self.assertTrue('x_object_public' in objects[0])
216

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

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

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

    
240
        # Assert listing the container public contents to shared users
241
        r = self.get('%s?public=&format=json' % url, user='alice')
242
        self.assertEqual(r.status_code, 200)
243
        try:
244
            objects = json.loads(r.content)
245
        except:
246
            self.fail('json format expected')
247
        # TODO
248
        #self.assertEqual([oname], [o['name'] for o in objects])
249
        self.assertTrue('x_object_sharing' in objects[0])
250
        # assert public is not returned though
251
        self.assertTrue('x_object_public' not in objects[0])
252

    
253
        # create child object
254
        descendant = strnextling(oname)
255
        self.upload_object(cname, descendant)
256
        # request public and assert child obejct is not listed
257
        r = self.get('%s?public=' % url)
258
        objects = r.content.split('\n')
259
        objects.remove('')
260
        self.assertEqual(r.status_code, 200)
261
        self.assertTrue(oname in objects)
262
        (self.assertTrue(o not in objects) for o in o_names[1:])
263

    
264
        # test folder inheritance
265
        oname, _ = self.create_folder(cname, HTTP_X_OBJECT_PUBLIC='true')
266
        # create child object
267
        descendant = '%s/%s' % (oname, get_random_word(8))
268
        self.upload_object(cname, descendant)
269
        # request public
270
        r = self.get('%s?public=' % url)
271
        self.assertEqual(r.status_code, 200)
272
        objects = r.content.split('\n')
273
        self.assertTrue(oname in objects)
274
        self.assertTrue(descendant not in objects)
275

    
276
#    def test_list_shared_public(self):
277
#        # publish an object
278
#        cname = self.cnames[0]
279
#        onames = self.objects[cname].keys()
280
#        oname = onames.pop()
281
#        r = self.post('/v1/%s/%s/%s' % (self.user, cname, oname),
282
#                      content_type='',
283
#                      HTTP_X_OBJECT_PUBLIC='true')
284
#        self.assertEqual(r.status_code, 202)
285
#
286
#        # share another
287
#        other = onames.pop()
288
#        r = self.post('/v1/%s/%s/%s' % (self.user, cname, other),
289
#                      content_type='',
290
#                      HTTP_X_OBJECT_SHARING='read=alice')
291
#        self.assertEqual(r.status_code, 202)
292
#
293
#        # list shared and public objects and assert object is listed
294
#        r = self.get('/v1/%s/%s?shared=&public=&format=json' % (
295
#            self.user, cname))
296
#        self.assertEqual(r.status_code, 200)
297
#        objects = json.loads(r.content)
298
#        self.assertEqual([o['name'] for o in objects], sorted([oname, other]))
299
#        for o in objects:
300
#            if o['name'] == oname:
301
#                self.assertTrue('x_object_public' in objects[0])
302
#            elif o['name'] == other:
303
#                self.assertTrue('x_object_sharing' in objects[1])
304
#
305
#        # assert not listing shared and public to a not shared user
306
#        r = self.get('/v1/%s/%s?shared=&public=&format=json' % (
307
#            self.user, cname), user='bob')
308
#        self.assertEqual(r.status_code, 403)
309
#
310
#        # assert listing shared and public to a shared user
311
#        r = self.get('/v1/%s/%s?shared=&public=&format=json' % (
312
#            self.user, cname), user='alice')
313
#        self.assertEqual(r.status_code, 200)
314
#        try:
315
#            objects = json.loads(r.content)
316
#        except:
317
#            self.fail('json format expected')
318
#        self.assertEqual([o['name'] for o in objects], sorted([oname, other]))
319
#
320
#        # create child object
321
#        descentant1 = strnextling(oname)
322
#        self.upload_object(cname, descendant1)
323
#        descentant2 = strnextling(other)
324
#        self.upload_object(cname, descendant2)
325
#        r = self.get('/v1/%s/%s?shared=&public=&format=json' % (
326
#            self.user, cname), user='alice')
327
#        self.assertEqual(r.status_code, 200)
328
#        try:
329
#            objects = json.loads(r.content)
330
#        except:
331
#            self.fail('json format expected')
332
#        self.assertEqual([o['name'] for o in objects], [oname])
333
#
334
#        # test inheritance
335
#        oname1, _ = self.create_folder(cname,
336
#                                       HTTP_X_OBJECT_SHARING='read=alice')
337
#        # create child object
338
#        descendant1 = '%s/%s' % (oname, get_random_word(8))
339
#        self.upload_object(cname, descendant1)
340
#
341
#        oname2, _ = self.create_folder(cname,
342
#                                       HTTP_X_OBJECT_PUBLIC='true')
343
#        # create child object
344
#        descendant2 = '%s/%s' % (oname, get_random_word(8))
345
#        self.upload_object(cname, descendant2)
346
#
347
#
348
#        o = self.upload_random_data(self.container[1], 'folder2/object')
349
#        objs = self.client.list_objects(
350
#            self.container[1], shared=True, public=True)
351
#        self.assertEqual(objs, ['folder1', 'folder1/object', 'folder2'])
352
#        objs = cl.list_objects(
353
#            self.container[1], shared=True, public=True, account=get_user()
354
#        )
355
#        self.assertEqual(objs, ['folder1', 'folder1/object'])
356
#
357
    def test_list_objects(self):
358
        cname = self.cnames[0]
359
        url = join_urls(self.pithos_path, self.user, cname)
360
        r = self.get(url)
361
        self.assertTrue(r.status_code, 200)
362
        objects = r.content.split('\n')
363
        if '' in objects:
364
            objects.remove('')
365
        self.assertEqual(objects, sorted(self.objects[cname].keys()))
366

    
367
    def test_list_objects_containing_slash(self):
368
        self.create_container('test')
369
        self.upload_object('test', '/objectname')
370

    
371
        url = join_urls(self.pithos_path, self.user, 'test')
372

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

    
379
        r = self.get('%s?format=json' % url)
380
        try:
381
            objects = json.loads(r.content)
382
        except:
383
            self.fail('json format expected')
384
        self.assertEqual([o['name'] for o in objects], ['/objectname'])
385

    
386
        r = self.get('%s?format=xml' % url)
387
        try:
388
            objects = minidom.parseString(r.content)
389
        except:
390
            self.fail('xml format expected')
391
        self.assertEqual(
392
            [n.firstChild.data for n in objects.getElementsByTagName('name')],
393
            ['/objectname'])
394

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

    
401
        r = self.get('%s?limit=2' % url)
402
        self.assertEqual(r.status_code, 200)
403
        objects = r.content.split('\n')
404
        if '' in objects:
405
            objects.remove('')
406

    
407
        onames = sorted(self.objects[cname].keys())
408
        self.assertEqual(objects, onames[:2])
409

    
410
        markers = ['How To Win Friends And Influence People.pdf',
411
                   'moms_birthday.jpg']
412
        limit = 4
413
        for m in markers:
414
            r = self.get('%s?limit=%s&marker=%s' % (url, limit, m))
415
            objects = r.content.split('\n')
416
            if '' in objects:
417
                objects.remove('')
418
            start = onames.index(m) + 1
419
            end = start + limit
420
            end = end if len(onames) >= end else len(onames)
421
            self.assertEqual(objects, onames[start:end])
422

    
423
    @pithos_test_settings(API_LIST_LIMIT=10)
424
    def test_list_limit_exceeds(self):
425
        self.create_container('container')
426
        url = join_urls(self.pithos_path, self.user, 'container')
427

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

    
431
        r = self.get('%s?format=json' % url)
432
        try:
433
            objects = json.loads(r.content)
434
        except:
435
            self.fail('json format expected')
436
        self.assertEqual(pithos_settings.API_LIST_LIMIT,
437
                         len(objects))
438

    
439
    def test_list_pseudo_hierarchical_folders(self):
440
        url = join_urls(self.pithos_path, self.user, 'apples')
441
        r = self.get('%s?prefix=photos&delimiter=/' % url)
442
        self.assertEqual(r.status_code, 200)
443
        objects = r.content.split('\n')
444
        if '' in objects:
445
            objects.remove('')
446
        self.assertEquals(
447
            ['photos/animals/', 'photos/me.jpg', 'photos/plants/'],
448
            objects)
449

    
450
        r = self.get('%s?prefix=photos/animals&delimiter=/' % url)
451
        objects = r.content.split('\n')
452
        if '' in objects:
453
            objects.remove('')
454
        self.assertEquals(
455
            ['photos/animals/cats/', 'photos/animals/dogs/'], objects)
456

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

    
463
    def test_extended_list_json(self):
464
        url = join_urls(self.pithos_path, self.user, 'apples')
465
        params = {'format': 'json', 'limit': 2, 'prefix': 'photos/animals',
466
                  'delimiter': '/'}
467
        r = self.get('%s?%s' % (url, urlencode(params)))
468
        self.assertEqual(r.status_code, 200)
469
        try:
470
            objects = json.loads(r.content)
471
        except:
472
            self.fail('json format expected')
473
        self.assertEqual(objects[0]['subdir'], 'photos/animals/cats/')
474
        self.assertEqual(objects[1]['subdir'], 'photos/animals/dogs/')
475

    
476
    def test_extended_list_xml(self):
477
        url = join_urls(self.pithos_path, self.user, 'apples')
478
        params = {'format': 'xml', 'limit': 4, 'prefix': 'photos',
479
                  'delimiter': '/'}
480
        r = self.get('%s?%s' % (url, urlencode(params)))
481
        self.assertEqual(r.status_code, 200)
482
        try:
483
            xml = minidom.parseString(r.content)
484
        except:
485
            self.fail('xml format expected')
486
        self.assert_extended(xml, 'xml', 'object', size=4)
487
        dirs = xml.getElementsByTagName('subdir')
488
        self.assertEqual(len(dirs), 2)
489
        self.assertEqual(dirs[0].attributes['name'].value, 'photos/animals/')
490
        self.assertEqual(dirs[1].attributes['name'].value, 'photos/plants/')
491

    
492
        objects = xml.getElementsByTagName('name')
493
        self.assertEqual(len(objects), 1)
494
        self.assertEqual(objects[0].childNodes[0].data, 'photos/me.jpg')
495

    
496
    def test_list_meta_double_matching(self):
497
        # update object meta
498
        cname = 'apples'
499
        container_url = join_urls(self.pithos_path, self.user, cname)
500
        oname = self.objects[cname].keys().pop()
501
        meta = {'quality': 'aaa', 'stock': 'true'}
502
        headers = dict(('HTTP_X_OBJECT_META_%s' % k.upper(), v)
503
                       for k, v in meta.iteritems())
504
        object_url = join_urls(container_url, oname)
505
        self.post(object_url, content_type='', **headers)
506

    
507
        # list objects that satisfy the criteria
508
        r = self.get('%s?meta=Quality,Stock' % container_url)
509
        self.assertEqual(r.status_code, 200)
510
        objects = r.content.split('\n')
511
        if '' in objects:
512
            objects.remove('')
513
        self.assertEqual(objects, [oname])
514

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

    
520
        oname1 = self.objects[cname].keys().pop()
521
        url = join_urls(container_url, oname1)
522
        self.post(url, content_type='', HTTP_X_OBJECT_META_QUALITY='aaa')
523

    
524
        oname2 = self.objects[cname].keys().pop()
525
        url = join_urls(container_url, cname, oname2)
526
        self.post(url, content_type='', HTTP_X_OBJECT_META_QUALITY='ab')
527

    
528
        oname3 = self.objects[cname].keys().pop()
529
        url = join_urls(container_url, oname3)
530
        self.post(url, content_type='', HTTP_X_OBJECT_META_STOCK='100')
531

    
532
        oname4 = self.objects[cname].keys().pop()
533
        url = join_urls(container_url, oname4)
534
        self.post(url, content_type='', HTTP_X_OBJECT_META_STOCK='200')
535

    
536
        # test multiple existence criteria matches
537
        r = self.get('%s?meta=Quality,Stock' % container_url)
538
        self.assertEqual(r.status_code, 200)
539
        objects = r.content.split('\n')
540
        if '' in objects:
541
            objects.remove('')
542
        self.assertTrue(objects, sorted([oname1, oname2, oname3, oname4]))
543

    
544
        # list objects that satisfy the existence criteria
545
        r = self.get('%s?meta=Stock' % container_url)
546
        self.assertEqual(r.status_code, 200)
547
        objects = r.content.split('\n')
548
        if '' in objects:
549
            objects.remove('')
550
        self.assertTrue(objects, sorted([oname3, oname4]))
551

    
552
        # test case insensitive existence criteria matching
553
        r = self.get('%s?meta=quality' % container_url)
554
        self.assertEqual(r.status_code, 200)
555
        objects = r.content.split('\n')
556
        if '' in objects:
557
            objects.remove('')
558
        self.assertTrue(objects, sorted([oname1, oname2]))
559

    
560
        # test do not all existencecriteria match
561
        r = self.get('%s?meta=Quality,Foo' % container_url)
562
        self.assertEqual(r.status_code, 200)
563
        objects = r.content.split('\n')
564
        if '' in objects:
565
            objects.remove('')
566
        self.assertTrue(objects, sorted([oname1, oname2]))
567

    
568
        # test equals criteria
569
        r = self.get('%s?meta=%s' % (container_url, quote('Quality=aaa')))
570
        self.assertEqual(r.status_code, 200)
571
        objects = r.content.split('\n')
572
        if '' in objects:
573
            objects.remove('')
574
        self.assertTrue(objects, [oname1])
575

    
576
        # test not equals criteria
577
        r = self.get('%s?meta=%s' % (container_url, urlencode('Quality!=aaa')))
578
        self.assertEqual(r.status_code, 200)
579
        objects = r.content.split('\n')
580
        if '' in objects:
581
            objects.remove()
582
        self.assertTrue(objects, [oname2])
583

    
584
        # test lte criteria
585
        r = self.get('%s?meta=%s' % (container_url, urlencode('Stock<=120')))
586
        self.assertEqual(r.status_code, 200)
587
        objects = r.content.split('\n')
588
        if '' in objects:
589
            objects.remove('')
590
        self.assertTrue(objects, [oname3])
591

    
592
        # test gte criteria
593
        r = self.get('%s?meta=%s' % (container_url, urlencode('Stock>=200')))
594
        self.assertEqual(r.status_code, 200)
595
        objects = r.content.split('\n')
596
        if '' in objects:
597
            objects.remove('')
598
        self.assertTrue(objects, [oname4])
599

    
600
#
601
#    def test_if_modified_since(self):
602
#        t = datetime.datetime.utcnow()
603
#        t2 = t - datetime.timedelta(minutes=10)
604
#
605
#        #add a new object
606
#        self.upload_random_data(self.container[0], o_names[0])
607
#
608
#        for f in DATE_FORMATS:
609
#            past = t2.strftime(f)
610
#            try:
611
#                o = self.client.list_objects(self.container[0],
612
#                                            if_modified_since=past)
613
#                self.assertEqual(o,
614
#                                 self.client.list_objects(self.container[0]))
615
#            except Fault, f:
616
#                self.failIf(f.status == 304) #fail if not modified
617
#
618
#    def test_if_modified_since_invalid_date(self):
619
#        headers = {'if-modified-since':''}
620
#        o = self.client.list_objects(self.container[0], if_modified_since='')
621
#        self.assertEqual(o, self.client.list_objects(self.container[0]))
622
#
623
#    def test_if_not_modified_since(self):
624
#        now = datetime.datetime.utcnow()
625
#        since = now + datetime.timedelta(1)
626
#
627
#        for f in DATE_FORMATS:
628
#            args = {'if_modified_since':'%s' %since.strftime(f)}
629
#
630
#            #assert not modified
631
#            self.assert_raises_fault(304, self.client.list_objects,
632
#                                     self.container[0], **args)
633
#
634
#    def test_if_unmodified_since(self):
635
#        now = datetime.datetime.utcnow()
636
#        since = now + datetime.timedelta(1)
637
#
638
#        for f in DATE_FORMATS:
639
#            obj = self.client.list_objects(
640
#                self.container[0], if_unmodified_since=since.strftime(f))
641
#
642
#            #assert unmodified
643
#            self.assertEqual(obj, self.client.list_objects(self.container[0]))
644
#
645
#    def test_if_unmodified_since_precondition_failed(self):
646
#        t = datetime.datetime.utcnow()
647
#        t2 = t - datetime.timedelta(minutes=10)
648
#
649
#        #add a new container
650
#        self.client.create_container('dummy')
651
#
652
#        for f in DATE_FORMATS:
653
#            past = t2.strftime(f)
654
#
655
#            args = {'if_unmodified_since':'%s' %past}
656
#
657
#            #assert precondition failed
658
#            self.assert_raises_fault(412, self.client.list_objects,
659
#                                     self.container[0], **args)
660
#
661
#class ContainerPut(BaseTestCase):
662
#    def setUp(self):
663
#        BaseTestCase.setUp(self)
664
#        self.containers = list(set(self.initial_containers + ['c1', 'c2']))
665
#        self.containers.sort()
666
#
667
#    def test_create(self):
668
#        self.client.create_container(self.containers[0])
669
#        containers = self.client.list_containers()
670
#        self.assertTrue(self.containers[0] in containers)
671
#        self.assert_container_exists(self.containers[0])
672
#
673
#    def test_create_twice(self):
674
#        self.client.create_container(self.containers[0])
675
#        self.assertTrue(not self.client.create_container(self.containers[0]))
676
#
677
#    def test_quota(self):
678
#        self.client.create_container(self.containers[0])
679
#
680
#        policy = {'quota':100}
681
#        self.client.set_container_policies(self.containers[0], **policy)
682
#
683
#        meta = self.client.retrieve_container_metadata(self.containers[0])
684
#        self.assertTrue('x-container-policy-quota' in meta)
685
#        self.assertEqual(meta['x-container-policy-quota'], '100')
686
#
687
#        args = [self.containers[0], 'o1']
688
#        kwargs = {'length':101}
689
#        self.assert_raises_fault(
690
#            413, self.upload_random_data, *args, **kwargs)
691
#
692
#        #reset quota
693
#        policy = {'quota':0}
694
#        self.client.set_container_policies(self.containers[0], **policy)