test_file.py

# coding: utf-8

# -------------------------------------------------------------------------
# Copyright (c) Microsoft Corporation. All rights reserved.
# Licensed under the MIT License. See License.txt in the project root for
# license information.
# --------------------------------------------------------------------------
import base64
import os
import unittest
from datetime import datetime, timedelta

import requests
from azure.common import (
    AzureHttpError,
    AzureMissingResourceHttpError,
)

from azure.storage.common import (
    AccessPolicy,
    ResourceTypes,
    AccountPermissions,
)
from azure.storage.file import (
    File,
    FileService,
    ContentSettings,
    FilePermissions,
    DeleteSnapshot,
)
from tests.testcase import (
    StorageTestCase,
    TestMode,
    record,
)

# ------------------------------------------------------------------------------
TEST_SHARE_PREFIX = 'share'
TEST_DIRECTORY_PREFIX = 'dir'
TEST_FILE_PREFIX = 'file'
INPUT_FILE_PATH = 'file_input.temp.dat'
OUTPUT_FILE_PATH = 'file_output.temp.dat'
LARGE_FILE_SIZE = 64 * 1024 + 5


# ------------------------------------------------------------------------------

class StorageFileTest(StorageTestCase):
    def setUp(self):
        super(StorageFileTest, self).setUp()

        self.fs = self._create_storage_service(FileService, self.settings)

        # test chunking functionality by reducing the threshold
        # for chunking and the size of each chunk, otherwise
        # the tests would take too long to execute
        self.fs.MAX_RANGE_SIZE = 4 * 1024

        self.share_name = self.get_resource_name('utshare')

        if not self.is_playback():
            self.fs.create_share(self.share_name)

        self.short_byte_data = self.get_random_bytes(1024)

        self.fs2 = self._create_remote_storage_service(FileService, self.settings)
        self.remote_share_name = None

    def tearDown(self):
        if not self.is_playback():
            try:
                self.fs.delete_share(self.share_name, delete_snapshots=DeleteSnapshot.Include)
            except:
                pass

            if self.remote_share_name:
                try:
                    self.fs2.delete_share(self.remote_share_name, delete_snapshots=DeleteSnapshot.Include)
                except:
                    pass

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

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

        return super(StorageFileTest, self).tearDown()

    # --Helpers-----------------------------------------------------------------
    def _get_file_reference(self):
        return self.get_resource_name(TEST_FILE_PREFIX)

    def _create_file(self):
        file_name = self._get_file_reference()
        self.fs.create_file_from_bytes(self.share_name, None, file_name, self.short_byte_data)
        return file_name

    def _create_remote_share(self):
        self.remote_share_name = self.get_resource_name('remoteshare')
        self.fs2.create_share(self.remote_share_name)

    def _create_remote_file(self, file_data=None):
        if not file_data:
            file_data = b'12345678' * 1024 * 1024
        source_file_name = self._get_file_reference()
        self.fs2.create_file_from_bytes(self.remote_share_name, None, source_file_name, file_data)
        return source_file_name

    def _wait_for_async_copy(self, share_name, dir_name, file_name):
        count = 0
        file = self.fs.get_file_properties(share_name, dir_name, file_name)
        while file.properties.copy.status != 'success':
            count = count + 1
            if count > 10:
                self.fail('Timed out waiting for async copy to complete.')
            self.sleep(6)
            file = self.fs.get_file_properties(share_name, dir_name, file_name)
        self.assertEqual(file.properties.copy.status, 'success')

    def assertFileEqual(self, share_name, dir_name, file_name, expected_data):
        actual_data = self.fs.get_file_to_bytes(share_name, dir_name, file_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 files ----------------------------------------------
    @record
    def test_make_file_url(self):
        # Arrange

        # Act
        res = self.fs.make_file_url('vhds', 'vhd_dir', 'my.vhd')

        # Assert
        self.assertEqual(res, 'https://' + self.settings.STORAGE_ACCOUNT_NAME
                         + '.file.core.windows.net/vhds/vhd_dir/my.vhd')

    @record
    def test_make_file_url_no_directory(self):
        # Arrange

        # Act
        res = self.fs.make_file_url('vhds', None, 'my.vhd')

        # Assert
        self.assertEqual(res, 'https://' + self.settings.STORAGE_ACCOUNT_NAME
                         + '.file.core.windows.net/vhds/my.vhd')

    @record
    def test_make_file_url_with_protocol(self):
        # Arrange

        # Act
        res = self.fs.make_file_url('vhds', 'vhd_dir', 'my.vhd', protocol='http')

        # Assert
        self.assertEqual(res, 'http://' + self.settings.STORAGE_ACCOUNT_NAME
                         + '.file.core.windows.net/vhds/vhd_dir/my.vhd')

    @record
    def test_make_file_url_with_sas(self):
        # Arrange

        # Act
        res = self.fs.make_file_url(
            'vhds', 'vhd_dir', 'my.vhd', sas_token='sas')

        # Assert
        self.assertEqual(res, 'https://' + self.settings.STORAGE_ACCOUNT_NAME +
                         '.file.core.windows.net/vhds/vhd_dir/my.vhd?sas')

    @record
    def test_create_file(self):
        # Arrange
        file_name = self._get_file_reference()

        # Act
        self.fs.create_file(self.share_name, None, file_name, 1024)

        # Assert
        self.fs.exists(self.share_name, None, file_name)

    @record
    def test_create_file_with_metadata(self):
        # Arrange
        metadata = {'hello': 'world', 'number': '42'}
        file_name = self._get_file_reference()

        # Act
        self.fs.create_file(self.share_name, None, file_name, 1024, metadata=metadata)

        # Assert
        md = self.fs.get_file_metadata(self.share_name, None, file_name)
        self.assertDictEqual(md, metadata)

    @record
    def test_file_exists(self):
        # Arrange
        file_name = self._create_file()

        # Act
        exists = self.fs.exists(self.share_name, None, file_name)

        # Assert
        self.assertTrue(exists)

    @record
    def test_file_not_exists(self):
        # Arrange
        file_name = self._get_file_reference()

        # Act
        exists = self.fs.exists(self.share_name, 'missingdir', file_name)

        # Assert
        self.assertFalse(exists)

    @record
    def test_file_exists_with_snapshot(self):
        # Arrange
        file_name = self._create_file()
        snapshot = self.fs.snapshot_share(self.share_name)
        self.fs.delete_file(self.share_name, None, file_name)

        # Act
        exists = self.fs.exists(self.share_name, None, file_name, snapshot=snapshot.snapshot)

        # Assert
        self.assertTrue(exists)

    @record
    def test_file_not_exists_with_snapshot(self):
        # Arrange
        snapshot = self.fs.snapshot_share(self.share_name)
        file_name = self._create_file()

        # Act
        exists = self.fs.exists(self.share_name, None, file_name, snapshot=snapshot.snapshot)

        # Assert
        self.assertFalse(exists)

    @record
    def test_resize_file(self):
        # Arrange
        file_name = self._create_file()

        # Act
        self.fs.resize_file(self.share_name, None, file_name, 5)

        # Assert
        file = self.fs.get_file_properties(self.share_name, None, file_name)
        self.assertEqual(file.properties.content_length, 5)

    @record
    def test_set_file_properties(self):
        # Arrange
        file_name = self._create_file()

        # Act
        content_settings = ContentSettings(
            content_language='spanish',
            content_disposition='inline')
        resp = self.fs.set_file_properties(
            self.share_name,
            None,
            file_name,
            content_settings=content_settings
        )

        # Assert
        properties = self.fs.get_file_properties(self.share_name, None, file_name).properties
        self.assertEqual(properties.content_settings.content_language, content_settings.content_language)
        self.assertEqual(properties.content_settings.content_disposition, content_settings.content_disposition)

    @record
    def test_get_file_properties(self):
        # Arrange
        file_name = self._create_file()

        # Act
        file = self.fs.get_file_properties(self.share_name, None, file_name)

        # Assert
        self.assertIsNotNone(file)
        self.assertEqual(file.properties.content_length, len(self.short_byte_data))

    @record
    def test_get_file_properties_with_snapshot(self):
        # Arrange
        file_name = self._create_file()
        metadata = {"test1": "foo", "test2": "bar"}
        self.fs.set_file_metadata(self.share_name, None, file_name, metadata)
        snapshot = self.fs.snapshot_share(self.share_name)
        metadata2 = {"test100": "foo100", "test200": "bar200"}
        self.fs.set_file_metadata(self.share_name, None, file_name, metadata2)

        # Act
        file = self.fs.get_file_properties(self.share_name, None, file_name)
        file_snapshot = self.fs.get_file_properties(self.share_name, None, file_name, snapshot=snapshot.snapshot)

        # Assert
        self.assertIsNotNone(file)
        self.assertIsNotNone(file_snapshot)
        self.assertEqual(file.properties.content_length, file_snapshot.properties.content_length)
        self.assertDictEqual(metadata, file_snapshot.metadata)

    @record
    def test_get_file_metadata_with_snapshot(self):
        # Arrange
        file_name = self._create_file()
        metadata = {"test1": "foo", "test2": "bar"}
        self.fs.set_file_metadata(self.share_name, None, file_name, metadata)
        snapshot = self.fs.snapshot_share(self.share_name)
        metadata2 = {"test100": "foo100", "test200": "bar200"}
        self.fs.set_file_metadata(self.share_name, None, file_name, metadata2)

        # Act
        file_metadata = self.fs.get_file_metadata(self.share_name, None, file_name)
        file_snapshot_metadata = self.fs.get_file_metadata(self.share_name, None, file_name, snapshot=snapshot.snapshot)

        # Assert
        self.assertDictEqual(metadata2, file_metadata)
        self.assertDictEqual(metadata, file_snapshot_metadata)

    @record
    def test_get_file_properties_with_non_existing_file(self):
        # Arrange
        file_name = self._get_file_reference()

        # Act
        with self.assertRaises(AzureMissingResourceHttpError):
            self.fs.get_file_properties(self.share_name, None, file_name)

            # Assert

    @record
    def test_get_file_metadata(self):
        # Arrange
        file_name = self._create_file()

        # Act
        md = self.fs.get_file_metadata(self.share_name, None, file_name)

        # Assert
        self.assertIsNotNone(md)
        self.assertEqual(0, len(md))

    @record
    def test_set_file_metadata_with_upper_case(self):
        # Arrange
        metadata = {'hello': 'world', 'number': '42', 'UP': 'UPval'}
        file_name = self._create_file()

        # Act
        self.fs.set_file_metadata(self.share_name, None, file_name, metadata)

        # Assert
        md = self.fs.get_file_metadata(self.share_name, None, file_name)
        self.assertEqual(3, len(md))
        self.assertEqual(md['hello'], 'world')
        self.assertEqual(md['number'], '42')
        self.assertEqual(md['UP'], 'UPval')
        self.assertFalse('up' in md)

    @record
    def test_delete_file_with_existing_file(self):
        # Arrange
        file_name = self._create_file()

        # Act
        self.fs.delete_file(self.share_name, None, file_name)

        # Assert
        self.assertFalse(self.fs.exists(self.share_name, None, file_name))

    @record
    def test_delete_file_with_non_existing_file(self):
        # Arrange
        file_name = self._get_file_reference()

        # Act
        with self.assertRaises(AzureMissingResourceHttpError):
            self.fs.delete_file(self.share_name, None, file_name)

            # Assert

    @record
    def test_update_range(self):
        # Arrange
        file_name = self._create_file()

        # Act
        data = b'abcdefghijklmnop' * 32
        self.fs.update_range(self.share_name, None, file_name, data, 0, 511)

        # Assert
        file = self.fs.get_file_to_bytes(self.share_name, None, file_name)
        self.assertEqual(data, file.content[:512])
        self.assertEqual(self.short_byte_data[512:], file.content[512:])

    @record
    def test_update_range_with_md5(self):
        # Arrange
        file_name = self._create_file()

        # Act
        data = b'abcdefghijklmnop' * 32
        self.fs.update_range(self.share_name, None, file_name, data, 0, 511, validate_content=True)

        # Assert

    @record
    def test_clear_range(self):
        # Arrange
        file_name = self._create_file()

        # Act
        resp = self.fs.clear_range(self.share_name, None, file_name, 0, 511)

        # Assert
        file = self.fs.get_file_to_bytes(self.share_name, None, file_name)
        self.assertEqual(b'\x00' * 512, file.content[:512])
        self.assertEqual(self.short_byte_data[512:], file.content[512:])

    @record
    def test_update_file_unicode(self):
        # Arrange
        file_name = self._create_file()

        # Act
        data = u'abcdefghijklmnop' * 32
        with self.assertRaises(TypeError):
            self.fs.update_range(self.share_name, None, file_name,
                                 data, 0, 511)

            # Assert

    @record
    def test_list_ranges_none(self):
        # Arrange
        file_name = self._get_file_reference()
        self.fs.create_file(self.share_name, None, file_name, 1024)

        # Act
        ranges = self.fs.list_ranges(self.share_name, None, file_name)

        # Assert
        self.assertIsNotNone(ranges)
        self.assertEqual(len(ranges), 0)

    @record
    def test_list_ranges_2(self):
        # Arrange
        file_name = self._get_file_reference()
        self.fs.create_file(self.share_name, None, file_name, 2048)

        data = b'abcdefghijklmnop' * 32
        resp1 = self.fs.update_range(self.share_name, None, file_name, data, 0, 511)
        resp2 = self.fs.update_range(self.share_name, None, file_name, data, 1024, 1535)

        # Act
        ranges = self.fs.list_ranges(self.share_name, None, file_name)

        # Assert
        self.assertIsNotNone(ranges)
        self.assertEqual(len(ranges), 2)
        self.assertEqual(ranges[0].start, 0)
        self.assertEqual(ranges[0].end, 511)
        self.assertEqual(ranges[1].start, 1024)
        self.assertEqual(ranges[1].end, 1535)

    @record
    def test_list_ranges_none_from_snapshot(self):
        # Arrange
        file_name = self._get_file_reference()
        self.fs.create_file(self.share_name, None, file_name, 1024)
        share_snapshot = self.fs.snapshot_share(self.share_name)
        self.fs.delete_file(self.share_name, None, file_name)

        # Act
        ranges = self.fs.list_ranges(self.share_name, None, file_name, snapshot=share_snapshot.snapshot)

        # Assert
        self.assertIsNotNone(ranges)
        self.assertEqual(len(ranges), 0)

    @record
    def test_list_ranges_2_from_snapshot(self):
        # Arrange
        file_name = self._get_file_reference()
        self.fs.create_file(self.share_name, None, file_name, 2048)

        data = b'abcdefghijklmnop' * 32
        resp1 = self.fs.update_range(self.share_name, None, file_name, data, 0, 511)
        resp2 = self.fs.update_range(self.share_name, None, file_name, data, 1024, 1535)
        share_snapshot = self.fs.snapshot_share(self.share_name)
        self.fs.delete_file(self.share_name, None, file_name)

        # Act
        ranges = self.fs.list_ranges(self.share_name, None, file_name, snapshot=share_snapshot.snapshot)

        # Assert
        self.assertIsNotNone(ranges)
        self.assertEqual(len(ranges), 2)
        self.assertEqual(ranges[0].start, 0)
        self.assertEqual(ranges[0].end, 511)
        self.assertEqual(ranges[1].start, 1024)
        self.assertEqual(ranges[1].end, 1535)

    @record
    def test_copy_file_with_existing_file(self):
        # Arrange
        file_name = self._create_file()

        # Act
        sourcefile = self.fs.make_file_url(self.share_name, None, file_name)
        copy = self.fs.copy_file(self.share_name, None, 'file1copy', sourcefile)

        # Assert
        self.assertIsNotNone(copy)
        self.assertEqual(copy.status, 'success')
        self.assertIsNotNone(copy.id)
        copy_file = self.fs.get_file_to_bytes(self.share_name, None, 'file1copy')
        self.assertEqual(copy_file.content, self.short_byte_data)

    @record
    def test_copy_file_async_private_file(self):
        # Arrange
        self._create_remote_share()
        source_file_name = self._create_remote_file()
        source_file_url = self.fs2.make_file_url(self.remote_share_name, None, source_file_name)

        # Act
        target_file_name = 'targetfile'
        with self.assertRaises(AzureMissingResourceHttpError):
            self.fs.copy_file(self.share_name, None, target_file_name, source_file_url)

            # Assert

    @record
    def test_copy_file_async_private_file_with_sas(self):
        # Arrange
        data = b'12345678' * 1024 * 1024
        self._create_remote_share()
        source_file_name = self._create_remote_file(file_data=data)

        sas_token = self.fs2.generate_file_shared_access_signature(
            self.remote_share_name,
            None,
            source_file_name,
            permission=FilePermissions.READ,
            expiry=datetime.utcnow() + timedelta(hours=1),
        )

        source_file_url = self.fs2.make_file_url(
            self.remote_share_name,
            None,
            source_file_name,
            sas_token=sas_token,
        )

        # Act
        target_file_name = 'targetfile'
        copy_resp = self.fs.copy_file(self.share_name, None, target_file_name, source_file_url)

        # Assert
        self.assertEqual(copy_resp.status, 'pending')
        self._wait_for_async_copy(self.share_name, None, target_file_name)
        actual_data = self.fs.get_file_to_bytes(self.share_name, None, target_file_name)
        self.assertEqual(actual_data.content, data)

    @record
    def test_abort_copy_file(self):
        # Arrange
        data = b'12345678' * 1024 * 1024
        self._create_remote_share()
        source_file_name = self._create_remote_file(file_data=data)

        sas_token = self.fs2.generate_file_shared_access_signature(
            self.remote_share_name,
            None,
            source_file_name,
            permission=FilePermissions.READ,
            expiry=datetime.utcnow() + timedelta(hours=1),
        )

        source_file_url = self.fs2.make_file_url(
            self.remote_share_name,
            None,
            source_file_name,
            sas_token=sas_token,
        )

        # Act
        target_file_name = 'targetfile'
        copy_resp = self.fs.copy_file(self.share_name, None, target_file_name, source_file_url)
        self.assertEqual(copy_resp.status, 'pending')
        self.fs.abort_copy_file(self.share_name, None, 'targetfile', copy_resp.id)

        # Assert
        target_file = self.fs.get_file_to_bytes(self.share_name, None, target_file_name)
        self.assertEqual(target_file.content, b'')
        self.assertEqual(target_file.properties.copy.status, 'aborted')

    @record
    def test_abort_copy_file_with_synchronous_copy_fails(self):
        # Arrange
        source_file_name = self._create_file()
        source_file_url = self.fs.make_file_url(self.share_name, None, source_file_name)

        # Act
        target_file_name = 'targetfile'
        copy_resp = self.fs.copy_file(self.share_name, None, target_file_name, source_file_url)
        with self.assertRaises(AzureHttpError):
            self.fs.abort_copy_file(
                self.share_name,
                None,
                target_file_name,
                copy_resp.id)

        # Assert
        self.assertEqual(copy_resp.status, 'success')

    @record
    def test_unicode_get_file_unicode_name(self):
        # Arrange
        file_name = '啊齄丂狛狜'
        self.fs.create_file_from_bytes(self.share_name, None, file_name, b'hello world')

        # Act
        file = self.fs.get_file_to_bytes(self.share_name, None, file_name)

        # Assert
        self.assertIsInstance(file, File)
        self.assertEqual(file.content, b'hello world')

    @record
    def test_file_unicode_data(self):
        # Arrange
        file_name = self._get_file_reference()

        # Act
        data = u'hello world啊齄丂狛狜'.encode('utf-8')
        self.fs.create_file_from_bytes(self.share_name, None, file_name, data)

        # Assert
        file = self.fs.get_file_to_bytes(self.share_name, None, file_name)
        self.assertIsInstance(file, File)
        self.assertEqual(file.content, data)

    @record
    def test_unicode_get_file_binary_data(self):
        # Arrange
        base64_data = 'AAECAwQFBgcICQoLDA0ODxAREhMUFRYXGBkaGxwdHh8gISIjJCUmJygpKissLS4vMDEyMzQ1Njc4OTo7PD0+P0BBQkNERUZHSElKS0xNTk9QUVJTVFVWV1hZWltcXV5fYGFiY2RlZmdoaWprbG1ub3BxcnN0dXZ3eHl6e3x9fn+AgYKDhIWGh4iJiouMjY6PkJGSk5SVlpeYmZqbnJ2en6ChoqOkpaanqKmqq6ytrq+wsbKztLW2t7i5uru8vb6/wMHCw8TFxsfIycrLzM3Oz9DR0tPU1dbX2Nna29zd3t/g4eLj5OXm5+jp6uvs7e7v8PHy8/T19vf4+fr7/P3+/wABAgMEBQYHCAkKCwwNDg8QERITFBUWFxgZGhscHR4fICEiIyQlJicoKSorLC0uLzAxMjM0NTY3ODk6Ozw9Pj9AQUJDREVGR0hJSktMTU5PUFFSU1RVVldYWVpbXF1eX2BhYmNkZWZnaGlqa2xtbm9wcXJzdHV2d3h5ent8fX5/gIGCg4SFhoeIiYqLjI2Oj5CRkpOUlZaXmJmam5ydnp+goaKjpKWmp6ipqqusra6vsLGys7S1tre4ubq7vL2+v8DBwsPExcbHyMnKy8zNzs/Q0dLT1NXW19jZ2tvc3d7f4OHi4+Tl5ufo6err7O3u7/Dx8vP09fb3+Pn6+/z9/v8AAQIDBAUGBwgJCgsMDQ4PEBESExQVFhcYGRobHB0eHyAhIiMkJSYnKCkqKywtLi8wMTIzNDU2Nzg5Ojs8PT4/QEFCQ0RFRkdISUpLTE1OT1BRUlNUVVZXWFlaW1xdXl9gYWJjZGVmZ2hpamtsbW5vcHFyc3R1dnd4eXp7fH1+f4CBgoOEhYaHiImKi4yNjo+QkZKTlJWWl5iZmpucnZ6foKGio6SlpqeoqaqrrK2ur7CxsrO0tba3uLm6u7y9vr/AwcLDxMXGx8jJysvMzc7P0NHS09TV1tfY2drb3N3e3+Dh4uPk5ebn6Onq6+zt7u/w8fLz9PX29/j5+vv8/f7/AAECAwQFBgcICQoLDA0ODxAREhMUFRYXGBkaGxwdHh8gISIjJCUmJygpKissLS4vMDEyMzQ1Njc4OTo7PD0+P0BBQkNERUZHSElKS0xNTk9QUVJTVFVWV1hZWltcXV5fYGFiY2RlZmdoaWprbG1ub3BxcnN0dXZ3eHl6e3x9fn+AgYKDhIWGh4iJiouMjY6PkJGSk5SVlpeYmZqbnJ2en6ChoqOkpaanqKmqq6ytrq+wsbKztLW2t7i5uru8vb6/wMHCw8TFxsfIycrLzM3Oz9DR0tPU1dbX2Nna29zd3t/g4eLj5OXm5+jp6uvs7e7v8PHy8/T19vf4+fr7/P3+/w=='
        binary_data = base64.b64decode(base64_data)

        file_name = self._get_file_reference()
        self.fs.create_file_from_bytes(self.share_name, None, file_name, binary_data)

        # Act
        file = self.fs.get_file_to_bytes(self.share_name, None, file_name)

        # Assert
        self.assertIsInstance(file, File)
        self.assertEqual(file.content, binary_data)

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

        # Arrange
        file_name = self._get_file_reference()
        data = self.get_random_bytes(LARGE_FILE_SIZE)

        # Act
        progress = []

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

        self.fs.create_file_from_bytes(self.share_name, None, file_name, data, progress_callback=callback)

        # Assert
        self.assertFileEqual(self.share_name, None, file_name, data)

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

        # Arrange
        file_name = self._get_file_reference()
        data = self.get_random_bytes(LARGE_FILE_SIZE)
        index = 1024

        # Act
        self.fs.create_file_from_bytes(self.share_name, None, file_name, data, index)

        # Assert
        self.assertFileEqual(self.share_name, None, file_name, data[1024:])

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

        # Arrange
        file_name = self._get_file_reference()
        data = self.get_random_bytes(LARGE_FILE_SIZE)
        index = 512
        count = 1024

        # Act
        resp = self.fs.create_file_from_bytes(self.share_name, None, file_name, data, index, count)

        # Assert
        self.assertFileEqual(self.share_name, None, file_name, data[index:index + count])

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

        # Arrange        
        file_name = self._get_file_reference()
        data = self.get_random_bytes(LARGE_FILE_SIZE)
        with open(INPUT_FILE_PATH, 'wb') as stream:
            stream.write(data)

        # Act
        self.fs.create_file_from_path(self.share_name, None, file_name, INPUT_FILE_PATH)

        # Assert
        self.assertFileEqual(self.share_name, None, file_name, data)

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

        # Arrange        
        file_name = self._get_file_reference()
        data = self.get_random_bytes(LARGE_FILE_SIZE)
        with open(INPUT_FILE_PATH, 'wb') as stream:
            stream.write(data)

        # Act
        progress = []

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

        self.fs.create_file_from_path(self.share_name, None, file_name, INPUT_FILE_PATH,
                                      progress_callback=callback)

        # Assert
        self.assertFileEqual(self.share_name, None, file_name, data)
        self.assert_upload_progress(len(data), self.fs.MAX_RANGE_SIZE, progress, unknown_size=False)

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

        # Arrange       
        file_name = self._get_file_reference()
        data = self.get_random_bytes(LARGE_FILE_SIZE)
        with open(INPUT_FILE_PATH, 'wb') as stream:
            stream.write(data)

        # Act
        file_size = len(data)
        with open(INPUT_FILE_PATH, 'rb') as stream:
            self.fs.create_file_from_stream(self.share_name, None, file_name, stream, file_size)

        # Assert
        self.assertFileEqual(self.share_name, None, file_name, data[:file_size])

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

        # Arrange      
        file_name = self._get_file_reference()
        data = self.get_random_bytes(LARGE_FILE_SIZE)
        with open(INPUT_FILE_PATH, 'wb') as stream:
            stream.write(data)

        # Act
        file_size = len(data)
        with open(INPUT_FILE_PATH, 'rb') as stream:
            non_seekable_file = StorageFileTest.NonSeekableFile(stream)
            self.fs.create_file_from_stream(self.share_name, None, file_name,
                                            non_seekable_file, file_size, max_connections=1)

        # Assert
        self.assertFileEqual(self.share_name, None, file_name, data[:file_size])

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

        # Arrange      
        file_name = self._get_file_reference()
        data = self.get_random_bytes(LARGE_FILE_SIZE)
        with open(INPUT_FILE_PATH, 'wb') as stream:
            stream.write(data)

        # Act
        progress = []

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

        file_size = len(data)
        with open(INPUT_FILE_PATH, 'rb') as stream:
            self.fs.create_file_from_stream(self.share_name, None, file_name, stream,
                                            file_size, progress_callback=callback)

        # Assert
        self.assertFileEqual(self.share_name, None, file_name, data[:file_size])
        self.assert_upload_progress(len(data), self.fs.MAX_RANGE_SIZE, progress, unknown_size=False)

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

        # Arrange       
        file_name = self._get_file_reference()
        data = self.get_random_bytes(LARGE_FILE_SIZE)
        with open(INPUT_FILE_PATH, 'wb') as stream:
            stream.write(data)

        # Act
        file_size = len(data) - 512
        with open(INPUT_FILE_PATH, 'rb') as stream:
            self.fs.create_file_from_stream(self.share_name, None, file_name, stream, file_size)

        # Assert
        self.assertFileEqual(self.share_name, None, file_name, data[:file_size])

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

        # Arrange       
        file_name = self._get_file_reference()
        data = self.get_random_bytes(LARGE_FILE_SIZE)
        with open(INPUT_FILE_PATH, 'wb') as stream:
            stream.write(data)

        # Act
        progress = []

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

        file_size = len(data) - 5
        with open(INPUT_FILE_PATH, 'rb') as stream:
            self.fs.create_file_from_stream(self.share_name, None, file_name, stream,
                                            file_size, progress_callback=callback)

        # Assert
        self.assertFileEqual(self.share_name, None, file_name, data[:file_size])
        self.assert_upload_progress(file_size, self.fs.MAX_RANGE_SIZE, progress, unknown_size=False)

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

        # Act
        self.fs.create_file_from_text(self.share_name, None, file_name, text)

        # Assert
        self.assertFileEqual(self.share_name, None, file_name, data)

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

        # Act
        self.fs.create_file_from_text(self.share_name, None, file_name, text, 'utf-16')

        # Assert
        self.assertFileEqual(self.share_name, None, file_name, data)

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

        # Arrange
        file_name = self._get_file_reference()
        data = self.get_random_text_data(LARGE_FILE_SIZE)
        encoded_data = data.encode('utf-8')

        # Act
        self.fs.create_file_from_text(self.share_name, None, file_name, data)

        # Assert
        self.assertFileEqual(self.share_name, None, file_name, encoded_data)

    @record
    def test_create_file_with_md5_small(self):
        # Arrange
        file_name = self._get_file_reference()
        data = self.get_random_bytes(512)

        # Act
        self.fs.create_file_from_bytes(self.share_name, None, file_name, data,
                                       validate_content=True)

        # Assert

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

        # Arrange
        file_name = self._get_file_reference()
        data = self.get_random_bytes(LARGE_FILE_SIZE)

        # Act
        self.fs.create_file_from_bytes(self.share_name, None, file_name, data,
                                       validate_content=True)

        # Assert

    # --Test cases for sas & acl ------------------------------------------------
    @record
    def test_sas_access_file(self):
        # SAS URL is calculated from storage key, so this test runs live only
        if TestMode.need_recording_file(self.test_mode):
            return

        # Arrange
        file_name = self._create_file()

        token = self.fs.generate_file_shared_access_signature(
            self.share_name,
            None,
            file_name,
            permission=FilePermissions.READ,
            expiry=datetime.utcnow() + timedelta(hours=1),
        )

        # Act
        service = FileService(
            self.settings.STORAGE_ACCOUNT_NAME,
            sas_token=token,
            request_session=requests.Session(),
        )
        self._set_test_proxy(service, self.settings)
        result = service.get_file_to_bytes(self.share_name, None, file_name)

        # Assert
        self.assertEqual(self.short_byte_data, result.content)

    @record
    def test_sas_signed_identifier(self):
        # SAS URL is calculated from storage key, so this test runs live only
        if TestMode.need_recording_file(self.test_mode):
            return

        # Arrange
        file_name = self._create_file()

        access_policy = AccessPolicy()
        access_policy.start = datetime.utcnow() - timedelta(hours=1)
        access_policy.expiry = datetime.utcnow() + timedelta(hours=1)
        access_policy.permission = FilePermissions.READ
        identifiers = {'testid': access_policy}

        resp = self.fs.set_share_acl(self.share_name, identifiers)

        token = self.fs.generate_file_shared_access_signature(
            self.share_name,
            None,
            file_name,
            id='testid'
        )

        # Act
        service = FileService(
            self.settings.STORAGE_ACCOUNT_NAME,
            sas_token=token,
            request_session=requests.Session(),
        )
        self._set_test_proxy(service, self.settings)
        result = service.get_file_to_bytes(self.share_name, None, file_name)

        # Assert
        self.assertEqual(self.short_byte_data, result.content)

    @record
    def test_account_sas(self):
        # SAS URL is calculated from storage key, so this test runs live only
        if TestMode.need_recording_file(self.test_mode):
            return

        # Arrange
        file_name = self._create_file()

        token = self.fs.generate_account_shared_access_signature(
            ResourceTypes.OBJECT,
            AccountPermissions.READ,
            datetime.utcnow() + timedelta(hours=1),
        )

        # Act
        url = self.fs.make_file_url(
            self.share_name,
            None,
            file_name,
            sas_token=token,
        )
        response = requests.get(url)

        # Assert
        self.assertTrue(response.ok)
        self.assertEqual(self.short_byte_data, response.content)

    @record
    def test_shared_read_access_file(self):
        # SAS URL is calculated from storage key, so this test runs live only
        if TestMode.need_recording_file(self.test_mode):
            return

        # Arrange
        file_name = self._create_file()

        token = self.fs.generate_file_shared_access_signature(
            self.share_name,
            None,
            file_name,
            permission=FilePermissions.READ,
            expiry=datetime.utcnow() + timedelta(hours=1),
        )

        # Act
        url = self.fs.make_file_url(
            self.share_name,
            None,
            file_name,
            sas_token=token,
        )
        response = requests.get(url)

        # Assert
        self.assertTrue(response.ok)
        self.assertEqual(self.short_byte_data, response.content)

    @record
    def test_shared_read_access_file_with_content_query_params(self):
        # SAS URL is calculated from storage key, so this test runs live only
        if TestMode.need_recording_file(self.test_mode):
            return

        # Arrange
        file_name = self._create_file()

        token = self.fs.generate_file_shared_access_signature(
            self.share_name,
            None,
            file_name,
            permission=FilePermissions.READ,
            expiry=datetime.utcnow() + timedelta(hours=1),
            cache_control='no-cache',
            content_disposition='inline',
            content_encoding='utf-8',
            content_language='fr',
            content_type='text',
        )
        url = self.fs.make_file_url(
            self.share_name,
            None,
            file_name,
            sas_token=token,
        )

        # Act
        response = requests.get(url)

        # Assert
        self.assertEqual(self.short_byte_data, response.content)
        self.assertEqual(response.headers['cache-control'], 'no-cache')
        self.assertEqual(response.headers['content-disposition'], 'inline')
        self.assertEqual(response.headers['content-encoding'], 'utf-8')
        self.assertEqual(response.headers['content-language'], 'fr')
        self.assertEqual(response.headers['content-type'], 'text')

    @record
    def test_shared_write_access_file(self):
        # SAS URL is calculated from storage key, so this test runs live only
        if TestMode.need_recording_file(self.test_mode):
            return

        # Arrange
        updated_data = b'updated file data'
        file_name = self._create_file()

        token = self.fs.generate_file_shared_access_signature(
            self.share_name,
            None,
            file_name,
            permission=FilePermissions.WRITE,
            expiry=datetime.utcnow() + timedelta(hours=1),
        )
        url = self.fs.make_file_url(
            self.share_name,
            None,
            file_name,
            sas_token=token,
        )

        # Act
        headers = {'x-ms-range': 'bytes=0-16', 'x-ms-write': 'update'}
        response = requests.put(url + '&comp=range', headers=headers, data=updated_data)

        # Assert
        self.assertTrue(response.ok)
        file = self.fs.get_file_to_bytes(self.share_name, None, file_name)
        self.assertEqual(updated_data, file.content[:len(updated_data)])

    @record
    def test_shared_delete_access_file(self):
        # SAS URL is calculated from storage key, so this test runs live only
        if TestMode.need_recording_file(self.test_mode):
            return

        # Arrange
        file_name = self._create_file()

        token = self.fs.generate_file_shared_access_signature(
            self.share_name,
            None,
            file_name,
            permission=FilePermissions.DELETE,
            expiry=datetime.utcnow() + timedelta(hours=1),
        )
        url = self.fs.make_file_url(
            self.share_name,
            None,
            file_name,
            sas_token=token,
        )

        # Act
        response = requests.delete(url)

        # Assert
        self.assertTrue(response.ok)
        with self.assertRaises(AzureMissingResourceHttpError):
            file = self.fs.get_file_to_bytes(self.share_name, None, file_name)


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