# 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 pithos.lib.client import Client, Fault
+import unittest
from django.utils import simplejson as json
from xml.dom import minidom
import types
import random
import datetime
import string
-from pithos.backends import backend
+import re
+
+#from pithos.backends import backend
DATE_FORMATS = ["%a %b %d %H:%M:%S %Y",
"%A, %d-%b-%y %H:%M:%S GMT",
"%a, %d %b %Y %H:%M:%S GMT"]
-class AaiClient(Client):
- def request(self, **request):
- request['HTTP_X_AUTH_TOKEN'] = '0000'
- return super(AaiClient, self).request(**request)
+DEFAULT_HOST = 'pithos.dev.grnet.gr'
+#DEFAULT_HOST = '127.0.0.1:8000'
+DEFAULT_API = 'v1'
+DEFAULT_USER = 'papagian'
+DEFAULT_AUTH = '0004'
-class BaseTestCase(TestCase):
+class BaseTestCase(unittest.TestCase):
#TODO unauthorized request
def setUp(self):
- self.client = AaiClient()
+ self.client = Client(DEFAULT_HOST, DEFAULT_AUTH, DEFAULT_USER, DEFAULT_API)
+ self.invalid_client = Client(DEFAULT_HOST, DEFAULT_AUTH, 'non-existing', DEFAULT_API)
+ self.unauthorised_client = Client(DEFAULT_HOST, '', DEFAULT_USER, DEFAULT_API)
self.headers = {
- 'account':(
- 'X-Account-Container-Count',
- 'X-Account-Bytes-Used',
- 'Last-Modified',
- 'Content-Length',
- 'Date',
- 'Content-Type',),
- 'container':(
- 'X-Container-Object-Count',
- 'X-Container-Bytes-Used',
- 'Content-Type',
- 'Last-Modified',
- 'Content-Length',
- 'Date',
- 'X-Container-Block-Size',
- 'X-Container-Block-Hash',
- 'X-Container-Policy-Quota',
- 'X-Container-Policy-Versioning',),
- 'object':(
- 'ETag',
- 'Content-Length',
- 'Content-Type',
- 'Content-Encoding',
- 'Last-Modified',
- 'Date',
- 'X-Object-Manifest',
- 'Content-Range',
- 'X-Object-Modified-By',
- 'X-Object-Version',
- 'X-Object-Version-Timestamp',)}
+ 'account': ('x-account-container-count',
+ 'x-account-bytes-used',
+ 'last-modified',
+ 'content-length',
+ 'date',
+ 'content-type',
+ 'server',),
+ 'object': ('etag',
+ 'content-length',
+ 'content-type',
+ 'content-encoding',
+ 'last-modified',
+ 'date',
+ 'x-object-manifest',
+ 'content-range',
+ 'x-object-modified-by',
+ 'x-object-version',
+ 'x-object-version-timestamp',
+ 'server',),
+ 'container': ('x-container-object-count',
+ 'x-container-bytes-used',
+ 'content-type',
+ 'last-modified',
+ 'content-length',
+ 'date',
+ 'x-container-block-size',
+ 'x-container-block-hash',
+ 'x-container-policy-quota',
+ 'x-container-policy-versioning',
+ 'server',
+ 'x-container-object-meta',
+ 'x-container-policy-versioning',
+ 'server',)}
+
self.contentTypes = {'xml':'application/xml',
'json':'application/json',
'':'text/plain'}
'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')
-
- def assertUnauthorized(self, response):
- self.assertFault(response, 401, 'unauthorized')
-
- def assertServiceUnavailable(self, response):
- self.assertFault(response, 503, 'serviceUnavailable')
-
- def assertNonEmpty(self, response):
- self.assertFault(response, 409, 'nonEmpty')
-
- def assert_status(self, response, codes):
+ def assert_status(self, status, codes):
l = [elem for elem in self.return_codes]
if type(codes) == types.ListType:
l.extend(codes)
else:
l.append(codes)
- self.assertTrue(response.status_code in l)
-
- def get_account_meta(self, account, exp_meta={}):
- path = '/v1/%s' % account
- response = self.client.head(path)
- self.assert_status(response, 204)
- self.assert_headers(response, 'account', exp_meta)
- return response
-
- 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, **headers)
- self.assert_status(response, [200, 204, 304, 412])
- response.content = response.content.strip()
- if format:
- self.assert_extended(response, format, 'container', limit)
+ self.assertTrue(status in l)
+
+ def assert_list(self, path, entity, limit=10000, format='text', params=None, **headers):
+ status, headers, data = self.client.get(path, format=format,
+ headers=headers, params=params)
+
+ self.assert_status(status, [200, 204, 304, 412])
+ if format == 'text':
+ data = data.strip().split('\n') if data else []
+ self.assertTrue(len(data) <= limit)
else:
- names = get_content_splitted(response)
- self.assertTrue(len(names) <= limit)
- return response
-
- def update_account_meta(self, account, **metadata):
- path = '/v1/%s' % account
- response = self.client.post(path, **metadata)
- response.content = response.content
- self.assert_status(response, 202)
- return response
-
- def get_container_meta(self, account, container, exp_meta={}):
+ exp_content_type = self.contentTypes[format]
+ self.assertEqual(headers['content-type'].find(exp_content_type), 0)
+ #self.assert_extended(data, format, entity, limit)
+ if format == 'json':
+ data = json.loads(data) if data else []
+ elif format == 'xml':
+ data = minidom.parseString(data)
+ return status, headers, data
+
+ def list_containers(self, limit=10000, marker='', format='text', **headers):
params = locals()
params.pop('self')
- params.pop('account')
- params.pop('container')
- path = '/v1/%s/%s' %(account, container)
- response = self.client.head(path, params)
- 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='',
+ return self.assert_list('', 'account', limit, format, params, **headers)
+
+ def list_objects(self, 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, **headers)
- response.content = response.content.strip()
- if format:
- self.assert_extended(response, format, 'object', limit)
- 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
- 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=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
- 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
- self.assert_status(response, 200)
- return response
-
- def get_object(self, account, container, name, format='', **headers):
- path = '/v1/%s/%s/%s' %(account, container, name)
- 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='',
- **headers):
- path = '/v1/%s/%s/%s' %(account, container, name)
- response = self.client.put(path, data, content_type, **headers)
- response.content = response.content
- self.assert_status(response, [201, 411, 422])
- if response.status_code == 201:
- self.assertTrue(response['Etag'])
- return response
-
- def copy_object(self, account, container, name, src, **headers):
- path = '/v1/%s/%s/%s' %(account, container, name)
- headers['HTTP_X_COPY_FROM'] = src
- response = self.client.put(path, **headers)
- response.content = response.content
- self.assert_status(response, 201)
- return response
-
- def move_object(self, account, container, name, src, **headers):
- path = '/v1/%s/%s/%s' % (account, container, name)
- headers['HTTP_X_MOVE_FROM'] = src
- response = self.client.put(path, **headers)
- response.content = response.content
- self.assert_status(response, 201)
- return response
-
- def update_object(self, account, container, name, data='',
- content_type='', **headers):
- path = '/v1/%s/%s/%s' %(account, container, name)
- 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
- self.assert_status(response, 204)
- return response
-
- def assert_headers(self, response, type, exp_meta={}):
- 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()]
- 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):
- 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':
- self.assert_json(response, type, size)
-
- def assert_json(self, response, type, size):
- convert = lambda s: s.lower()
- info = [convert(elem) for elem in self.extended[type]]
- data = json.loads(response.content)
- self.assertTrue(len(data) <= size)
- for item in info:
- for i in data:
- if 'subdir' in i.keys():
- continue
- self.assertTrue(item in i.keys())
-
- def assert_xml(self, response, type, size):
- convert = lambda s: s.lower()
- info = [convert(elem) for elem in self.extended[type]]
- try:
- info.remove('content_encoding')
- except ValueError:
- pass
- xml = minidom.parseString(response.content)
- for item in info:
- nodes = xml.getElementsByTagName(item)
- self.assertTrue(nodes)
- self.assertTrue(len(nodes) <= size)
-
+ path = '/' + container
+ format = 'text' if format == '' else format
+ return self.assert_list(path, 'container', limit, format, params, **headers)
+
+ def _assert_get_meta(self, path, entity, params=None, **exp_meta):
+ status, headers, data = self.client.head(path, params)
+ self.assert_status(status, 204)
+ #self.assert_headers(headers, entity, **exp_meta)
+ return status, headers, data
+
+ def get_account_meta(self, params=None, **exp_meta):
+ return self._assert_get_meta('', 'account', params, **exp_meta)
- def upload_os_file(self, account, container, fullpath, meta={}):
- try:
- f = open(fullpath, 'r')
- data = f.read()
- name = os.path.split(fullpath)[-1]
- return self.upload_data(account, container, name, data)
- except IOError:
- return
+ def get_container_meta(self, container, params=None, **exp_meta):
+ path = '/%s' % container
+ return self._assert_get_meta(path, 'container', params, **exp_meta)
- def upload_random_data(self, account, container, name, length=1024,
- meta={}):
+ def create_container(self, name, **meta):
+ headers = {}
+ for k,v in meta.items():
+ headers['x-container-meta-%s' %k.strip().upper()] = v.strip()
+ status, header, data = self.client.put('/' + name, headers=headers)
+ self.assert_status(status, [201, 202])
+ return status, header, data
+
+ def get_object(self, container, name, format='', version=None, **headers):
+ path = '/%s/%s' % (container, name)
+ params = {'version':version} if version else None
+ status, headers, data = self.client.get(path, format, headers, params)
+ self.assert_status(status, [200, 206, 304, 412, 416])
+ #if status in [200, 206]:
+ # self.assert_headers(headers, 'object')
+ return status, headers, data
+
+ def update_object(self, container, name, data='', content_type='', **headers):
+ if content_type != '':
+ headers['content-type'] = content_type
+ status, headers, data = self.client.update_object_data(container,
+ name,
+ data,
+ headers)
+ self.assert_status(status, [202, 204, 416])
+ return status, headers, data
+
+ def assert_headers(self, headers, type, **exp_meta):
+ prefix = 'x-%s-meta-' %type
+ system_headers = [h for h in headers if not h.startswith(prefix)]
+ for k,v in headers.items():
+ if k in system_headers:
+ self.assertTrue(k in headers[type])
+ elif exp_meta:
+ k = k.split(prefix)[-1]
+ self.assertEqual(v, exp_meta[k])
+
+ #def assert_extended(self, data, format, type, size):
+ # if format == 'xml':
+ # self._assert_xml(data, type, size)
+ # elif format == 'json':
+ # self._assert_json(data, type, size)
+ #
+ #def _assert_json(self, data, type, size):
+ # print '#', data
+ # convert = lambda s: s.lower()
+ # info = [convert(elem) for elem in self.extended[type]]
+ # data = json.loads(data)
+ # self.assertTrue(len(data) <= size)
+ # for item in info:
+ # for i in data:
+ # if 'subdir' in i.keys():
+ # continue
+ # self.assertTrue(item in i.keys())
+ #
+ #def _assert_xml(self, data, type, size):
+ # print '#', data
+ # convert = lambda s: s.lower()
+ # info = [convert(elem) for elem in self.extended[type]]
+ # try:
+ # info.remove('content_encoding')
+ # except ValueError:
+ # pass
+ # xml = minidom.parseString(data)
+ # entities = xml.getElementsByTagName(type)
+ # self.assertTrue(len(entities) <= size)
+ # for e in entities:
+ # for item in info:
+ # self.assertTrue(e.hasAttribute(item))
+
+ def assert_raises_fault(self, status, callableObj, *args, **kwargs):
+ """
+ asserts that a Fault with a specific status is raised
+ when callableObj is called with the specific arguments
+ """
+ try:
+ callableObj(*args, **kwargs)
+ self.fail('Should never reach here')
+ except Fault, f:
+ self.failUnless(f.status == status)
+
+ def assert_object_exists(self, container, object):
+ """
+ asserts the existence of an object
+ """
+ try:
+ self.client.retrieve_object_metadata(container, object)
+ except Fault, f:
+ self.failIf(f.status == 404)
+
+ def assert_object_not_exists(self, container, object):
+ """
+ asserts there is no such an object
+ """
+ self.assert_raises_fault(404, self.client.retrieve_object_metadata,
+ container, object)
+
+ def upload_random_data(self, container, name, length=1024, type=None,
+ enc=None, **meta):
data = get_random_data(length)
- return self.upload_data(account, container, name, data, meta)
+ return self.upload_data(container, name, data, type, enc, **meta)
- def upload_data(self, account, container, name, data, meta={}):
+ def upload_data(self, container, name, data, type=None, enc=None, etag=None,
+ **meta):
obj = {}
obj['name'] = name
try:
obj['data'] = data
obj['hash'] = compute_md5_hash(obj['data'])
- meta.update({'HTTP_X_OBJECT_META_TEST':'test1',
- 'HTTP_ETAG':obj['hash']})
- type, enc = mimetypes.guess_type(name)
- meta['HTTP_CONTENT_TYPE'] = type and type or 'plain/text'
- if enc:
- meta['HTTP_CONTENT_ENCODING'] = enc
- obj['meta'] = meta
- r = self.upload_object(account,
- container,
- obj['name'],
- obj['data'],
- meta['HTTP_CONTENT_TYPE'],
- **meta)
- if r.status_code == 201:
+ headers = {}
+ for k,v in meta.items():
+ key = 'x-object-meta-%s' % k
+ headers[key] = v
+ headers['etag'] = etag if etag else obj['hash']
+ guess = mimetypes.guess_type(name)
+ type = type if type else guess[0]
+ enc = enc if enc else guess[1]
+ headers['content-type'] = type if type else 'plain/text'
+ headers['content-encoding'] = enc if enc else None
+ obj['meta'] = headers
+
+ path = '/%s/%s' % (container, name)
+ status, headers, data = self.client.put(path, obj['data'],
+ headers=headers)
+ if status == 201:
+ self.assertTrue('etag' in headers)
+ self.assertEqual(obj['hash'], headers['etag'])
return obj
except IOError:
return
self.account = 'test'
self.containers = ['apples', 'bananas', 'kiwis', 'oranges', 'pears']
for item in self.containers:
- self.create_container(self.account, item)
+ self.create_container(item)
def tearDown(self):
- for c in get_content_splitted(self.list_containers(self.account)):
- self.delete_container(self.account, c)
-
+ for c in self.list_containers()[2]:
+ self.client.delete_container(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)
+ headers = self.get_account_meta()[1]
+
+ containers = self.list_containers()[2]
l = str(len(containers))
- self.assertEqual(response['X-Account-Container-Count'], l)
+ self.assertEqual(headers['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))
+ h = self.get_container_meta(c)[1]
+ size = size + int(h['x-container-bytes-used'])
+ self.assertEqual(headers['x-account-bytes-used'], str(size))
#def test_get_account_401(self):
# response = self.get_account_meta('non-existing-account')
#create some containers
self.containers = ['apples', 'bananas', 'kiwis', 'oranges', 'pears']
for item in self.containers:
- self.create_container(self.account, item)
+ self.create_container(item)
def tearDown(self):
- for c in get_content_splitted(self.list_containers(self.account)):
- response = self.delete_container(self.account, c)
+ for c in self.list_containers()[2]:
+ for o in self.list_objects(c)[2]:
+ self.client.delete_object(c, o)
+ self.client.delete_container(c)
def test_list(self):
#list containers
- response = self.list_containers(self.account)
- containers = get_content_splitted(response)
+ containers = self.list_containers()[2]
self.assertEquals(self.containers, containers)
#def test_list_204(self):
def test_list_with_limit(self):
limit = 2
- response = self.list_containers(self.account, limit=limit)
- containers = get_content_splitted(response)
+ containers = self.list_containers(limit=limit)[2]
self.assertEquals(len(containers), limit)
self.assertEquals(self.containers[:2], containers)
def test_list_with_marker(self):
l = 2
m = 'bananas'
- response = self.list_containers(self.account, limit=l, marker=m)
- containers = get_content_splitted(response)
+ containers = self.list_containers(limit=l, marker=m)[2]
i = self.containers.index(m) + 1
self.assertEquals(self.containers[i:(i+l)], containers)
m = 'oranges'
- response = self.list_containers(self.account, limit=l, marker=m)
- containers = get_content_splitted(response)
+ containers = self.list_containers(limit=l, marker=m)[2]
i = self.containers.index(m) + 1
self.assertEquals(self.containers[i:(i+l)], containers)
def test_list_json_with_marker(self):
l = 2
m = 'bananas'
- response = self.list_containers(self.account, limit=l, marker=m,
+ status, headers, containers = self.list_containers(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):
l = 2
m = 'oranges'
- response = self.list_containers(self.account, limit=l, marker=m,
+ status, headers, xml = self.list_containers(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')
t2 = t - datetime.timedelta(minutes=10)
#add a new container
- self.create_container(self.account,
- 'dummy')
+ self.create_container('dummy')
for f in DATE_FORMATS:
past = t2.strftime(f)
- headers = {'HTTP_IF_MODIFIED_SINCE':'%s' %past}
- r = self.list_containers(self.account, **headers)
+ headers = {'if-modified-since':'%s' %past}
+ status, headers, data = self.list_containers(**headers)
#assert get success
- self.assertEqual(r.status_code, 200)
+ self.assertEqual(status, 200)
def test_if_modified_since_invalid_date(self):
- headers = {'HTTP_IF_MODIFIED_SINCE':''}
- r = self.list_containers(self.account, **headers)
+ headers = {'if-modified-since':''}
+ status, headers, data = self.list_containers(**headers)
#assert get success
- self.assertEqual(r.status_code, 200)
+ self.assertEqual(status, 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)
-
+ headers = {'if-modified-since':'%s' %since.strftime(f)}
#assert not modified
- self.assertEqual(r.status_code, 304)
+ self.assert_raises_fault(304, self.list_containers, **headers)
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)
+ headers = {'if-unmodified-since':'%s' %since.strftime(f)}
+ status, headers, data = self.list_containers(**headers)
#assert success
- self.assertEqual(r.status_code, 200)
- self.assertEqual(self.containers, get_content_splitted(r))
+ self.assertEqual(status, 200)
+ self.assertEqual(self.containers, data)
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')
+ self.create_container('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)
+ headers = {'if-unmodified-since':'%s' %past}
+ #assert precondition failed
+ self.assert_raises_fault(412, self.list_containers, **headers)
class AccountPost(BaseTestCase):
def setUp(self):
self.account = 'test'
self.containers = ['apples', 'bananas', 'kiwis', 'oranges', 'pears']
for item in self.containers:
- self.create_container(self.account, item)
+ self.create_container(item)
def tearDown(self):
- for c in get_content_splitted(self.list_containers(self.account)):
- self.delete_container(self.account, c)
+ containers = self.list_containers()[2]
+ for c in containers:
+ self.client.delete_container(c)
def test_update_meta(self):
- 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():
- key = '-'.join(elem.capitalize() for elem in k.split('_')[1:])
- self.assertTrue(response[key])
- self.assertEqual(response[key], v)
-
+ meta = {'test':'test', 'tost':'tost'}
+ status, headers, data = self.get_account_meta(**meta)
+
#def test_invalid_account_update_meta(self):
- # with AssertInvariant(self.get_account_meta, self.account):
+ # with AssertMappingInvariant(self.get_account_meta, self.account):
# meta = {'HTTP_X_ACCOUNT_META_TEST':'test',
# 'HTTP_X_ACCOUNT_META_TOST':'tost'}
# response = self.update_account_meta('non-existing-account', **meta)
-
+
class ContainerHead(BaseTestCase):
def setUp(self):
BaseTestCase.setUp(self)
self.account = 'test'
self.container = 'apples'
- self.create_container(self.account, self.container)
+ status = self.create_container(self.container)[0]
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)
+ for o in self.list_objects(self.container)[2]:
+ self.client.delete_object(self.container, o)
+ self.client.delete_container(self.container)
def test_get_meta(self):
- headers = {'HTTP_X_OBJECT_META_TRASH':'true'}
+ meta = {'trash':'true'}
t1 = datetime.datetime.utcnow()
- o = self.upload_random_data(self.account,
- self.container,
- 'McIntosh.jpg',
- meta=headers)
+ o = self.upload_random_data(self.container, o_names[0], **meta)
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])
+ status, headers, data = self.get_container_meta(self.container)
+ self.assertEqual(headers['x-container-object-count'], '1')
+ self.assertEqual(headers['x-container-bytes-used'], str(len(o['data'])))
+ t2 = datetime.datetime.strptime(headers['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'])
+ self.assertTrue(headers['x-container-object-meta'])
+ self.assertTrue('Trash' in headers['x-container-object-meta'])
class ContainerGet(BaseTestCase):
def setUp(self):
self.account = 'test'
self.container = ['pears', 'apples']
for c in self.container:
- self.create_container(self.account, c)
+ self.create_container(c)
self.obj = []
for o in o_names[:8]:
- self.obj.append(self.upload_random_data(self.account,
- self.container[0],
- o))
+ self.obj.append(self.upload_random_data(self.container[0], o))
for o in o_names[8:]:
- self.obj.append(self.upload_random_data(self.account,
- self.container[1],
- o))
+ self.obj.append(self.upload_random_data(self.container[1], o))
def tearDown(self):
for c in self.container:
- 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 self.list_objects(c)[2]:
+ self.client.delete_object(c, obj)
+ self.client.delete_container(c)
def test_list_objects(self):
- response = self.list_objects(self.account, self.container[0])
- objects = get_content_splitted(response)
+ objects = self.list_objects(self.container[0])[2]
l = [elem['name'] for elem in self.obj[:8]]
l.sort()
self.assertEqual(objects, l)
def test_list_objects_with_limit_marker(self):
- response = self.list_objects(self.account, self.container[0], limit=2)
- objects = get_content_splitted(response)
+ objects = self.list_objects(self.container[0], limit=2)[2]
l = [elem['name'] for elem in self.obj[:8]]
l.sort()
self.assertEqual(objects, l[:2])
'moms_birthday.jpg']
limit = 4
for m in markers:
- response = self.list_objects(self.account, self.container[0],
- limit=limit, marker=m)
- objects = get_content_splitted(response)
+ objects = self.list_objects(self.container[0], limit=limit,
+ marker=m)[2]
l = [elem['name'] for elem in self.obj[:8]]
l.sort()
start = l.index(m) + 1
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='/')
- objects = get_content_splitted(response)
+ objects = self.list_objects(self.container[1], prefix='photos',
+ delimiter='/')[2]
self.assertEquals(['photos/animals/', 'photos/me.jpg',
'photos/plants/'], objects)
- response = self.list_objects(self.account, self.container[1],
- prefix='photos/animals', delimiter='/')
- objs = get_content_splitted(response)
+ objects = self.list_objects(self.container[1], prefix='photos/animals',
+ delimiter='/')[2]
l = ['photos/animals/cats/', 'photos/animals/dogs/']
- self.assertEquals(l, objs)
+ self.assertEquals(l, objects)
- response = self.list_objects(self.account, self.container[1],
- path='photos')
- objects = get_content_splitted(response)
+ objects = self.list_objects(self.container[1], path='photos')[2]
self.assertEquals(['photos/me.jpg'], objects)
def test_extended_list_json(self):
- response = self.list_objects(self.account,
- self.container[1],
+ objects = self.list_objects(self.container[1],
format='json', limit=2,
prefix='photos/animals',
- delimiter='/')
- objects = json.loads(response.content)
+ delimiter='/')[2]
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='/')
- xml = minidom.parseString(response.content)
+ xml = self.list_objects(self.container[1], format='xml', limit=4,
+ prefix='photos', delimiter='/')[2]
dirs = xml.getElementsByTagName('subdir')
self.assertEqual(len(dirs), 2)
self.assertEqual(dirs[0].attributes['name'].value, 'photos/animals/')
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)
+ meta = {'quality':'aaa', 'stock':'true'}
+ self.client.update_object_metadata(self.container[0],
+ self.obj[0]['name'], **meta)
+ obj = self.list_objects(self.container[0], meta='Quality,Stock')[2]
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'}
+ meta = {'quality':'aaa'}
for o in self.obj[:2]:
- r = self.update_object(self.account,
- self.container[0],
- o['name'],
- **meta)
- meta = {'HTTP_X_OBJECT_META_STOCK':'true'}
+ self.client.update_object_metadata(self.container[0], o['name'],
+ **meta)
+ meta = {'stock':'true'}
for o in self.obj[3:5]:
- 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]])
+ self.client.update_object_metadata(self.container[0], o['name'],
+ **meta)
+
+ status, headers, data = self.list_objects(self.container[0],
+ meta='Quality')
+ self.assertEqual(status, 200)
+ self.assertEqual(len(data), 2)
+ self.assertTrue(data, [o['name'] for o in self.obj[:2]])
# test case insensitive
- r = self.list_objects(self.account,
- self.container[0],
- meta='quality')
- self.assertEqual(r.status_code, 200)
- obj = get_content_splitted(r)
+ status, headers, obj = self.list_objects(self.container[0],
+ meta='quality')
+ self.assertEqual(status, 200)
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')
- self.assertEqual(r.status_code, 200)
- obj = get_content_splitted(r)
+ status, headers, obj = self.list_objects(self.container[0],
+ meta='Quality,Stock')
+ self.assertEqual(status, 200)
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')
- self.assertEqual(r.status_code, 200)
- obj = get_content_splitted(r)
+ status, headers, obj = self.list_objects(self.container[0],
+ meta='Quality,aaaa')
+ self.assertEqual(status, 200)
self.assertEqual(len(obj), 2)
self.assertTrue(obj, [o['name'] for o in self.obj[:2]])
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')
+ #add a new object
+ self.upload_random_data(self.container[0], o_names[0])
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)
+ headers = {'if-modified-since':'%s' %past}
+ status, headers, data = self.list_objects(self.container[0],
+ **headers)
#assert get success
- self.assertEqual(r.status_code, 200)
+ self.assertEqual(status, 200)
def test_if_modified_since_invalid_date(self):
- headers = {'HTTP_IF_MODIFIED_SINCE':''}
- r = self.list_objects(self.account,
- self.container[0], **headers)
+ headers = {'if-modified-since':''}
+ status, headers, data = self.list_objects(self.container[0], **headers)
#assert get success
- self.assertEqual(r.status_code, 200)
+ self.assertEqual(status, 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_objects(self.account,
- self.container[0], **headers)
-
+ headers = {'if-modified-since':'%s' %since.strftime(f)}
#assert not modified
- self.assertEqual(r.status_code, 304)
-
+ self.assert_raises_fault(304, self.list_objects, self.container[0],
+ **headers)
+
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)
-
+ headers = {'if-unmodified-since':'%s' %since.strftime(f)}
+ status, headers, data = self.list_objects(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))
+ self.assertEqual(status, 200)
+ objlist = self.list_objects(self.container[0])[2]
+ self.assertEqual(data, 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')
+ self.create_container('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)
+ headers = {'if-unmodified-since':'%s' %past}
+ #assert precondition failed
+ self.assert_raises_fault(412, self.list_objects, self.container[0],
+ **headers)
class ContainerPut(BaseTestCase):
def setUp(self):
self.containers = ['c1', 'c2']
def tearDown(self):
- for c in self.containers:
- r = self.delete_container(self.account, c)
+ for c in self.list_containers()[2]:
+ r = self.client.delete_container(c)
def test_create(self):
- response = self.create_container(self.account, self.containers[0])
- if response.status_code == 201:
- response = self.list_containers(self.account)
- 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)
+ status = self.create_container(self.containers[0])[0]
+ self.assertEqual(status, 201)
+
+ containers = self.list_containers()[2]
+ self.assertTrue(self.containers[0] in containers)
+ status = self.get_container_meta(self.containers[0])[0]
+ self.assertEqual(status, 204)
def test_create_twice(self):
- response = self.create_container(self.account, self.containers[0])
- if response.status_code == 201:
- r = self.create_container(self.account, self.containers[0])
- self.assertTrue(r.status_code, 202)
+ status, header, data = self.create_container(self.containers[0])
+ if status == 201:
+ status, header, data = self.create_container(self.containers[0])
+ self.assertTrue(status, 202)
class ContainerPost(BaseTestCase):
def setUp(self):
BaseTestCase.setUp(self)
self.account = 'test'
self.container = 'apples'
- self.create_container(self.account, self.container)
+ self.create_container(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)
+ for o in self.list_objects(self.container)[2]:
+ self.client.delete_object(self.account, self.container, o)
+ self.client.delete_container(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)
+ meta = {'test':'test33',
+ 'tost':'tost22'}
+ self.client.update_container_metadata(self.container, **meta)
+ headers = self.get_container_meta(self.container)[1]
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)
+ k = 'x-container-meta-%s' % k
+ self.assertTrue(headers[k])
+ self.assertEqual(headers[k], v)
class ContainerDelete(BaseTestCase):
def setUp(self):
self.account = 'test'
self.containers = ['c1', 'c2']
for c in self.containers:
- self.create_container(self.account, c)
- self.upload_random_data(self.account,
- self.containers[1],
- 'nice.jpg')
+ self.create_container(c)
+ self.upload_random_data(self.containers[1], o_names[0])
def tearDown(self):
- for c in self.containers:
- for o in get_content_splitted(self.list_objects(self.account, c)):
- self.delete_object(self.account, c, o)
- self.delete_container(self.account, c)
+ for c in self.list_containers()[2]:
+ for o in self.list_objects(c)[2]:
+ self.client.delete_object(c, o)
+ self.client.delete_container(c)
def test_delete(self):
- r = self.delete_container(self.account, self.containers[0])
- self.assertEqual(r.status_code, 204)
+ status = self.client.delete_container(self.containers[0])[0]
+ self.assertEqual(status, 204)
def test_delete_non_empty(self):
- r = self.delete_container(self.account, self.containers[1])
- self.assertNonEmpty(r)
+ self.assert_raises_fault(409, self.client.delete_container,
+ self.containers[1])
def test_delete_invalid(self):
- self.assertItemNotFound(self.delete_container(self.account, 'c3'))
+ self.assert_raises_fault(404, self.client.delete_container, 'c3')
class ObjectHead(BaseTestCase):
pass
self.containers = ['c1', 'c2']
#create some containers
for c in self.containers:
- self.create_container(self.account, c)
+ self.create_container(c)
#upload a file
+ names = ('obj1', 'obj2')
self.objects = []
- self.objects.append(self.upload_os_file(self.account,
- self.containers[1],
- './api/tests.py'))
- self.objects.append(self.upload_os_file(self.account,
- self.containers[1],
- 'settings.py'))
+ for n in names:
+ self.objects.append(self.upload_random_data(self.containers[1], n))
def tearDown(self):
for c in self.containers:
- for o in get_content_splitted(self.list_objects(self.account, c)):
- self.delete_object(self.account, c, o)
- self.delete_container(self.account, c)
+ for o in self.list_objects(c)[2]:
+ self.client.delete_object(c, o)
+ self.client.delete_container(c)
def test_get(self):
#perform get
- r = self.get_object(self.account,
- self.containers[1],
+ status, headers, data = self.get_object(self.containers[1],
self.objects[0]['name'],
self.objects[0]['meta'])
#assert success
- self.assertEqual(r.status_code, 200)
+ self.assertEqual(status, 200)
#assert content-type
- self.assertEqual(r['Content-Type'],
- self.objects[0]['meta']['HTTP_CONTENT_TYPE'])
+ self.assertEqual(headers['content-type'],
+ self.objects[0]['meta']['content-type'])
def test_get_invalid(self):
- r = self.get_object(self.account,
- self.containers[0],
- self.objects[0]['name'])
- self.assertItemNotFound(r)
+ self.assert_raises_fault(404, self.get_object, self.containers[0],
+ self.objects[0]['name'])
def test_get_partial(self):
#perform get with range
- headers = {'HTTP_RANGE':'bytes=0-499'}
- r = self.get_object(self.account,
- self.containers[1],
- self.objects[0]['name'],
- **headers)
+ headers = {'range':'bytes=0-499'}
+ status, headers, data = self.get_object(self.containers[1],
+ self.objects[0]['name'],
+ **headers)
#assert successful partial content
- self.assertEqual(r.status_code, 206)
+ self.assertEqual(status, 206)
#assert content-type
- self.assertEqual(r['Content-Type'],
- self.objects[0]['meta']['HTTP_CONTENT_TYPE'])
+ self.assertEqual(headers['content-type'],
+ self.objects[0]['meta']['content-type'])
#assert content length
- self.assertEqual(int(r['Content-Length']), 500)
+ self.assertEqual(int(headers['content-length']), 500)
#assert content
- self.assertEqual(self.objects[0]['data'][:500], r.content)
+ self.assertEqual(self.objects[0]['data'][:500], data)
def test_get_final_500(self):
#perform get with range
- headers = {'HTTP_RANGE':'bytes=-500'}
- r = self.get_object(self.account,
- self.containers[1],
- self.objects[0]['name'],
- **headers)
+ headers = {'range':'bytes=-500'}
+ status, headers, data = self.get_object(self.containers[1],
+ self.objects[0]['name'],
+ **headers)
#assert successful partial content
- self.assertEqual(r.status_code, 206)
+ self.assertEqual(status, 206)
#assert content-type
- self.assertEqual(r['Content-Type'],
- self.objects[0]['meta']['HTTP_CONTENT_TYPE'])
+ self.assertEqual(headers['content-type'],
+ self.objects[0]['meta']['content-type'])
#assert content length
- self.assertEqual(int(r['Content-Length']), 500)
+ self.assertEqual(int(headers['content-length']), 500)
#assert content
- self.assertTrue(self.objects[0]['data'][-500:], r.content)
+ self.assertTrue(self.objects[0]['data'][-500:], data)
def test_get_rest(self):
#perform get with range
offset = len(self.objects[0]['data']) - 500
- headers = {'HTTP_RANGE':'bytes=%s-' %offset}
- r = self.get_object(self.account,
- self.containers[1],
- self.objects[0]['name'],
- **headers)
+ headers = {'range':'bytes=%s-' %offset}
+ status, headers, data = self.get_object(self.containers[1],
+ self.objects[0]['name'],
+ **headers)
#assert successful partial content
- self.assertEqual(r.status_code, 206)
+ self.assertEqual(status, 206)
#assert content-type
- self.assertEqual(r['Content-Type'],
- self.objects[0]['meta']['HTTP_CONTENT_TYPE'])
+ self.assertEqual(headers['content-type'],
+ self.objects[0]['meta']['content-type'])
#assert content length
- self.assertEqual(int(r['Content-Length']), 500)
+ self.assertEqual(int(headers['content-length']), 500)
#assert content
- self.assertTrue(self.objects[0]['data'][-500:], r.content)
+ self.assertTrue(self.objects[0]['data'][-500:], data)
def test_get_range_not_satisfiable(self):
#perform get with range
offset = len(self.objects[0]['data']) + 1
- headers = {'HTTP_RANGE':'bytes=0-%s' %offset}
- r = self.get_object(self.account,
- self.containers[1],
- self.objects[0]['name'],
- **headers)
+ headers = {'range':'bytes=0-%s' %offset}
- #assert Range Not Satisfiable
- self.assertEqual(r.status_code, 416)
+ #assert range not satisfiable
+ self.assert_raises_fault(416, self.get_object, self.containers[1],
+ self.objects[0]['name'], **headers)
def test_multiple_range(self):
#perform get with multiple range
ranges = ['0-499', '-500', '1000-']
- headers = {'HTTP_RANGE' : 'bytes=%s' % ','.join(ranges)}
- r = self.get_object(self.account,
- self.containers[1],
- self.objects[0]['name'],
- **headers)
+ headers = {'range' : 'bytes=%s' % ','.join(ranges)}
+ status, headers, data = self.get_object(self.containers[1],
+ self.objects[0]['name'],
+ **headers)
# assert partial content
- self.assertEqual(r.status_code, 206)
+ self.assertEqual(status, 206)
# assert Content-Type of the reply will be multipart/byteranges
- self.assertTrue(r['Content-Type'])
- content_type_parts = r['Content-Type'].split()
+ self.assertTrue(headers['content-type'])
+ content_type_parts = headers['content-type'].split()
self.assertEqual(content_type_parts[0], ('multipart/byteranges;'))
boundary = '--%s' %content_type_parts[1].split('=')[-1:][0]
- cparts = r.content.split(boundary)[1:-1]
+ cparts = data.split(boundary)[1:-1]
# assert content parts are exactly 2
self.assertEqual(len(cparts), len(ranges))
#perform get with multiple range
out_of_range = len(self.objects[0]['data']) + 1
ranges = ['0-499', '-500', '%d-' %out_of_range]
- headers = {'HTTP_RANGE' : 'bytes=%s' % ','.join(ranges)}
- r = self.get_object(self.account,
- self.containers[1],
- self.objects[0]['name'],
- **headers)
+ headers = {'range' : 'bytes=%s' % ','.join(ranges)}
# assert partial content
- self.assertEqual(r.status_code, 416)
+ self.assert_raises_fault(416, self.get_object, self.containers[1],
+ self.objects[0]['name'], **headers)
+
def test_get_with_if_match(self):
#perform get with If-Match
- headers = {'HTTP_IF_MATCH':self.objects[0]['hash']}
- r = self.get_object(self.account,
- self.containers[1],
- self.objects[0]['name'],
- **headers)
+ headers = {'if-match':self.objects[0]['hash']}
+ status, headers, data = self.get_object(self.containers[1],
+ self.objects[0]['name'],
+ **headers)
#assert get success
- self.assertEqual(r.status_code, 200)
+ self.assertEqual(status, 200)
#assert content-type
- self.assertEqual(r['Content-Type'],
- self.objects[0]['meta']['HTTP_CONTENT_TYPE'])
+ self.assertEqual(headers['content-type'],
+ self.objects[0]['meta']['content-type'])
#assert response content
- self.assertEqual(self.objects[0]['data'].strip(), r.content.strip())
+ self.assertEqual(self.objects[0]['data'], data)
def test_get_with_if_match_star(self):
#perform get with If-Match *
- headers = {'HTTP_IF_MATCH':'*'}
- r = self.get_object(self.account,
- self.containers[1],
- self.objects[0]['name'],
- **headers)
+ headers = {'if-match':'*'}
+ status, headers, data = self.get_object(self.containers[1],
+ self.objects[0]['name'],
+ **headers)
#assert get success
- self.assertEqual(r.status_code, 200)
+ self.assertEqual(status, 200)
#assert content-type
- self.assertEqual(r['Content-Type'],
- self.objects[0]['meta']['HTTP_CONTENT_TYPE'])
+ self.assertEqual(headers['content-type'],
+ self.objects[0]['meta']['content-type'])
#assert response content
- self.assertEqual(self.objects[0]['data'].strip(), r.content.strip())
+ self.assertEqual(self.objects[0]['data'], data)
def test_get_with_multiple_if_match(self):
#perform get with If-Match
etags = [i['hash'] for i in self.objects if i]
etags = ','.join('"%s"' % etag for etag in etags)
- headers = {'HTTP_IF_MATCH':etags}
- r = self.get_object(self.account,
- self.containers[1],
- self.objects[0]['name'],
- **headers)
+ headers = {'if-match':etags}
+ status, headers, data = self.get_object(self.containers[1],
+ self.objects[0]['name'],
+ **headers)
#assert get success
- self.assertEqual(r.status_code, 200)
+ self.assertEqual(status, 200)
#assert content-type
- self.assertEqual(r['Content-Type'],
- self.objects[0]['meta']['HTTP_CONTENT_TYPE'])
+ self.assertEqual(headers['content-type'],
+ self.objects[0]['meta']['content-type'])
#assert content-type
- self.assertEqual(r['Content-Type'],
- self.objects[0]['meta']['HTTP_CONTENT_TYPE'])
+ self.assertEqual(headers['content-type'],
+ self.objects[0]['meta']['content-type'])
#assert response content
- self.assertEqual(self.objects[0]['data'].strip(), r.content.strip())
+ self.assertEqual(self.objects[0]['data'], data)
def test_if_match_precondition_failed(self):
#perform get with If-Match
- headers = {'HTTP_IF_MATCH':'123'}
- r = self.get_object(self.account,
- self.containers[1],
- self.objects[0]['name'],
- **headers)
- #assert precondition failed
- self.assertEqual(r.status_code, 412)
+ headers = {'if-match':'123'}
+
+ #assert precondition failed
+ self.assert_raises_fault(412, self.get_object, self.containers[1],
+ self.objects[0]['name'], **headers)
+
def test_if_none_match(self):
#perform get with If-None-Match
- headers = {'HTTP_IF_NONE_MATCH':'123'}
- r = self.get_object(self.account,
- self.containers[1],
- self.objects[0]['name'],
- **headers)
+ headers = {'if-none-match':'123'}
+ status, headers, data = self.get_object(self.containers[1],
+ self.objects[0]['name'],
+ **headers)
#assert get success
- self.assertEqual(r.status_code, 200)
+ self.assertEqual(status, 200)
#assert content-type
- self.assertEqual(r['Content-Type'],
- self.objects[0]['meta']['HTTP_CONTENT_TYPE'])
+ self.assertEqual(headers['content-type'],
+ self.objects[0]['meta']['content-type'])
def test_if_none_match(self):
#perform get with If-None-Match *
- headers = {'HTTP_IF_NONE_MATCH':'*'}
- r = self.get_object(self.account,
- self.containers[1],
- self.objects[0]['name'],
- **headers)
+ headers = {'if-none-match':'*'}
- #assert get success
- self.assertEqual(r.status_code, 304)
+ #assert not modified
+ self.assert_raises_fault(304, self.get_object, self.containers[1],
+ self.objects[0]['name'],
+ **headers)
def test_if_none_match_not_modified(self):
#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],
- self.objects[0]['name'],
- **headers)
+ headers = {'if-none-match':'%s' %self.objects[0]['hash']}
#assert not modified
- self.assertEqual(r.status_code, 304)
- self.assertEqual(r['ETag'], self.objects[0]['hash'])
+ self.assert_raises_fault(304, self.get_object, self.containers[1],
+ self.objects[0]['name'],
+ **headers)
+
+ headers = self.get_object(self.containers[1],
+ self.objects[0]['name'])[1]
+ self.assertEqual(headers['etag'], self.objects[0]['hash'])
def test_if_modified_since(self):
t = datetime.datetime.utcnow()
t2 = t - datetime.timedelta(minutes=10)
#modify the object
- self.upload_object(self.account,
- self.containers[1],
+ self.upload_data(self.containers[1],
self.objects[0]['name'],
self.objects[0]['data'][:200])
for f in DATE_FORMATS:
past = t2.strftime(f)
- headers = {'HTTP_IF_MODIFIED_SINCE':'%s' %past}
- r = self.get_object(self.account,
- self.containers[1],
- self.objects[0]['name'],
- **headers)
+ headers = {'if-modified-since':'%s' %past}
+ status, headers, data = self.get_object(self.containers[1],
+ self.objects[0]['name'],
+ **headers)
#assert get success
- self.assertEqual(r.status_code, 200)
+ self.assertEqual(status, 200)
#assert content-type
- self.assertEqual(r['Content-Type'],
- self.objects[0]['meta']['HTTP_CONTENT_TYPE'])
+ self.assertEqual(headers['content-type'],
+ self.objects[0]['meta']['content-type'])
def test_if_modified_since_invalid_date(self):
- headers = {'HTTP_IF_MODIFIED_SINCE':''}
- r = self.get_object(self.account,
- self.containers[1],
- self.objects[0]['name'],
- **headers)
+ headers = {'if-modified-since':''}
+ status, headers, data = self.get_object(self.containers[1],
+ self.objects[0]['name'],
+ **headers)
#assert get success
- self.assertEqual(r.status_code, 200)
+ self.assertEqual(status, 200)
#assert content-type
- self.assertEqual(r['Content-Type'],
- self.objects[0]['meta']['HTTP_CONTENT_TYPE'])
+ self.assertEqual(headers['content-type'],
+ self.objects[0]['meta']['content-type'])
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.get_object(self.account,
- self.containers[1],
- self.objects[0]['name'],
- **headers)
+ headers = {'if-modified-since':'%s' %since.strftime(f)}
#assert not modified
- self.assertEqual(r.status_code, 304)
+ self.assert_raises_fault(304, self.get_object, self.containers[1],
+ self.objects[0]['name'], **headers)
+
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)
+ headers = {'if-unmodified-since':'%s' %since.strftime(f)}
+ status, headers, data = self.get_object(self.containers[1],
+ self.objects[0]['name'],
+ **headers)
#assert success
- self.assertEqual(r.status_code, 200)
- self.assertEqual(self.objects[0]['data'], r.content)
+ self.assertEqual(status, 200)
+ self.assertEqual(self.objects[0]['data'], data)
#assert content-type
- self.assertEqual(r['Content-Type'],
- self.objects[0]['meta']['HTTP_CONTENT_TYPE'])
+ self.assertEqual(headers['content-type'],
+ self.objects[0]['meta']['content-type'])
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.upload_data(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)
+ headers = {'if-unmodified-since':'%s' %past}
+
+ #assert precondition failed
+ self.assert_raises_fault(412, self.get_object, self.containers[1],
+ self.objects[0]['name'], **headers)
+
def test_hashes(self):
l = 8388609
fname = 'largefile'
- o = self.upload_random_data(self.account,
- self.containers[1],
- fname,
- l)
+ o = self.upload_random_data(self.containers[1], fname, l)
if o:
- r = self.get_object(self.account,
- self.containers[1],
+ data = self.get_object(self.containers[1],
fname,
- 'json')
- body = json.loads(r.content)
+ 'json')[2]
+ body = json.loads(data)
hashes = body['hashes']
block_size = body['block_size']
block_hash = body['block_hash']
BaseTestCase.setUp(self)
self.account = 'test'
self.container = 'c1'
- self.create_container(self.account, self.container)
-
- self.src = os.path.join('.', 'api', 'tests.py')
- self.dest = os.path.join('.', 'api', 'chunked_update_test_file')
- create_chunked_update_test_file(self.src, self.dest)
-
+ self.create_container(self.container)
+
def tearDown(self):
- 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)
+ objects = self.list_objects(self.container)[2]
+ for o in objects:
+ self.client.delete_object(self.container, o)
+ self.client.delete_container(self.container)
def test_upload(self):
- filename = 'tests.py'
- fullpath = os.path.join('.', 'api', filename)
- f = open(fullpath, 'r')
- data = f.read()
- hash = compute_md5_hash(data)
- meta={'HTTP_ETAG':hash,
- 'HTTP_X_OBJECT_META_TEST':'test1'
- }
- 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,
- data,
- content_type=meta['HTTP_CONTENT_TYPE'],
- **meta)
- 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'])
+ name = o_names[0]
+ meta = {'test':'test1'}
+ o = self.upload_random_data(self.container, name, **meta)
+
+ headers = self.client.retrieve_object_metadata(self.container,
+ name,
+ restricted=True)
+ self.assertTrue('test' in headers.keys())
+ self.assertEqual(headers['test'], meta['test'])
#assert uploaded content
- r = self.get_object(self.account, self.container, filename)
- self.assertEqual(os.path.getsize(fullpath), int(r['Content-Length']))
- self.assertEqual(data.strip(), r.content.strip())
+ status, headers, content = self.get_object(self.container, name)
+ self.assertEqual(len(o['data']), int(headers['content-length']))
+ self.assertEqual(o['data'], content)
def test_upload_unprocessable_entity(self):
- filename = 'tests.py'
- fullpath = os.path.join('.', 'api', filename)
- f = open(fullpath, 'r')
- data = f.read()
- meta={'HTTP_ETAG':'123',
- 'HTTP_X_OBJECT_META_TEST':'test1'
- }
- 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,
- data,
- 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 = {}
- 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,
- objname,
- data,
- content_type = 'plain/text',
- **meta)
- self.assertEqual(r.status_code, 201)
-
- r = self.get_object(self.account,
- self.container,
- objname)
- uploaded_data = r.content
- f = open(self.src, 'r')
+ meta={'etag':'123', 'test':'test1'}
+
+ #assert unprocessable entity
+ self.assert_raises_fault(422, self.upload_random_data,self.container,
+ o_names[0], **meta)
+
+ def test_chucked_transfer(self):
+ fname = './api/tests.py'
+ objname = os.path.split(fname)[-1:][0]
+ f = open(fname, 'r')
+ status = self.client.create_object(self.container,
+ objname,
+ f,
+ chunked=True)[0]
+ self.assertEqual(status, 201)
+
+ uploaded_data = self.get_object(self.container,
+ objname)[2]
+ f = open(fname, 'r')
actual_data = f.read()
self.assertEqual(actual_data, uploaded_data)
self.account = 'test'
self.containers = ['c1', 'c2']
for c in self.containers:
- self.create_container(self.account, c)
- self.obj = self.upload_os_file(self.account,
- self.containers[0],
- './api/tests.py')
+ self.create_container(c)
+ self.obj = self.upload_random_data(self.containers[0], o_names[0])
def tearDown(self):
for c in self.containers:
- for o in get_content_splitted(self.list_objects(self.account, c)):
- self.delete_object(self.account, c, o)
- self.delete_container(self.account, c)
+ for o in self.list_objects(c)[2]:
+ self.client.delete_object(c, o)
+ self.client.delete_container(c)
def test_copy(self):
- with AssertInvariant(self.get_object_meta, self.account,
+ with AssertMappingInvariant(self.client.retrieve_object_metadata,
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'])
- r = self.copy_object(self.account,
- self.containers[0],
- 'testcopy',
- src_path,
- **meta)
+ meta = {'test':'testcopy'}
+ status = self.client.copy_object(self.containers[0],
+ self.obj['name'],
+ self.containers[0],
+ 'testcopy',
+ **meta)[0]
+
#assert copy success
- self.assertEqual(r.status_code, 201)
+ self.assertEqual(status, 201)
#assert access the new object
- r = self.get_object_meta(self.account,
- self.containers[0],
- 'testcopy')
- self.assertTrue(r['X-Object-Meta-Test'])
- self.assertTrue(r['X-Object-Meta-Test'], 'testcopy')
+ headers = self.client.retrieve_object_metadata(self.containers[0],
+ 'testcopy')
+ self.assertTrue('x-object-meta-test' in headers.keys())
+ self.assertTrue(headers['x-object-meta-test'], 'testcopy')
#assert etag is the same
- self.assertEqual(r['ETag'], self.obj['hash'])
+ self.assertEqual(headers['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, 200)
+ self.assert_object_exists(self.containers[0], self.obj['name'])
def test_copy_from_different_container(self):
- with AssertInvariant(self.get_object_meta,
- self.account,
- self.containers[0],
- self.obj['name']):
- meta = {'HTTP_X_OBJECT_META_TEST':'testcopy'}
- src_path = os.path.join('/', self.containers[0], self.obj['name'])
- r = self.copy_object(self.account,
- self.containers[1],
- 'testcopy',
- src_path,
- **meta)
- self.assertEqual(r.status_code, 201)
+ with AssertMappingInvariant(self.client.retrieve_object_metadata,
+ self.containers[0], self.obj['name']):
+ meta = {'test':'testcopy'}
+ status = self.client.copy_object(self.containers[0],
+ self.obj['name'],
+ self.containers[1],
+ 'testcopy',
+ **meta)[0]
+ self.assertEqual(status, 201)
# assert updated metadata
- r = self.get_object_meta(self.account,
- self.containers[1],
- 'testcopy')
- self.assertTrue(r['X-Object-Meta-Test'])
- self.assertTrue(r['X-Object-Meta-Test'], 'testcopy')
+ meta = self.client.retrieve_object_metadata(self.containers[1],
+ 'testcopy',
+ restricted=True)
+ self.assertTrue('test' in meta.keys())
+ self.assertTrue(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, 200)
+ self.assert_object_exists(self.containers[0], self.obj['name'])
def test_copy_invalid(self):
#copy from invalid object
- meta = {'HTTP_X_OBJECT_META_TEST':'testcopy'}
- r = self.copy_object(self.account,
- self.containers[1],
- 'testcopy',
- os.path.join('/', self.containers[0], 'test.py'),
- **meta)
- self.assertItemNotFound(r)
+ meta = {'test':'testcopy'}
+ self.assert_raises_fault(404, self.client.copy_object, self.containers[0],
+ 'test.py', self.containers[1], 'testcopy',
+ **meta)
#copy from invalid container
- meta = {'HTTP_X_OBJECT_META_TEST':'testcopy'}
- src_path = os.path.join('/', self.containers[1], self.obj['name'])
- r = self.copy_object(self.account,
- self.containers[1],
- 'testcopy',
- src_path,
- **meta)
- self.assertItemNotFound(r)
+ meta = {'test':'testcopy'}
+ self.assert_raises_fault(404, self.client.copy_object, self.containers[1],
+ self.obj['name'], self.containers[1],
+ 'testcopy', **meta)
+
class ObjectMove(ObjectCopy):
def test_move(self):
#perform move
- meta = {'HTTP_X_OBJECT_META_TEST':'testcopy'}
+ meta = {'test':'testcopy'}
src_path = os.path.join('/', self.containers[0], self.obj['name'])
- r = self.move_object(self.account,
- self.containers[0],
- 'testcopy',
- src_path,
- **meta)
+ status = self.client.move_object(self.containers[0], self.obj['name'],
+ self.containers[0], 'testcopy',
+ **meta)[0]
+
#assert successful move
- self.assertEqual(r.status_code, 201)
+ self.assertEqual(status, 201)
#assert updated metadata
- r = self.get_object_meta(self.account,
- self.containers[0],
- 'testcopy')
- self.assertTrue(r['X-Object-Meta-Test'])
- self.assertTrue(r['X-Object-Meta-Test'], 'testcopy')
+ meta = self.client.retrieve_object_metadata(self.containers[0],
+ 'testcopy',
+ restricted=True)
+ self.assertTrue('test' in meta.keys())
+ self.assertTrue(meta['test'], 'testcopy')
#assert src object no more exists
- r = self.get_object_meta(self.account, self.containers[0],
- self.obj['name'])
- self.assertItemNotFound(r)
+ self.assert_object_not_exists(self.containers[0], self.obj['name'])
class ObjectPost(BaseTestCase):
def setUp(self):
self.account = 'test'
self.containers = ['c1', 'c2']
for c in self.containers:
- self.create_container(self.account, c)
- self.obj = self.upload_os_file(self.account,
- self.containers[0],
- './api/tests.py')
+ self.create_container(c)
+ self.obj = self.upload_random_data(self.containers[0], o_names[0])
def tearDown(self):
for c in self.containers:
- for o in get_content_splitted(self.list_objects(self.account, c)):
- self.delete_object(self.account, c, o)
- self.delete_container(self.account, c)
+ for o in self.list_objects(c)[2]:
+ self.client.delete_object(c, o)
+ self.client.delete_container(c)
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(self.account,
- self.containers[0],
- self.obj['name'],
- **more)
+ more = {'foo':'foo', 'bar':'bar'}
+ status = self.client.update_object_metadata(self.containers[0],
+ self.obj['name'],
+ **more)[0]
#assert request accepted
- self.assertEqual(r.status_code, 202)
+ self.assertEqual(status, 202)
#assert old metadata are still there
- r = self.get_object_meta(self.account, self.containers[0],
- self.obj['name'])
- self.assertTrue('X-Object-Meta-Test' not in r.items())
-
+ headers = self.client.retrieve_object_metadata(self.containers[0],
+ self.obj['name'],
+ restricted=True)
#assert new metadata have been updated
for k,v in more.items():
- key = '-'.join(elem.capitalize() for elem in k.split('_')[1:])
- self.assertTrue(r[key])
- self.assertTrue(r[key], v)
+ self.assertTrue(k in headers.keys())
+ self.assertTrue(headers[k], v)
def test_update_object(self,
first_byte_pos=0,
instance_length = True,
content_length = 500):
l = len(self.obj['data'])
- length = instance_length and l or '*'
+ length = l if instance_length else '*'
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}
+ headers = {'content-range':range,
+ 'content-type':'application/octet-stream'}
if content_length:
- more.update({'CONTENT_LENGTH':'%s' % content_length})
+ headers.update({'content-length':'%s' % content_length})
+
+ status = self.client.update_object_data(self.containers[0],
+ self.obj['name'],
+ data,
+ headers)[0]
- 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)
+ self.assertEqual(status, 202)
else:
- self.assertEqual(r.status_code, 204)
+ self.assertEqual(status, 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])
+ content = self.get_object(self.containers[0], self.obj['name'])[2]
+ self.assertEqual(content[0:partial], data)
+ self.assertEqual(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)
+
+ #fails if the server resets the content-legth
+ #def test_update_object_invalid_content_length(self):
+ # with AssertContentInvariant(self.get_object, 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']):
+ with AssertContentInvariant(self.get_object, self.containers[0],
+ self.obj['name']):
self.test_update_object(499, 0, True)
+ #no use if the server resets the content-legth
def test_update_object_invalid_range_and_length(self):
- with AssertContentInvariant(self.get_object, self.account, self.containers[0],
- self.obj['name']):
+ with AssertContentInvariant(self.get_object, self.containers[0],
+ self.obj['name']):
self.test_update_object(499, 0, True, -1)
+ #no use if the server resets the content-legth
def test_update_object_invalid_range_with_no_content_length(self):
- with AssertContentInvariant(self.get_object, self.account, self.containers[0],
- self.obj['name']):
+ with AssertContentInvariant(self.get_object, 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']):
+ with AssertContentInvariant(self.get_object, self.containers[0],
+ self.obj['name']):
l = len(self.obj['data'])
- self.test_update_object(0, l+1, True)
+ self.assert_raises_fault(416, 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 */*'}
+ headers = {'content-type':'application/octet-stream',
+ 'content-length':'500'}
+ status = self.client.update_object_data(self.containers[0],
+ self.obj['name'],
+ data, headers)[0]
- r = self.update_object(self.account,
- self.containers[0],
- self.obj['name'],
- data,
- 'application/octet-stream',
- **more)
+ self.assertEqual(status, 204)
- 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'])
+ content = self.get_object(self.containers[0], self.obj['name'])[2]
+ self.assertEqual(len(content), len(self.obj['data']) + 500)
+ self.assertEqual(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)
+ meta = {'transfer-encoding':'chunked',
+ 'content-range':'bytes 0-/%d' %fl}
+ self.update_object(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])
+ content = self.get_object(self.containers[0], self.obj['name'])[2]
+ self.assertEqual(content[0:dl], pure)
+ self.assertEqual(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)
+ meta = {'transfer-encoding':'chunked',
+ 'content-range':'bytes 0-%d/%d' %(dl, fl)}
+ self.update_object(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])
+ content = self.get_object(self.containers[0], self.obj['name'])[2]
+ self.assertEqual(content[0:dl+1], pure)
+ self.assertEqual(content[dl+1:fl], self.obj['data'][dl+1:fl])
class ObjectDelete(BaseTestCase):
def setUp(self):
self.account = 'test'
self.containers = ['c1', 'c2']
for c in self.containers:
- self.create_container(self.account, c)
- self.obj = self.upload_os_file(self.account,
- self.containers[0],
- './api/tests.py')
+ self.create_container(c)
+ self.obj = self.upload_random_data(self.containers[0], o_names[0])
def tearDown(self):
for c in self.containers:
- for o in get_content_splitted(self.list_objects(self.account, c)):
- self.delete_object(self.account, c, o)
- self.delete_container(self.account, c)
+ for o in self.list_objects(c)[2]:
+ self.client.delete_object(c, o)
+ self.client.delete_container(c)
def test_delete(self):
#perform delete object
- r = self.delete_object(self.account, self.containers[0],
- self.obj['name'])
+ self.client.delete_object(self.containers[0], self.obj['name'])[0]
- #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'])
-
- #assert failure
- self.assertItemNotFound(r)
+ #assert item not found
+ self.assert_raises_fault(404, self.client.delete_object, self.containers[1],
+ self.obj['name'])
-class AssertInvariant(object):
+class AssertMappingInvariant(object):
def __init__(self, callable, *args, **kwargs):
self.callable = callable
self.args = args
self.kwargs = kwargs
def __enter__(self):
- self.items = self.callable(*self.args, **self.kwargs).items()
- return self.items
+ self.map = self.callable(*self.args, **self.kwargs)
+ return self.map
def __exit__(self, type, value, tb):
- items = self.callable(*self.args, **self.kwargs).items()
- assert self.items == items
+ map = self.callable(*self.args, **self.kwargs)
+ for k in self.map.keys():
+ if is_date(map[k]):
+ continue
+ assert map[k] == self.map[k]
class AssertContentInvariant(object):
def __init__(self, callable, *args, **kwargs):
self.kwargs = kwargs
def __enter__(self):
- self.content = self.callable(*self.args, **self.kwargs).content
+ self.content = self.callable(*self.args, **self.kwargs)[2]
return self.content
def __exit__(self, type, value, tb):
- content = self.callable(*self.args, **self.kwargs).content
+ content = self.callable(*self.args, **self.kwargs)[2]
assert self.content == content
def get_content_splitted(response):
char_set = string.ascii_uppercase + string.digits
return ''.join(random.choice(char_set) for x in range(length))
+def is_date(date):
+ MONTHS = 'jan feb mar apr may jun jul aug sep oct nov dec'.split()
+ __D = r'(?P<day>\d{2})'
+ __D2 = r'(?P<day>[ \d]\d)'
+ __M = r'(?P<mon>\w{3})'
+ __Y = r'(?P<year>\d{4})'
+ __Y2 = r'(?P<year>\d{2})'
+ __T = r'(?P<hour>\d{2}):(?P<min>\d{2}):(?P<sec>\d{2})'
+ RFC1123_DATE = re.compile(r'^\w{3}, %s %s %s %s GMT$' % (__D, __M, __Y, __T))
+ RFC850_DATE = re.compile(r'^\w{6,9}, %s-%s-%s %s GMT$' % (__D, __M, __Y2, __T))
+ ASCTIME_DATE = re.compile(r'^\w{3} %s %s %s %s$' % (__M, __D2, __T, __Y))
+ for regex in RFC1123_DATE, RFC850_DATE, ASCTIME_DATE:
+ m = regex.match(date)
+ if m is not None:
+ return True
+ return False
+
o_names = ['kate.jpg',
'kate_beckinsale.jpg',
'How To Win Friends And Influence People.pdf',
'photos/plants/fern.jpg',
'photos/plants/rose.jpg',
'photos/me.jpg']
+
+if __name__ == "__main__":
+ unittest.main()
projects / pithos / commitdiff