Revision bcbf3cf0 kamaki/clients/test/pithos.py
b/kamaki/clients/test/pithos.py | ||
---|---|---|
86 | 86 |
class FR(object): |
87 | 87 |
"""FR stands for Fake Response""" |
88 | 88 |
json = dict() |
89 |
headers = {}
|
|
89 |
headers = dict()
|
|
90 | 90 |
content = json |
91 | 91 |
status = None |
92 | 92 |
status_code = 200 |
93 |
headers = dict() |
|
94 | 93 |
|
95 | 94 |
def release(self): |
96 | 95 |
pass |
... | ... | |
115 | 114 |
def tearDown(self): |
116 | 115 |
self.FR.headers = dict() |
117 | 116 |
self.FR.status_code = 200 |
117 |
self.FR.json = dict() |
|
118 | 118 |
|
119 | 119 |
def test_get_account_info(self): |
120 | 120 |
self.FR.headers = account_info |
... | ... | |
194 | 194 |
r = self.client.list_containers() |
195 | 195 |
for i in range(len(r)): |
196 | 196 |
self.assert_dicts_are_equal(r[i], container_list[i]) |
197 |
|
|
198 |
def test_upload_object(self): |
|
199 |
PC.get_container_info = Mock(return_value=container_info) |
|
200 |
PC.container_post = Mock(return_value=self.FR()) |
|
201 |
PC.object_put = Mock(return_value=self.FR()) |
|
202 |
from tempfile import NamedTemporaryFile |
|
203 |
from os import urandom |
|
204 |
tmpFile = NamedTemporaryFile() |
|
205 |
num_of_blocks = 8 |
|
206 |
file_size = num_of_blocks * 4 * 1024 * 1024 |
|
207 |
print('\n\tCreate tmp file') |
|
208 |
tmpFile.write(urandom(file_size)) |
|
209 |
tmpFile.flush() |
|
210 |
tmpFile.seek(0) |
|
211 |
print('\t\tDone') |
|
212 |
obj = 'objectName' |
|
213 |
self.client.upload_object(obj, tmpFile) |
|
214 |
tmpFile.close() |
|
215 |
self.assertEqual(PC.get_container_info.mock_calls, [call()]) |
|
216 |
[call1, call2] = PC.object_put.mock_calls |
|
217 |
|
|
218 |
(args, kwargs) = call1[1:3] |
|
219 |
self.assertEqual(args, (obj,)) |
|
220 |
expected = dict( |
|
221 |
hashmap=True, |
|
222 |
success=(201, 409), |
|
223 |
format='json', |
|
224 |
json=dict( |
|
225 |
hashes=['s0m3h@5h'] * num_of_blocks, |
|
226 |
bytes=file_size), |
|
227 |
etag=None, |
|
228 |
content_encoding=None, |
|
229 |
content_type='application/octet-stream', |
|
230 |
content_disposition=None, |
|
231 |
public=None, |
|
232 |
permissions=None) |
|
233 |
for k, v in expected.items(): |
|
234 |
if k == 'json': |
|
235 |
self.assertEqual(len(v['hashes']), len(kwargs[k]['hashes'])) |
|
236 |
self.assertEqual(v['bytes'], kwargs[k]['bytes']) |
|
237 |
else: |
|
238 |
self.assertEqual(v, kwargs[k]) |
|
239 |
# TO BE CONTINUED |
|
240 |
(args, kwargs) = call2[1:3] |
Also available in: Unified diff