+# Copyright 2011 GRNET S.A. All rights reserved.
+#
+# Redistribution and use in source and binary forms, with or
+# without modification, are permitted provided that the following
+# conditions are met:
+#
+# 1. Redistributions of source code must retain the above
+# copyright notice, this list of conditions and the following
+# disclaimer.
+#
+# 2. Redistributions in binary form must reproduce the above
+# copyright notice, this list of conditions and the following
+# disclaimer in the documentation and/or other materials
+# provided with the distribution.
+#
+# THIS SOFTWARE IS PROVIDED BY GRNET S.A. ``AS IS'' AND ANY EXPRESS
+# OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED
+# WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR
+# PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL GRNET S.A OR
+# CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
+# SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
+# LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF
+# USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED
+# AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT
+# LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN
+# ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
+# POSSIBILITY OF SUCH DAMAGE.
+#
+# The views and conclusions contained in the software and
+# documentation are those of the authors and should not be
+# interpreted as representing official policies, either expressed
+# or implied, of GRNET S.A.
+
from django.test.client import Client
from django.test import TestCase
from django.utils import simplejson as json
import mimetypes
import random
import datetime
+import string
+from pithos.backends import backend
DATE_FORMATS = ["%a %b %d %H:%M:%S %Y",
"%A, %d-%b-%y %H:%M:%S GMT",
class AaiClient(Client):
def request(self, **request):
- request['HTTP_X_AUTH_TOKEN'] = '46e427d657b20defe352804f0eb6f8a2'
+ request['HTTP_X_AUTH_TOKEN'] = '0000'
return super(AaiClient, self).request(**request)
class BaseTestCase(TestCase):
'Content-Type',
'Last-Modified',
'Content-Length',
- 'Date',),
+ 'Date',
+ 'X-Container-Block-Size',
+ 'X-Container-Block-Hash',),
'object':(
'ETag',
'Content-Length',
'Last-Modified',
'Date',
'X-Object-Manifest',
- 'Content-Range',)}
+ 'Content-Range',
+ 'X-Object-Modified-By',
+ 'X-Object-Version',
+ 'X-Object-Version-Timestamp',)}
self.contentTypes = {'xml':'application/xml',
'json':'application/json',
'':'text/plain'}
'content_encoding',
'last_modified',)}
self.return_codes = (400, 401, 404, 503,)
-
+
def assertFault(self, response, status_code, name):
self.assertEqual(response.status_code, status_code)
-
+
def assertBadRequest(self, response):
self.assertFault(response, 400, 'badRequest')
-
+
def assertItemNotFound(self, response):
self.assertFault(response, 404, 'itemNotFound')
self.assert_headers(response, 'account', exp_meta)
return response
- def list_containers(self, account, limit=10000, marker='', format=''):
+ def list_containers(self, account, limit=10000, marker='', format='',
+ **headers):
params = locals()
params.pop('self')
params.pop('account')
path = '/v1/%s' % account
- response = self.client.get(path, params)
- self.assert_status(response, [200, 204])
+ response = self.client.get(path, params, **headers)
+ self.assert_status(response, [200, 204, 304, 412])
response.content = response.content.strip()
if format:
self.assert_extended(response, format, 'container', limit)
def update_account_meta(self, account, **metadata):
path = '/v1/%s' % account
response = self.client.post(path, **metadata)
- response.content = response.content.strip()
+ response.content = response.content
self.assert_status(response, 202)
return response
params.pop('container')
path = '/v1/%s/%s' %(account, container)
response = self.client.head(path, params)
- response.content = response.content.strip()
+ response.content = response.content
self.assert_status(response, 204)
if response.status_code == 204:
self.assert_headers(response, 'container', exp_meta)
return response
- def list_objects(self, account, container, limit=10000, marker='', prefix='', format='', path='', delimiter='', meta=''):
+ def list_objects(self, account, container, limit=10000, marker='',
+ prefix='', format='', path='', delimiter='', meta='',
+ **headers):
params = locals()
params.pop('self')
params.pop('account')
params.pop('container')
path = '/v1/%s/%s' % (account, container)
- response = self.client.get(path, params)
+ response = self.client.get(path, params, **headers)
response.content = response.content.strip()
if format:
self.assert_extended(response, format, 'object', limit)
- self.assert_status(response, [200, 204])
+ self.assert_status(response, [200, 204, 304, 412])
return response
def create_container(self, account, name, **meta):
path = '/v1/%s/%s' %(account, name)
response = self.client.put(path, **meta)
- response.content = response.content.strip()
+ response.content = response.content
self.assert_status(response, [201, 202])
return response
def update_container_meta(self, account, name, **meta):
path = '/v1/%s/%s' %(account, name)
- response = self.client.post(path, data={}, content_type='text/xml', follow=False, **meta)
- response.content = response.content.strip()
+ response = self.client.post(path,
+ data=None,
+ content_type='text/xml',
+ follow=False, **meta)
+ response.content = response.content
self.assert_status(response, 202)
return response
def delete_container(self, account, container):
path = '/v1/%s/%s' %(account, container)
response = self.client.delete(path)
- response.content = response.content.strip()
+ response.content = response.content
self.assert_status(response, [204, 409])
return response
def get_object_meta(self, account, container, name):
path = '/v1/%s/%s/%s' %(account, container, name)
response = self.client.head(path)
- response.content = response.content.strip()
- self.assert_status(response, 204)
+ response.content = response.content
+ self.assert_status(response, 200)
return response
- def get_object(self, account, container, name, exp_meta={}, **headers):
+ def get_object(self, account, container, name, format='', **headers):
path = '/v1/%s/%s/%s' %(account, container, name)
- response = self.client.get(path, **headers)
- response.content = response.content.strip()
+ response = self.client.get(path, {'format':format}, **headers)
+ response.content = response.content
self.assert_status(response, [200, 206, 304, 412, 416])
if response.status_code in [200, 206]:
self.assert_headers(response, 'object')
return response
- def upload_object(self, account, container, name, data, content_type='application/json', **headers):
+ def upload_object(self, account, container, name, data, content_type='',
+ **headers):
path = '/v1/%s/%s/%s' %(account, container, name)
response = self.client.put(path, data, content_type, **headers)
- response.content = response.content.strip()
+ response.content = response.content
self.assert_status(response, [201, 411, 422])
if response.status_code == 201:
self.assertTrue(response['Etag'])
path = '/v1/%s/%s/%s' %(account, container, name)
headers['HTTP_X_COPY_FROM'] = src
response = self.client.put(path, **headers)
- response.content = response.content.strip()
+ response.content = response.content
self.assert_status(response, 201)
return response
path = '/v1/%s/%s/%s' % (account, container, name)
headers['HTTP_X_MOVE_FROM'] = src
response = self.client.put(path, **headers)
- response.content = response.content.strip()
+ response.content = response.content
self.assert_status(response, 201)
return response
- def update_object_meta(self, account, container, name, **headers):
+ def update_object(self, account, container, name, data='',
+ content_type='', **headers):
path = '/v1/%s/%s/%s' %(account, container, name)
- response = self.client.post(path, **headers)
- response.content = response.content.strip()
- self.assert_status(response, 202)
+ response = self.client.post(path, data, content_type, **headers)
+ response.content = response.content
+ self.assert_status(response, [202, 204, 416])
return response
def delete_object(self, account, container, name):
path = '/v1/%s/%s/%s' %(account, container, name)
response = self.client.delete(path)
- response.content = response.content.strip()
+ response.content = response.content
self.assert_status(response, 204)
return response
entities = ['Account', 'Container', 'Container-Object', 'Object']
user_defined_meta = ['X-%s-Meta' %elem for elem in entities]
headers = [item for item in response._headers.values()]
- system_headers = [h for h in headers if not h[0].startswith(tuple(user_defined_meta))]
+ t = tuple(user_defined_meta)
+ system_headers = [h for h in headers if not h[0].startswith(t)]
for h in system_headers:
self.assertTrue(h[0] in self.headers[type])
if exp_meta:
self.assertEqual(h[1], exp_meta[h[0]])
def assert_extended(self, response, format, type, size):
- self.assertEqual(response['Content-Type'].find(self.contentTypes[format]), 0)
+ exp_content_type = self.contentTypes[format]
+ self.assertEqual(response['Content-Type'].find(exp_content_type), 0)
if format == 'xml':
self.assert_xml(response, type, size)
elif format == 'json':
except IOError:
return
- def upload_random_data(self, account, container, name, length=1024, meta={}):
- data = str(random.getrandbits(length))
+ def upload_random_data(self, account, container, name, length=1024,
+ meta={}):
+ data = get_random_data(length)
return self.upload_data(account, container, name, data, meta)
def upload_data(self, account, container, name, data, meta={}):
obj['name'] = name
try:
obj['data'] = data
- obj['hash'] = compute_hash(obj['data'])
+ obj['hash'] = compute_md5_hash(obj['data'])
meta.update({'HTTP_X_OBJECT_META_TEST':'test1',
'HTTP_ETAG':obj['hash']})
- meta['HTTP_CONTENT_TYPE'], enc = mimetypes.guess_type(name)
+ type, enc = mimetypes.guess_type(name)
+ meta['HTTP_CONTENT_TYPE'] = type and type or 'plain/text'
if enc:
- meta['HTTP_CONTENT_TYPE'] = enc
+ meta['HTTP_CONTENT_ENCODING'] = enc
+
obj['meta'] = meta
r = self.upload_object(account,
container,
except IOError:
return
-class ListContainers(BaseTestCase):
+class AccountHead(BaseTestCase):
+ def setUp(self):
+ BaseTestCase.setUp(self)
+ self.account = 'test'
+ self.containers = ['apples', 'bananas', 'kiwis', 'oranges', 'pears']
+ for item in self.containers:
+ self.create_container(self.account, item)
+
+ def tearDown(self):
+ for c in get_content_splitted(self.list_containers(self.account)):
+ self.delete_container(self.account, c)
+
+ def test_get_account_meta(self):
+ response = self.get_account_meta(self.account)
+ r2 = self.list_containers(self.account)
+ containers = get_content_splitted(r2)
+ l = str(len(containers))
+ self.assertEqual(response['X-Account-Container-Count'], l)
+ size = 0
+ for c in containers:
+ r = self.get_container_meta(self.account, c)
+ size = size + int(r['X-Container-Bytes-Used'])
+ self.assertEqual(response['X-Account-Bytes-Used'], str(size))
+
+ #def test_get_account_401(self):
+ # response = self.get_account_meta('non-existing-account')
+ # print response
+ # self.assertEqual(response.status_code, 401)
+
+class AccountGet(BaseTestCase):
def setUp(self):
BaseTestCase.setUp(self)
self.account = 'test'
self.assertEquals(self.containers[:2], containers)
def test_list_with_marker(self):
- limit = 2
- marker = 'bananas'
- response = self.list_containers(self.account, limit=limit, marker=marker)
+ l = 2
+ m = 'bananas'
+ response = self.list_containers(self.account, limit=l, marker=m)
containers = get_content_splitted(response)
- i = self.containers.index(marker) + 1
- self.assertEquals(self.containers[i:(i+limit)], containers)
+ i = self.containers.index(m) + 1
+ self.assertEquals(self.containers[i:(i+l)], containers)
- marker = 'oranges'
- response = self.list_containers(self.account, limit=limit, marker=marker)
+ m = 'oranges'
+ response = self.list_containers(self.account, limit=l, marker=m)
containers = get_content_splitted(response)
- i = self.containers.index(marker) + 1
- self.assertEquals(self.containers[i:(i+limit)], containers)
+ i = self.containers.index(m) + 1
+ self.assertEquals(self.containers[i:(i+l)], containers)
#def test_extended_list(self):
# self.list_containers(self.account, limit=3, format='xml')
# self.list_containers(self.account, limit=3, format='json')
def test_list_json_with_marker(self):
- limit = 2
- marker = 'bananas'
- response = self.list_containers(self.account, limit=limit, marker=marker, format='json')
+ l = 2
+ m = 'bananas'
+ response = self.list_containers(self.account, limit=l, marker=m,
+ format='json')
containers = json.loads(response.content)
self.assertEqual(containers[0]['name'], 'kiwis')
self.assertEqual(containers[1]['name'], 'oranges')
def test_list_xml_with_marker(self):
- limit = 2
- marker = 'oranges'
- response = self.list_containers(self.account, limit=limit, marker=marker, format='xml')
+ l = 2
+ m = 'oranges'
+ response = self.list_containers(self.account, limit=l, marker=m,
+ format='xml')
xml = minidom.parseString(response.content)
nodes = xml.getElementsByTagName('name')
self.assertEqual(len(nodes), 1)
self.assertEqual(nodes[0].childNodes[0].data, 'pears')
-class AccountMetadata(BaseTestCase):
+ def test_if_modified_since(self):
+ t = datetime.datetime.utcnow()
+ t2 = t - datetime.timedelta(minutes=10)
+
+ #add a new container
+ self.create_container(self.account,
+ 'dummy')
+
+ for f in DATE_FORMATS:
+ past = t2.strftime(f)
+
+ headers = {'HTTP_IF_MODIFIED_SINCE':'%s' %past}
+ r = self.list_containers(self.account, **headers)
+
+ #assert get success
+ self.assertEqual(r.status_code, 200)
+
+ def test_if_modified_since_invalid_date(self):
+ headers = {'HTTP_IF_MODIFIED_SINCE':''}
+ r = self.list_containers(self.account, **headers)
+
+ #assert get success
+ self.assertEqual(r.status_code, 200)
+
+ def test_if_not_modified_since(self):
+ now = datetime.datetime.utcnow()
+ since = now + datetime.timedelta(1)
+
+ for f in DATE_FORMATS:
+ headers = {'HTTP_IF_MODIFIED_SINCE':'%s' %since.strftime(f)}
+ r = self.list_containers(self.account, **headers)
+
+ #assert not modified
+ self.assertEqual(r.status_code, 304)
+
+ def test_if_unmodified_since(self):
+ now = datetime.datetime.utcnow()
+ since = now + datetime.timedelta(1)
+
+ for f in DATE_FORMATS:
+ headers = {'HTTP_IF_UNMODIFIED_SINCE':'%s' %since.strftime(f)}
+ r = self.list_containers(self.account, **headers)
+
+ #assert success
+ self.assertEqual(r.status_code, 200)
+ self.assertEqual(self.containers, get_content_splitted(r))
+
+ def test_if_unmodified_since_precondition_failed(self):
+ t = datetime.datetime.utcnow()
+ t2 = t - datetime.timedelta(minutes=10)
+
+ #add a new container
+ self.create_container(self.account,
+ 'dummy')
+
+ for f in DATE_FORMATS:
+ past = t2.strftime(f)
+
+ headers = {'HTTP_IF_UNMODIFIED_SINCE':'%s' %past}
+ r = self.list_containers(self.account, **headers)
+
+ #assert get success
+ self.assertEqual(r.status_code, 412)
+
+class AccountPost(BaseTestCase):
def setUp(self):
BaseTestCase.setUp(self)
self.account = 'test'
for c in get_content_splitted(self.list_containers(self.account)):
self.delete_container(self.account, c)
- def test_get_account_meta(self):
- response = self.get_account_meta(self.account)
- r2 = self.list_containers(self.account)
- containers = get_content_splitted(r2)
- self.assertEqual(response['X-Account-Container-Count'], str(len(containers)))
- size = 0
- for c in containers:
- r = self.get_container_meta(self.account, c)
- size = size + int(r['X-Container-Bytes-Used'])
- self.assertEqual(response['X-Account-Bytes-Used'], str(size))
-
- #def test_get_account_401(self):
- # response = self.get_account_meta('non-existing-account')
- # print response
- # self.assertEqual(response.status_code, 401)
-
def test_update_meta(self):
- meta = {'HTTP_X_ACCOUNT_META_TEST':'test', 'HTTP_X_ACCOUNT_META_TOST':'tost'}
+ meta = {'HTTP_X_ACCOUNT_META_TEST':'test',
+ 'HTTP_X_ACCOUNT_META_TOST':'tost'}
response = self.update_account_meta(self.account, **meta)
response = self.get_account_meta(self.account)
for k,v in meta.items():
#def test_invalid_account_update_meta(self):
# with AssertInvariant(self.get_account_meta, self.account):
- # meta = {'HTTP_X_ACCOUNT_META_TEST':'test', 'HTTP_X_ACCOUNT_META_TOST':'tost'}
+ # meta = {'HTTP_X_ACCOUNT_META_TEST':'test',
+ # 'HTTP_X_ACCOUNT_META_TOST':'tost'}
# response = self.update_account_meta('non-existing-account', **meta)
-class ListObjects(BaseTestCase):
+class ContainerHead(BaseTestCase):
+ def setUp(self):
+ BaseTestCase.setUp(self)
+ self.account = 'test'
+ self.container = 'apples'
+ self.create_container(self.account, self.container)
+
+ def tearDown(self):
+ for o in self.list_objects(self.account, self.container):
+ self.delete_object(self.account, self.container, o)
+ self.delete_container(self.account, self.container)
+
+ def test_get_meta(self):
+ headers = {'HTTP_X_OBJECT_META_TRASH':'true'}
+ t1 = datetime.datetime.utcnow()
+ o = self.upload_random_data(self.account,
+ self.container,
+ 'McIntosh.jpg',
+ meta=headers)
+ if o:
+ r = self.get_container_meta(self.account,
+ self.container)
+ self.assertEqual(r['X-Container-Object-Count'], '1')
+ self.assertEqual(r['X-Container-Bytes-Used'], str(len(o['data'])))
+ t2 = datetime.datetime.strptime(r['Last-Modified'], DATE_FORMATS[2])
+ delta = (t2 - t1)
+ threashold = datetime.timedelta(seconds=1)
+ self.assertTrue(delta < threashold)
+ self.assertTrue(r['X-Container-Object-Meta'])
+ self.assertTrue('Trash' in r['X-Container-Object-Meta'])
+
+class ContainerGet(BaseTestCase):
def setUp(self):
BaseTestCase.setUp(self)
self.account = 'test'
def tearDown(self):
for c in self.container:
- self.delete_container_recursively(c)
-
- def delete_container_recursively(self, c):
- for obj in get_content_splitted(self.list_objects(self.account, c)):
- self.delete_object(self.account, c, obj)
- self.delete_container(self.account, c)
+ for obj in get_content_splitted(self.list_objects(self.account, c)):
+ self.delete_object(self.account, c, obj)
+ self.delete_container(self.account, c)
def test_list_objects(self):
response = self.list_objects(self.account, self.container[0])
'moms_birthday.jpg']
limit = 4
for m in markers:
- response = self.list_objects(self.account, self.container[0], limit=limit, marker=m)
+ response = self.list_objects(self.account, self.container[0],
+ limit=limit, marker=m)
objects = get_content_splitted(response)
l = [elem['name'] for elem in self.obj[:8]]
l.sort()
self.assertEqual(objects, l[start:end])
def test_list_pseudo_hierarchical_folders(self):
- response = self.list_objects(self.account, self.container[1], prefix='photos', delimiter='/')
+ response = self.list_objects(self.account, self.container[1],
+ prefix='photos', delimiter='/')
objects = get_content_splitted(response)
- self.assertEquals(['photos/animals/', 'photos/me.jpg', 'photos/plants/'], objects)
+ self.assertEquals(['photos/animals/', 'photos/me.jpg',
+ 'photos/plants/'], objects)
- response = self.list_objects(self.account, self.container[1], prefix='photos/animals', delimiter='/')
- objects = get_content_splitted(response)
- self.assertEquals(['photos/animals/cats/', 'photos/animals/dogs/'], objects)
+ response = self.list_objects(self.account, self.container[1],
+ prefix='photos/animals', delimiter='/')
+ objs = get_content_splitted(response)
+ l = ['photos/animals/cats/', 'photos/animals/dogs/']
+ self.assertEquals(l, objs)
- response = self.list_objects(self.account, self.container[1], path='photos')
+ response = self.list_objects(self.account, self.container[1],
+ path='photos')
objects = get_content_splitted(response)
self.assertEquals(['photos/me.jpg'], objects)
def test_extended_list_json(self):
- response = self.list_objects(self.account, self.container[1], format='json', limit=2, prefix='photos/animals', delimiter='/')
+ response = self.list_objects(self.account,
+ self.container[1],
+ format='json', limit=2,
+ prefix='photos/animals',
+ delimiter='/')
objects = json.loads(response.content)
self.assertEqual(objects[0]['subdir'], 'photos/animals/cats/')
self.assertEqual(objects[1]['subdir'], 'photos/animals/dogs/')
def test_extended_list_xml(self):
- response = self.list_objects(self.account, self.container[1], format='xml', limit=4, prefix='photos', delimiter='/')
+ response = self.list_objects(self.account, self.container[1],
+ format='xml', limit=4, prefix='photos',
+ delimiter='/')
xml = minidom.parseString(response.content)
dirs = xml.getElementsByTagName('subdir')
self.assertEqual(len(dirs), 2)
self.assertEqual(len(objects), 1)
self.assertEqual(objects[0].childNodes[0].data, 'photos/me.jpg')
+ def test_list_meta_double_matching(self):
+ meta = {'HTTP_X_OBJECT_META_QUALITY':'aaa',
+ 'HTTP_X_OBJECT_META_STOCK':'true'}
+ r = self.update_object(self.account,
+ self.container[0],
+ self.obj[0]['name'],
+
+ **meta)
+ r = self.list_objects(self.account,
+ self.container[0],
+ meta='Quality,Stock')
+ self.assertEqual(r.status_code, 200)
+ obj = get_content_splitted(r)
+ self.assertEqual(len(obj), 1)
+ self.assertTrue(obj, self.obj[0]['name'])
+
def test_list_using_meta(self):
meta = {'HTTP_X_OBJECT_META_QUALITY':'aaa'}
for o in self.obj[:2]:
- r = self.update_object_meta(self.account,
+ r = self.update_object(self.account,
self.container[0],
o['name'],
**meta)
meta = {'HTTP_X_OBJECT_META_STOCK':'true'}
for o in self.obj[3:5]:
- r = self.update_object_meta(self.account,
+ r = self.update_object(self.account,
self.container[0],
o['name'],
**meta)
-
+
r = self.list_objects(self.account,
self.container[0],
meta='Quality')
self.assertEqual(r.status_code, 200)
obj = get_content_splitted(r)
self.assertEqual(len(obj), 2)
+ self.assertTrue(obj, [o['name'] for o in self.obj[:2]])
# test case insensitive
r = self.list_objects(self.account,
self.assertEqual(r.status_code, 200)
obj = get_content_splitted(r)
self.assertEqual(len(obj), 2)
+ self.assertTrue(obj, [o['name'] for o in self.obj[:2]])
# test multiple matches
r = self.list_objects(self.account,
self.container[0],
- meta='Quality, Stock')
+ meta='Quality,Stock')
self.assertEqual(r.status_code, 200)
obj = get_content_splitted(r)
self.assertEqual(len(obj), 4)
+ self.assertTrue(obj, [o['name'] for o in self.obj[:4]])
# test non 1-1 multiple match
r = self.list_objects(self.account,
self.container[0],
- meta='Quality, aaaa')
+ meta='Quality,aaaa')
self.assertEqual(r.status_code, 200)
obj = get_content_splitted(r)
self.assertEqual(len(obj), 2)
+ self.assertTrue(obj, [o['name'] for o in self.obj[:2]])
+
+ def test_if_modified_since(self):
+ t = datetime.datetime.utcnow()
+ t2 = t - datetime.timedelta(minutes=10)
+ #add a new container
+ self.upload_random_data(self.account,
+ self.container[0],
+ 'dummy.txt')
-class ContainerMeta(BaseTestCase):
- def setUp(self):
- BaseTestCase.setUp(self)
- self.account = 'test'
- self.container = 'apples'
- self.create_container(self.account, self.container)
+ for f in DATE_FORMATS:
+ past = t2.strftime(f)
+
+ headers = {'HTTP_IF_MODIFIED_SINCE':'%s' %past}
+ r = self.list_objects(self.account,
+ self.container[0], **headers)
+
+ #assert get success
+ self.assertEqual(r.status_code, 200)
- def tearDown(self):
- for o in self.list_objects(self.account, self.container):
- self.delete_object(self.account, self.container, o)
- self.delete_container(self.account, self.container)
-
- def test_get_meta(self):
- headers = {'HTTP_X_OBJECT_META_TRASH':'true'}
- t1 = datetime.datetime.utcnow()
- o = self.upload_random_data(self.account,
- self.container,
- 'McIntosh.jpg',
- meta=headers)
- if o:
- r = self.get_container_meta(self.account,
- self.container)
- self.assertEqual(r['X-Container-Object-Count'], '1')
- self.assertEqual(r['X-Container-Bytes-Used'], str(len(o['data'])))
- t2 = datetime.datetime.strptime(r['Last-Modified'], DATE_FORMATS[2])
- delta = (t2 - t1)
- threashold = datetime.timedelta(seconds=1)
- self.assertTrue(delta < threashold)
- self.assertTrue(r['X-Container-Object-Meta'])
- self.assertTrue('Trash' in r['X-Container-Object-Meta'])
+ def test_if_modified_since_invalid_date(self):
+ headers = {'HTTP_IF_MODIFIED_SINCE':''}
+ r = self.list_objects(self.account,
+ self.container[0], **headers)
+
+ #assert get success
+ self.assertEqual(r.status_code, 200)
- def test_update_meta(self):
- meta = {'HTTP_X_CONTAINER_META_TEST':'test33', 'HTTP_X_CONTAINER_META_TOST':'tost22'}
- response = self.update_container_meta(self.account, self.container, **meta)
- response = self.get_container_meta(self.account, self.container)
- for k,v in meta.items():
- key = '-'.join(elem.capitalize() for elem in k.split('_')[1:])
- self.assertTrue(response[key])
- self.assertEqual(response[key], v)
+ def test_if_not_modified_since(self):
+ now = datetime.datetime.utcnow()
+ since = now + datetime.timedelta(1)
+
+ for f in DATE_FORMATS:
+ headers = {'HTTP_IF_MODIFIED_SINCE':'%s' %since.strftime(f)}
+ r = self.list_objects(self.account,
+ self.container[0], **headers)
+
+ #assert not modified
+ self.assertEqual(r.status_code, 304)
+
+ def test_if_unmodified_since(self):
+ now = datetime.datetime.utcnow()
+ since = now + datetime.timedelta(1)
+
+ for f in DATE_FORMATS:
+ headers = {'HTTP_IF_UNMODIFIED_SINCE':'%s' %since.strftime(f)}
+ r = self.list_objects(self.account,
+ self.container[0], **headers)
+
+ #assert success
+ self.assertEqual(r.status_code, 200)
+ objlist = self.list_objects(self.account, self.container[0])
+ self.assertEqual(get_content_splitted(r),
+ get_content_splitted(objlist))
+
+ def test_if_unmodified_since_precondition_failed(self):
+ t = datetime.datetime.utcnow()
+ t2 = t - datetime.timedelta(minutes=10)
+
+ #add a new container
+ self.create_container(self.account,
+ 'dummy')
+
+ for f in DATE_FORMATS:
+ past = t2.strftime(f)
+
+ headers = {'HTTP_IF_UNMODIFIED_SINCE':'%s' %past}
+ r = self.list_objects(self.account,
+ self.container[0], **headers)
+
+ #assert get success
+ self.assertEqual(r.status_code, 412)
-class CreateContainer(BaseTestCase):
+class ContainerPut(BaseTestCase):
def setUp(self):
BaseTestCase.setUp(self)
self.account = 'test'
response = self.create_container(self.account, self.containers[0])
if response.status_code == 201:
response = self.list_containers(self.account)
- self.assertTrue(self.containers[0] in get_content_splitted(response))
+ content = get_content_splitted(response)
+ self.assertTrue(self.containers[0] in content)
r = self.get_container_meta(self.account, self.containers[0])
self.assertEqual(r.status_code, 204)
def test_create_twice(self):
response = self.create_container(self.account, self.containers[0])
if response.status_code == 201:
- self.assertTrue(self.create_container(self.account, self.containers[0]).status_code, 202)
+ r = self.create_container(self.account, self.containers[0])
+ self.assertTrue(r.status_code, 202)
-class DeleteContainer(BaseTestCase):
+class ContainerPost(BaseTestCase):
+ def setUp(self):
+ BaseTestCase.setUp(self)
+ self.account = 'test'
+ self.container = 'apples'
+ self.create_container(self.account, self.container)
+
+ def tearDown(self):
+ for o in self.list_objects(self.account, self.container):
+ self.delete_object(self.account, self.container, o)
+ self.delete_container(self.account, self.container)
+
+ def test_update_meta(self):
+ meta = {'HTTP_X_CONTAINER_META_TEST':'test33',
+ 'HTTP_X_CONTAINER_META_TOST':'tost22'}
+ response = self.update_container_meta(self.account, self.container,
+ **meta)
+ response = self.get_container_meta(self.account, self.container)
+ for k,v in meta.items():
+ key = '-'.join(elem.capitalize() for elem in k.split('_')[1:])
+ self.assertTrue(response[key])
+ self.assertEqual(response[key], v)
+
+class ContainerDelete(BaseTestCase):
def setUp(self):
BaseTestCase.setUp(self)
self.account = 'test'
def test_delete_invalid(self):
self.assertItemNotFound(self.delete_container(self.account, 'c3'))
-class GetObjects(BaseTestCase):
+class ObjectHead(BaseTestCase):
+ pass
+
+class ObjectGet(BaseTestCase):
def setUp(self):
BaseTestCase.setUp(self)
self.account = 'test'
self.objects[0]['meta'])
#assert success
self.assertEqual(r.status_code, 200)
+
+ #assert content-type
+ self.assertEqual(r['Content-Type'],
+ self.objects[0]['meta']['HTTP_CONTENT_TYPE'])
def test_get_invalid(self):
r = self.get_object(self.account,
#assert successful partial content
self.assertEqual(r.status_code, 206)
+ #assert content-type
+ self.assertEqual(r['Content-Type'],
+ self.objects[0]['meta']['HTTP_CONTENT_TYPE'])
+
#assert content length
self.assertEqual(int(r['Content-Length']), 500)
#assert successful partial content
self.assertEqual(r.status_code, 206)
+ #assert content-type
+ self.assertEqual(r['Content-Type'],
+ self.objects[0]['meta']['HTTP_CONTENT_TYPE'])
+
#assert content length
self.assertEqual(int(r['Content-Length']), 500)
#assert successful partial content
self.assertEqual(r.status_code, 206)
+ #assert content-type
+ self.assertEqual(r['Content-Type'],
+ self.objects[0]['meta']['HTTP_CONTENT_TYPE'])
+
#assert content length
self.assertEqual(int(r['Content-Length']), 500)
#assert content
self.assertTrue(self.objects[0]['data'][-500:], r.content)
-
+
def test_get_range_not_satisfiable(self):
#perform get with range
offset = len(self.objects[0]['data']) + 1
#assert get success
self.assertEqual(r.status_code, 200)
+ #assert content-type
+ self.assertEqual(r['Content-Type'],
+ self.objects[0]['meta']['HTTP_CONTENT_TYPE'])
+
#assert response content
self.assertEqual(self.objects[0]['data'].strip(), r.content.strip())
#assert get success
self.assertEqual(r.status_code, 200)
+ #assert content-type
+ self.assertEqual(r['Content-Type'],
+ self.objects[0]['meta']['HTTP_CONTENT_TYPE'])
+
#assert response content
self.assertEqual(self.objects[0]['data'].strip(), r.content.strip())
#assert get success
self.assertEqual(r.status_code, 200)
+ #assert content-type
+ self.assertEqual(r['Content-Type'],
+ self.objects[0]['meta']['HTTP_CONTENT_TYPE'])
+
+ #assert content-type
+ self.assertEqual(r['Content-Type'],
+ self.objects[0]['meta']['HTTP_CONTENT_TYPE'])
+
#assert response content
self.assertEqual(self.objects[0]['data'].strip(), r.content.strip())
self.containers[1],
self.objects[0]['name'],
**headers)
-
#assert precondition failed
self.assertEqual(r.status_code, 412)
def test_if_none_match(self):
- #perform get with If-Match
+ #perform get with If-None-Match
headers = {'HTTP_IF_NONE_MATCH':'123'}
r = self.get_object(self.account,
self.containers[1],
#assert get success
self.assertEqual(r.status_code, 200)
+
+ #assert content-type
+ self.assertEqual(r['Content-Type'],
+ self.objects[0]['meta']['HTTP_CONTENT_TYPE'])
def test_if_none_match(self):
- #perform get with If-Match *
+ #perform get with If-None-Match *
headers = {'HTTP_IF_NONE_MATCH':'*'}
r = self.get_object(self.account,
self.containers[1],
#assert get success
self.assertEqual(r.status_code, 304)
-
+
def test_if_none_match_not_modified(self):
- #perform get with If-Match
+ #perform get with If-None-Match
headers = {'HTTP_IF_NONE_MATCH':'%s' %self.objects[0]['hash']}
r = self.get_object(self.account,
self.containers[1],
#assert not modified
self.assertEqual(r.status_code, 304)
self.assertEqual(r['ETag'], self.objects[0]['hash'])
-
- def test_get_with_if_modified_since(self):
+
+ def test_if_modified_since(self):
t = datetime.datetime.utcnow()
t2 = t - datetime.timedelta(minutes=10)
#assert get success
self.assertEqual(r.status_code, 200)
- def test_get_modified_since_invalid_date(self):
+ #assert content-type
+ self.assertEqual(r['Content-Type'],
+ self.objects[0]['meta']['HTTP_CONTENT_TYPE'])
+
+ def test_if_modified_since_invalid_date(self):
headers = {'HTTP_IF_MODIFIED_SINCE':''}
r = self.get_object(self.account,
self.containers[1],
#assert get success
self.assertEqual(r.status_code, 200)
+
+ #assert content-type
+ self.assertEqual(r['Content-Type'],
+ self.objects[0]['meta']['HTTP_CONTENT_TYPE'])
- def test_get_not_modified_since(self):
+ def test_if_not_modified_since(self):
now = datetime.datetime.utcnow()
since = now + datetime.timedelta(1)
#assert not modified
self.assertEqual(r.status_code, 304)
- def test_get_with_if_unmodified_since(self):
- return
+ def test_if_unmodified_since(self):
+ now = datetime.datetime.utcnow()
+ since = now + datetime.timedelta(1)
+
+ for f in DATE_FORMATS:
+ headers = {'HTTP_IF_UNMODIFIED_SINCE':'%s' %since.strftime(f)}
+ r = self.get_object(self.account,
+ self.containers[1],
+ self.objects[0]['name'],
+ **headers)
+ #assert success
+ self.assertEqual(r.status_code, 200)
+ self.assertEqual(self.objects[0]['data'], r.content)
+
+ #assert content-type
+ self.assertEqual(r['Content-Type'],
+ self.objects[0]['meta']['HTTP_CONTENT_TYPE'])
-class UploadObject(BaseTestCase):
+ def test_if_unmodified_since_precondition_failed(self):
+ t = datetime.datetime.utcnow()
+ t2 = t - datetime.timedelta(minutes=10)
+
+ #modify the object
+ self.upload_object(self.account,
+ self.containers[1],
+ self.objects[0]['name'],
+ self.objects[0]['data'][:200])
+
+ for f in DATE_FORMATS:
+ past = t2.strftime(f)
+
+ headers = {'HTTP_IF_UNMODIFIED_SINCE':'%s' %past}
+ r = self.get_object(self.account,
+ self.containers[1],
+ self.objects[0]['name'],
+ **headers)
+ #assert get success
+ self.assertEqual(r.status_code, 412)
+
+ def test_hashes(self):
+ l = 8388609
+ fname = 'largefile'
+ o = self.upload_random_data(self.account,
+ self.containers[1],
+ fname,
+ l)
+ if o:
+ r = self.get_object(self.account,
+ self.containers[1],
+ fname,
+ 'json')
+ body = json.loads(r.content)
+ hashes = body['hashes']
+ block_size = body['block_size']
+ block_hash = body['block_hash']
+ block_num = l/block_size == 0 and l/block_size or l/block_size + 1
+ self.assertTrue(len(hashes), block_num)
+ i = 0
+ for h in hashes:
+ start = i * block_size
+ end = (i + 1) * block_size
+ hash = compute_block_hash(o['data'][start:end], block_hash)
+ self.assertEqual(h, hash)
+ i += 1
+
+class ObjectPut(BaseTestCase):
def setUp(self):
BaseTestCase.setUp(self)
self.account = 'test'
create_chunked_update_test_file(self.src, self.dest)
def tearDown(self):
- for o in get_content_splitted(self.list_objects(self.account, self.container)):
+ r = self.list_objects(self.account, self.container)
+ for o in get_content_splitted(r):
self.delete_object(self.account, self.container, o)
self.delete_container(self.account, self.container)
# delete test file
os.remove(self.dest)
-
+
def test_upload(self):
filename = 'tests.py'
fullpath = os.path.join('.', 'api', filename)
f = open(fullpath, 'r')
data = f.read()
- hash = compute_hash(data)
+ hash = compute_md5_hash(data)
meta={'HTTP_ETAG':hash,
- 'HTTP_X_OBJECT_MANIFEST':123,
'HTTP_X_OBJECT_META_TEST':'test1'
}
- meta['HTTP_CONTENT_TYPE'], meta['HTTP_CONTENT_ENCODING'] = mimetypes.guess_type(fullpath)
+ type, enc = mimetypes.guess_type(fullpath)
+ meta['HTTP_CONTENT_TYPE'] = type and type or 'plain/text'
+ if enc:
+ meta['HTTP_CONTENT_ENCODING'] = enc
r = self.upload_object(self.account,
self.container,
filename,
self.assertEqual(r.status_code, 201)
r = self.get_object_meta(self.account, self.container, filename)
self.assertTrue(r['X-Object-Meta-Test'])
- self.assertEqual(r['X-Object-Meta-Test'], meta['HTTP_X_OBJECT_META_TEST'])
+ self.assertEqual(r['X-Object-Meta-Test'],
+ meta['HTTP_X_OBJECT_META_TEST'])
#assert uploaded content
r = self.get_object(self.account, self.container, filename)
f = open(fullpath, 'r')
data = f.read()
meta={'HTTP_ETAG':'123',
- 'HTTP_X_OBJECT_MANIFEST':123,
'HTTP_X_OBJECT_META_TEST':'test1'
}
- meta['HTTP_CONTENT_TYPE'], meta['HTTP_CONTENT_ENCODING'] = mimetypes.guess_type(fullpath)
+ type, enc = mimetypes.guess_type(fullpath)
+ meta['HTTP_CONTENT_TYPE'] = type and type or 'plain/text'
+ if enc:
+ meta['HTTP_CONTENT_ENCODING'] = enc
r = self.upload_object(self.account,
self.container,
filename,
content_type = meta['HTTP_CONTENT_TYPE'],
**meta)
self.assertEqual(r.status_code, 422)
-
+
def test_chucked_update(self):
objname = os.path.split(self.src)[-1:][0]
f = open(self.dest, 'r')
data = f.read()
meta = {}
- meta['HTTP_CONTENT_TYPE'], meta['HTTP_CONTENT_ENCODING'] = mimetypes.guess_type(objname)
+ type, enc = mimetypes.guess_type(self.dest)
+ meta['HTTP_CONTENT_TYPE'] = type and type or 'plain/text'
+ if enc:
+ meta['HTTP_CONTENT_ENCODING'] = enc
meta.update({'HTTP_TRANSFER_ENCODING':'chunked'})
r = self.upload_object(self.account,
self.container,
actual_data = f.read()
self.assertEqual(actual_data, uploaded_data)
-class CopyObject(BaseTestCase):
+class ObjectCopy(BaseTestCase):
def setUp(self):
BaseTestCase.setUp(self)
self.account = 'test'
self.delete_container(self.account, c)
def test_copy(self):
- with AssertInvariant(self.get_object_meta, self.account, self.containers[0], self.obj['name']):
+ with AssertInvariant(self.get_object_meta, self.account,
+ self.containers[0], self.obj['name']):
#perform copy
meta = {'HTTP_X_OBJECT_META_TEST':'testcopy'}
src_path = os.path.join('/', self.containers[0], self.obj['name'])
self.assertEqual(r['ETag'], self.obj['hash'])
#assert src object still exists
- r = self.get_object_meta(self.account, self.containers[0], self.obj['name'])
- self.assertEqual(r.status_code, 204)
+ r = self.get_object_meta(self.account, self.containers[0],
+ self.obj['name'])
+ self.assertEqual(r.status_code, 200)
def test_copy_from_different_container(self):
with AssertInvariant(self.get_object_meta,
self.assertTrue(r['X-Object-Meta-Test'], 'testcopy')
#assert src object still exists
- r = self.get_object_meta(self.account, self.containers[0], self.obj['name'])
- self.assertEqual(r.status_code, 204)
+ r = self.get_object_meta(self.account, self.containers[0],
+ self.obj['name'])
+ self.assertEqual(r.status_code, 200)
def test_copy_invalid(self):
#copy from invalid object
**meta)
self.assertItemNotFound(r)
-
#copy from invalid container
meta = {'HTTP_X_OBJECT_META_TEST':'testcopy'}
src_path = os.path.join('/', self.containers[1], self.obj['name'])
**meta)
self.assertItemNotFound(r)
-class MoveObject(CopyObject):
+class ObjectMove(ObjectCopy):
def test_move(self):
#perform move
meta = {'HTTP_X_OBJECT_META_TEST':'testcopy'}
self.assertTrue(r['X-Object-Meta-Test'], 'testcopy')
#assert src object no more exists
- r = self.get_object_meta(self.account, self.containers[0], self.obj['name'])
+ r = self.get_object_meta(self.account, self.containers[0],
+ self.obj['name'])
self.assertItemNotFound(r)
-class UpdateObjectMeta(BaseTestCase):
+class ObjectPost(BaseTestCase):
def setUp(self):
BaseTestCase.setUp(self)
self.account = 'test'
self.delete_object(self.account, c, o)
self.delete_container(self.account, c)
- def test_update(self):
+ def test_update_meta(self):
#perform update metadata
more = {'HTTP_X_OBJECT_META_FOO':'foo',
'HTTP_X_OBJECT_META_BAR':'bar'}
- r = self.update_object_meta(self.account,
+ r = self.update_object(self.account,
self.containers[0],
self.obj['name'],
**more)
self.assertEqual(r.status_code, 202)
#assert old metadata are still there
- r = self.get_object_meta(self.account, self.containers[0], self.obj['name'])
+ r = self.get_object_meta(self.account, self.containers[0],
+ self.obj['name'])
self.assertTrue('X-Object-Meta-Test' not in r.items())
#assert new metadata have been updated
self.assertTrue(r[key])
self.assertTrue(r[key], v)
-class DeleteObject(BaseTestCase):
+ def test_update_object(self,
+ first_byte_pos=0,
+ last_byte_pos=499,
+ instance_length = True,
+ content_length = 500):
+ l = len(self.obj['data'])
+ length = instance_length and l or '*'
+ range = 'bytes %d-%d/%s' %(first_byte_pos,
+ last_byte_pos,
+ length)
+ partial = last_byte_pos - first_byte_pos + 1
+ data = get_random_data(partial)
+ more = {'HTTP_CONTENT_RANGE':range}
+ if content_length:
+ more.update({'CONTENT_LENGTH':'%s' % content_length})
+
+ r = self.update_object(self.account,
+ self.containers[0],
+ self.obj['name'],
+ data,
+ 'application/octet-stream',
+ **more)
+
+ if partial < 0 or (instance_length and l <= last_byte_pos):
+ self.assertEqual(r.status_code, 416)
+ elif content_length and content_length != partial:
+ self.assertEqual(r.status_code, 400)
+ else:
+ self.assertEqual(r.status_code, 204)
+
+ #check modified object
+ r = self.get_object(self.account,
+ self.containers[0],
+ self.obj['name'])
+ self.assertEqual(r.content[0:partial], data)
+ self.assertEqual(r.content[partial:l], self.obj['data'][partial:l])
+
+ def test_update_object_no_content_length(self):
+ self.test_update_object(content_length = None)
+
+ def test_update_object_invalid_content_length(self):
+ with AssertContentInvariant(self.get_object, self.account, self.containers[0],
+ self.obj['name']):
+ self.test_update_object(content_length = 1000)
+
+ def test_update_object_with_unknown_instance_length(self):
+ self.test_update_object(instance_length = False)
+
+ def test_update_object_invalid_range(self):
+ with AssertContentInvariant(self.get_object, self.account, self.containers[0],
+ self.obj['name']):
+ self.test_update_object(499, 0, True)
+
+ def test_update_object_invalid_range_and_length(self):
+ with AssertContentInvariant(self.get_object, self.account, self.containers[0],
+ self.obj['name']):
+ self.test_update_object(499, 0, True, -1)
+
+ def test_update_object_invalid_range_with_no_content_length(self):
+ with AssertContentInvariant(self.get_object, self.account, self.containers[0],
+ self.obj['name']):
+ self.test_update_object(499, 0, True, content_length = None)
+
+ def test_update_object_out_of_limits(self):
+ with AssertContentInvariant(self.get_object, self.account, self.containers[0],
+ self.obj['name']):
+ l = len(self.obj['data'])
+ self.test_update_object(0, l+1, True)
+
+ def test_append(self):
+ data = get_random_data(500)
+ more = {'CONTENT_LENGTH':'500',
+ 'HTTP_CONTENT_RANGE':'bytes */*'}
+
+ r = self.update_object(self.account,
+ self.containers[0],
+ self.obj['name'],
+ data,
+ 'application/octet-stream',
+ **more)
+
+ self.assertEqual(r.status_code, 204)
+
+ r = self.get_object(self.account,
+ self.containers[0],
+ self.obj['name'])
+ self.assertEqual(len(r.content), len(self.obj['data']) + 500)
+ self.assertEqual(r.content[:-500], self.obj['data'])
+
+ def test_update_with_chunked_transfer(self):
+ data, pure = create_random_chunked_data()
+ dl = len(pure)
+ fl = len(self.obj['data'])
+ meta = {'HTTP_TRANSFER_ENCODING':'chunked',
+ 'HTTP_CONTENT_RANGE':'bytes 0-/%d' %fl}
+ r = self.update_object(self.account,
+ self.containers[0],
+ self.obj['name'],
+ data,
+ 'application/octet-stream',
+ **meta)
+
+ #check modified object
+ r = self.get_object(self.account,
+ self.containers[0],
+ self.obj['name'])
+ self.assertEqual(r.content[0:dl], pure)
+ self.assertEqual(r.content[dl:fl], self.obj['data'][dl:fl])
+
+ def test_update_with_chunked_transfer_strict_range(self):
+ data, pure = create_random_chunked_data()
+ dl = len(pure) - 1
+ fl = len(self.obj['data'])
+ meta = {'HTTP_TRANSFER_ENCODING':'chunked',
+ 'HTTP_CONTENT_RANGE':'bytes 0-%d/%d' %(dl, fl)}
+ r = self.update_object(self.account,
+ self.containers[0],
+ self.obj['name'],
+ data,
+ 'application/octet-stream',
+ **meta)
+
+ #check modified object
+ r = self.get_object(self.account,
+ self.containers[0],
+ self.obj['name'])
+ self.assertEqual(r.content[0:dl+1], pure)
+ self.assertEqual(r.content[dl+1:fl], self.obj['data'][dl+1:fl])
+
+class ObjectDelete(BaseTestCase):
def setUp(self):
BaseTestCase.setUp(self)
self.account = 'test'
def test_delete(self):
#perform delete object
- r = self.delete_object(self.account, self.containers[0], self.obj['name'])
+ r = self.delete_object(self.account, self.containers[0],
+ self.obj['name'])
#assert success
self.assertEqual(r.status_code, 204)
-
+
def test_delete_invalid(self):
#perform delete object
- r = self.delete_object(self.account, self.containers[1], self.obj['name'])
+ r = self.delete_object(self.account, self.containers[1],
+ self.obj['name'])
#assert failure
self.assertItemNotFound(r)
items = self.callable(*self.args, **self.kwargs).items()
assert self.items == items
+class AssertContentInvariant(object):
+ def __init__(self, callable, *args, **kwargs):
+ self.callable = callable
+ self.args = args
+ self.kwargs = kwargs
+
+ def __enter__(self):
+ self.content = self.callable(*self.args, **self.kwargs).content
+ return self.content
+
+ def __exit__(self, type, value, tb):
+ content = self.callable(*self.args, **self.kwargs).content
+ assert self.content == content
+
def get_content_splitted(response):
if response:
return response.content.split('\n')
-def compute_hash(data):
+def compute_md5_hash(data):
md5 = hashlib.md5()
offset = 0
md5.update(data)
return md5.hexdigest().lower()
+def compute_block_hash(data, algorithm):
+ h = hashlib.new(algorithm)
+ h.update(data.rstrip('\x00'))
+ return h.hexdigest()
+
def create_chunked_update_test_file(src, dest):
fr = open(src, 'r')
fw = open(dest, 'w')
fw.write(hex(0))
fw.write('\r\n')
+def create_random_chunked_data(rows=5):
+ i = 0
+ out = []
+ pure= []
+ while i < rows:
+ data = get_random_data(random.randint(1, 100))
+ out.append(hex(len(data)))
+ out.append(data)
+ pure.append(data)
+ i+=1
+ out.append(hex(0))
+ out.append('\r\n')
+ return '\r\n'.join(out), ''.join(pure)
+
+def get_random_data(length=500):
+ char_set = string.ascii_uppercase + string.digits
+ return ''.join(random.choice(char_set) for x in range(length))
+
o_names = ['kate.jpg',
'kate_beckinsale.jpg',
'How To Win Friends And Influence People.pdf',
'photos/animals/cats/siamese.jpg',
'photos/plants/fern.jpg',
'photos/plants/rose.jpg',
- 'photos/me.jpg']
\ No newline at end of file
+ 'photos/me.jpg']
projects / pithos / blobdiff