+
+ @patch('%s.get_container_info' % pithos_pkg, return_value=container_info)
+ @patch('%s.object_post' % pithos_pkg, return_value=FR())
+ def test_append_object(self, post, GCI):
+ num_of_blocks = 4
+ tmpFile = self._create_temp_file(4)
+ tmpFile.seek(0, 2)
+ file_size = tmpFile.tell()
+ for turn in range(2):
+ tmpFile.seek(0, 0)
+
+ try:
+ from progress.bar import ShadyBar
+ apn_bar = ShadyBar('Mock append')
+ except ImportError:
+ apn_bar = None
+
+ if apn_bar:
+
+ def append_gen(n):
+ for i in apn_bar.iter(range(n)):
+ yield
+ yield
+
+ else:
+ append_gen = None
+
+ self.client.append_object(
+ obj, tmpFile,
+ upload_cb=append_gen if turn else None)
+ self.assertEqual((turn + 1) * num_of_blocks, len(post.mock_calls))
+ (args, kwargs) = post.mock_calls[-1][1:3]
+ self.assertEqual(args, (obj,))
+ self.assertEqual(kwargs['content_length'], len(kwargs['data']))
+ fsize = num_of_blocks * int(kwargs['content_length'])
+ self.assertEqual(fsize, file_size)
+ self.assertEqual(kwargs['content_range'], 'bytes */*')
+ exp = 'application/octet-stream'
+ self.assertEqual(kwargs['content_type'], exp)
+ self.assertEqual(kwargs['update'], True)