Statistics
| Branch: | Tag: | Revision:

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

History | View | Annotate | Download (17.7 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, AssertMappingInvariant,\
38
    DATE_FORMATS
39

    
40
import time as _time
41
import datetime
42

    
43

    
44
class AccountHead(PithosAPITest):
45
    def test_get_account_meta(self):
46
        cnames = ['apples', 'bananas', 'kiwis', 'oranges', 'pears']
47

    
48
        # create containers
49
        uploaded_bytes = 0
50
        for cname in cnames:
51
            self.create_container(cname)
52

    
53
            # upload object
54
            name, data, resp = self.upload_object(cname)
55
            uploaded_bytes += len(data)
56

    
57
        # set account meta
58
        self.update_account_meta({'foo': 'bar'})
59

    
60
        account_info = self.get_account_info()
61
        self.assertTrue('X-Account-Meta-Foo' in account_info)
62
        self.assertEqual(account_info['X-Account-Meta-Foo'], 'bar')
63

    
64
        # list containers
65
        containers = self.list_containers()
66
        self.assertEqual(int(account_info['X-Account-Container-Count']),
67
                         len(containers))
68
        usage = 0
69
        for c in containers:
70
            # list objects
71
            objects = self.list_objects(c['name'])
72
            self.assertEqual(c['count'], len(objects))
73
            csum = sum([o['bytes'] for o in objects])
74
            self.assertEqual(int(c['bytes']), csum)
75
            usage += int(c['bytes'])
76

    
77
        self.assertEqual(
78
            int(account_info['x-account-bytes-used']) + uploaded_bytes,
79
            usage)
80

    
81
    def test_get_account_meta_until(self):
82
        self.update_account_meta({'foo': 'bar'})
83

    
84
        account_info = self.get_account_info()
85
        t = datetime.datetime.strptime(account_info['Last-Modified'],
86
                                       DATE_FORMATS[2])
87
        t1 = t + datetime.timedelta(seconds=1)
88
        until = int(_time.mktime(t1.timetuple()))
89

    
90
        _time.sleep(2)
91
        self.update_account_meta({'quality': 'AAA'})
92

    
93
        account_info = self.get_account_info()
94
        t = datetime.datetime.strptime(account_info['Last-Modified'],
95
                                       DATE_FORMATS[-1])
96
        last_modified = int(_time.mktime(t.timetuple()))
97
        assert until < last_modified
98

    
99
        self.assertTrue('X-Account-Meta-Quality' in account_info)
100
        self.assertTrue('X-Account-Meta-Foo' in account_info)
101

    
102
        account_info = self.get_account_info(until=until)
103
        self.assertTrue('X-Account-Meta-Quality' not in account_info)
104
        self.assertTrue('X-Account-Meta-Foo' in account_info)
105
        self.assertTrue('X-Account-Until-Timestamp' in account_info)
106
        t = datetime.datetime.strptime(
107
            account_info['X-Account-Until-Timestamp'], DATE_FORMATS[2])
108
        self.assertTrue(int(_time.mktime(t1.timetuple())) <= until)
109

    
110
    def test_get_account_meta_until_invalid_date(self):
111
        self.update_account_meta({'quality': 'AAA'})
112
        meta = self.get_account_meta(until='-1')
113
        self.assertTrue('X-Account-Meta-Quality' in meta)
114

    
115

    
116
class AccountGet(PithosAPITest):
117
    def setUp(self):
118
        PithosAPITest.setUp(self)
119
        cnames = ['apples', 'bananas', 'kiwis', 'oranges', 'pears']
120

    
121
        # create containers
122
        uploaded_bytes = 0
123
        for cname in cnames:
124
            self.create_container(cname)
125

    
126
            # upload object
127
            name, data, resp = self.upload_object(cname)
128
            uploaded_bytes += len(data)
129

    
130
    def test_list(self):
131
        #list containers: row format
132
        containers = self.list_containers(format=None)
133
        self.assertEquals(containers,
134
                          ['apples', 'bananas', 'kiwis', 'oranges', 'pears'])
135

    
136
    def test_list_with_limit(self):
137
        containers = self.list_containers(format=None, limit=2)
138
        self.assertEquals(len(containers), 2)
139
        self.assertEquals(containers, ['apples', 'bananas'])
140

    
141
    def test_list_with_marker(self):
142
        containers = self.list_containers(format=None, limit=2,
143
                                          marker='bananas')
144
        self.assertEquals(containers, ['kiwis', 'oranges'])
145

    
146
        containers = self.list_containers(format=None, limit=2,
147
                                          marker='oranges')
148
        self.assertEquals(containers, ['pears'])
149

    
150
    def test_list_json_with_marker(self):
151
        containers = self.list_containers(format='json', limit=2,
152
                                          marker='bananas')
153
        self.assert_extended(containers, 'json', 'container', 2)
154
        self.assertEqual(containers[0]['name'], 'kiwis')
155
        self.assertEqual(containers[1]['name'], 'oranges')
156

    
157
        containers = self.list_containers(format='json', limit=2,
158
                                          marker='oranges')
159
        self.assert_extended(containers, 'json', 'container', 1)
160
        self.assertEqual(containers[0]['name'], 'pears')
161

    
162
    def test_list_xml_with_marker(self):
163
        xml = self.list_containers(format='xml', limit=2, marker='bananas')
164
        self.assert_extended(xml, 'xml', 'container', 2)
165
        nodes = xml.getElementsByTagName('name')
166
        self.assertTrue(len(nodes) <= 2)
167
        names = [n.childNodes[0].data for n in nodes]
168
        self.assertEqual(names, ['kiwis', 'oranges'])
169

    
170
        xml = self.list_containers(format='xml', limit=2, marker='oranges')
171
        self.assert_extended(xml, 'xml', 'container', 1)
172
        nodes = xml.getElementsByTagName('name')
173
        self.assertTrue(len(nodes) <= 2)
174
        names = [n.childNodes[0].data for n in nodes]
175
        self.assertEqual(names, ['pears'])
176

    
177
    def test_if_modified_since(self):
178
        account_info = self.get_account_info()
179
        last_modified = account_info['Last-Modified']
180
        t1 = datetime.datetime.strptime(last_modified, DATE_FORMATS[-1])
181
        t1_formats = map(t1.strftime, DATE_FORMATS)
182

    
183
        # Check not modified
184
        for t in t1_formats:
185
            r = self.get('/v1/%s' % self.user, HTTP_IF_MODIFIED_SINCE=t)
186
            self.assertEqual(r.status_code, 304)
187

    
188
        # modify account: add container
189
        _time.sleep(1)
190
        self.create_container('c1')
191

    
192
        # Check modified
193
        for t in t1_formats:
194
            r = self.get('/v1/%s' % self.user, HTTP_IF_MODIFIED_SINCE=t)
195
            self.assertEqual(r.status_code, 200)
196
            self.assertEqual(
197
                r.content.split('\n')[:-1],
198
                ['apples', 'bananas', 'c1', 'kiwis', 'oranges', 'pears'])
199

    
200
        account_info = self.get_account_info()
201
        last_modified = account_info['Last-Modified']
202
        t2 = datetime.datetime.strptime(last_modified, DATE_FORMATS[-1])
203
        t2_formats = map(t2.strftime, DATE_FORMATS)
204

    
205
        # modify account: update account meta
206
        _time.sleep(1)
207
        self.update_account_meta({'foo': 'bar'})
208

    
209
        # Check modified
210
        for t in t2_formats:
211
            r = self.get('/v1/%s' % self.user, HTTP_IF_MODIFIED_SINCE=t)
212
            self.assertEqual(r.status_code, 200)
213
            self.assertEqual(
214
                r.content.split('\n')[:-1],
215
                ['apples', 'bananas', 'c1', 'kiwis', 'oranges', 'pears'])
216

    
217
    def test_if_modified_since_invalid_date(self):
218
        r = self.get('/v1/%s' % self.user, HTTP_IF_MODIFIED_SINCE='Monday')
219
        self.assertEqual(r.status_code, 200)
220
        self.assertEqual(
221
            r.content.split('\n')[:-1],
222
            ['apples', 'bananas', 'kiwis', 'oranges', 'pears'])
223

    
224
    def test_if_not_modified_since(self):
225
        account_info = self.get_account_info()
226
        last_modified = account_info['Last-Modified']
227
        t = datetime.datetime.strptime(last_modified, DATE_FORMATS[-1])
228

    
229
        # Check unmodified
230
        t1 = t + datetime.timedelta(seconds=1)
231
        t1_formats = map(t1.strftime, DATE_FORMATS)
232
        for t in t1_formats:
233
            r = self.get('/v1/%s' % self.user, HTTP_IF_UNMODIFIED_SINCE=t)
234
            self.assertEqual(r.status_code, 200)
235
            self.assertEqual(
236
                r.content.split('\n')[:-1],
237
                ['apples', 'bananas', 'kiwis', 'oranges', 'pears'])
238

    
239
        # modify account: add container
240
        _time.sleep(2)
241
        self.create_container('c1')
242

    
243
        account_info = self.get_account_info()
244
        last_modified = account_info['Last-Modified']
245
        t = datetime.datetime.strptime(last_modified, DATE_FORMATS[-1])
246
        t2 = t - datetime.timedelta(seconds=1)
247
        t2_formats = map(t2.strftime, DATE_FORMATS)
248

    
249
        # Check modified
250
        for t in t2_formats:
251
            r = self.get('/v1/%s' % self.user, HTTP_IF_UNMODIFIED_SINCE=t)
252
            self.assertEqual(r.status_code, 412)
253

    
254
        # modify account: update account meta
255
        _time.sleep(1)
256
        self.update_account_meta({'foo': 'bar'})
257

    
258
        account_info = self.get_account_info()
259
        last_modified = account_info['Last-Modified']
260
        t = datetime.datetime.strptime(last_modified, DATE_FORMATS[-1])
261
        t3 = t - datetime.timedelta(seconds=1)
262
        t3_formats = map(t3.strftime, DATE_FORMATS)
263

    
264
        # Check modified
265
        for t in t3_formats:
266
            r = self.get('/v1/%s' % self.user, HTTP_IF_UNMODIFIED_SINCE=t)
267
            self.assertEqual(r.status_code, 412)
268

    
269
    def test_if_unmodified_since_invalid_date(self):
270
        r = self.get('/v1/%s' % self.user, HTTP_IF_UNMODIFIED_SINCE='Monday')
271
        self.assertEqual(r.status_code, 200)
272
        self.assertEqual(
273
            r.content.split('\n')[:-1],
274
            ['apples', 'bananas', 'kiwis', 'oranges', 'pears'])
275

    
276

    
277
class AccountPost(PithosAPITest):
278
    def setUp(self):
279
        PithosAPITest.setUp(self)
280
        cnames = ['apples', 'bananas', 'kiwis', 'oranges', 'pears']
281

    
282
        # create containers
283
        uploaded_bytes = 0
284
        for cname in cnames:
285
            self.create_container(cname)
286

    
287
            # upload object
288
            name, data, resp = self.upload_object(cname)
289
            uploaded_bytes += len(data)
290

    
291
        # set account meta
292
        self.update_account_meta({'foo': 'bar'})
293

    
294
    def test_update_meta(self):
295
        with AssertMappingInvariant(self.get_account_groups):
296
            initial = self.get_account_meta()
297

    
298
            meta = {'test': 'tost', 'ping': 'pong'}
299
            kwargs = dict(('HTTP_X_ACCOUNT_META_%s' % k, str(v))
300
                          for k, v in meta.items())
301
            r = self.post('/v1/%s?update=' % self.user, **kwargs)
302
            self.assertEqual(r.status_code, 202)
303

    
304
            meta.update(initial)
305
            account_meta = self.get_account_meta()
306
            (self.assertTrue('X-Account-Meta-%s' % k in account_meta) for
307
                k in meta.keys())
308
            (self.assertEqual(account_meta['X-Account-Meta-%s' % k], v) for
309
                k, v in meta.items())
310

    
311
    def test_reset_meta(self):
312
        with AssertMappingInvariant(self.get_account_groups):
313
            meta = {'test': 'tost', 'ping': 'pong'}
314
            self.update_account_meta(meta)
315

    
316
            new_meta = {'test': 'test33'}
317
            kwargs = dict((
318
                'HTTP_X_ACCOUNT_META_%s' % k, str(v)
319
            ) for k, v in new_meta.items())
320
            r = self.post('/v1/%s' % self.user, **kwargs)
321
            self.assertEqual(r.status_code, 202)
322

    
323
            account_meta = self.get_account_meta()
324
            (self.assertTrue('X-Account-Meta-%s' % k in account_meta) for
325
                k in new_meta.keys())
326
            (self.assertEqual(account_meta['X-Account-Meta-%s' % k], v) for
327
                k, v in new_meta.items())
328

    
329
            (self.assertTrue('X-Account-Meta-%s' % k not in account_meta) for
330
                k in meta.keys())
331

    
332
    def test_delete_meta(self):
333
        with AssertMappingInvariant(self.get_account_groups):
334
            meta = {'test': 'tost', 'ping': 'pong'}
335
            self.update_account_meta(meta)
336

    
337
            kwargs = dict(
338
                ('HTTP_X_ACCOUNT_META_%s' % k, '') for k, v in meta.items())
339
            r = self.post('/v1/%s?update=' % self.user, **kwargs)
340
            self.assertEqual(r.status_code, 202)
341

    
342
            account_meta = self.get_account_meta()
343

    
344
            (self.assertTrue('X-Account-Meta-%s' % k not in account_meta) for
345
                k in meta.keys())
346

    
347
    def test_set_account_groups(self):
348
        with AssertMappingInvariant(self.get_account_meta):
349
            pithosdevs = ['verigak', 'gtsouk', 'chazapis']
350
            r = self.post(
351
                '/v1/%s?update=' % self.user,
352
                HTTP_X_ACCOUNT_GROUP_PITHOSDEV=','.join(pithosdevs))
353
            self.assertEqual(r.status_code, 202)
354

    
355
            account_groups = self.get_account_groups()
356
            self.assertTrue(
357
                'X-Account-Group-Pithosdev' in self.get_account_groups())
358
            self.assertEqual(
359
                account_groups['X-Account-Group-Pithosdev'],
360
                ','.join(sorted(pithosdevs)))
361

    
362
            clientdevs = ['pkanavos', 'mvasilak']
363
            r = self.post(
364
                '/v1/%s?update=' % self.user,
365
                HTTP_X_ACCOUNT_GROUP_CLIENTSDEV=','.join(clientdevs))
366
            self.assertEqual(r.status_code, 202)
367

    
368
            account_groups = self.get_account_groups()
369
            self.assertTrue(
370
                'X-Account-Group-Pithosdev' in account_groups)
371
            self.assertTrue(
372
                'X-Account-Group-Clientsdev' in account_groups)
373
            self.assertEqual(
374
                account_groups['X-Account-Group-Pithosdev'],
375
                ','.join(sorted(pithosdevs)))
376
            self.assertEqual(
377
                account_groups['X-Account-Group-Clientsdev'],
378
                ','.join(sorted(clientdevs)))
379

    
380
            clientdevs = ['mvasilak']
381
            r = self.post(
382
                '/v1/%s?update=' % self.user,
383
                HTTP_X_ACCOUNT_GROUP_CLIENTSDEV=''.join(clientdevs))
384
            self.assertEqual(r.status_code, 202)
385

    
386
            account_groups = self.get_account_groups()
387
            self.assertTrue(
388
                'X-Account-Group-Pithosdev' in account_groups)
389
            self.assertTrue(
390
                'X-Account-Group-Clientsdev' in account_groups)
391
            self.assertEqual(
392
                account_groups['X-Account-Group-Pithosdev'],
393
                ','.join(sorted(pithosdevs)))
394
            self.assertEqual(
395
                account_groups['X-Account-Group-Clientsdev'],
396
                ','.join(sorted(clientdevs)))
397

    
398
    def test_reset_account_groups(self):
399
        with AssertMappingInvariant(self.get_account_meta):
400
            groups = {'pithosdev': ['verigak', 'gtsouk', 'chazapis'],
401
                      'clientsdev': ['pkanavos', 'mvasilak']}
402
            headers = dict(('HTTP_X_ACCOUNT_GROUP_%s' % k, ','.join(v))
403
                           for k, v in groups.iteritems())
404
            r = self.post('/v1/%s?update=' % self.user, **headers)
405
            self.assertEqual(r.status_code, 202)
406

    
407
            groups = {'pithosdev': ['verigak',
408
                                    'gtsouk',
409
                                    'chazapis',
410
                                    'papagian']}
411
            headers = dict(('HTTP_X_ACCOUNT_GROUP_%s' % k, ','.join(v))
412
                           for k, v in groups.iteritems())
413
            account_meta = self.get_account_meta()
414
            headers.update(dict(('HTTP_%s' % k.upper().replace('-', '_'), v)
415
                           for k, v in account_meta.iteritems()))
416
            r = self.post('/v1/%s' % self.user, **headers)
417
            self.assertEqual(r.status_code, 202)
418

    
419
            account_groups = self.get_account_groups()
420
            self.assertTrue(
421
                'X-Account-Group-Pithosdev' in account_groups)
422
            self.assertTrue(
423
                'X-Account-Group-Clientsdev' not in account_groups)
424
            self.assertEqual(
425
                account_groups['X-Account-Group-Pithosdev'],
426
                ','.join(sorted(groups['pithosdev'])))
427

    
428
    def test_delete_account_groups(self):
429
        with AssertMappingInvariant(self.get_account_meta):
430
            groups = {'pithosdev': ['verigak', 'gtsouk', 'chazapis'],
431
                      'clientsdev': ['pkanavos', 'mvasilak']}
432
            headers = dict(('HTTP_X_ACCOUNT_GROUP_%s' % k, ','.join(v))
433
                           for k, v in groups.iteritems())
434
            self.post('/v1/%s?update=' % self.user, **headers)
435

    
436
            kwargs = dict(('HTTP_X_ACCOUNT_GROUP_%s' % k, '')
437
                          for k, v in groups.items())
438
            r = self.post('/v1/%s?update=' % self.user, **kwargs)
439
            self.assertEqual(r.status_code, 202)
440

    
441
            account_groups = self.get_account_groups()
442
            self.assertTrue(
443
                'X-Account-Group-Pithosdev' not in account_groups)
444
            self.assertTrue(
445
                'X-Account-Group-Clientsdev' not in account_groups)