# Copyright 2016 Amazon.com, Inc. or its affiliates. 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. A copy of
# the License is located at
#
# http://aws.amazon.com/apache2.0/
#
# or in the "license" file accompanying this file. This file 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.path
import shutil
import tempfile
import threading
import random
import time
import io

from tests import unittest
from tests import RecordingSubscriber
from tests import NonSeekableWriter
from s3transfer.compat import six
from s3transfer.futures import TransferFuture
from s3transfer.futures import TransferMeta
from s3transfer.utils import get_callbacks
from s3transfer.utils import random_file_extension
from s3transfer.utils import invoke_progress_callbacks
from s3transfer.utils import calculate_range_parameter
from s3transfer.utils import CallArgs
from s3transfer.utils import FunctionContainer
from s3transfer.utils import CountCallbackInvoker
from s3transfer.utils import OSUtils
from s3transfer.utils import DeferredOpenFile
from s3transfer.utils import ReadFileChunk
from s3transfer.utils import StreamReaderProgress
from s3transfer.utils import TaskSemaphore
from s3transfer.utils import SlidingWindowSemaphore
from s3transfer.utils import NoResourcesAvailable
from s3transfer.utils import ChunksizeAdjuster
from s3transfer.utils import MIN_UPLOAD_CHUNKSIZE, MAX_SINGLE_UPLOAD_SIZE
from s3transfer.utils import MAX_PARTS


class TestGetCallbacks(unittest.TestCase):
    def setUp(self):
        self.subscriber = RecordingSubscriber()
        self.second_subscriber = RecordingSubscriber()
        self.call_args = CallArgs(subscribers=[
            self.subscriber, self.second_subscriber]
        )
        self.transfer_meta = TransferMeta(self.call_args)
        self.transfer_future = TransferFuture(self.transfer_meta)

    def test_get_callbacks(self):
        callbacks = get_callbacks(self.transfer_future, 'queued')
        # Make sure two callbacks were added as both subscribers had
        # an on_queued method.
        self.assertEqual(len(callbacks), 2)

        # Ensure that the callback was injected with the future by calling
        # one of them and checking that the future was used in the call.
        callbacks[0]()
        self.assertEqual(
            self.subscriber.on_queued_calls,
            [{'future': self.transfer_future}]
        )

    def test_get_callbacks_for_missing_type(self):
        callbacks = get_callbacks(self.transfer_future, 'fake_state')
        # There should be no callbacks as the subscribers will not have the
        # on_fake_state method
        self.assertEqual(len(callbacks), 0)


class TestCallArgs(unittest.TestCase):
    def test_call_args(self):
        call_args = CallArgs(foo='bar', biz='baz')
        self.assertEqual(call_args.foo, 'bar')
        self.assertEqual(call_args.biz, 'baz')


class TestFunctionContainer(unittest.TestCase):
    def get_args_kwargs(self, *args, **kwargs):
        return args, kwargs

    def test_call(self):
        func_container = FunctionContainer(
            self.get_args_kwargs, 'foo', bar='baz')
        self.assertEqual(func_container(), (('foo',), {'bar': 'baz'}))

    def test_repr(self):
        func_container = FunctionContainer(
            self.get_args_kwargs, 'foo', bar='baz')
        self.assertEqual(
            str(func_container), 'Function: %s with args %s and kwargs %s' % (
                self.get_args_kwargs, ('foo',), {'bar': 'baz'}))


