# coding: utf-8

#-------------------------------------------------------------------------
# Copyright (c) Microsoft.  All rights reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#   http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
#--------------------------------------------------------------------------
import os
import unittest

from azure.storage.blob import (
    AppendBlobService,
)
from tests.testcase import (
    StorageTestCase,
    TestMode,
    record,
)

#------------------------------------------------------------------------------
TEST_BLOB_PREFIX = 'blob'
FILE_PATH = 'blob_input.temp.dat'
LARGE_BLOB_SIZE = 64 * 1024
#------------------------------------------------------------------------------

class StorageAppendBlobTest(StorageTestCase):

    def setUp(self):
        super(StorageAppendBlobTest, self).setUp()

        self.bs = self._create_storage_service(AppendBlobService, self.settings)
        self.container_name = self.get_resource_name('utcontainer')

        if not self.is_playback():
            self.bs.create_container(self.container_name)

        # test chunking functionality by reducing the size of each chunk,
        # otherwise the tests would take too long to execute
        self.bs.MAX_BLOCK_SIZE = 4 * 1024

    def tearDown(self):
        if not self.is_playback():
            try:
                self.bs.delete_container(self.container_name)
            except:
                pass

        if os.path.isfile(FILE_PATH):
            try:
                os.remove(FILE_PATH)
            except:
                pass

        return super(StorageAppendBlobTest, self).tearDown()

    #--Helpers-----------------------------------------------------------------
    def _get_blob_reference(self):
        return self.get_resource_name(TEST_BLOB_PREFIX)

    def _create_blob(self):
        blob_name = self._get_blob_reference()
        self.bs.create_blob(self.container_name, blob_name)
        return blob_name

    def assertBlobEqual(self, container_name, blob_name, expected_data):
        actual_data = self.bs.get_blob_to_bytes(container_name, blob_name)
        self.assertEqual(actual_data.content, expected_data)

    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 block blobs --------------------------------------------

    @record
    def test_create_blob(self):
        # Arrange
        blob_name = self._get_blob_reference()

        # Act
        create_resp = self.bs.create_blob(self.container_name, blob_name)
        blob = self.bs.get_blob_properties(self.container_name, blob_name)

        # Assert
        self.assertTrue(self.bs.exists(self.container_name, blob_name))
        self.assertEqual(blob.properties.etag, create_resp.etag)
        self.assertEqual(blob.properties.last_modified, create_resp.last_modified)

    @record
    def test_create_blob_with_lease_id(self):
        # Arrange
        blob_name = self._create_blob()
        lease_id = self.bs.acquire_blob_lease(self.container_name, blob_name)

        # Act
        self.bs.create_blob(self.container_name, blob_name, lease_id=lease_id)

        # Assert
        self.assertTrue(self.bs.exists(self.container_name, blob_name))

    @record
    def test_create_blob_with_metadata(self):
        # Arrange
        metadata = {'hello': 'world', 'number': '42'}
        blob_name = self._get_blob_reference()

        # Act
        self.bs.create_blob(self.container_name, blob_name, metadata=metadata)

        # Assert
        md = self.bs.get_blob_metadata(self.container_name, blob_name)
        self.assertDictEqual(md, metadata)

    @record
    def test_append_block(self):
        # Arrange
        blob_name = self._create_blob()

        # Act
        for i in range(5):
            resp = self.bs.append_block(self.container_name, blob_name, 
                                        u'block {0}'.format(i).encode('utf-8'))          
            self.assertEqual(resp.append_offset, 7 * i)
            self.assertEqual(resp.committed_block_count, i + 1)
            self.assertIsNotNone(resp.etag)
            self.assertIsNotNone(resp.last_modified)

        # Assert
        self.assertBlobEqual(self.container_name, blob_name, b'block 0block 1block 2block 3block 4')

    @record
    def test_append_block_unicode(self):
        # Arrange
        blob_name = self._create_blob()

        # Act
        with self.assertRaises(TypeError):
            resp = self.bs.append_block(self.container_name, blob_name, u'啊齄丂狛狜')

        # Assert

    @record
    def test_append_block_with_md5(self):
        # Arrange
        blob_name = self._create_blob()

        # Act
        resp = self.bs.append_block(self.container_name, blob_name, 
                                    b'block',
                                    validate_content=True)

        # Assert

    @record
    def test_append_blob_from_bytes(self):
        # Arrange
        blob_name = self._create_blob()

        # Act
        data = b'abcdefghijklmnopqrstuvwxyz'
        append_resp = self.bs.append_blob_from_bytes(self.container_name, blob_name, data)
        blob = self.bs.get_blob_properties(self.container_name, blob_name)

        # Assert
        self.assertBlobEqual(self.container_name, blob_name, data)
        self.assertEqual(blob.properties.etag, append_resp.etag)
        self.assertEqual(blob.properties.last_modified, append_resp.last_modified)

    @record
    def test_append_blob_from_0_bytes(self):
        # Arrange
        blob_name = self._create_blob()

        # Act
        data = b''
        append_resp = self.bs.append_blob_from_bytes(self.container_name, blob_name, data)

        # Assert
        self.assertBlobEqual(self.container_name, blob_name, data)
        # appending nothing should not make any network call
        self.assertIsNone(append_resp.etag)
        self.assertIsNone(append_resp.last_modified)

    @record
    def test_append_blob_from_bytes_with_progress(self):
        # Arrange
        blob_name = self._create_blob()
        data = b'abcdefghijklmnopqrstuvwxyz'

        # Act
        progress = []

        def callback(current, total):
            progress.append((current, total))

        self.bs.append_blob_from_bytes(self.container_name, blob_name, data, progress_callback=callback)

        # Assert
        self.assertBlobEqual(self.container_name, blob_name, data)
        self.assert_upload_progress(len(data), self.bs.MAX_BLOCK_SIZE, progress)

    @record
    def test_append_blob_from_bytes_with_index(self):
        # Arrange
        blob_name = self._create_blob()

        # Act
        data = b'abcdefghijklmnopqrstuvwxyz'
        self.bs.append_blob_from_bytes(self.container_name, blob_name, data, 3)

        # Assert
        self.assertBlobEqual(self.container_name, blob_name, data[3:])

    @record
    def test_append_blob_from_bytes_with_index_and_count(self):
        # Arrange
        blob_name = self._create_blob()

        # Act
        data = b'abcdefghijklmnopqrstuvwxyz'
        self.bs.append_blob_from_bytes(self.container_name, blob_name, data, 3, 5)

        # Assert
        self.assertBlobEqual(self.container_name, blob_name, data[3:8])

    @record
    def test_append_blob_from_bytes_chunked_upload(self):
        # Arrange
        blob_name = self._create_blob()
        data = self.get_random_bytes(LARGE_BLOB_SIZE)

        # Act
        append_resp = self.bs.append_blob_from_bytes(self.container_name, blob_name, data)
        blob = self.bs.get_blob_properties(self.container_name, blob_name)

        # Assert
        self.assertBlobEqual(self.container_name, blob_name, data)
        self.assertEqual(blob.properties.etag, append_resp.etag)
        self.assertEqual(blob.properties.last_modified, append_resp.last_modified)

    @record
    def test_append_blob_from_bytes_with_progress_chunked_upload(self):
        # Arrange
        blob_name = self._create_blob()
        data = self.get_random_bytes(LARGE_BLOB_SIZE)

        # Act
        progress = []

        def callback(current, total):
            progress.append((current, total))

        self.bs.append_blob_from_bytes(self.container_name, blob_name, data, progress_callback=callback)

        # Assert
        self.assertBlobEqual(self.container_name, blob_name, data)
        self.assert_upload_progress(len(data), self.bs.MAX_BLOCK_SIZE, progress)

    @record
    def test_append_blob_from_bytes_chunked_upload_with_index_and_count(self):
        # Arrange
        blob_name = self._create_blob()
        data = self.get_random_bytes(LARGE_BLOB_SIZE)
        index = 33
        blob_size = len(data) - 66

        # Act
        self.bs.append_blob_from_bytes(self.container_name, blob_name, data, index, blob_size)

        # Assert
        self.assertBlobEqual(self.container_name, blob_name, data[index:index + blob_size])

    @record
    def test_append_blob_from_path_chunked_upload(self):
        # Arrange
        blob_name = self._create_blob()
        data = self.get_random_bytes(LARGE_BLOB_SIZE)
        with open(FILE_PATH, 'wb') as stream:
            stream.write(data)

        # Act
        append_resp = self.bs.append_blob_from_path(self.container_name, blob_name, FILE_PATH)
        blob = self.bs.get_blob_properties(self.container_name, blob_name)

        # Assert
        self.assertBlobEqual(self.container_name, blob_name, data)
        self.assertEqual(blob.properties.etag, append_resp.etag)
        self.assertEqual(blob.properties.last_modified, append_resp.last_modified)

    @record
    def test_append_blob_from_path_with_progress_chunked_upload(self):
        # Arrange
        blob_name = self._create_blob()
        data = self.get_random_bytes(LARGE_BLOB_SIZE)
        with open(FILE_PATH, 'wb') as stream:
            stream.write(data)

        # Act
        progress = []

        def callback(current, total):
            progress.append((current, total))

        self.bs.append_blob_from_path(self.container_name, blob_name, FILE_PATH, progress_callback=callback)

        # Assert
        self.assertBlobEqual(self.container_name, blob_name, data)
        self.assert_upload_progress(len(data), self.bs.MAX_BLOCK_SIZE, progress)

    @record
    def test_append_blob_from_stream_chunked_upload(self):
        # Arrange
        blob_name = self._create_blob()
        data = self.get_random_bytes(LARGE_BLOB_SIZE)
        with open(FILE_PATH, 'wb') as stream:
            stream.write(data)

        # Act
        with open(FILE_PATH, 'rb') as stream:
            append_resp = self.bs.append_blob_from_stream(self.container_name, blob_name, stream)
        blob = self.bs.get_blob_properties(self.container_name, blob_name)

        # Assert
        self.assertBlobEqual(self.container_name, blob_name, data)
        self.assertEqual(blob.properties.etag, append_resp.etag)
        self.assertEqual(blob.properties.last_modified, append_resp.last_modified)

    @record
    def test_append_blob_from_stream_non_seekable_chunked_upload_known_size(self):
        # Arrange
        blob_name = self._create_blob()
        data = self.get_random_bytes(LARGE_BLOB_SIZE)
        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 = StorageAppendBlobTest.NonSeekableFile(stream)
            self.bs.append_blob_from_stream(self.container_name, blob_name, 
                                            non_seekable_file, count=blob_size)

        # Assert
        self.assertBlobEqual(self.container_name, blob_name, data[:blob_size])

    @record
    def test_append_blob_from_stream_non_seekable_chunked_upload_unknown_size(self):
        # Arrange
        blob_name = self._create_blob()
        data = self.get_random_bytes(LARGE_BLOB_SIZE)
        with open(FILE_PATH, 'wb') as stream:
            stream.write(data)

        # Act
        with open(FILE_PATH, 'rb') as stream:
            non_seekable_file = StorageAppendBlobTest.NonSeekableFile(stream)
            self.bs.append_blob_from_stream(self.container_name, blob_name, non_seekable_file)

        # Assert
        self.assertBlobEqual(self.container_name, blob_name, data)

    @record
    def test_append_blob_from_stream_with_multiple_appends(self):
        # Arrange
        blob_name = self._create_blob()
        data = self.get_random_bytes(LARGE_BLOB_SIZE)
        with open(FILE_PATH, 'wb') as stream1:
            stream1.write(data)
        with open(FILE_PATH, 'wb') as stream2:
            stream2.write(data)

        # Act
        with open(FILE_PATH, 'rb') as stream1:
            self.bs.append_blob_from_stream(self.container_name, blob_name, stream1)
        with open(FILE_PATH, 'rb') as stream2:
            self.bs.append_blob_from_stream(self.container_name, blob_name, stream2)

        # Assert
        data = data * 2
        self.assertBlobEqual(self.container_name, blob_name, data)

    @record
    def test_append_blob_from_stream_chunked_upload_with_count(self):
        # Arrange
        blob_name = self._create_blob()
        data = self.get_random_bytes(LARGE_BLOB_SIZE)
        with open(FILE_PATH, 'wb') as stream:
            stream.write(data)

        # Act
        blob_size = len(data) - 301
        with open(FILE_PATH, 'rb') as stream:
            self.bs.append_blob_from_stream(self.container_name, blob_name, stream, blob_size)

        # Assert
        self.assertBlobEqual(self.container_name, blob_name, data[:blob_size])

    def test_append_blob_from_stream_chunked_upload_with_count_parallel(self):
        # parallel tests introduce random order of requests, can only run live
        if TestMode.need_recording_file(self.test_mode):
            return

        # Arrange
        blob_name = self._create_blob()
        data = self.get_random_bytes(LARGE_BLOB_SIZE)
        with open(FILE_PATH, 'wb') as stream:
            stream.write(data)

        # Act
        blob_size = len(data) - 301
        with open(FILE_PATH, 'rb') as stream:
            append_resp = self.bs.append_blob_from_stream(self.container_name, blob_name, stream, blob_size)
        blob = self.bs.get_blob_properties(self.container_name, blob_name)

        # Assert
        self.assertBlobEqual(self.container_name, blob_name, data[:blob_size])
        self.assertEqual(blob.properties.etag, append_resp.etag)
        self.assertEqual(blob.properties.last_modified, append_resp.last_modified)

    @record
    def test_append_blob_from_text(self):
        # Arrange
        blob_name = self._create_blob()
        text = u'hello 啊齄丂狛狜 world'
        data = text.encode('utf-8')

        # Act
        append_resp = self.bs.append_blob_from_text(self.container_name, blob_name, text)
        blob = self.bs.get_blob_properties(self.container_name, blob_name)

        # Assert
        self.assertBlobEqual(self.container_name, blob_name, data)
        self.assertEqual(blob.properties.etag, append_resp.etag)
        self.assertEqual(blob.properties.last_modified, append_resp.last_modified)

    @record
    def test_append_blob_from_text_with_encoding(self):
        # Arrange
        blob_name = self._create_blob()
        text = u'hello 啊齄丂狛狜 world'
        data = text.encode('utf-16')

        # Act
        self.bs.append_blob_from_text(self.container_name, blob_name, text, 'utf-16')

        # Assert
        self.assertBlobEqual(self.container_name, blob_name, data)

    @record
    def test_append_blob_from_text_with_encoding_and_progress(self):
        # Arrange
        blob_name = self._create_blob()
        text = u'hello 啊齄丂狛狜 world'
        data = text.encode('utf-16')

        # Act
        progress = []

        def callback(current, total):
            progress.append((current, total))

        self.bs.append_blob_from_text(self.container_name, blob_name, text, 'utf-16', 
                                      progress_callback=callback)

        # Assert
        self.assert_upload_progress(len(data), self.bs.MAX_BLOCK_SIZE, progress)

    @record
    def test_append_blob_from_text_chunked_upload(self):
        # Arrange
        blob_name = self._create_blob()
        data = self.get_random_text_data(LARGE_BLOB_SIZE)
        encoded_data = data.encode('utf-8')

        # Act
        self.bs.append_blob_from_text(self.container_name, blob_name, data)

        # Assert
        self.assertBlobEqual(self.container_name, blob_name, encoded_data)

    @record
    def test_append_blob_with_md5(self):
        # Arrange
        blob_name = self._create_blob()
        data = b'hello world'

        # Act
        self.bs.append_blob_from_bytes(self.container_name, blob_name, data, 
                                       validate_content=True)

        # Assert

#------------------------------------------------------------------------------
if __name__ == '__main__':
    unittest.main()
