# -*- coding: utf-8 -*-
#
# Copyright (C) 2019 Radim Rehurek <me@radimrehurek.com>
#
# This code is distributed under the terms and conditions
# from the MIT License (MIT).
#
from collections import defaultdict
import functools
import random
import gzip
import io
import re
import logging
import os
import tempfile
import unittest
import warnings
from contextlib import contextmanager
from unittest import mock
import sys

import boto3
import botocore.client
import botocore.endpoint
import botocore.exceptions
import pytest

# See https://github.com/piskvorky/smart_open/issues/800
# This supports moto 4 & 5 until v4 is no longer used by distros.
try:
    from moto import mock_s3
except ImportError:
    from moto import mock_aws as mock_s3

import smart_open
import smart_open.s3

# To reduce spurious errors due to S3's eventually-consistent behavior
# we create this bucket once before running these tests and then
# remove it when we're done.  The bucket has a random name so that we
# can run multiple instances of this suite in parallel and not have
# them conflict with one another. Travis, for example, runs the Python
# 2.7, 3.6, and 3.7 suites concurrently.
BUCKET_NAME = 'test-smartopen'
KEY_NAME = 'test-key'
WRITE_KEY_NAME = 'test-write-key'
ENABLE_MOTO_SERVER = os.environ.get("SO_ENABLE_MOTO_SERVER") == "1"

#
# This is a hack to keep moto happy
# See https://github.com/spulec/moto/issues/1941
#
os.environ["AWS_ACCESS_KEY_ID"] = "test"
os.environ["AWS_SECRET_ACCESS_KEY"] = "test"

logger = logging.getLogger(__name__)

_resource = functools.partial(boto3.resource, region_name='us-east-1')


def ignore_resource_warnings():
    #
    # https://github.com/boto/boto3/issues/454
    # Py2 doesn't have ResourceWarning, so do nothing.
    #
    warnings.filterwarnings("ignore", category=ResourceWarning, message="unclosed.*<ssl.SSLSocket.*>")  # noqa


@contextmanager
def patch_invalid_range_response(actual_size):
    """ Work around a bug in moto (https://github.com/spulec/moto/issues/2981) where the
     API response doesn't match when requesting an invalid range of bytes from an S3 GetObject. """
    _real_get = smart_open.s3._get

    def mock_get(*args, **kwargs):
        try:
            return _real_get(*args, **kwargs)
        except IOError as ioe:
            error_response = smart_open.s3._unwrap_ioerror(ioe)
            if error_response and error_response.get('Message') == 'Requested Range Not Satisfiable':
                error_response['ActualObjectSize'] = actual_size
                error_response['Code'] = 'InvalidRange'
                error_response['Message'] = 'The requested range is not satisfiable'
            if actual_size is None:
                error_response.pop('ActualObjectSize', None)
            raise

    with mock.patch('smart_open.s3._get', new=mock_get):
        yield


class BaseTest(unittest.TestCase):
    @contextmanager
    def assertApiCalls(self, **expected_api_calls):
        """ Track calls to S3 in self.api_calls by patching botocore.endpoint.Endpoint.make_request. """
        _real_make_request = botocore.endpoint.Endpoint.make_request
        api_calls = defaultdict(int)

        def mock_make_request(self, operation_model, *args, **kwargs):
            api_calls[operation_model.name] += 1
            return _real_make_request(self, operation_model, *args, **kwargs)

        patcher = mock.patch('botocore.endpoint.Endpoint.make_request', new=mock_make_request)
        patcher.start()
        try:
            yield api_calls
            self.assertDictEqual(expected_api_calls, api_calls)
        finally:
            patcher.stop()


@unittest.skipUnless(
    ENABLE_MOTO_SERVER,
    'The test case needs a Moto server running on the local 5000 port.'
)
class SeekableRawReaderTest(unittest.TestCase):
    def setUp(self):
        self._body = b'123456'
        self._local_resource = boto3.resource('s3', endpoint_url='http://localhost:5000')
        self._local_resource.Bucket(BUCKET_NAME).create()
        self._local_resource.Object(BUCKET_NAME, KEY_NAME).put(Body=self._body)
        self._local_client = boto3.client('s3', endpoint_url='http://localhost:5000')

    def tearDown(self):
        self._local_resource.Object(BUCKET_NAME, KEY_NAME).delete()
        self._local_resource.Bucket(BUCKET_NAME).delete()

    def test_read_from_a_closed_body(self):
        reader = smart_open.s3._SeekableRawReader(self._local_client, BUCKET_NAME, KEY_NAME)
        self.assertEqual(reader.read(1), b'1')
        reader._body.close()
        self.assertEqual(reader.read(2), b'23')