class TestCountCallbackInvoker(unittest.TestCase):
    def invoke_callback(self):
        self.ref_results.append('callback invoked')

    def assert_callback_invoked(self):
        self.assertEqual(self.ref_results, ['callback invoked'])

    def assert_callback_not_invoked(self):
        self.assertEqual(self.ref_results, [])

    def setUp(self):
        self.ref_results = []
        self.invoker = CountCallbackInvoker(self.invoke_callback)

    def test_increment(self):
        self.invoker.increment()
        self.assertEqual(self.invoker.current_count, 1)

    def test_decrement(self):
        self.invoker.increment()
        self.invoker.increment()
        self.invoker.decrement()
        self.assertEqual(self.invoker.current_count, 1)

    def test_count_cannot_go_below_zero(self):
        with self.assertRaises(RuntimeError):
            self.invoker.decrement()

    def test_callback_invoked_only_once_finalized(self):
        self.invoker.increment()
        self.invoker.decrement()
        self.assert_callback_not_invoked()
        self.invoker.finalize()
        # Callback should only be invoked once finalized
        self.assert_callback_invoked()

    def test_callback_invoked_after_finalizing_and_count_reaching_zero(self):
        self.invoker.increment()
        self.invoker.finalize()
        # Make sure that it does not get invoked immediately after
        # finalizing as the count is currently one
        self.assert_callback_not_invoked()
        self.invoker.decrement()
        self.assert_callback_invoked()

    def test_cannot_increment_after_finalization(self):
        self.invoker.finalize()
        with self.assertRaises(RuntimeError):
            self.invoker.increment()


class TestRandomFileExtension(unittest.TestCase):
    def test_has_proper_length(self):
        self.assertEqual(
            len(random_file_extension(num_digits=4)), 4)


class TestInvokeProgressCallbacks(unittest.TestCase):
    def test_invoke_progress_callbacks(self):
        recording_subscriber = RecordingSubscriber()
        invoke_progress_callbacks([recording_subscriber.on_progress], 2)
        self.assertEqual(recording_subscriber.calculate_bytes_seen(), 2)

    def test_invoke_progress_callbacks_with_no_progress(self):
        recording_subscriber = RecordingSubscriber()
        invoke_progress_callbacks([recording_subscriber.on_progress], 0)
        self.assertEqual(len(recording_subscriber.on_progress_calls), 0)


class TestCalculateRangeParameter(unittest.TestCase):
    def setUp(self):
        self.part_size = 5
        self.part_index = 1
        self.num_parts = 3

    def test_calculate_range_paramter(self):
        range_val = calculate_range_parameter(
            self.part_size, self.part_index, self.num_parts)
        self.assertEqual(range_val, 'bytes=5-9')

    def test_last_part_with_no_total_size(self):
        range_val = calculate_range_parameter(
            self.part_size, self.part_index, num_parts=2)
        self.assertEqual(range_val, 'bytes=5-')

    def test_last_part_with_total_size(self):
        range_val = calculate_range_parameter(
            self.part_size, self.part_index, num_parts=2, total_size=8)
        self.assertEqual(range_val, 'bytes=5-7')


class BaseUtilsTest(unittest.TestCase):
    def setUp(self):
        self.tempdir = tempfile.mkdtemp()
        self.filename = os.path.join(self.tempdir, 'foo')
        self.content = b'abc'
        with open(self.filename, 'wb') as f:
            f.write(self.content)
        self.amounts_seen = []
        self.num_close_callback_calls = 0

    def tearDown(self):
        shutil.rmtree(self.tempdir)

    def callback(self, bytes_transferred):
        self.amounts_seen.append(bytes_transferred)

    def close_callback(self):
        self.num_close_callback_calls += 1


