Revision cf687343
b/kamaki/clients/test/pithos.py | ||
---|---|---|
115 | 115 |
self.FR.headers = dict() |
116 | 116 |
self.FR.status_code = 200 |
117 | 117 |
self.FR.json = dict() |
118 |
for f in self.files: |
|
119 |
f.close() |
|
118 | 120 |
|
119 | 121 |
def test_get_account_info(self): |
120 | 122 |
self.FR.headers = account_info |
... | ... | |
201 | 203 |
PC.object_put = Mock(return_value=self.FR()) |
202 | 204 |
from tempfile import NamedTemporaryFile |
203 | 205 |
from os import urandom |
204 |
tmpFile = NamedTemporaryFile() |
|
206 |
self.files.append(NamedTemporaryFile()) |
|
207 |
tmpFile = self.files[-1] |
|
205 | 208 |
num_of_blocks = 8 |
206 | 209 |
file_size = num_of_blocks * 4 * 1024 * 1024 |
207 | 210 |
print('\n\tCreate tmp file') |
... | ... | |
210 | 213 |
tmpFile.seek(0) |
211 | 214 |
print('\t\tDone') |
212 | 215 |
obj = 'objectName' |
216 |
|
|
217 |
# No special args |
|
213 | 218 |
self.client.upload_object(obj, tmpFile) |
214 |
tmpFile.close() |
|
215 | 219 |
self.assertEqual(PC.get_container_info.mock_calls, [call()]) |
216 | 220 |
[call1, call2] = PC.object_put.mock_calls |
217 | 221 |
|
218 |
(args, kwargs) = call1[1:3] |
|
219 |
self.assertEqual(args, (obj,)) |
|
220 |
expected = dict( |
|
222 |
(args1, kwargs1) = call1[1:3] |
|
223 |
(args2, kwargs2) = call2[1:3] |
|
224 |
self.assertEqual(args1, (obj,)) |
|
225 |
expected1 = dict( |
|
221 | 226 |
hashmap=True, |
222 | 227 |
success=(201, 409), |
223 | 228 |
format='json', |
... | ... | |
230 | 235 |
content_disposition=None, |
231 | 236 |
public=None, |
232 | 237 |
permissions=None) |
233 |
for k, v in expected.items(): |
|
238 |
for k, v in expected1.items(): |
|
239 |
if k == 'json': |
|
240 |
self.assertEqual(len(v['hashes']), len(kwargs1[k]['hashes'])) |
|
241 |
self.assertEqual(v['bytes'], kwargs1[k]['bytes']) |
|
242 |
else: |
|
243 |
self.assertEqual(v, kwargs1[k]) |
|
244 |
|
|
245 |
(args2, kwargs2) = call2[1:3] |
|
246 |
self.assertEqual(args2, (obj,)) |
|
247 |
expected2 = dict( |
|
248 |
json=dict( |
|
249 |
hashes=['s0m3h@5h'] * num_of_blocks, |
|
250 |
bytes=file_size), |
|
251 |
content_type='application/octet-stream', |
|
252 |
hashmap=True, |
|
253 |
success=201, |
|
254 |
format='json') |
|
255 |
for k, v in expected2.items(): |
|
234 | 256 |
if k == 'json': |
235 |
self.assertEqual(len(v['hashes']), len(kwargs[k]['hashes'])) |
|
236 |
self.assertEqual(v['bytes'], kwargs[k]['bytes']) |
|
257 |
self.assertEqual(len(v['hashes']), len(kwargs2[k]['hashes']))
|
|
258 |
self.assertEqual(v['bytes'], kwargs2[k]['bytes'])
|
|
237 | 259 |
else: |
238 |
self.assertEqual(v, kwargs[k]) |
|
239 |
# TO BE CONTINUED |
|
240 |
(args, kwargs) = call2[1:3] |
|
260 |
self.assertEqual(v, kwargs2[k]) |
|
261 |
|
|
262 |
OP = PC.object_put |
|
263 |
mock_offset = 2 |
|
264 |
|
|
265 |
# With progress bars |
|
266 |
try: |
|
267 |
from progress.bar import ShadyBar |
|
268 |
blck_bar = ShadyBar('Mock blck calc.') |
|
269 |
upld_bar = ShadyBar('Mock uplds') |
|
270 |
except ImportError: |
|
271 |
blck_bar = None |
|
272 |
upld_bar = None |
|
273 |
|
|
274 |
if blck_bar and upld_bar: |
|
275 |
|
|
276 |
def blck_gen(n): |
|
277 |
for i in blck_bar.iter(range(n)): |
|
278 |
yield |
|
279 |
yield |
|
280 |
|
|
281 |
def upld_gen(n): |
|
282 |
for i in upld_bar.iter(range(n)): |
|
283 |
yield |
|
284 |
yield |
|
285 |
|
|
286 |
tmpFile.seek(0) |
|
287 |
self.client.upload_object( |
|
288 |
obj, tmpFile, |
|
289 |
hash_cb=blck_gen, upload_cb=upld_gen) |
|
290 |
|
|
291 |
for i, c in enumerate(OP.mock_calls[-mock_offset:]): |
|
292 |
self.assertEqual(OP.mock_calls[i], c) |
|
293 |
|
|
294 |
# With content-type |
|
295 |
tmpFile.seek(0) |
|
296 |
ctype = 'video/mpeg' |
|
297 |
sharing = dict(read=['u1', 'g1', 'u2'], write=['u1']) |
|
298 |
self.client.upload_object(obj, tmpFile, |
|
299 |
content_type=ctype, sharing=sharing) |
|
300 |
self.assertEqual(OP.mock_calls[-1][2]['content_type'], ctype) |
|
301 |
self.assert_dicts_are_equal( |
|
302 |
OP.mock_calls[-2][2]['permissions'], |
|
303 |
sharing) |
|
304 |
|
|
305 |
# With other args |
|
306 |
tmpFile.seek(0) |
|
307 |
kwargs = dict( |
|
308 |
etag='s0m3E74g', |
|
309 |
content_type=ctype, |
|
310 |
content_disposition=ctype + 'd15p051710n', |
|
311 |
public=True, |
|
312 |
content_encoding='802.11') |
|
313 |
self.client.upload_object(obj, tmpFile, **kwargs) |
|
314 |
for arg, val in kwargs.items(): |
|
315 |
self.assertEqual(OP.mock_calls[-2][2][arg], val) |
Also available in: Unified diff