class CrapStream(io.BytesIO):
    """Raises an exception on every second read call."""
    def __init__(self, *args, modulus=2, **kwargs):
        super().__init__(*args, **kwargs)
        self._count = 0
        self._modulus = modulus

    def read(self, size=-1):
        self._count += 1
        if self._count % self._modulus == 0:
            raise botocore.exceptions.BotoCoreError()
        the_bytes = super().read(size)
        return the_bytes

    def close(self):
        # we slightly bogusly return the same stream object from multiple
        # mock "requests" so we need to avoid .close from the first request
        # preventing us from using the stream in the second request.
        pass


class CrapClient:
    def __init__(self, data, modulus=2):
        self._datasize = len(data)
        self._body = CrapStream(data, modulus=modulus)

    def get_object(self, *args, Range=None, **kwargs):
        start = 0
        end = self._datasize
        if Range is not None:
            m = re.fullmatch(r'bytes=(\d+)?-(\d+)?', Range)
            assert m, Range

            if m.group(2) is not None and m.group(1) is None:
                start = max(0, self._datasize - int(m.group(2)))
                end = self._datasize
            else:
                if m.group(1) is not None:
                    start = min(int(m.group(1)), self._datasize)
                else:
                    start = 0

                if m.group(2) is not None:
                    end = min(int(m.group(2)) + 1, self._datasize)
                else:
                    end = self._datasize

        self._body.seek(start)
        return {
            'ActualObjectSize': self._datasize,
            'ContentLength': self._datasize,
            'ContentRange': 'bytes %d-%d/%d' % (start, end - 1, self._datasize),
            'Body': self._body,
            'ResponseMetadata': {'RetryAttempts': 1, 'HTTPStatusCode': 206},
        }


class IncrementalBackoffTest(unittest.TestCase):
    def test_every_read_fails(self):
        reader = smart_open.s3._SeekableRawReader(CrapClient(b'hello', 1), 'bucket', 'key')
        with mock.patch('time.sleep') as mock_sleep:
            with self.assertRaises(IOError):
                reader.read()

            #
            # Make sure our incremental backoff is actually happening here.
            #
            mock_sleep.assert_has_calls([mock.call(s) for s in (1, 2, 4, 8, 16)])

    def test_every_second_read_fails(self):
        """Can we read from a stream that raises exceptions from time to time?"""
        reader = smart_open.s3._SeekableRawReader(CrapClient(b'hello'), 'bucket', 'key')
        with mock.patch('time.sleep') as mock_sleep:
            assert reader.read(1) == b'h'
            mock_sleep.assert_not_called()

            assert reader.read(1) == b'e'
            mock_sleep.assert_called_with(1)
            mock_sleep.reset_mock()

            assert reader.read(1) == b'l'
            mock_sleep.reset_mock()

            assert reader.read(1) == b'l'
            mock_sleep.assert_called_with(1)
            mock_sleep.reset_mock()

            assert reader.read(1) == b'o'
            mock_sleep.assert_called_with(1)
            mock_sleep.reset_mock()


