from tempfile import NamedTemporaryFile
from os import urandom
from itertools import product
+from random import randint
try:
from collections import OrderedDict
except ImportError:
- from kamaki.clients.commisioning.utils.ordereddict import OrderedDict
+ from kamaki.clients.utils.ordereddict import OrderedDict
-from kamaki.clients import ClientError
-from kamaki.clients.pithos import PithosClient, PithosRestClient
+from kamaki.clients import pithos, ClientError
rest_pkg = 'kamaki.clients.pithos.rest_api.PithosRestClient'
status = None
status_code = 200
- def release(self):
- pass
-
-class PithosRest(TestCase):
+class PithosRestClient(TestCase):
def setUp(self):
self.url = 'https://www.example.com/pithos'
self.token = 'p17h0570k3n'
- self.client = PithosRestClient(self.url, self.token)
+ self.client = pithos.PithosRestClient(self.url, self.token)
self.client.account = user_id
self.client.container = 'c0nt@1n3r_i'
(None, 'v3r510n1ng'),
((), ('someval',), ('v1', 'v2',)),
(dict(), dict(success=200), dict(k='v', v='k'))):
- args, kwargs = pm[-2], pm[-1]
+ args, kwargs = pm[-2:]
pm = pm[:-2]
self.client.account_post(*(pm + args), **kwargs)
upd = pm[0]
success=kwargs.pop('success', 202),
**kwargs))
+ @patch('%s.set_param' % rest_pkg)
+ @patch('%s.set_header' % rest_pkg)
+ @patch('%s.head' % rest_pkg, return_value=FR())
+ def test_container_head(self, head, SH, SP):
+ for pm in product(
+ (None, '4-d473'),
+ (None, '47h3r-d473'),
+ (None, 'y37-4n47h3r'),
+ ((), ('someval',)),
+ (dict(), dict(success=200), dict(k='v', v='k'))):
+ args, kwargs = pm[-2:]
+ pm = pm[:-2]
+ self.client.container_head(*(pm + args), **kwargs)
+ unt, ims, ius = pm[0:3]
+ self.assertEqual(SP.mock_calls[-1], call('until', unt, iff=unt))
+ self.assertEqual(SH.mock_calls[-2:], [
+ call('If-Modified-Since', ims),
+ call('If-Unmodified-Since', ius)])
+ self.assertEqual(head.mock_calls[-1], call(
+ '/%s/%s' % (self.client.account, self.client.container),
+ *args,
+ success=kwargs.pop('success', 204),
+ **kwargs))
+
+ @patch('%s.set_param' % rest_pkg)
+ @patch('%s.set_header' % rest_pkg)
+ @patch('%s.get' % rest_pkg, return_value=FR())
+ def test_container_get(self, get, SH, SP):
+ for pm in product(
+ (None, 42),
+ (None, 'X'),
+ (None, 'some/prefix'),
+ (None, 'delimiter'),
+ (None, '/some/path'),
+ ('json', 'some-format'),
+ ([], ['k1', 'k2', 'k3']),
+ (False, True),
+ (None, 'unt1l-d473'),
+ (None, 'y37-4n47h3r'),
+ (None, '4n47h3r-d473'),
+ ((), ('someval',)),
+ (dict(), dict(success=400), dict(k='v', v='k'))):
+ args, kwargs = pm[-2:]
+ pm = pm[:-2]
+ self.client.container_get(*(pm + args), **kwargs)
+ lmt, mrk, prfx, dlm, path, frmt, meta, shr, unt = pm[:-2]
+ exp = [call('limit', lmt, iff=lmt), call('marker', mrk, iff=mrk)]
+ exp += [call('path', path)] if path else [
+ call('prefix', prfx, iff=prfx),
+ call('delimiter', dlm, iff=dlm)]
+ exp += [call('format', frmt, iff=frmt), call('shared', iff=shr)]
+ if meta:
+ exp += [call('meta', ','.join(meta))]
+ exp += [call('until', unt, iff=unt)]
+ self.assertEqual(SP.mock_calls[- len(exp):], exp)
+ ims, ius = pm[-2:]
+ self.assertEqual(SH.mock_calls[-2:], [
+ call('If-Modified-Since', ims),
+ call('If-Unmodified-Since', ius)])
+ self.assertEqual(get.mock_calls[-1], call(
+ '/%s/%s' % (self.client.account, self.client.container),
+ *args,
+ success=kwargs.pop('success', 200),
+ **kwargs))
+
+ @patch('%s.set_header' % rest_pkg)
+ @patch('%s.put' % rest_pkg, return_value=FR())
+ def test_container_put(self, put, SH):
+ for pm in product(
+ (None, 42),
+ (None, 'v3r51on1ng'),
+ (dict(), dict(k1='v2'), dict(k2='v2', k3='v3')),
+ ((), ('someval',)),
+ (dict(), dict(success=400), dict(k='v', v='k'))):
+ args, kwargs = pm[-2:]
+ pm = pm[:-2]
+ self.client.container_put(*(pm + args), **kwargs)
+ quota, versioning, metas = pm[-3:]
+ exp = [
+ call('X-Container-Policy-Quota', quota),
+ call('X-Container-Policy-Versioning', versioning)] + [
+ call('X-Container-Meta-%s' % k, v) for k, v in metas.items()]
+ self.assertEqual(SH.mock_calls[- len(exp):], exp)
+ self.assertEqual(put.mock_calls[-1], call(
+ '/%s/%s' % (self.client.account, self.client.container),
+ *args,
+ success=kwargs.pop('success', (201, 202)),
+ **kwargs))
+
+ @patch('%s.set_param' % rest_pkg)
+ @patch('%s.set_header' % rest_pkg)
+ @patch('%s.post' % rest_pkg, return_value=FR())
+ def test_container_post(self, post, SH, SP):
+ for pm in product(
+ (True, False),
+ ('json', 'some-format'),
+ (None, 'quota'),
+ (None, 'v3r51on1ng'),
+ (dict(), dict(k1='v2'), dict(k2='v2', k3='v3')),
+ (None, 'content-type'),
+ (None, 42),
+ (None, 'transfer-encoding'),
+ ((), ('someval',)),
+ (dict(), dict(success=400), dict(k='v', v='k'))):
+ args, kwargs = pm[-2:]
+ pm = pm[:-2]
+ self.client.container_post(*(pm + args), **kwargs)
+ upd, frmt = pm[:2]
+ self.assertEqual(SP.mock_calls[-2:], [
+ call('update', iff=upd),
+ call('format', frmt, iff=frmt)])
+ qta, vrs, metas, ctype, clen, trenc = pm[2:]
+ prfx = 'X-Container-Meta-'
+ exp = [
+ call('X-Container-Policy-Quota', qta),
+ call('X-Container-Policy-Versioning', vrs)] + [
+ call('%s%s' % (prfx, k), v) for k, v in metas.items()] + [
+ call('Content-Type', ctype),
+ call('Content-Length', clen),
+ call('Transfer-Encoding', trenc)]
+ self.assertEqual(SH.mock_calls[- len(exp):], exp)
+ ims, ius = pm[-2:]
+ self.assertEqual(post.mock_calls[-1], call(
+ '/%s/%s' % (self.client.account, self.client.container),
+ *args,
+ success=kwargs.pop('success', 202),
+ **kwargs))
+
+ @patch('%s.set_param' % rest_pkg)
+ @patch('%s.delete' % rest_pkg, return_value=FR())
+ def test_container_delete(self, delete, SP):
+ for pm in product(
+ (None, 'd473'),
+ (None, 'd3l1m'),
+ ((), ('someval',)),
+ (dict(), dict(success=400), dict(k='v', v='k'))):
+ args, kwargs = pm[-2:]
+ pm = pm[:-2]
+ self.client.container_delete(*(pm + args), **kwargs)
+ unt, dlm = pm[-2:]
+ self.assertEqual(SP.mock_calls[-2:], [
+ call('until', unt, iff=unt),
+ call('delimiter', dlm, iff=dlm)])
+ self.assertEqual(delete.mock_calls[-1], call(
+ '/%s/%s' % (self.client.account, self.client.container),
+ *args,
+ success=kwargs.pop('success', 204),
+ **kwargs))
+
+ @patch('%s.set_param' % rest_pkg)
+ @patch('%s.set_header' % rest_pkg)
+ @patch('%s.head' % rest_pkg, return_value=FR())
+ def test_object_head(self, head, SH, SP):
+ for pm in product(
+ (None, 'v3r510n'),
+ (None, '1f-374g'),
+ (None, '1f-n0-74g'),
+ (None, '1f-m0d-51nc3'),
+ (None, '1f-unm0d-51nc3'),
+ ((), ('someval',)),
+ (dict(), dict(success=400), dict(k='v', v='k'))):
+ args, kwargs = pm[-2:]
+ pm = pm[:-2]
+ self.client.object_head(obj, *(pm + args), **kwargs)
+ vrs, etag, netag, ims, ius = pm[:5]
+ self.assertEqual(
+ SP.mock_calls[-1],
+ call('version', vrs, iff=vrs))
+ self.assertEqual(SH.mock_calls[-4:], [
+ call('If-Match', etag),
+ call('If-None-Match', netag),
+ call('If-Modified-Since', ims),
+ call('If-Unmodified-Since', ius)])
+ acc, cont = self.client.account, self.client.container
+ self.assertEqual(head.mock_calls[-1], call(
+ '/%s/%s/%s' % (acc, cont, obj),
+ *args,
+ success=kwargs.pop('success', 200),
+ **kwargs))
+
+ @patch('%s.set_param' % rest_pkg)
+ @patch('%s.set_header' % rest_pkg)
+ @patch('%s.get' % rest_pkg, return_value=FR())
+ def test_object_get(self, get, SH, SP):
+ for pm in product(
+ ('json', 'f0rm47'),
+ (False, True),
+ (None, 'v3r510n'),
+ (None, 'range=74-63'),
+ (False, True),
+ (None, '3746'),
+ (None, 'non-3746'),
+ (None, '1f-m0d'),
+ (None, '1f-unm0d'),
+ ((), ('someval',)),
+ (dict(), dict(success=400), dict(k='v', v='k'))):
+ args, kwargs = pm[-2:]
+ pm = pm[:-2]
+ self.client.object_get(obj, *(pm + args), **kwargs)
+ format, hashmap, version = pm[:3]
+ self.assertEqual(SP.mock_calls[-3:], [
+ call('format', format, iff=format),
+ call('hashmap', hashmap, iff=hashmap),
+ call('version', version, iff=version)])
+ rng, ifrng, im, inm, ims, ius = pm[-6:]
+ self.assertEqual(SH.mock_calls[-6:], [
+ call('Range', rng),
+ call('If-Range', '', ifrng and rng),
+ call('If-Match', im),
+ call('If-None-Match', inm),
+ call('If-Modified-Since', ims),
+ call('If-Unmodified-Since', ius)])
+ acc, cont = self.client.account, self.client.container
+ self.assertEqual(get.mock_calls[-1], call(
+ '/%s/%s/%s' % (acc, cont, obj),
+ *args,
+ success=kwargs.pop('success', 200),
+ **kwargs))
+
+ @patch('%s.set_param' % rest_pkg)
+ @patch('%s.set_header' % rest_pkg)
+ @patch('%s.put' % rest_pkg, return_value=FR())
+ def test_object_put(self, put, SH, SP):
+ for pm in product(
+ ('json', 'f0rm47'),
+ (False, True),
+ (None, 'delim',),
+ (dict(), dict(read=['u1', 'g2'], write=['u1'])),
+ (False, True),
+ (dict(), dict(k2='v2', k3='v3')),
+ ((), ('someval',)),
+ (dict(), dict(success=400), dict(k='v', v='k'))):
+ args, kwargs = pm[-2:]
+ pm = pm[:-2]
+ terms = [None] * 13
+ for i in range(len(terms)):
+ if randint(0, 2):
+ terms[i] = 'val_%s' % randint(13, 1024)
+ self.client.object_put(
+ obj,
+ *(pm[:3] + tuple(terms) + pm[3:] + args),
+ **kwargs)
+ format, hashmap, delimiter = pm[:3]
+ self.assertEqual(SP.mock_calls[-3:], [
+ call('format', format, iff=format),
+ call('hashmap', hashmap, iff=hashmap),
+ call('delimiter', delimiter, iff=delimiter)])
+ (
+ im, inm, etag, clen, ctype, trenc,
+ cp, mv, srcacc, srcvrs, conenc, condis, mnf) = terms
+ perms, public, metas = pm[3:]
+ exp = [
+ call('If-Match', im),
+ call('If-None-Match', inm),
+ call('ETag', etag),
+ call('Content-Length', clen),
+ call('Content-Type', ctype),
+ call('Transfer-Encoding', trenc),
+ call('X-Copy-From', cp),
+ call('X-Move-From', mv),
+ call('X-Source-Account', srcacc),
+ call('X-Source-Version', srcvrs),
+ call('Content-Encoding', conenc),
+ call('Content-Disposition', condis),
+ call('X-Object-Manifest', mnf)]
+ if perms:
+ perm_str = ''
+ for ptype, pval in perms.items():
+ if pval:
+ perm_str += ';' if perm_str else ''
+ perm_str += '%s=%s' % (ptype, ','.join(pval))
+ exp += [call('X-Object-Sharing', perm_str)]
+ exp += [call('X-Object-Public', public)]
+ for k, v in metas.items():
+ exp += [call('X-Object-Meta-%s' % k, v)]
+ self.assertEqual(SH.mock_calls[- len(exp):], exp)
+ acc, cont = self.client.account, self.client.container
+ self.assertEqual(put.mock_calls[-1], call(
+ '/%s/%s/%s' % (acc, cont, obj),
+ *args,
+ success=kwargs.pop('success', 201),
+ **kwargs))
+
+ @patch('%s.set_param' % rest_pkg)
+ @patch('%s.set_header' % rest_pkg)
+ @patch('%s.copy' % rest_pkg, return_value=FR())
+ def test_object_copy(self, copy, SH, SP):
+ dest = 'dest1n4710n'
+ for pm in product(
+ ('json', 'f0rm47'),
+ (False, True),
+ (None, 'ifmatch'),
+ (None, 'ifnonematch'),
+ (None, 'destinationaccount'),
+ (None, 'content-type'),
+ (None, 'content-encoding'),
+ (None, 'content-disp'),
+ (None, 'source-version'),
+ (dict(), dict(read=['u1', 'g2'], write=['u1'])),
+ (False, True),
+ (dict(), dict(k2='v2', k3='v3')),
+ ((), ('someval',)),
+ (dict(), dict(success=400), dict(k='v', v='k'))):
+ args, kwargs = pm[-2:]
+ pm = pm[:-2]
+ self.client.object_copy(obj, dest, *(pm + args), **kwargs)
+ format, ict = pm[:2]
+ self.assertEqual(SP.mock_calls[-2:], [
+ call('format', format, iff=format),
+ call('ignore_content_type', iff=ict)])
+ im, inm, da, ct, ce, cd, sv, perms, public, metas = pm[2:]
+ exp = [call('If-Match', im),
+ call('If-None-Match', inm),
+ call('Destination', dest),
+ call('Destination-Account', da),
+ call('Content-Type', ct),
+ call('Content-Encoding', ce),
+ call('Content-Disposition', cd),
+ call('X-Source-Version', sv)]
+ if perms:
+ perm_str = ''
+ for ptype, pval in perms.items():
+ if pval:
+ perm_str += ';' if perm_str else ''
+ perm_str += '%s=%s' % (ptype, ','.join(pval))
+ exp += [call('X-Object-Sharing', perm_str)]
+ exp += [call('X-Object-Public', public)]
+ for k, v in metas.items():
+ exp += [call('X-Object-Meta-%s' % k, v)]
+ self.assertEqual(SH.mock_calls[- len(exp):], exp)
+ acc, cont = self.client.account, self.client.container
+ self.assertEqual(copy.mock_calls[-1], call(
+ '/%s/%s/%s' % (acc, cont, obj),
+ *args,
+ success=kwargs.pop('success', 201),
+ **kwargs))
+
+ @patch('%s.set_param' % rest_pkg)
+ @patch('%s.set_header' % rest_pkg)
+ @patch('%s.move' % rest_pkg, return_value=FR())
+ def test_object_move(self, move, SH, SP):
+ for pm in product(
+ ('json', 'f0rm47'),
+ (False, True),
+ (None, 'ifmatch'),
+ (None, 'ifnonematch'),
+ (None, 'destination'),
+ (None, 'destinationaccount'),
+ (None, 'content-type'),
+ (None, 'content-encoding'),
+ (None, 'content-disp'),
+ (dict(), dict(read=['u1', 'g2'], write=['u1'])),
+ (False, True),
+ (dict(), dict(k2='v2', k3='v3')),
+ ((), ('someval',)),
+ (dict(), dict(success=400), dict(k='v', v='k'))):
+ args, kwargs = pm[-2:]
+ pm = pm[:-2]
+ self.client.object_move(obj, *(pm + args), **kwargs)
+ format, ict = pm[:2]
+ self.assertEqual(SP.mock_calls[-2:], [
+ call('format', format, iff=format),
+ call('ignore_content_type', iff=ict)])
+ im, inm, d, da, ct, ce, cd, perms, public, metas = pm[2:]
+ exp = [call('If-Match', im),
+ call('If-None-Match', inm),
+ call('Destination', d),
+ call('Destination-Account', da),
+ call('Content-Type', ct),
+ call('Content-Encoding', ce),
+ call('Content-Disposition', cd)]
+ if perms:
+ perm_str = ''
+ for ptype, pval in perms.items():
+ if pval:
+ perm_str += ';' if perm_str else ''
+ perm_str += '%s=%s' % (ptype, ','.join(pval))
+ exp += [call('X-Object-Sharing', perm_str)]
+ exp += [call('X-Object-Public', public)]
+ for k, v in metas.items():
+ exp += [call('X-Object-Meta-%s' % k, v)]
+ self.assertEqual(SH.mock_calls[- len(exp):], exp)
+ acc, cont = self.client.account, self.client.container
+ self.assertEqual(move.mock_calls[-1], call(
+ '/%s/%s/%s' % (acc, cont, obj),
+ *args,
+ success=kwargs.pop('success', 201),
+ **kwargs))
+
+ @patch('%s.set_param' % rest_pkg)
+ @patch('%s.set_header' % rest_pkg)
+ @patch('%s.post' % rest_pkg, return_value=FR())
+ def test_object_post(self, post, SH, SP):
+ for pm in product(
+ ('json', 'f0rm47'),
+ (False, True),
+ (dict(), dict(read=['u1', 'g2'], write=['u1'])),
+ (False, True),
+ (dict(), dict(k2='v2', k3='v3')),
+ ((), ('someval',)),
+ (dict(), dict(success=400), dict(k='v', v='k'))):
+ args, kwargs = pm[-2:]
+ pm = pm[:-2]
+ terms = [None] * 13
+ for i in range(len(terms)):
+ if randint(0, 2):
+ terms[i] = 'val_%s' % randint(13, 1024)
+ self.client.object_post(
+ obj,
+ *(pm[:2] + tuple(terms) + pm[2:] + args),
+ **kwargs)
+ format, update = pm[:2]
+ self.assertEqual(SP.mock_calls[-2:], [
+ call('format', format, iff=format),
+ call('update', iff=update)])
+ (
+ im, inm, clen, ctype, crng, trenc, cenc,
+ condis, srcobj, srcacc, srcvrs, obytes, mnfs) = terms
+ exp = [
+ call('If-Match', im),
+ call('If-None-Match', inm),
+ call('Content-Length', clen, iff=not trenc),
+ call('Content-Type', ctype),
+ call('Content-Range', crng),
+ call('Transfer-Encoding', trenc),
+ call('Content-Encoding', cenc),
+ call('Content-Disposition', condis),
+ call('X-Source-Object', srcobj),
+ call('X-Source-Account', srcacc),
+ call('X-Source-Version', srcvrs),
+ call('X-Object-Bytes', obytes),
+ call('X-Object-Manifest', mnfs)]
+ perms, public, metas = pm[2:]
+ if perms:
+ perm_str = ''
+ for ptype, pval in perms.items():
+ if pval:
+ perm_str += ';' if perm_str else ''
+ perm_str += '%s=%s' % (ptype, ','.join(pval))
+ exp += [call('X-Object-Sharing', perm_str)]
+ exp += [call('X-Object-Public', public)]
+ for k, v in metas.items():
+ exp += [call('X-Object-Meta-%s' % k, v)]
+ self.assertEqual(SH.mock_calls[- len(exp):], exp)
+ acc, cont = self.client.account, self.client.container
+ self.assertEqual(post.mock_calls[-1], call(
+ '/%s/%s/%s' % (acc, cont, obj),
+ *args,
+ success=kwargs.pop('success', (202, 204)),
+ **kwargs))
+
+ @patch('%s.set_param' % rest_pkg)
+ @patch('%s.delete' % rest_pkg, return_value=FR())
+ def test_object_delete(self, delete, SP):
+ for pm in product(
+ (None, 'until'),
+ (None, 'delim'),
+ ((), ('someval',)),
+ (dict(), dict(success=400), dict(k='v', v='k'))):
+ args, kwargs = pm[-2:]
+ pm = pm[:-2]
+ self.client.object_delete(
+ obj,
+ *(pm + args),
+ **kwargs)
+ until, dlm = pm[-2:]
+ self.assertEqual(SP.mock_calls[-2:], [
+ call('until', until, iff=until),
+ call('delimiter', dlm, iff=dlm)])
+ acc, cont = self.client.account, self.client.container
+ self.assertEqual(delete.mock_calls[-1], call(
+ '/%s/%s/%s' % (acc, cont, obj),
+ *args,
+ success=kwargs.pop('success', 204),
+ **kwargs))
+
-class Pithos(TestCase):
+class PithosClient(TestCase):
files = []
def setUp(self):
self.url = 'https://www.example.com/pithos'
self.token = 'p17h0570k3n'
- self.client = PithosClient(self.url, self.token)
+ self.client = pithos.PithosClient(self.url, self.token)
self.client.account = user_id
self.client.container = 'c0nt@1n3r_i'
json=dict(
hashes=['s0m3h@5h'] * num_of_blocks,
bytes=num_of_blocks * 4 * 1024 * 1024),
- etag=None,
content_encoding=None,
content_type='application/octet-stream',
content_disposition=None,
tmpFile.seek(0)
kwargs = dict(
etag='s0m3E74g',
+ if_etag_match='if etag match',
+ if_not_exist=True,
content_type=ctype,
content_disposition=ctype + 'd15p051710n',
public=True,
content_encoding='802.11')
self.client.upload_object(obj, tmpFile, **kwargs)
+ kwargs.pop('if_not_exist')
+ ematch = kwargs.pop('if_etag_match')
+ etag = kwargs.pop('etag')
for arg, val in kwargs.items():
self.assertEqual(OP.mock_calls[-2][2][arg], val)
+ self.assertEqual(OP.mock_calls[-1][2]['if_etag_match'], ematch)
+ self.assertEqual(OP.mock_calls[-1][2]['if_etag_not_match'], '*')
+ self.assertEqual(OP.mock_calls[-1][2]['etag'], etag)
def test_get_object_info(self):
FR.headers = object_info
version = 'v3r510n'
with patch.object(
- PithosClient, 'object_head',
+ pithos.PithosClient, 'object_head',
return_value=FR()) as head:
r = self.client.get_object_info(obj)
self.assertEqual(r, object_info)
call(obj, version=None),
call(obj, version=version)])
with patch.object(
- PithosClient, 'object_head',
+ pithos.PithosClient, 'object_head',
side_effect=ClientError('Obj not found', 404)):
self.assertRaises(
ClientError,
self.assertEqual(GET.mock_calls[-1][2][k], v)
# ALl options on no tty
-
def foo():
return True
FR.json = object_hashmap
for empty in (304, 412):
with patch.object(
- PithosClient, 'object_get',
+ pithos.PithosClient, 'object_get',
side_effect=ClientError('Empty', status=empty)):
r = self.client.get_object_hashmap(obj)
self.assertEqual(r, {})
if_unmodified_since='some date here',
data_range='10-20')
with patch.object(
- PithosClient, 'object_get',
+ pithos.PithosClient, 'object_get',
return_value=FR()) as get:
r = self.client.get_object_hashmap(obj)
self.assertEqual(r, object_hashmap)
def test_get_account_meta(self):
key = 'x-account-meta-'
with patch.object(
- PithosClient, 'get_account_info',
+ pithos.PithosClient, 'get_account_info',
return_value=account_info):
r = self.client.get_account_meta()
keys = [k for k in r if k.startswith(key)]
acc_info['%sk2' % key] = 'v2'
acc_info['%sk3' % key] = 'v3'
with patch.object(
- PithosClient, 'get_account_info',
+ pithos.PithosClient, 'get_account_info',
return_value=acc_info):
r = self.client.get_account_meta()
for k in [k for k in acc_info if k.startswith(key)]:
def test_get_account_group(self):
key = 'x-account-group-'
with patch.object(
- PithosClient, 'get_account_info',
+ pithos.PithosClient, 'get_account_info',
return_value=account_info):
r = self.client.get_account_group()
keys = [k for k in r if k.startswith(key)]
acc_info['%sk2' % key] = 'g2'
acc_info['%sk3' % key] = 'g3'
with patch.object(
- PithosClient, 'get_account_info',
+ pithos.PithosClient, 'get_account_info',
return_value=acc_info):
r = self.client.get_account_group()
for k in [k for k in acc_info if k.startswith(key)]:
container_plus[key] = metaval
for ret in ((container_info, {}), (container_plus, {key: metaval})):
with patch.object(
- PithosClient,
+ pithos.PithosClient,
'get_container_info',
return_value=ret[0]) as GCI:
for until in (None, somedate):
(container_info, {key: ''}),
(container_plus, {key: metaval})):
with patch.object(
- PithosClient,
+ pithos.PithosClient,
'get_container_info',
return_value=ret[0]) as GCI:
for until in (None, somedate):
AP.assert_called_once_with(update=True, metadata={'somekey': ''})
@patch('%s.container_post' % pithos_pkg, return_value=FR())
- def test_set_container_quota(self, post):
+ def test_set_container_limit(self, post):
qu = 1024
- self.client.set_container_quota(qu)
+ self.client.set_container_limit(qu)
post.assert_called_once_with(update=True, quota=qu)
@patch('%s.container_post' % pithos_pkg, return_value=FR())
val = 'pubL1c'
oinfo['x-object-public'] = val
with patch.object(
- PithosClient, 'get_object_info',
+ pithos.PithosClient, 'get_object_info',
return_value=oinfo) as GOF:
r = self.client.publish_object(obj)
self.assertEqual(
info['x-object-sharing'] = '; '.join(
['%s=%s' % (k, v) for k, v in expected.items()])
with patch.object(
- PithosClient, 'get_object_info',
+ pithos.PithosClient, 'get_object_info',
return_value=info) as GOF:
r = self.client.get_object_sharing(obj)
self.assertEqual(GOF.mock_calls[-1], call(obj))
info['content-length'] = file_size
block_size = container_info['x-container-block-size']
with patch.object(
- PithosClient, 'get_object_info',
+ pithos.PithosClient, 'get_object_info',
return_value=info) as GOI:
for start, end in (
(0, file_size + 1),
from sys import argv
from kamaki.clients.test import runTestCase
not_found = True
- if not argv[1:] or argv[1] == 'Pithos':
+ if not argv[1:] or argv[1] == 'PithosClient':
not_found = False
- runTestCase(Pithos, 'Pithos Client', argv[2:])
- if not argv[1:] or argv[1] == 'PithosRest':
+ runTestCase(PithosClient, 'Pithos Client', argv[2:])
+ if not argv[1:] or argv[1] == 'PithosRestClient':
not_found = False
- runTestCase(PithosRest, 'PithosRest Client', argv[2:])
+ runTestCase(PithosRestClient, 'PithosRest Client', argv[2:])
if not_found:
print('TestCase %s not found' % argv[1])
projects / kamaki / blobdiff