Revision cf687343 kamaki/clients/test/pithos.py

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