class TestOSUtils(BaseUtilsTest):
    def test_get_file_size(self):
        self.assertEqual(
            OSUtils().get_file_size(self.filename), len(self.content))

    def test_open_file_chunk_reader(self):
        reader = OSUtils().open_file_chunk_reader(
            self.filename, 0, 3, [self.callback])

        # The returned reader should be a ReadFileChunk.
        self.assertIsInstance(reader, ReadFileChunk)
        # The content of the reader should be correct.
        self.assertEqual(reader.read(), self.content)
        # Callbacks should be disabled depspite being passed in.
        self.assertEqual(self.amounts_seen, [])

    def test_open_file_chunk_reader_from_fileobj(self):
        with open(self.filename, 'rb') as f:
            reader = OSUtils().open_file_chunk_reader_from_fileobj(
                f, len(self.content), len(self.content), [self.callback])

            # The returned reader should be a ReadFileChunk.
            self.assertIsInstance(reader, ReadFileChunk)
            # The content of the reader should be correct.
            self.assertEqual(reader.read(), self.content)
            reader.close()
            # Callbacks should be disabled depspite being passed in.
            self.assertEqual(self.amounts_seen, [])
            self.assertEqual(self.num_close_callback_calls, 0)

    def test_open_file(self):
        fileobj = OSUtils().open(os.path.join(self.tempdir, 'foo'), 'w')
        self.assertTrue(hasattr(fileobj, 'write'))

    def test_remove_file_ignores_errors(self):
        non_existent_file = os.path.join(self.tempdir, 'no-exist')
        # This should not exist to start.
        self.assertFalse(os.path.exists(non_existent_file))
        try:
            OSUtils().remove_file(non_existent_file)
        except OSError as e:
            self.fail('OSError should have been caught: %s' % e)

    def test_remove_file_proxies_remove_file(self):
        OSUtils().remove_file(self.filename)
        self.assertFalse(os.path.exists(self.filename))

    def test_rename_file(self):
        new_filename = os.path.join(self.tempdir, 'newfoo')
        OSUtils().rename_file(self.filename, new_filename)
        self.assertFalse(os.path.exists(self.filename))
        self.assertTrue(os.path.exists(new_filename))

    def test_is_special_file_for_normal_file(self):
        self.assertFalse(OSUtils().is_special_file(self.filename))

    def test_is_special_file_for_non_existant_file(self):
        non_existant_filename = os.path.join(self.tempdir, 'no-exist')
        self.assertFalse(os.path.exists(non_existant_filename))
        self.assertFalse(OSUtils().is_special_file(non_existant_filename))


