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