Statistics
| Branch: | Tag: | Revision:

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

History | View | Annotate | Download (18 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
from synnefo.lib import join_urls
41

    
42
import time as _time
43
import datetime
44

    
45

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

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

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

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

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

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

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

    
83
    def test_get_account_meta_until(self):
84
        self.update_account_meta({'foo': 'bar'})
85

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

    
92
        _time.sleep(2)
93
        self.update_account_meta({'quality': 'AAA'})
94

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

    
101
        self.assertTrue('X-Account-Meta-Quality' in account_info)
102
        self.assertTrue('X-Account-Meta-Foo' in account_info)
103

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

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

    
117

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

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

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

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

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

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

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

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

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

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

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

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

    
185
        # Check not modified
186
        url = join_urls(self.pithos_path, self.user)
187
        for t in t1_formats:
188
            r = self.get(url, HTTP_IF_MODIFIED_SINCE=t)
189
            self.assertEqual(r.status_code, 304)
190

    
191
        # modify account: add container
192
        _time.sleep(1)
193
        self.create_container('c1')
194

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

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

    
208
        # modify account: update account meta
209
        _time.sleep(1)
210
        self.update_account_meta({'foo': 'bar'})
211

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

    
220
    def test_if_modified_since_invalid_date(self):
221
        url = join_urls(self.pithos_path, self.user)
222
        r = self.get('%s' % url, HTTP_IF_MODIFIED_SINCE='Monday')
223
        self.assertEqual(r.status_code, 200)
224
        self.assertEqual(
225
            r.content.split('\n')[:-1],
226
            ['apples', 'bananas', 'kiwis', 'oranges', 'pears'])
227

    
228
    def test_if_not_modified_since(self):
229
        url = join_urls(self.pithos_path, self.user)
230
        account_info = self.get_account_info()
231
        last_modified = account_info['Last-Modified']
232
        t = datetime.datetime.strptime(last_modified, DATE_FORMATS[-1])
233

    
234
        # Check unmodified
235
        t1 = t + datetime.timedelta(seconds=1)
236
        t1_formats = map(t1.strftime, DATE_FORMATS)
237
        for t in t1_formats:
238
            r = self.get(url, HTTP_IF_UNMODIFIED_SINCE=t)
239
            self.assertEqual(r.status_code, 200)
240
            self.assertEqual(
241
                r.content.split('\n')[:-1],
242
                ['apples', 'bananas', 'kiwis', 'oranges', 'pears'])
243

    
244
        # modify account: add container
245
        _time.sleep(2)
246
        self.create_container('c1')
247

    
248
        account_info = self.get_account_info()
249
        last_modified = account_info['Last-Modified']
250
        t = datetime.datetime.strptime(last_modified, DATE_FORMATS[-1])
251
        t2 = t - datetime.timedelta(seconds=1)
252
        t2_formats = map(t2.strftime, DATE_FORMATS)
253

    
254
        # Check modified
255
        for t in t2_formats:
256
            r = self.get(url, HTTP_IF_UNMODIFIED_SINCE=t)
257
            self.assertEqual(r.status_code, 412)
258

    
259
        # modify account: update account meta
260
        _time.sleep(1)
261
        self.update_account_meta({'foo': 'bar'})
262

    
263
        account_info = self.get_account_info()
264
        last_modified = account_info['Last-Modified']
265
        t = datetime.datetime.strptime(last_modified, DATE_FORMATS[-1])
266
        t3 = t - datetime.timedelta(seconds=1)
267
        t3_formats = map(t3.strftime, DATE_FORMATS)
268

    
269
        # Check modified
270
        for t in t3_formats:
271
            r = self.get(url, HTTP_IF_UNMODIFIED_SINCE=t)
272
            self.assertEqual(r.status_code, 412)
273

    
274
    def test_if_unmodified_since_invalid_date(self):
275
        url = join_urls(self.pithos_path, self.user)
276
        r = self.get(url, HTTP_IF_UNMODIFIED_SINCE='Monday')
277
        self.assertEqual(r.status_code, 200)
278
        self.assertEqual(
279
            r.content.split('\n')[:-1],
280
            ['apples', 'bananas', 'kiwis', 'oranges', 'pears'])
281

    
282

    
283
class AccountPost(PithosAPITest):
284
    def setUp(self):
285
        PithosAPITest.setUp(self)
286
        cnames = ['apples', 'bananas', 'kiwis', 'oranges', 'pears']
287

    
288
        # create containers
289
        uploaded_bytes = 0
290
        for cname in cnames:
291
            self.create_container(cname)
292

    
293
            # upload object
294
            name, data, resp = self.upload_object(cname)
295
            uploaded_bytes += len(data)
296

    
297
        # set account meta
298
        self.update_account_meta({'foo': 'bar'})
299

    
300
    def test_update_meta(self):
301
        url = join_urls(self.pithos_path, self.user)
302
        with AssertMappingInvariant(self.get_account_groups):
303
            initial = self.get_account_meta()
304

    
305
            meta = {'test': 'tost', 'ping': 'pong'}
306
            kwargs = dict(('HTTP_X_ACCOUNT_META_%s' % k, str(v))
307
                          for k, v in meta.items())
308
            r = self.post('%s?update=' % url, **kwargs)
309
            self.assertEqual(r.status_code, 202)
310

    
311
            meta.update(initial)
312
            account_meta = self.get_account_meta()
313
            (self.assertTrue('X-Account-Meta-%s' % k in account_meta) for
314
                k in meta.keys())
315
            (self.assertEqual(account_meta['X-Account-Meta-%s' % k], v) for
316
                k, v in meta.items())
317

    
318
    def test_reset_meta(self):
319
        url = join_urls(self.pithos_path, self.user)
320
        with AssertMappingInvariant(self.get_account_groups):
321
            meta = {'test': 'tost', 'ping': 'pong'}
322
            self.update_account_meta(meta)
323

    
324
            new_meta = {'test': 'test33'}
325
            kwargs = dict((
326
                'HTTP_X_ACCOUNT_META_%s' % k, str(v)
327
            ) for k, v in new_meta.items())
328
            r = self.post(url, **kwargs)
329
            self.assertEqual(r.status_code, 202)
330

    
331
            account_meta = self.get_account_meta()
332
            (self.assertTrue('X-Account-Meta-%s' % k in account_meta) for
333
                k in new_meta.keys())
334
            (self.assertEqual(account_meta['X-Account-Meta-%s' % k], v) for
335
                k, v in new_meta.items())
336

    
337
            (self.assertTrue('X-Account-Meta-%s' % k not in account_meta) for
338
                k in meta.keys())
339

    
340
    def test_delete_meta(self):
341
        url = join_urls(self.pithos_path, self.user)
342
        with AssertMappingInvariant(self.get_account_groups):
343
            meta = {'test': 'tost', 'ping': 'pong'}
344
            self.update_account_meta(meta)
345

    
346
            kwargs = dict(
347
                ('HTTP_X_ACCOUNT_META_%s' % k, '') for k, v in meta.items())
348
            r = self.post('%s?update=' % url, **kwargs)
349
            self.assertEqual(r.status_code, 202)
350

    
351
            account_meta = self.get_account_meta()
352

    
353
            (self.assertTrue('X-Account-Meta-%s' % k not in account_meta) for
354
                k in meta.keys())
355

    
356
    def test_set_account_groups(self):
357
        url = join_urls(self.pithos_path, self.user)
358
        with AssertMappingInvariant(self.get_account_meta):
359
            pithosdevs = ['verigak', 'gtsouk', 'chazapis']
360
            r = self.post('%s?update=' % url,
361
                          HTTP_X_ACCOUNT_GROUP_PITHOSDEV=','.join(pithosdevs))
362
            self.assertEqual(r.status_code, 202)
363

    
364
            account_groups = self.get_account_groups()
365
            self.assertTrue(
366
                'X-Account-Group-Pithosdev' in self.get_account_groups())
367
            self.assertEqual(
368
                account_groups['X-Account-Group-Pithosdev'],
369
                ','.join(sorted(pithosdevs)))
370

    
371
            clientdevs = ['pkanavos', 'mvasilak']
372
            r = self.post('%s?update=' % url,
373
                          HTTP_X_ACCOUNT_GROUP_CLIENTSDEV=','.join(clientdevs))
374
            self.assertEqual(r.status_code, 202)
375

    
376
            account_groups = self.get_account_groups()
377
            self.assertTrue(
378
                'X-Account-Group-Pithosdev' in account_groups)
379
            self.assertTrue(
380
                'X-Account-Group-Clientsdev' in account_groups)
381
            self.assertEqual(
382
                account_groups['X-Account-Group-Pithosdev'],
383
                ','.join(sorted(pithosdevs)))
384
            self.assertEqual(
385
                account_groups['X-Account-Group-Clientsdev'],
386
                ','.join(sorted(clientdevs)))
387

    
388
            clientdevs = ['mvasilak']
389
            r = self.post('%s?update=' % url,
390
                          HTTP_X_ACCOUNT_GROUP_CLIENTSDEV=''.join(clientdevs))
391
            self.assertEqual(r.status_code, 202)
392

    
393
            account_groups = self.get_account_groups()
394
            self.assertTrue(
395
                'X-Account-Group-Pithosdev' in account_groups)
396
            self.assertTrue(
397
                'X-Account-Group-Clientsdev' in account_groups)
398
            self.assertEqual(
399
                account_groups['X-Account-Group-Pithosdev'],
400
                ','.join(sorted(pithosdevs)))
401
            self.assertEqual(
402
                account_groups['X-Account-Group-Clientsdev'],
403
                ','.join(sorted(clientdevs)))
404

    
405
    def test_reset_account_groups(self):
406
        url = join_urls(self.pithos_path, self.user)
407
        with AssertMappingInvariant(self.get_account_meta):
408
            groups = {'pithosdev': ['verigak', 'gtsouk', 'chazapis'],
409
                      'clientsdev': ['pkanavos', 'mvasilak']}
410
            headers = dict(('HTTP_X_ACCOUNT_GROUP_%s' % k, ','.join(v))
411
                           for k, v in groups.iteritems())
412
            r = self.post('%s?update=' % url, **headers)
413
            self.assertEqual(r.status_code, 202)
414

    
415
            groups = {'pithosdev': ['verigak',
416
                                    'gtsouk',
417
                                    'chazapis',
418
                                    'papagian']}
419
            headers = dict(('HTTP_X_ACCOUNT_GROUP_%s' % k, ','.join(v))
420
                           for k, v in groups.iteritems())
421
            account_meta = self.get_account_meta()
422
            headers.update(dict(('HTTP_%s' % k.upper().replace('-', '_'), v)
423
                           for k, v in account_meta.iteritems()))
424
            r = self.post(url, **headers)
425
            self.assertEqual(r.status_code, 202)
426

    
427
            account_groups = self.get_account_groups()
428
            self.assertTrue(
429
                'X-Account-Group-Pithosdev' in account_groups)
430
            self.assertTrue(
431
                'X-Account-Group-Clientsdev' not in account_groups)
432
            self.assertEqual(
433
                account_groups['X-Account-Group-Pithosdev'],
434
                ','.join(sorted(groups['pithosdev'])))
435

    
436
    def test_delete_account_groups(self):
437
        url = join_urls(self.pithos_path, self.user)
438
        with AssertMappingInvariant(self.get_account_meta):
439
            groups = {'pithosdev': ['verigak', 'gtsouk', 'chazapis'],
440
                      'clientsdev': ['pkanavos', 'mvasilak']}
441
            headers = dict(('HTTP_X_ACCOUNT_GROUP_%s' % k, ','.join(v))
442
                           for k, v in groups.iteritems())
443
            self.post('%s?update=' % url, **headers)
444

    
445
            kwargs = dict(('HTTP_X_ACCOUNT_GROUP_%s' % k, '')
446
                          for k, v in groups.items())
447
            r = self.post('%s?update=' % url, **kwargs)
448
            self.assertEqual(r.status_code, 202)
449

    
450
            account_groups = self.get_account_groups()
451
            self.assertTrue(
452
                'X-Account-Group-Pithosdev' not in account_groups)
453
            self.assertTrue(
454
                'X-Account-Group-Clientsdev' not in account_groups)