class TestDeferredOpenFile(BaseUtilsTest):
    def setUp(self):
        super(TestDeferredOpenFile, self).setUp()
        self.filename = os.path.join(self.tempdir, 'foo')
        self.contents = b'my contents'
        with open(self.filename, 'wb') as f:
            f.write(self.contents)
        self.deferred_open_file = DeferredOpenFile(
            self.filename, open_function=self.recording_open_function)
        self.open_call_args = []

    def tearDown(self):
        self.deferred_open_file.close()
        super(TestDeferredOpenFile, self).tearDown()

    def recording_open_function(self, filename, mode):
        self.open_call_args.append((filename, mode))
        return open(filename, mode)

    def open_nonseekable(self, filename, mode):
        self.open_call_args.append((filename, mode))
        return NonSeekableWriter(six.BytesIO(self.content))

    def test_instantiation_does_not_open_file(self):
        DeferredOpenFile(
            self.filename, open_function=self.recording_open_function)
        self.assertEqual(len(self.open_call_args), 0)

    def test_name(self):
        self.assertEqual(self.deferred_open_file.name, self.filename)

    def test_read(self):
        content = self.deferred_open_file.read(2)
        self.assertEqual(content, self.contents[0:2])
        content = self.deferred_open_file.read(2)
        self.assertEqual(content, self.contents[2:4])
        self.assertEqual(len(self.open_call_args), 1)

    def test_write(self):
        self.deferred_open_file = DeferredOpenFile(
            self.filename, mode='wb',
            open_function=self.recording_open_function)

        write_content = b'foo'
        self.deferred_open_file.write(write_content)
        self.deferred_open_file.write(write_content)
        self.deferred_open_file.close()
        # Both of the writes should now be in the file.
        with open(self.filename, 'rb') as f:
            self.assertEqual(f.read(), write_content*2)
        # Open should have only been called once.
        self.assertEqual(len(self.open_call_args), 1)

    def test_seek(self):
        self.deferred_open_file.seek(2)
        content = self.deferred_open_file.read(2)
        self.assertEqual(content, self.contents[2:4])
        self.assertEqual(len(self.open_call_args), 1)

    def test_open_does_not_seek_with_zero_start_byte(self):
        self.deferred_open_file = DeferredOpenFile(
            self.filename, mode='wb', start_byte=0,
            open_function=self.open_nonseekable)

        try:
            # If this seeks, an UnsupportedOperation error will be raised.
            self.deferred_open_file.write(b'data')
        except io.UnsupportedOperation:
            self.fail('DeferredOpenFile seeked upon opening')

    def test_open_seeks_with_nonzero_start_byte(self):
        self.deferred_open_file = DeferredOpenFile(
            self.filename, mode='wb', start_byte=5,
            open_function=self.open_nonseekable)

        # Since a non-seekable file is being opened, calling Seek will raise
        # an UnsupportedOperation error.
        with self.assertRaises(io.UnsupportedOperation):
            self.deferred_open_file.write(b'data')

    def test_tell(self):
        self.deferred_open_file.tell()
        # tell() should not have opened the file if it has not been seeked
        # or read because we know the start bytes upfront.
        self.assertEqual(len(self.open_call_args), 0)

        self.deferred_open_file.seek(2)
        self.assertEqual(self.deferred_open_file.tell(), 2)
        self.assertEqual(len(self.open_call_args), 1)

    def test_open_args(self):
        self.deferred_open_file = DeferredOpenFile(
            self.filename, mode='ab+',
            open_function=self.recording_open_function)
        # Force an open
        self.deferred_open_file.write(b'data')
        self.assertEqual(len(self.open_call_args), 1)
        self.assertEqual(self.open_call_args[0], (self.filename, 'ab+'))

    def test_context_handler(self):
        with self.deferred_open_file:
            self.assertEqual(len(self.open_call_args), 1)


