Add a test case for it
# limitations under the License.
import sys
+import errno
import hashlib
from libcloud.utils.py3 import httplib
@@ -92,7 +93,6 @@ class BaseStorageTests(unittest.TestCase):
response_streamed = mock_response.request.stream
assert response_streamed is False
-
def test__get_hash_function(self):
self.driver1.hash_type = 'md5'
func = self.driver1._get_hash_function()
@@ -141,7 +141,8 @@ class BaseStorageTests(unittest.TestCase):
def test_upload_object_hash_calculation_is_efficient(self, mock_read_in_chunks,
mock_exhaust_iterator):
# Verify that we don't buffer whole file in memory when calculating
- # object has when iterator has __next__ method, but instead read and calculate hash in chunks
+ # object has when iterator has __next__ method, but instead read and
+ # calculate hash in chunks
size = 100
self.driver1.connection = Mock()
@@ -210,6 +211,50 @@ class BaseStorageTests(unittest.TestCase):
self.assertEqual(mock_read_in_chunks.call_count, 2)
self.assertEqual(mock_exhaust_iterator.call_count, 0)
+ def test_upload_object_via_stream_illegal_seek_errors_are_ignored(self):
+ # Illegal seek errors should be ignored
+ size = 100
+
+ self.driver1.connection = Mock()
+
+ seek_error = OSError('Illegal seek')
+ seek_error.errno = 29
+ assert errno.ESPIPE == 29
+
+ iterator = BodyStream('a' * size)
+ iterator.seek = mock.Mock(side_effect=seek_error)
+ self.assertTrue(hasattr(iterator, '__next__'))
+ self.assertTrue(hasattr(iterator, 'next'))
+
+ result = self.driver1._upload_object(object_name='test1',
+ content_type=None,
+ request_path='/',
+ stream=iterator)
+
+ hasher = hashlib.md5()
+ hasher.update(b('a') * size)
+ expected_hash = hasher.hexdigest()
+
+ self.assertEqual(result['data_hash'], expected_hash)
+ self.assertEqual(result['bytes_transferred'], size)
+
+ # But others shouldn't
+
+ self.driver1.connection = Mock()
+
+ seek_error = OSError('Other error')
+ seek_error.errno = 21
+
+ iterator = BodyStream('b' * size)
+ iterator.seek = mock.Mock(side_effect=seek_error)
+
+ self.assertRaisesRegex(OSError, 'Other error',
+ self.driver1._upload_object,
+ object_name='test1',
+ content_type=None,
+ request_path='/',
+ stream=iterator)
+
if __name__ == '__main__':
sys.exit(unittest.main())
