1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
|
import base64
import os
import subprocess
import unittest
import urlparse
import threading
from funkload.FunkLoadTestCase import FunkLoadTestCase
from webunit.utility import Upload
SIZE = 100 # KB
DELETE_ON_SETUP = False
DELETE_ON_TEARDOWN = False
def _get_auth_header(user_id):
encoded = base64.b64encode('%s:%s-token' % (user_id, user_id))
return 'Token %s' % encoded
def _ensure_template(templates_dir, size):
fname = '%dK-0.blob' % size
fpath = os.path.join(templates_dir, fname)
if not os.path.isfile(fpath):
dirname = os.path.dirname(os.path.realpath(__file__))
executable = os.path.join(dirname, 'templates.py')
code = subprocess.check_call([
executable,
'--amount', '1',
'--size', str(size),
templates_dir,
])
assert code == 0, 'failed creating template'
return fpath
class Blobs(FunkLoadTestCase):
next_user_id = 0
lock = threading.Lock()
def _get_next_user_id(self):
with Blobs.lock:
user_id = Blobs.next_user_id
Blobs.next_user_id += 1
Blobs.next_user_id %= 5000
return user_id
def setUp(self):
user_id = self._get_next_user_id()
base_url = self.conf_get('main', 'url')
self.url = urlparse.urljoin(base_url, 'blobs/%s/0' % user_id)
self.setHeader('Authorization', _get_auth_header(user_id))
templates_dir = self.conf_get('main', 'templates_dir')
size = self.conf_getInt('main', 'size')
fpath = _ensure_template(templates_dir, size)
self.upload = ['file', Upload(fpath)]
if DELETE_ON_SETUP:
ret = self.delete(self.url, description='Delete blob on setUp')
self.assert_(ret.code in [404, 200], 'expected 404 or 200')
def tearDown(self):
if DELETE_ON_TEARDOWN:
ret = self.delete(self.url, description='Delete blob on tearDown')
self.assert_(ret.code in [404, 200], 'expected 404 or 200')
self.clearHeaders()
def test_upload(self):
ret = self.put(self.url, params=[self.upload],
description='Upload blob')
self.assert_(ret.code == 200, "expecting a 200")
def test_download(self):
ret = self.get(self.url, description='Download blob')
self.assert_(ret.code == 200, "expecting a 200")
if __name__ == '__main__':
unittest.main()
|