class TestReadFileChunk(BaseUtilsTest):
    def test_read_entire_chunk(self):
        filename = os.path.join(self.tempdir, 'foo')
        with open(filename, 'wb') as f:
            f.write(b'onetwothreefourfivesixseveneightnineten')
        chunk = ReadFileChunk.from_filename(
            filename, start_byte=0, chunk_size=3)
        self.assertEqual(chunk.read(), b'one')
        self.assertEqual(chunk.read(), b'')

    def test_read_with_amount_size(self):
        filename = os.path.join(self.tempdir, 'foo')
        with open(filename, 'wb') as f:
            f.write(b'onetwothreefourfivesixseveneightnineten')
        chunk = ReadFileChunk.from_filename(
            filename, start_byte=11, chunk_size=4)
        self.assertEqual(chunk.read(1), b'f')
        self.assertEqual(chunk.read(1), b'o')
        self.assertEqual(chunk.read(1), b'u')
        self.assertEqual(chunk.read(1), b'r')
        self.assertEqual(chunk.read(1), b'')

    def test_reset_stream_emulation(self):
        filename = os.path.join(self.tempdir, 'foo')
        with open(filename, 'wb') as f:
            f.write(b'onetwothreefourfivesixseveneightnineten')
        chunk = ReadFileChunk.from_filename(
            filename, start_byte=11, chunk_size=4)
        self.assertEqual(chunk.read(), b'four')
        chunk.seek(0)
        self.assertEqual(chunk.read(), b'four')

    def test_read_past_end_of_file(self):
        filename = os.path.join(self.tempdir, 'foo')
        with open(filename, 'wb') as f:
            f.write(b'onetwothreefourfivesixseveneightnineten')
        chunk = ReadFileChunk.from_filename(
            filename, start_byte=36, chunk_size=100000)
        self.assertEqual(chunk.read(), b'ten')
        self.assertEqual(chunk.read(), b'')
        self.assertEqual(len(chunk), 3)

    def test_tell_and_seek(self):
        filename = os.path.join(self.tempdir, 'foo')
        with open(filename, 'wb') as f:
            f.write(b'onetwothreefourfivesixseveneightnineten')
        chunk = ReadFileChunk.from_filename(
            filename, start_byte=36, chunk_size=100000)
        self.assertEqual(chunk.tell(), 0)
        self.assertEqual(chunk.read(), b'ten')
        self.assertEqual(chunk.tell(), 3)
        chunk.seek(0)
        self.assertEqual(chunk.tell(), 0)

    def test_file_chunk_supports_context_manager(self):
        filename = os.path.join(self.tempdir, 'foo')
        with open(filename, 'wb') as f:
            f.write(b'abc')
        with ReadFileChunk.from_filename(filename,
                                         start_byte=0,
                                         chunk_size=2) as chunk:
            val = chunk.read()
            self.assertEqual(val, b'ab')

    def test_iter_is_always_empty(self):
        # This tests the workaround for the httplib bug (see
        # the source for more info).
        filename = os.path.join(self.tempdir, 'foo')
        open(filename, 'wb').close()
        chunk = ReadFileChunk.from_filename(
            filename, start_byte=0, chunk_size=10)
        self.assertEqual(list(chunk), [])

    def test_callback_is_invoked_on_read(self):
        chunk = ReadFileChunk.from_filename(
            self.filename, start_byte=0, chunk_size=3,
            callbacks=[self.callback])
        chunk.read(1)
        chunk.read(1)
        chunk.read(1)
        self.assertEqual(self.amounts_seen, [1, 1, 1])

    def test_all_callbacks_invoked_on_read(self):
        chunk = ReadFileChunk.from_filename(
            self.filename, start_byte=0, chunk_size=3,
            callbacks=[self.callback, self.callback])
        chunk.read(1)
        chunk.read(1)
        chunk.read(1)
        # The list should be twice as long because there are two callbacks
        # recording the amount read.
        self.assertEqual(self.amounts_seen, [1, 1, 1, 1, 1, 1])

    def test_callback_can_be_disabled(self):
        chunk = ReadFileChunk.from_filename(
            self.filename, start_byte=0, chunk_size=3,
            callbacks=[self.callback])
        chunk.disable_callback()
        # Now reading from the ReadFileChunk should not invoke
        # the callback.
        chunk.read()
        self.assertEqual(self.amounts_seen, [])

    def test_callback_will_also_be_triggered_by_seek(self):
        chunk = ReadFileChunk.from_filename(
            self.filename, start_byte=0, chunk_size=3,
            callbacks=[self.callback])
        chunk.read(2)
        chunk.seek(0)
        chunk.read(2)
        chunk.seek(1)
        chunk.read(2)
        self.assertEqual(self.amounts_seen, [2, -2, 2, -1, 2])

    def test_close_callbacks(self):
        with open(self.filename) as f:
            chunk = ReadFileChunk(f, chunk_size=1, full_file_size=3,
                                  close_callbacks=[self.close_callback])
            chunk.close()
            self.assertEqual(self.num_close_callback_calls, 1)

    def test_close_callbacks_when_not_enabled(self):
        with open(self.filename) as f:
            chunk = ReadFileChunk(f, chunk_size=1, full_file_size=3,
                                  enable_callbacks=False,
                                  close_callbacks=[self.close_callback])
            chunk.close()
            self.assertEqual(self.num_close_callback_calls, 0)

    def test_close_callbacks_when_context_handler_is_used(self):
        with open(self.filename) as f:
            with ReadFileChunk(f, chunk_size=1, full_file_size=3,
                               close_callbacks=[self.close_callback]) as chunk:
                chunk.read(1)
            self.assertEqual(self.num_close_callback_calls, 1)