@mock_s3
class ReaderTest(BaseTest):
    def setUp(self):
        # lower the multipart upload size, to speed up these tests
        self.old_min_part_size = smart_open.s3.DEFAULT_PART_SIZE
        smart_open.s3.DEFAULT_PART_SIZE = 5 * 1024**2

        ignore_resource_warnings()

        super().setUp()

        s3 = _resource('s3')
        s3.create_bucket(Bucket=BUCKET_NAME).wait_until_exists()

        self.body = u"hello wořld\nhow are you?".encode('utf8')
        s3.Object(BUCKET_NAME, KEY_NAME).put(Body=self.body)

    def tearDown(self):
        smart_open.s3.DEFAULT_PART_SIZE = self.old_min_part_size

    def test_iter(self):
        """Are S3 files iterated over correctly?"""
        # connect to fake s3 and read from the fake key we filled above
        with self.assertApiCalls(GetObject=1):
            fin = smart_open.s3.Reader(BUCKET_NAME, KEY_NAME)
            output = [line.rstrip(b'\n') for line in fin]
        self.assertEqual(output, self.body.split(b'\n'))

    def test_iter_context_manager(self):
        # same thing but using a context manager
        _resource('s3').create_bucket(Bucket=BUCKET_NAME).wait_until_exists()

        with self.assertApiCalls(GetObject=1):
            with smart_open.s3.Reader(BUCKET_NAME, KEY_NAME) as fin:
                output = [line.rstrip(b'\n') for line in fin]
        self.assertEqual(output, self.body.split(b'\n'))

    def test_read(self):
        """Are S3 files read correctly?"""
        with self.assertApiCalls(GetObject=1):
            fin = smart_open.s3.Reader(BUCKET_NAME, KEY_NAME)
            self.assertEqual(self.body[:6], fin.read(6))
            self.assertEqual(self.body[6:14], fin.read(8))  # ř is 2 bytes
            self.assertEqual(self.body[14:], fin.read())  # read the rest

    def test_seek_beginning(self):
        """Does seeking to the beginning of S3 files work correctly?"""
        with self.assertApiCalls(GetObject=1):
            fin = smart_open.s3.Reader(BUCKET_NAME, KEY_NAME)
            self.assertEqual(self.body[:6], fin.read(6))
            self.assertEqual(self.body[6:14], fin.read(8))  # ř is 2 bytes

        with self.assertApiCalls(GetObject=1):
            fin.seek(0)
            self.assertEqual(self.body, fin.read())  # no size given => read whole file

        with self.assertApiCalls(GetObject=1):
            fin.seek(0)
            self.assertEqual(self.body, fin.read(-1))  # same thing

    def test_seek_start(self):
        """Does seeking from the start of S3 files work correctly?"""
        with self.assertApiCalls(GetObject=1):
            fin = smart_open.s3.Reader(BUCKET_NAME, KEY_NAME, defer_seek=True)
            seek = fin.seek(6)
            self.assertEqual(seek, 6)
            self.assertEqual(fin.tell(), 6)
            self.assertEqual(fin.read(6), u'wořld'.encode('utf-8'))

    def test_seek_current(self):
        """Does seeking from the middle of S3 files work correctly?"""
        with self.assertApiCalls(GetObject=1):
            fin = smart_open.s3.Reader(BUCKET_NAME, KEY_NAME)
            self.assertEqual(fin.read(5), b'hello')

        with self.assertApiCalls():
            seek = fin.seek(1, whence=smart_open.constants.WHENCE_CURRENT)
            self.assertEqual(seek, 6)
            self.assertEqual(fin.read(6), u'wořld'.encode('utf-8'))

    def test_seek_end(self):
        """Does seeking from the end of S3 files work correctly?"""
        with self.assertApiCalls(GetObject=1):
            fin = smart_open.s3.Reader(BUCKET_NAME, KEY_NAME, defer_seek=True)
            seek = fin.seek(-4, whence=smart_open.constants.WHENCE_END)
            self.assertEqual(seek, len(self.body) - 4)
            self.assertEqual(fin.read(), b'you?')

    def test_seek_end_tell_without_defer_seek(self):
        """Does seek(0, SEEK_END) + tell() work correctly without defer_seek?"""
        with self.assertApiCalls(GetObject=1):  # One call during __init__
            fin = smart_open.s3.Reader(BUCKET_NAME, KEY_NAME, defer_seek=False)
            pos = fin.seek(0, whence=smart_open.constants.WHENCE_END)
            self.assertEqual(pos, len(self.body))
            self.assertEqual(fin.tell(), len(self.body))

    def test_seek_end_tell_with_defer_seek(self):
        """Does seek(0, SEEK_END) + tell() work correctly with defer_seek=True?"""
        with self.assertApiCalls(GetObject=1):  # One call on first seek
            fin = smart_open.s3.Reader(BUCKET_NAME, KEY_NAME, defer_seek=True)
            pos = fin.seek(0, whence=smart_open.constants.WHENCE_END)
            self.assertEqual(pos, len(self.body))
            self.assertEqual(fin.tell(), len(self.body))

    def test_seek_past_end(self):
        with self.assertApiCalls(GetObject=1), patch_invalid_range_response(str(len(self.body))):
            fin = smart_open.s3.Reader(BUCKET_NAME, KEY_NAME, defer_seek=True)
            seek = fin.seek(60)
            self.assertEqual(seek, len(self.body))

    def test_seek_past_end_from_end(self):
        """Test seeking from end with offset larger than file size."""
        body_len = len(self.body)
        with self.assertApiCalls(GetObject=1), patch_invalid_range_response(str(body_len)):
            fin = smart_open.s3.Reader(BUCKET_NAME, KEY_NAME, defer_seek=True)
            seek = fin.seek(-(body_len + 10), whence=smart_open.constants.WHENCE_END)
            self.assertEqual(seek, 0)  # Should clamp to start of file

    def test_seek_forward_within_buffer(self):
        """Does forward seeking within buffered data avoid additional GET requests?"""
        with self.assertApiCalls(GetObject=1):
            fin = smart_open.s3.Reader(BUCKET_NAME, KEY_NAME, buffer_size=32)
            self.assertEqual(fin.read(5), b'hello')

            # Forward seek within buffer using WHENCE_CURRENT - should NOT make a new GET request
            seek = fin.seek(1, whence=smart_open.constants.WHENCE_CURRENT)
            self.assertEqual(seek, 6)
            self.assertEqual(fin.read(6), u'wořld'.encode('utf-8'))

            # Forward seek within buffer using WHENCE_START - should NOT make a new GET request
            seek = fin.seek(13, whence=smart_open.constants.WHENCE_START)
            self.assertEqual(seek, 13)
            self.assertEqual(fin.read(3), b'how')

    def test_detect_eof(self):
        with self.assertApiCalls(GetObject=1):
            fin = smart_open.s3.Reader(BUCKET_NAME, KEY_NAME)
            fin.read()
            eof = fin.tell()
            self.assertEqual(eof, len(self.body))
            fin.seek(0, whence=smart_open.constants.WHENCE_END)
            self.assertEqual(eof, fin.tell())
            fin.seek(eof)
            self.assertEqual(eof, fin.tell())

    def test_read_gzip(self):
        expected = u'раcцветали яблони и груши, поплыли туманы над рекой...'.encode('utf-8')
        buf = io.BytesIO()
        buf.close = lambda: None  # keep buffer open so that we can .getvalue()
        with gzip.GzipFile(fileobj=buf, mode='w') as zipfile:
            zipfile.write(expected)

        _resource('s3').Object(BUCKET_NAME, KEY_NAME).put(Body=buf.getvalue())

        #
        # Make sure we're reading things correctly.
        #
        with smart_open.s3.Reader(BUCKET_NAME, KEY_NAME) as fin:
            self.assertEqual(fin.read(), buf.getvalue())

        #
        # Make sure the buffer we wrote is legitimate gzip.
        #
        sanity_buf = io.BytesIO(buf.getvalue())
        with gzip.GzipFile(fileobj=sanity_buf) as zipfile:
            self.assertEqual(zipfile.read(), expected)

        logger.debug('starting actual test')
        with self.assertApiCalls(GetObject=1):
            with smart_open.s3.Reader(BUCKET_NAME, KEY_NAME) as fin:
                with gzip.GzipFile(fileobj=fin) as zipfile:
                    actual = zipfile.read()

        self.assertEqual(expected, actual)

    def test_readline(self):
        content = b'englishman\nin\nnew\nyork\n'
        _resource('s3').Object(BUCKET_NAME, KEY_NAME).put(Body=content)

        with self.assertApiCalls(GetObject=2):
            with smart_open.s3.Reader(BUCKET_NAME, KEY_NAME) as fin:
                fin.readline()
                self.assertEqual(fin.tell(), content.index(b'\n')+1)

                fin.seek(0)
                actual = list(fin)
                self.assertEqual(fin.tell(), len(content))

        expected = [b'englishman\n', b'in\n', b'new\n', b'york\n']
        self.assertEqual(expected, actual)

    def test_readline_tiny_buffer(self):
        content = b'englishman\nin\nnew\nyork\n'
        _resource('s3').Object(BUCKET_NAME, KEY_NAME).put(Body=content)

        with self.assertApiCalls(GetObject=1):
            with smart_open.s3.Reader(BUCKET_NAME, KEY_NAME, buffer_size=8) as fin:
                actual = list(fin)

        expected = [b'englishman\n', b'in\n', b'new\n', b'york\n']
        self.assertEqual(expected, actual)

    def test_read0_does_not_return_data(self):
        with self.assertApiCalls():
            # set defer_seek to verify that read(0) doesn't trigger an unnecessary API call
            with smart_open.s3.Reader(BUCKET_NAME, KEY_NAME, defer_seek=True) as fin:
                data = fin.read(0)

        self.assertEqual(data, b'')

    def test_to_boto3(self):
        with self.assertApiCalls():
            # set defer_seek to verify that to_boto3() doesn't trigger an unnecessary API call
            with smart_open.s3.Reader(BUCKET_NAME, KEY_NAME, defer_seek=True) as fin:
                returned_obj = fin.to_boto3(_resource('s3'))

        boto3_body = returned_obj.get()['Body'].read()
        self.assertEqual(self.body, boto3_body)

    def test_binary_iterator(self):
        expected = u"выйду ночью в поле с конём".encode('utf-8').split(b' ')
        _resource('s3').Object(BUCKET_NAME, KEY_NAME).put(Body=b'\n'.join(expected))

        # test the __iter__ method
        with self.assertApiCalls(GetObject=1):
            with smart_open.s3.open(BUCKET_NAME, KEY_NAME, 'rb') as fin:
                actual = [line.rstrip() for line in fin]
        self.assertEqual(expected, actual)

        # test the __next__ method
        with self.assertApiCalls(GetObject=1):
            with smart_open.s3.open(BUCKET_NAME, KEY_NAME, 'rb') as fin:
                first = next(fin).rstrip()
        self.assertEqual(expected[0], first)

    def test_text_iterator(self):
        expected = u"выйду ночью в поле с конём".split(' ')
        uri = f's3://{BUCKET_NAME}/{KEY_NAME}.gz'

        with smart_open.open(uri, 'w', encoding='utf-8') as fout:
            fout.write('\n'.join(expected))

        # test the __iter__ method
        with self.assertApiCalls(GetObject=1):
            with smart_open.open(uri, 'r', encoding='utf-8') as fin:
                actual = [line.rstrip() for line in fin]
        self.assertEqual(expected, actual)

        # test the __next__ method
        with self.assertApiCalls(GetObject=1):
            with smart_open.open(uri, 'r', encoding='utf-8') as fin:
                first = next(fin).rstrip()
        self.assertEqual(expected[0], first)

    def test_defer_seek(self):
        content = b'englishman\nin\nnew\nyork\n'
        _resource('s3').Object(BUCKET_NAME, KEY_NAME).put(Body=content)

        with self.assertApiCalls():
            fin = smart_open.s3.Reader(BUCKET_NAME, KEY_NAME, defer_seek=True)
        with self.assertApiCalls(GetObject=1):
            self.assertEqual(fin.read(), content)

        with self.assertApiCalls():
            fin = smart_open.s3.Reader(BUCKET_NAME, KEY_NAME, defer_seek=True)
        with self.assertApiCalls(GetObject=1):
            fin.seek(10)
            self.assertEqual(fin.read(), content[10:])

    def test_read_empty_file(self):
        _resource('s3').Object(BUCKET_NAME, KEY_NAME).put(Body=b'')

        with self.assertApiCalls(GetObject=1), patch_invalid_range_response('0'):
            with smart_open.s3.Reader(BUCKET_NAME, KEY_NAME) as fin:
                self.assertEqual(fin.read(), b'')

    def test_read_empty_file_no_actual_size(self):
        _resource('s3').Object(BUCKET_NAME, KEY_NAME).put(Body=b'')

        with self.assertApiCalls(GetObject=1), patch_invalid_range_response(None):
            with smart_open.s3.Reader(BUCKET_NAME, KEY_NAME) as fin:
                self.assertEqual(fin.read(), b'')
                # a subsequent read does not call _open_body
                self.assertEqual(fin.read(), b'')

    def test_seek_empty_file_from_end(self):
        """Test seeking from end on an empty file."""
        _resource('s3').Object(BUCKET_NAME, KEY_NAME).put(Body=b'')
        with self.assertApiCalls(GetObject=1), patch_invalid_range_response('0'):
            with smart_open.s3.Reader(BUCKET_NAME, KEY_NAME, defer_seek=True) as fin:
                seek = fin.seek(-10, whence=smart_open.constants.WHENCE_END)
                self.assertEqual(seek, 0)  # Should be at position 0 for empty file

