Repository: azure/azure-sdk-for-python
Issue: Version 0.10.0 fails with 'SocketReader' object has no attribute 'tell'

In version 0.10.0 there is a bug in `azure/storage/__init__.py` causing 
`AttributeError: 'SocketReader' object has no attribute 'tell'`


Changed files:
tests/test_blobservice.py, azure/storage/__init__.py, azure/storage/blobservice.py

diff --git a/azure/storage/__init__.py b/azure/storage/__init__.py
index 408a07d928..acb3050509 100644
--- a/azure/storage/__init__.py
+++ b/azure/storage/__init__.py
@@ -900,7 +900,7 @@ class _BlobChunkDownloader(object):
         self.blob_size = blob_size
         self.chunk_size = chunk_size
         self.stream = stream
-        self.stream_start = stream.tell()
+        self.stream_start = stream.tell() if parallel else None
         self.stream_lock = threading.Lock() if parallel else None
         self.progress_callback = progress_callback
         self.progress_total = 0
@@ -938,7 +938,6 @@ class _BlobChunkDownloader(object):
                 self.stream.seek(self.stream_start + chunk_offset)
                 self.stream.write(chunk_data)
         else:
-            self.stream.seek(self.stream_start + chunk_offset)
             self.stream.write(chunk_data)
 
     def _download_chunk_with_retries(self, chunk_offset):
@@ -969,7 +968,7 @@ class _BlobChunkUploader(object):
         self.blob_size = blob_size
         self.chunk_size = chunk_size
         self.stream = stream
-        self.stream_start = stream.tell()
+        self.stream_start = stream.tell() if parallel else None
         self.stream_lock = threading.Lock() if parallel else None
         self.progress_callback = progress_callback
         self.progress_total = 0
@@ -1001,13 +1000,27 @@ class _BlobChunkUploader(object):
         chunk_data = self._read_from_stream(chunk_offset, size)
         return self._upload_chunk_with_retries(chunk_offset, chunk_data)
 
+    def process_all_unknown_size(self):
+        assert self.stream_lock is None
+        range_ids = []
+        index = 0
+        while True:
+            data = self._read_from_stream(None, self.chunk_size)
+            if data:
+                index += len(data)
+                range_id = self._upload_chunk_with_retries(index, data)
+                range_ids.append(range_id)
+            else:
+                break
+
+        return range_ids
+
     def _read_from_stream(self, offset, count):
         if self.stream_lock is not None:
             with self.stream_lock:
                 self.stream.seek(self.stream_start + offset)
                 data = self.stream.read(count)
         else:
-            self.stream.seek(self.stream_start + offset)
             data = self.stream.read(count)
         return data
 