class TestStreamReaderProgress(BaseUtilsTest):
    def test_proxies_to_wrapped_stream(self):
        original_stream = six.StringIO('foobarbaz')
        wrapped = StreamReaderProgress(original_stream)
        self.assertEqual(wrapped.read(), 'foobarbaz')

    def test_callback_invoked(self):
        original_stream = six.StringIO('foobarbaz')
        wrapped = StreamReaderProgress(
            original_stream, [self.callback, self.callback])
        self.assertEqual(wrapped.read(), 'foobarbaz')
        self.assertEqual(self.amounts_seen, [9, 9])


class TestTaskSemaphore(unittest.TestCase):
    def setUp(self):
        self.semaphore = TaskSemaphore(1)

    def test_should_block_at_max_capacity(self):
        self.semaphore.acquire('a', blocking=False)
        with self.assertRaises(NoResourcesAvailable):
            self.semaphore.acquire('a', blocking=False)

    def test_release_capacity(self):
        acquire_token = self.semaphore.acquire('a', blocking=False)
        self.semaphore.release('a', acquire_token)
        try:
            self.semaphore.acquire('a', blocking=False)
        except NoResourcesAvailable:
            self.fail(
                'The release of the semaphore should have allowed for '
                'the second acquire to not be blocked'
            )