@mock_s3
class MultipartWriterTest(unittest.TestCase):
    """
    Test writing into s3 files.

    """
    def setUp(self):
        ignore_resource_warnings()

        _resource('s3').create_bucket(Bucket=BUCKET_NAME).wait_until_exists()

    def test_write_01(self):
        """Does writing into s3 work correctly?"""
        test_string = u"žluťoučký koníček".encode('utf8')

        # write into key
        with smart_open.s3.MultipartWriter(BUCKET_NAME, WRITE_KEY_NAME) as fout:
            fout.write(test_string)

        # read key and test content
        output = list(smart_open.s3.open(BUCKET_NAME, WRITE_KEY_NAME, "rb"))

        self.assertEqual(output, [test_string])

    def test_write_01a(self):
        """Does s3 write fail on incorrect input?"""
        try:
            with smart_open.s3.MultipartWriter(BUCKET_NAME, WRITE_KEY_NAME) as fin:
                fin.write(None)
        except TypeError:
            pass
        else:
            self.fail()

    def test_write_02(self):
        """Does s3 write unicode-utf8 conversion work?"""
        smart_open_write = smart_open.s3.MultipartWriter(BUCKET_NAME, WRITE_KEY_NAME)
        smart_open_write.tell()
        logger.info("smart_open_write: %r", smart_open_write)
        with smart_open_write as fout:
            fout.write(u"testžížáč".encode("utf-8"))
            self.assertEqual(fout.tell(), 14)

    #
    # Nb. Under Windows, the byte offsets are different for some reason
    #
    @pytest.mark.skipif(condition=sys.platform == 'win32', reason="does not run on windows")
    def test_write_03(self):
        """Does s3 multipart chunking work correctly?"""
        #
        # generate enough test data for a single multipart upload part.
        # We need this because moto behaves like S3; it refuses to upload
        # parts smaller than 5MB.
        #
        data_dir = os.path.join(os.path.dirname(__file__), "test_data")
        with open(os.path.join(data_dir, "crime-and-punishment.txt"), "rb") as fin:
            crime = fin.read()
        data = b''
        ps = 5 * 1024 * 1024
        while len(data) < ps:
            data += crime

        title = "Преступление и наказание\n\n".encode()
        to_be_continued = "\n\n... продолжение следует ...\n\n".encode()

        with smart_open.s3.MultipartWriter(BUCKET_NAME, WRITE_KEY_NAME, part_size=ps) as fout:
            #
            # Write some data without triggering an upload
            #
            fout.write(title)
            assert fout._total_parts == 0
            assert fout._buf.tell() == 48

            #
            # Trigger a part upload
            #
            fout.write(data)
            assert fout._total_parts == 1
            assert fout._buf.tell() == 661

            #
            # Write _without_ triggering a part upload
            #
            fout.write(to_be_continued)
            assert fout._total_parts == 1
            assert fout._buf.tell() == 710

        #
        # We closed the writer, so the final part must have been uploaded
        #
        assert fout._buf.tell() == 0
        assert fout._total_parts == 2

        #
        # read back the same key and check its content
        #
        with smart_open.s3.open(BUCKET_NAME, WRITE_KEY_NAME, 'rb') as fin:
            got = fin.read()
        want = title + data + to_be_continued
        assert want == got

    def test_write_04(self):
        """Does writing no data cause key with an empty value to be created?"""
        smart_open_write = smart_open.s3.MultipartWriter(BUCKET_NAME, WRITE_KEY_NAME)
        with smart_open_write as fout:  # noqa
            pass

        # read back the same key and check its content
        with patch_invalid_range_response('0'):
            output = list(smart_open.s3.open(BUCKET_NAME, WRITE_KEY_NAME, 'rb'))

        self.assertEqual(output, [])

    def test_gzip(self):
        expected = u'а не спеть ли мне песню... о любви'.encode('utf-8')
        with smart_open.s3.MultipartWriter(BUCKET_NAME, WRITE_KEY_NAME) as fout:
            with gzip.GzipFile(fileobj=fout, mode='w') as zipfile:
                zipfile.write(expected)

        with smart_open.s3.Reader(BUCKET_NAME, WRITE_KEY_NAME) as fin:
            with gzip.GzipFile(fileobj=fin) as zipfile:
                actual = zipfile.read()

        self.assertEqual(expected, actual)

    def test_buffered_writer_wrapper_works(self):
        """
        Ensure that we can wrap a smart_open s3 stream in a BufferedWriter, which
        passes a memoryview object to the underlying stream in python >= 2.7
        """
        expected = u'не думай о секундах свысока'

        with smart_open.s3.MultipartWriter(BUCKET_NAME, WRITE_KEY_NAME) as fout:
            with io.BufferedWriter(fout) as sub_out:
                sub_out.write(expected.encode('utf-8'))

        with smart_open.s3.open(BUCKET_NAME, WRITE_KEY_NAME, 'rb') as fin:
            with io.TextIOWrapper(fin, encoding='utf-8') as text:
                actual = text.read()

        self.assertEqual(expected, actual)

    def test_nonexisting_bucket(self):
        expected = u"выйду ночью в поле с конём".encode('utf-8')
        with self.assertRaisesRegex(ValueError, "does not exist"):
            with smart_open.s3.open('thisbucketdoesntexist', 'mykey', 'wb') as fout:
                fout.write(expected)

        with self.assertRaisesRegex(ValueError, "does not exist"):
            with smart_open.s3.open('thisbucketdoesntexist', 'mykey', 'wb', multipart_upload=False) as fout:
                fout.write(expected)

    def test_read_nonexisting_key(self):
        with self.assertRaises(IOError):
            with smart_open.s3.open(BUCKET_NAME, 'my_nonexisting_key', 'rb') as fin:
                fin.read()

    def test_double_close(self):
        text = u'там за туманами, вечными, пьяными'.encode('utf-8')
        fout = smart_open.s3.open(BUCKET_NAME, 'key', 'wb')
        fout.write(text)
        fout.close()
        fout.close()

    def test_flush_close(self):
        text = u'там за туманами, вечными, пьяными'.encode('utf-8')
        fout = smart_open.s3.open(BUCKET_NAME, 'key', 'wb')
        fout.write(text)
        fout.flush()
        fout.close()

    def test_to_boto3(self):
        contents = b'the spice melange\n'

        with smart_open.s3.open(BUCKET_NAME, KEY_NAME, 'wb') as fout:
            fout.write(contents)
            returned_obj = fout.to_boto3(_resource('s3'))

        boto3_body = returned_obj.get()['Body'].read()
        self.assertEqual(contents, boto3_body)

    def test_writebuffer(self):
        """Does the MultipartWriter support writing to a custom buffer?"""
        contents = b'get ready for a surprise'

        with tempfile.NamedTemporaryFile(mode='rb+') as f:
            with smart_open.s3.MultipartWriter(BUCKET_NAME, WRITE_KEY_NAME, writebuffer=f) as fout:
                fout.write(contents)

            with smart_open.s3.open(BUCKET_NAME, WRITE_KEY_NAME, 'rb') as fin:
                actual = fin.read()

            assert actual == contents

    def test_write_gz_using_context_manager(self):
        """Does s3 multipart upload create a compressed file using context manager?"""
        contents = b'get ready for a surprise'
        with smart_open.open(
                f's3://{BUCKET_NAME}/{WRITE_KEY_NAME}.gz',
                mode="wb",
                transport_params={
                    "multipart_upload": True,
                    "min_part_size": 10,
                }
        ) as fout:
            fout.write(contents)

        with smart_open.open(f's3://{BUCKET_NAME}/{WRITE_KEY_NAME}.gz', 'rb') as fin:
            actual = fin.read()

        assert actual == contents

    def test_write_gz_not_using_context_manager(self):
        """Does s3 multipart upload create a compressed file not using context manager but close()?"""
        contents = b'get ready for a surprise'
        fout = smart_open.open(
            f's3://{BUCKET_NAME}/{WRITE_KEY_NAME}.gz',
            mode="wb",
            transport_params={
                "multipart_upload": True,
                "min_part_size": 10,
            }
        )
        fout.write(contents)
        fout.close()

        with smart_open.open(f's3://{BUCKET_NAME}/{WRITE_KEY_NAME}.gz', 'rb') as fin:
            actual = fin.read()

        assert actual == contents

    def test_write_gz_with_error(self):
        """Does s3 multipart upload abort for a failed compressed file upload?"""
        with self.assertRaises(ValueError):
            with smart_open.open(
                    f's3://{BUCKET_NAME}/{WRITE_KEY_NAME}',
                    mode="wb",
                    compression='.gz',
                    transport_params={
                        "multipart_upload": True,
                        "min_part_size": 10,
                    }
            ) as fout:
                fout.write(b"test12345678test12345678")
                fout.write(b"test\n")

                # FileLikeWrapper.__exit__ should cause a MultipartWriter.terminate()
                raise ValueError("some error")

        # no multipart upload was committed:
        # smart_open.s3.MultipartWriter.__exit__ was called
        with self.assertRaises(OSError) as cm:
            smart_open.s3.open(BUCKET_NAME, WRITE_KEY_NAME, 'rb')

        assert 'The specified key does not exist.' in cm.exception.args[0]

    def test_write_text_with_error(self):
        """Does s3 multipart upload abort for a failed text file upload?"""
        with self.assertRaises(ValueError):
            with smart_open.open(
                    f's3://{BUCKET_NAME}/{WRITE_KEY_NAME}',
                    mode="w",
                    transport_params={
                        "multipart_upload": True,
                        "min_part_size": 10,
                    }
            ) as fout:
                fout.write("test12345678test12345678")
                fout.write("test\n")

                # TextIOWrapper.__exit__ should not cause a self.buffer.close()
                # FileLikeWrapper.__exit__ should cause a MultipartWriter.terminate()
                raise ValueError("some error")

        # no multipart upload was committed:
        # smart_open.s3.MultipartWriter.__exit__ was called
        with self.assertRaises(OSError) as cm:
            smart_open.s3.open(BUCKET_NAME, WRITE_KEY_NAME, 'rb')

        assert 'The specified key does not exist.' in cm.exception.args[0]