@@ -1118,7 +1131,10 @@ def _upload_blob_chunks(blob_service, container_name, blob_name,
         executor = concurrent.futures.ThreadPoolExecutor(max_connections)
         range_ids = list(executor.map(uploader.process_chunk, uploader.get_chunk_offsets()))
     else:
-        range_ids = [uploader.process_chunk(start) for start in uploader.get_chunk_offsets()]
+        if blob_size is not None:
+            range_ids = [uploader.process_chunk(start) for start in uploader.get_chunk_offsets()]
+        else:
+            range_ids = uploader.process_all_unknown_size()
 
     return range_ids
 
diff --git a/azure/storage/blobservice.py b/azure/storage/blobservice.py
index 2696e9fd08..7074191395 100644
--- a/azure/storage/blobservice.py
+++ b/azure/storage/blobservice.py
@@ -911,6 +911,7 @@ class BlobService(_StorageClient):
             Set to 1 to upload the blob chunks sequentially.
             Set to 2 or more to upload the blob chunks in parallel. This uses
             more system resources but will upload faster.
+            Note that parallel upload requires the stream to be seekable.
         max_retries:
             Number of times to retry upload of blob chunk if an error occurs.
         retry_wait:
@@ -1392,6 +1393,7 @@ class BlobService(_StorageClient):
             Set to 1 to upload the blob chunks sequentially.
             Set to 2 or more to upload the blob chunks in parallel. This uses
             more system resources but will upload faster.
+            Note that parallel upload requires the stream to be seekable.
         max_retries:
             Number of times to retry upload of blob chunk if an error occurs.
         retry_wait:
diff --git a/tests/test_blobservice.py b/tests/test_blobservice.py
index 8fe06fba3a..e98e262e8a 100644
--- a/tests/test_blobservice.py
+++ b/tests/test_blobservice.py
@@ -350,6 +350,16 @@ class BlobServiceTest(AzureTestCase):
         finally:
             connection.close()
 
+    class NonSeekableFile(object):
+        def __init__(self, wrapped_file):
+            self.wrapped_file = wrapped_file
+
+        def write(self, data):
+            self.wrapped_file.write(data)
+
+        def read(self, count):
+            return self.wrapped_file.read(count)
+
     #--Test cases for blob service --------------------------------------------
     def test_create_blob_service_missing_arguments(self):
         # Arrange
@@ -3040,6 +3050,47 @@ class BlobServiceTest(AzureTestCase):
             actual = stream.read()
             self.assertEqual(data, actual)
 
+    def test_get_blob_to_file_non_seekable_chunked_download(self):
+        # Arrange
+        blob_name = 'blob1'
+        data = self._get_oversized_binary_data()
+        file_path = 'blob_output.temp.dat'
+        self._create_container_and_block_blob(
+            self.container_name, blob_name, data)
+
+        # Act
+        with open(file_path, 'wb') as stream:
+            non_seekable_stream = BlobServiceTest.NonSeekableFile(stream)
+            resp = self.bs.get_blob_to_file(
+                self.container_name, blob_name, non_seekable_stream,
+                max_connections=1)
+
+        # Assert
+        self.assertIsNone(resp)
+        with open(file_path, 'rb') as stream:
+            actual = stream.read()
+            self.assertEqual(data, actual)
+
+    def test_get_blob_to_file_non_seekable_chunked_download_parallel(self):
+        # Arrange
+        blob_name = 'blob1'
+        data = self._get_oversized_binary_data()
+        file_path = 'blob_output.temp.dat'
+        self._create_container_and_block_blob(
+            self.container_name, blob_name, data)
+
+        # Act
+        with open(file_path, 'wb') as stream:
+            non_seekable_stream = BlobServiceTest.NonSeekableFile(stream)
+
+            # Parallel downloads require that the file be seekable
+            with self.assertRaises(AttributeError):
+                resp = self.bs.get_blob_to_file(
+                    self.container_name, blob_name, non_seekable_stream,
+                    max_connections=10)
+
+        # Assert
+
     def test_get_blob_to_file_with_progress(self):
         # Arrange
         blob_name = 'blob1'
@@ -3659,6 +3710,70 @@ class BlobServiceTest(AzureTestCase):
         self.assertBlobLengthEqual(self.container_name, blob_name, len(data))
         self.assertBlobEqual(self.container_name, blob_name, data)
 
+    def test_put_block_blob_from_file_non_seekable_chunked_upload_known_size(self):
+        # Arrange
+        self._create_container(self.container_name)
+        blob_name = 'blob1'
+        data = self._get_oversized_binary_data()
+        file_path = 'blob_input.temp.dat'
+        with open(file_path, 'wb') as stream:
+            stream.write(data)
+        blob_size = len(data) - 66
+
+        # Act
+        with open(file_path, 'rb') as stream:
+            non_seekable_file = BlobServiceTest.NonSeekableFile(stream)
+            resp = self.bs.put_block_blob_from_file(
+                self.container_name, blob_name, non_seekable_file,
+                count=blob_size, max_connections=1)
+
+        # Assert
+        self.assertIsNone(resp)
+        self.assertBlobLengthEqual(self.container_name, blob_name, blob_size)
+        self.assertBlobEqual(self.container_name, blob_name, data[:blob_size])
+
+    def test_put_block_blob_from_file_non_seekable_chunked_upload_unknown_size(self):
+        # Arrange
+        self._create_container(self.container_name)
+        blob_name = 'blob1'
+        data = self._get_oversized_binary_data()
+        file_path = 'blob_input.temp.dat'
+        with open(file_path, 'wb') as stream:
+            stream.write(data)
+
+        # Act
+        with open(file_path, 'rb') as stream:
+            non_seekable_file = BlobServiceTest.NonSeekableFile(stream)
+            resp = self.bs.put_block_blob_from_file(
+                self.container_name, blob_name, non_seekable_file,
+                max_connections=1)
+
+        # Assert
+        self.assertIsNone(resp)
+        self.assertBlobLengthEqual(self.container_name, blob_name, len(data))
+        self.assertBlobEqual(self.container_name, blob_name, data)
+
+    def test_put_block_blob_from_file_non_seekable_chunked_upload_parallel(self):
+        # Arrange
+        self._create_container(self.container_name)
+        blob_name = 'blob1'
+        data = self._get_oversized_binary_data()
+        file_path = 'blob_input.temp.dat'
+        with open(file_path, 'wb') as stream:
+            stream.write(data)
+
+        # Act
+        with open(file_path, 'rb') as stream:
+            non_seekable_file = BlobServiceTest.NonSeekableFile(stream)
+
+            # Parallel uploads require that the file be seekable
+            with self.assertRaises(AttributeError):
+                resp = self.bs.put_block_blob_from_file(
+                    self.container_name, blob_name, non_seekable_file,
+                    max_connections=10)
+
+        # Assert
+
     def test_put_block_blob_from_file_with_progress_chunked_upload(self):
         # Arrange
         self._create_container(self.container_name)
@@ -4129,6 +4244,50 @@ class BlobServiceTest(AzureTestCase):
         self.assertBlobLengthEqual(self.container_name, blob_name, blob_size)
         self.assertBlobEqual(self.container_name, blob_name, data[:blob_size])
 
+    def test_put_page_blob_from_file_non_seekable_chunked_upload(self):
+        # Arrange
+        self._create_container(self.container_name)
+        blob_name = 'blob1'
+        data = self._get_oversized_page_blob_binary_data()
+        file_path = 'blob_input.temp.dat'
+        with open(file_path, 'wb') as stream:
+            stream.write(data)
+
+        # Act
+        blob_size = len(data)
+        with open(file_path, 'rb') as stream:
+            non_seekable_file = BlobServiceTest.NonSeekableFile(stream)
+            resp = self.bs.put_page_blob_from_file(
+                self.container_name, blob_name, non_seekable_file, blob_size,
+                max_connections=1)
+
+        # Assert
+        self.assertIsNone(resp)
+        self.assertBlobLengthEqual(self.container_name, blob_name, blob_size)
+        self.assertBlobEqual(self.container_name, blob_name, data[:blob_size])
+
+    def test_put_page_blob_from_file_non_seekable_chunked_upload_parallel(self):
+        # Arrange
+        self._create_container(self.container_name)
+        blob_name = 'blob1'
+        data = self._get_oversized_page_blob_binary_data()
+        file_path = 'blob_input.temp.dat'
+        with open(file_path, 'wb') as stream:
+            stream.write(data)
+
+        # Act
+        blob_size = len(data)
+        with open(file_path, 'rb') as stream:
+            non_seekable_file = BlobServiceTest.NonSeekableFile(stream)
+
+            # Parallel uploads require that the file be seekable
+            with self.assertRaises(AttributeError):
+                resp = self.bs.put_page_blob_from_file(
+                    self.container_name, blob_name, non_seekable_file, blob_size,
+                    max_connections=10)
+
+        # Assert
+
     def test_put_page_blob_from_file_with_progress_chunked_upload(self):
         # Arrange
         self._create_container(self.container_name)