class TestSlidingWindowSemaphore(unittest.TestCase):
    # These tests use block=False to tests will fail
    # instead of hang the test runner in the case of x
    # incorrect behavior.
    def test_acquire_release_basic_case(self):
        sem = SlidingWindowSemaphore(1)
        # Count is 1

        num = sem.acquire('a', blocking=False)
        self.assertEqual(num, 0)
        sem.release('a', 0)
        # Count now back to 1.

    def test_can_acquire_release_multiple_times(self):
        sem = SlidingWindowSemaphore(1)
        num = sem.acquire('a', blocking=False)
        self.assertEqual(num, 0)
        sem.release('a', num)

        num = sem.acquire('a', blocking=False)
        self.assertEqual(num, 1)
        sem.release('a', num)

    def test_can_acquire_a_range(self):
        sem = SlidingWindowSemaphore(3)
        self.assertEqual(sem.acquire('a', blocking=False), 0)
        self.assertEqual(sem.acquire('a', blocking=False), 1)
        self.assertEqual(sem.acquire('a', blocking=False), 2)
        sem.release('a', 0)
        sem.release('a', 1)
        sem.release('a', 2)
        # Now we're reset so we should be able to acquire the same
        # sequence again.
        self.assertEqual(sem.acquire('a', blocking=False), 3)
        self.assertEqual(sem.acquire('a', blocking=False), 4)
        self.assertEqual(sem.acquire('a', blocking=False), 5)
        self.assertEqual(sem.current_count(), 0)

    def test_counter_release_only_on_min_element(self):
        sem = SlidingWindowSemaphore(3)
        sem.acquire('a', blocking=False)
        sem.acquire('a', blocking=False)
        sem.acquire('a', blocking=False)

        # The count only increases when we free the min
        # element.  This means if we're currently failing to
        # acquire now:
        with self.assertRaises(NoResourcesAvailable):
            sem.acquire('a', blocking=False)

        # Then freeing a non-min element:
        sem.release('a', 1)

        # doesn't change anything.  We still fail to acquire.
        with self.assertRaises(NoResourcesAvailable):
            sem.acquire('a', blocking=False)
        self.assertEqual(sem.current_count(), 0)

    def test_raises_error_when_count_is_zero(self):
        sem = SlidingWindowSemaphore(3)
        sem.acquire('a', blocking=False)
        sem.acquire('a', blocking=False)
        sem.acquire('a', blocking=False)

        # Count is now 0 so trying to acquire should fail.
        with self.assertRaises(NoResourcesAvailable):
            sem.acquire('a', blocking=False)

    def test_release_counters_can_increment_counter_repeatedly(self):
        sem = SlidingWindowSemaphore(3)
        sem.acquire('a', blocking=False)
        sem.acquire('a', blocking=False)
        sem.acquire('a', blocking=False)

        # These two releases don't increment the counter
        # because we're waiting on 0.
        sem.release('a', 1)
        sem.release('a', 2)
        self.assertEqual(sem.current_count(), 0)
        # But as soon as we release 0, we free up 0, 1, and 2.
        sem.release('a', 0)
        self.assertEqual(sem.current_count(), 3)
        sem.acquire('a', blocking=False)
        sem.acquire('a', blocking=False)
        sem.acquire('a', blocking=False)

    def test_error_to_release_unknown_tag(self):
        sem = SlidingWindowSemaphore(3)
        with self.assertRaises(ValueError):
            sem.release('a', 0)

    def test_can_track_multiple_tags(self):
        sem = SlidingWindowSemaphore(3)
        self.assertEqual(sem.acquire('a', blocking=False), 0)
        self.assertEqual(sem.acquire('b', blocking=False), 0)
        self.assertEqual(sem.acquire('a', blocking=False), 1)

        # We're at our max of 3 even though 2 are for A and 1 is for B.
        with self.assertRaises(NoResourcesAvailable):
            sem.acquire('a', blocking=False)
        with self.assertRaises(NoResourcesAvailable):
            sem.acquire('b', blocking=False)

    def test_can_handle_multiple_tags_released(self):
        sem = SlidingWindowSemaphore(4)
        sem.acquire('a', blocking=False)
        sem.acquire('a', blocking=False)
        sem.acquire('b', blocking=False)
        sem.acquire('b', blocking=False)

        sem.release('b', 1)
        sem.release('a', 1)
        self.assertEqual(sem.current_count(), 0)

        sem.release('b', 0)
        self.assertEqual(sem.acquire('a', blocking=False), 2)

        sem.release('a', 0)
        self.assertEqual(sem.acquire('b', blocking=False), 2)

    def test_is_error_to_release_unknown_sequence_number(self):
        sem = SlidingWindowSemaphore(3)
        sem.acquire('a', blocking=False)
        with self.assertRaises(ValueError):
            sem.release('a', 1)

    def test_is_error_to_double_release(self):
        # This is different than other error tests because
        # we're verifying we can reset the state after an
        # acquire/release cycle.
        sem = SlidingWindowSemaphore(2)
        sem.acquire('a', blocking=False)
        sem.acquire('a', blocking=False)
        sem.release('a', 0)
        sem.release('a', 1)
        self.assertEqual(sem.current_count(), 2)
        with self.assertRaises(ValueError):
            sem.release('a', 0)

    def test_can_check_in_partial_range(self):
        sem = SlidingWindowSemaphore(4)
        sem.acquire('a', blocking=False)
        sem.acquire('a', blocking=False)
        sem.acquire('a', blocking=False)
        sem.acquire('a', blocking=False)

        sem.release('a', 1)
        sem.release('a', 3)
        sem.release('a', 0)
        self.assertEqual(sem.current_count(), 2)


