Statistics
| Branch: | Tag: | Revision:

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

History | View | Annotate | Download (26.6 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
import django.utils.simplejson as json
42
from django.http import urlencode
43

    
44
from xml.dom import minidom
45
from urllib import quote
46

    
47
import random
48
import datetime
49

    
50

    
51
class ContainerHead(PithosAPITest):
52
    def test_get_meta(self):
53
        self.create_container('apples')
54

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

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

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

    
82

    
83
class ContainerGet(PithosAPITest):
84
    def setUp(self):
85
        PithosAPITest.setUp(self)
86

    
87
        self.cnames = ['pears', 'apples']
88
        self.objects = {}
89
        for c in self.cnames:
90
            self.create_container(c)
91

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

    
101
    def test_list_shared(self):
102
        # share an object
103
        cname = self.cnames[0]
104
        onames = self.objects[cname].keys()
105
        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
        self.assertEqual(r.status_code, 202)
110

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

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

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

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

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

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

    
177
    def test_list_public(self):
178
        # publish an object
179
        cname = self.cnames[0]
180
        onames = self.objects[cname].keys()
181
        oname = onames.pop()
182
        r = self.post('/v1/%s/%s/%s' % (self.user, cname, oname),
183
                      content_type='',
184
                      HTTP_X_OBJECT_PUBLIC='true')
185
        self.assertEqual(r.status_code, 202)
186

    
187
        # share another
188
        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')
192
        self.assertEqual(r.status_code, 202)
193

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

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

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

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

    
234
        # 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')
237
        self.assertEqual(r.status_code, 200)
238
        try:
239
            objects = json.loads(r.content)
240
        except:
241
            self.fail('json format expected')
242
        # TODO
243
        #self.assertEqual([oname], [o['name'] for o in objects])
244
        self.assertTrue('x_object_sharing' in objects[0])
245
        # assert public is not returned though
246
        self.assertTrue('x_object_public' not in objects[0])
247

    
248
        # create child object
249
        descendant = strnextling(oname)
250
        self.upload_object(cname, descendant)
251
        # request public and assert child obejct is not listed
252
        r = self.get('/v1/%s/%s?public=' % (self.user, cname))
253
        objects = r.content.split('\n')
254
        objects.remove('')
255
        self.assertEqual(r.status_code, 200)
256
        self.assertTrue(oname in objects)
257
        (self.assertTrue(o not in objects) for o in o_names[1:])
258

    
259
        # test folder inheritance
260
        oname, _ = self.create_folder(cname, HTTP_X_OBJECT_PUBLIC='true')
261
        # create child object
262
        descendant = '%s/%s' % (oname, get_random_word(8))
263
        self.upload_object(cname, descendant)
264
        # request public
265
        r = self.get('/v1/%s/%s?public=' % (self.user, cname))
266
        self.assertEqual(r.status_code, 200)
267
        objects = r.content.split('\n')
268
        self.assertTrue(oname in objects)
269
        self.assertTrue(descendant not in objects)
270

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

    
360
    def test_list_objects_containing_slash(self):
361
        self.create_container('test')
362
        self.upload_object('test', '/objectname')
363

    
364
        r = self.get('/v1/%s/test' % self.user)
365
        objects = r.content.split('\n')
366
        if '' in objects:
367
            objects.remove('')
368
        self.assertEqual(objects, ['/objectname'])
369

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

    
377
        r = self.get('/v1/%s/test?format=xml' % self.user)
378
        try:
379
            objects = minidom.parseString(r.content)
380
        except:
381
            self.fail('xml format expected')
382
        self.assertEqual(
383
            [n.firstChild.data for n in objects.getElementsByTagName('name')],
384
            ['/objectname'])
385

    
386
    def test_list_objects_with_limit_marker(self):
387
        cname = self.cnames[0]
388
        r = self.get('/v1/%s/%s?limit=qwert' % (self.user, cname))
389
        self.assertTrue(r.status_code != 500)
390

    
391
        r = self.get('/v1/%s/%s?limit=2' % (self.user, cname))
392
        self.assertEqual(r.status_code, 200)
393
        objects = r.content.split('\n')
394
        if '' in objects:
395
            objects.remove('')
396

    
397
        onames = sorted(self.objects[cname].keys())
398
        self.assertEqual(objects, onames[:2])
399

    
400
        markers = ['How To Win Friends And Influence People.pdf',
401
                   'moms_birthday.jpg']
402
        limit = 4
403
        for m in markers:
404
            r = self.get('/v1/%s/%s?limit=%s&marker=%s' % (
405
                self.user, cname, limit, m))
406
            objects = r.content.split('\n')
407
            if '' in objects:
408
                objects.remove('')
409
            start = onames.index(m) + 1
410
            end = start + limit
411
            end = end if len(onames) >= end else len(onames)
412
            self.assertEqual(objects, onames[start:end])
413

    
414
    @pithos_test_settings(API_LIST_LIMIT=10)
415
    def test_list_limit_exceeds(self):
416
        self.create_container('container')
417

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

    
421
        r = self.get('/v1/%s/container?format=json' % self.user)
422
        try:
423
            objects = json.loads(r.content)
424
        except:
425
            self.fail('json format expected')
426
        self.assertEqual(pithos_settings.API_LIST_LIMIT,
427
                         len(objects))
428

    
429
    def test_list_pseudo_hierarchical_folders(self):
430
        r = self.get('/v1/%s/apples?prefix=photos&delimiter=/' % self.user)
431
        self.assertEqual(r.status_code, 200)
432
        objects = r.content.split('\n')
433
        if '' in objects:
434
            objects.remove('')
435
        self.assertEquals(
436
            ['photos/animals/', 'photos/me.jpg', 'photos/plants/'],
437
            objects)
438

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

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

    
453
    def test_extended_list_json(self):
454
        params = {'format': 'json', 'limit': 2, 'prefix': 'photos/animals',
455
                  'delimiter': '/'}
456
        r = self.get('/v1/%s/apples?%s' % (self.user, urlencode(params)))
457
        self.assertEqual(r.status_code, 200)
458
        try:
459
            objects = json.loads(r.content)
460
        except:
461
            self.fail('json format expected')
462
        self.assertEqual(objects[0]['subdir'], 'photos/animals/cats/')
463
        self.assertEqual(objects[1]['subdir'], 'photos/animals/dogs/')
464

    
465
    def test_extended_list_xml(self):
466
        params = {'format': 'xml', 'limit': 4, 'prefix': 'photos',
467
                  'delimiter': '/'}
468
        r = self.get('/v1/%s/apples?%s' % (self.user, urlencode(params)))
469
        self.assertEqual(r.status_code, 200)
470
        try:
471
            xml = minidom.parseString(r.content)
472
        except:
473
            self.fail('xml format expected')
474
        self.assert_extended(xml, 'xml', 'object', size=4)
475
        dirs = xml.getElementsByTagName('subdir')
476
        self.assertEqual(len(dirs), 2)
477
        self.assertEqual(dirs[0].attributes['name'].value, 'photos/animals/')
478
        self.assertEqual(dirs[1].attributes['name'].value, 'photos/plants/')
479

    
480
        objects = xml.getElementsByTagName('name')
481
        self.assertEqual(len(objects), 1)
482
        self.assertEqual(objects[0].childNodes[0].data, 'photos/me.jpg')
483

    
484
    def test_list_meta_double_matching(self):
485
        # update object meta
486
        cname = 'apples'
487
        oname = self.objects[cname].keys().pop()
488
        meta = {'quality': 'aaa', 'stock': 'true'}
489
        headers = dict(('HTTP_X_OBJECT_META_%s' % k.upper(), v)
490
                       for k, v in meta.iteritems())
491
        self.post('/v1/%s/%s/%s' % (self.user, cname, oname),
492
                  content_type='', **headers)
493

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

    
502
    def tearDown(self):
503
        pass
504

    
505
    def test_list_using_meta(self):
506
        # update object meta
507
        cname = 'apples'
508
        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')
511
        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')
514

    
515
        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')
518
        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')
521

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

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

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

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

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

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

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

    
581
        # test gte criteria
582
        r = self.get('/v1/%s/%s?meta=%s' % (self.user, cname,
583
                     urlencode('Stock>=200')))
584
        self.assertEqual(r.status_code, 200)
585
        objects = r.content.split('\n')
586
        if '' in objects:
587
            objects.remove('')
588
        self.assertTrue(objects, [oname4])
589

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