@mock_s3
class SinglepartWriterTest(unittest.TestCase):
    """
    Test writing into s3 files using single part upload.

    """
    def setUp(self):
        ignore_resource_warnings()

        _resource('s3').create_bucket(Bucket=BUCKET_NAME).wait_until_exists()

    def test_write_01(self):
        """Does writing into s3 work correctly?"""
        test_string = u"žluťoučký koníček".encode('utf8')

        # write into key
        with smart_open.s3.SinglepartWriter(BUCKET_NAME, WRITE_KEY_NAME) as fout:
            fout.write(test_string)

        # read key and test content
        output = list(smart_open.s3.open(BUCKET_NAME, WRITE_KEY_NAME, "rb"))

        self.assertEqual(output, [test_string])

    def test_write_01a(self):
        """Does s3 write fail on incorrect input?"""
        try:
            with smart_open.s3.SinglepartWriter(BUCKET_NAME, WRITE_KEY_NAME) as fin:
                fin.write(None)
        except TypeError:
            pass
        else:
            self.fail()

    def test_write_02(self):
        """Does s3 write unicode-utf8 conversion work?"""
        test_string = u"testžížáč".encode("utf-8")

        smart_open_write = smart_open.s3.SinglepartWriter(BUCKET_NAME, WRITE_KEY_NAME)