class TestThreadingPropertiesForSlidingWindowSemaphore(unittest.TestCase):
    # These tests focus on mutithreaded properties of the range
    # semaphore.  Basic functionality is tested in TestSlidingWindowSemaphore.
    def setUp(self):
        self.threads = []

    def tearDown(self):
        self.join_threads()

    def join_threads(self):
        for thread in self.threads:
            thread.join()
        self.threads = []

    def start_threads(self):
        for thread in self.threads:
            thread.start()

    def test_acquire_blocks_until_release_is_called(self):
        sem = SlidingWindowSemaphore(2)
        sem.acquire('a', blocking=False)
        sem.acquire('a', blocking=False)

        def acquire():
            # This next call to acquire will block.
            self.assertEqual(sem.acquire('a', blocking=True), 2)

        t = threading.Thread(target=acquire)
        self.threads.append(t)
        # Starting the thread will block the sem.acquire()
        # in the acquire function above.
        t.start()
        # This still will keep the thread blocked.
        sem.release('a', 1)
        # Releasing the min element will unblock the thread.
        sem.release('a', 0)
        t.join()
        sem.release('a', 2)

    def test_stress_invariants_random_order(self):
        sem = SlidingWindowSemaphore(100)
        for _ in range(10):
            recorded = []
            for _ in range(100):
                recorded.append(sem.acquire('a', blocking=False))
            # Release them in randomized order.  As long as we
            # eventually free all 100, we should have all the
            # resources released.
            random.shuffle(recorded)
            for i in recorded:
                sem.release('a', i)

        # Everything's freed so should be back at count == 100
        self.assertEqual(sem.current_count(), 100)

    def test_blocking_stress(self):
        sem = SlidingWindowSemaphore(5)
        num_threads = 10
        num_iterations = 50

        def acquire():
            for _ in range(num_iterations):
                num = sem.acquire('a', blocking=True)
                time.sleep(0.001)
                sem.release('a', num)

        for i in range(num_threads):
            t = threading.Thread(target=acquire)
            self.threads.append(t)
        self.start_threads()
        self.join_threads()
        # Should have all the available resources freed.
        self.assertEqual(sem.current_count(), 5)
        # Should have acquired num_threads * num_iterations
        self.assertEqual(sem.acquire('a', blocking=False),
                         num_threads * num_iterations)


class TestAdjustChunksize(unittest.TestCase):
    def setUp(self):
        self.adjuster = ChunksizeAdjuster()

    def test_valid_chunksize(self):
        chunksize = 7 * (1024 ** 2)
        file_size = 8 * (1024 ** 2)
        new_size = self.adjuster.adjust_chunksize(chunksize, file_size)
        self.assertEqual(new_size, chunksize)

    def test_chunksize_below_minimum(self):
        chunksize = MIN_UPLOAD_CHUNKSIZE - 1
        file_size = 3 * MIN_UPLOAD_CHUNKSIZE
        new_size = self.adjuster.adjust_chunksize(chunksize, file_size)
        self.assertEqual(new_size, MIN_UPLOAD_CHUNKSIZE)

    def test_chunksize_above_maximum(self):
        chunksize = MAX_SINGLE_UPLOAD_SIZE + 1
        file_size = MAX_SINGLE_UPLOAD_SIZE * 2
        new_size = self.adjuster.adjust_chunksize(chunksize, file_size)
        self.assertEqual(new_size, MAX_SINGLE_UPLOAD_SIZE)

    def test_chunksize_too_small(self):
        chunksize = 7 * (1024 ** 2)
        file_size = 5 * (1024 ** 4)
        # If we try to upload a 5TB file, we'll need to use 896MB part
        # sizes.
        new_size = self.adjuster.adjust_chunksize(chunksize, file_size)
        self.assertEqual(new_size, 896 * (1024 ** 2))
        num_parts = file_size / new_size
        self.assertLessEqual(num_parts, MAX_PARTS)

    def test_unknown_file_size_with_valid_chunksize(self):
        chunksize = 7 * (1024 ** 2)
        new_size = self.adjuster.adjust_chunksize(chunksize)
        self.assertEqual(new_size, chunksize)

    def test_unknown_file_size_below_minimum(self):
        chunksize = MIN_UPLOAD_CHUNKSIZE - 1
        new_size = self.adjuster.adjust_chunksize(chunksize)
        self.assertEqual(new_size, MIN_UPLOAD_CHUNKSIZE)

    def test_unknown_file_size_above_maximum(self):
        chunksize = MAX_SINGLE_UPLOAD_SIZE + 1
        new_size = self.adjuster.adjust_chunksize(chunksize)
        self.assertEqual(new_size, MAX_SINGLE_UPLOAD_SIZE)