# -*- 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).
#

import bz2
import csv
import contextlib
import functools
import io
import gzip
import hashlib
import logging
import os
from smart_open.compression import INFER_FROM_EXTENSION, NO_COMPRESSION
import tempfile
import unittest
from unittest import mock
import warnings

import boto3
import pytest
import responses

# 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
from smart_open import smart_open_lib
from smart_open import webhdfs
from smart_open.smart_open_lib import patch_pathlib, _patch_pathlib

from .test_s3 import patch_invalid_range_response

logger = logging.getLogger(__name__)

CURR_DIR = os.path.abspath(os.path.dirname(__file__))
SAMPLE_TEXT = 'Hello, world!'
SAMPLE_BYTES = SAMPLE_TEXT.encode('utf-8')

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


#
# For Windows platforms, under which tempfile.NamedTemporaryFile has some
# unwanted quirks.
#
# https://docs.python.org/3.14/library/tempfile.html#tempfile.NamedTemporaryFile
# https://stackoverflow.com/a/58955530
#
@contextlib.contextmanager
def named_temporary_file(mode='w+b', prefix=None, suffix=None, delete=True):
    filename = io.StringIO()
    if prefix:
        filename.write(prefix)
    filename.write(os.urandom(8).hex())
    if suffix:
        filename.write(suffix)
    pathname = os.path.join(tempfile.gettempdir(), filename.getvalue())

    with open(pathname, mode) as f:
        yield f

    if delete:
        try:
            os.unlink(pathname)
        except PermissionError as e:
            #
            # This can happen on Windows for unknown reasons.
            #
            logger.error(e)


def test_compression_extensions():
    for extension in smart_open.compression.get_supported_extensions():
        with named_temporary_file(suffix=extension) as tmp:
            with smart_open.open(tmp.name, "wt") as fout:
                print("hello world", file=fout)
                print("this is a test", file=fout)

            with smart_open.open(tmp.name, "rt") as fin:
                got = list(fin)

        assert got == ["hello world\n", "this is a test\n"], f"Error for {extension=}, mode='wt'"

        with named_temporary_file(suffix=extension) as tmp:
            with smart_open.open(tmp.name, "w") as fout:
                fout.write("hello world\n")
                fout.write("this is a test\n")

            with smart_open.open(tmp.name, "r") as fin:
                got = list(fin)

        assert got == ["hello world\n", "this is a test\n"], f"Error for {extension=}, mode='w'"

        with named_temporary_file(suffix=extension) as tmp:
            with smart_open.open(tmp.name, "wb") as fout:
                fout.write(b"hello world\n")
                fout.write(b"this is a test\n")

            with smart_open.open(tmp.name, "rb") as fin:
                got = list(fin)

        assert got == [b"hello world\n", b"this is a test\n"], f"Error for {extension=}, mode='wb'"


class ParseUriTest(unittest.TestCase):
    """
    Test ParseUri class.

    """
    def test_scheme(self):
        """Do URIs schemes parse correctly?"""
        # supported schemes
        for scheme in ("s3", "s3a", "s3n", "hdfs", "file", "http", "https", "gs", "azure"):
            parsed_uri = smart_open_lib._parse_uri(scheme + "://mybucket/mykey")
            self.assertEqual(parsed_uri.scheme, scheme)

        # unsupported scheme => NotImplementedError
        self.assertRaises(NotImplementedError, smart_open_lib._parse_uri, "foobar://mybucket/mykey")

        # unknown scheme => default_scheme
        parsed_uri = smart_open_lib._parse_uri("blah blah")
        self.assertEqual(parsed_uri.scheme, "file")

    def test_s3_uri(self):
        """Do S3 URIs parse correctly?"""
        # correct uri without credentials
        parsed_uri = smart_open_lib._parse_uri("s3://mybucket/mykey")
        self.assertEqual(parsed_uri.scheme, "s3")
        self.assertEqual(parsed_uri.bucket_id, "mybucket")
        self.assertEqual(parsed_uri.key_id, "mykey")
        self.assertEqual(parsed_uri.access_id, None)
        self.assertEqual(parsed_uri.access_secret, None)

    def test_s3_uri_contains_slash(self):
        parsed_uri = smart_open_lib._parse_uri("s3://mybucket/mydir/mykey")
        self.assertEqual(parsed_uri.scheme, "s3")
        self.assertEqual(parsed_uri.bucket_id, "mybucket")
        self.assertEqual(parsed_uri.key_id, "mydir/mykey")
        self.assertEqual(parsed_uri.access_id, None)
        self.assertEqual(parsed_uri.access_secret, None)

    def test_s3_uri_with_credentials(self):
        parsed_uri = smart_open_lib._parse_uri("s3://ACCESSID456:acces/sse_cr-et@mybucket/mykey")
        self.assertEqual(parsed_uri.scheme, "s3")
        self.assertEqual(parsed_uri.bucket_id, "mybucket")
        self.assertEqual(parsed_uri.key_id, "mykey")
        self.assertEqual(parsed_uri.access_id, "ACCESSID456")
        self.assertEqual(parsed_uri.access_secret, "acces/sse_cr-et")

    def test_s3_uri_with_credentials2(self):
        parsed_uri = smart_open_lib._parse_uri("s3://accessid:access/secret@mybucket/mykey")
        self.assertEqual(parsed_uri.scheme, "s3")
        self.assertEqual(parsed_uri.bucket_id, "mybucket")
        self.assertEqual(parsed_uri.key_id, "mykey")
        self.assertEqual(parsed_uri.access_id, "accessid")
        self.assertEqual(parsed_uri.access_secret, "access/secret")

    def test_s3_uri_has_atmark_in_key_name(self):
        parsed_uri = smart_open_lib._parse_uri("s3://accessid:access/secret@mybucket/my@ke@y")
        self.assertEqual(parsed_uri.scheme, "s3")
        self.assertEqual(parsed_uri.bucket_id, "mybucket")
        self.assertEqual(parsed_uri.key_id, "my@ke@y")
        self.assertEqual(parsed_uri.access_id, "accessid")
        self.assertEqual(parsed_uri.access_secret, "access/secret")

    #
    # Nb. should never happen in theory, but if it does, we should avoid crashing
    #
    def test_s3_uri_has_colon_in_secret(self):
        parsed_uri = smart_open_lib._parse_uri("s3://accessid:access/secret:totally@mybucket/my@ke@y")
        self.assertEqual(parsed_uri.scheme, "s3")
        self.assertEqual(parsed_uri.bucket_id, "mybucket")
        self.assertEqual(parsed_uri.key_id, "my@ke@y")
        self.assertEqual(parsed_uri.access_id, "accessid")
        self.assertEqual(parsed_uri.access_secret, "access/secret:totally")

    def test_s3_uri_has_atmark_in_key_name2(self):
        parsed_uri = smart_open_lib._parse_uri(
            "s3://accessid:access/secret@hostname:1234@mybucket/dir/my@ke@y"
        )
        self.assertEqual(parsed_uri.scheme, "s3")
        self.assertEqual(parsed_uri.bucket_id, "mybucket")
        self.assertEqual(parsed_uri.key_id, "dir/my@ke@y")
        self.assertEqual(parsed_uri.access_id, "accessid")
        self.assertEqual(parsed_uri.access_secret, "access/secret")
        self.assertEqual(parsed_uri.host, "hostname")
        self.assertEqual(parsed_uri.port, 1234)

    def test_s3_uri_has_atmark_in_key_name3(self):
        parsed_uri = smart_open_lib._parse_uri("s3://accessid:access/secret@hostname@mybucket/dir/my@ke@y")
        self.assertEqual(parsed_uri.scheme, "s3")
        self.assertEqual(parsed_uri.bucket_id, "mybucket")
        self.assertEqual(parsed_uri.key_id, "dir/my@ke@y")
        self.assertEqual(parsed_uri.access_id, "accessid")
        self.assertEqual(parsed_uri.access_secret, "access/secret")
        self.assertEqual(parsed_uri.host, "hostname")
        self.assertEqual(parsed_uri.port, 443)

    def test_s3_handles_fragments(self):
        uri_str = 's3://bucket-name/folder/picture #1.jpg'
        parsed_uri = smart_open_lib._parse_uri(uri_str)
        self.assertEqual(parsed_uri.key_id, "folder/picture #1.jpg")

    def test_s3_handles_querystring(self):
        uri_str = 's3://bucket-name/folder/picture1.jpg?bar'
        parsed_uri = smart_open_lib._parse_uri(uri_str)
        self.assertEqual(parsed_uri.key_id, "folder/picture1.jpg?bar")

    def test_s3_invalid_url_atmark_in_bucket_name(self):
        self.assertRaises(
            ValueError, smart_open_lib._parse_uri,
            "s3://access_id:access_secret@my@bucket@port/mykey",
        )

    def test_s3_invalid_uri_missing_colon(self):
        self.assertRaises(
            ValueError, smart_open_lib._parse_uri,
            "s3://access_id@access_secret@mybucket@port/mykey",
        )

    def test_webhdfs_uri_to_http(self):
        parsed_uri = smart_open_lib._parse_uri("webhdfs://host:14000/path/file")
        actual = webhdfs.convert_to_http_uri(parsed_uri)
        expected = "http://host:14000/webhdfs/v1/path/file"
        self.assertEqual(actual, expected)

    def test_webhdfs_uri_to_http_with_query(self):
        parsed_uri = smart_open_lib._parse_uri("webhdfs://host:14000/path/file?a=1")
        actual = webhdfs.convert_to_http_uri(parsed_uri)
        expected = "http://host:14000/webhdfs/v1/path/file?a=1"
        self.assertEqual(actual, expected)

    def test_webhdfs_uri_to_http_with_user(self):
        parsed_uri = smart_open_lib._parse_uri("webhdfs://user@host:14000/path")
        actual = webhdfs.convert_to_http_uri(parsed_uri)
        expected = "http://host:14000/webhdfs/v1/path?user.name=user"
        self.assertEqual(actual, expected)

    def test_webhdfs_uri_to_http_with_user_and_query(self):
        parsed_uri = smart_open_lib._parse_uri("webhdfs://user@host:14000/path?a=1")
        actual = webhdfs.convert_to_http_uri(parsed_uri)
        expected = "http://host:14000/webhdfs/v1/path?a=1&user.name=user"
        self.assertEqual(actual, expected)

    def test_uri_from_issue_223_works(self):
        uri = "s3://:@omax-mis/twilio-messages-media/final/MEcd7c36e75f87dc6dd9e33702cdcd8fb6"
        parsed_uri = smart_open_lib._parse_uri(uri)
        self.assertEqual(parsed_uri.scheme, "s3")
        self.assertEqual(parsed_uri.bucket_id, "omax-mis")
        self.assertEqual(parsed_uri.key_id, "twilio-messages-media/final/MEcd7c36e75f87dc6dd9e33702cdcd8fb6")
        self.assertEqual(parsed_uri.access_id, "")
        self.assertEqual(parsed_uri.access_secret, "")

    def test_s3_uri_with_colon_in_key_name(self):
        """ Correctly parse the s3 url if there is a colon in the key or dir """
        parsed_uri = smart_open_lib._parse_uri("s3://mybucket/mydir/my:key")
        self.assertEqual(parsed_uri.scheme, "s3")
        self.assertEqual(parsed_uri.bucket_id, "mybucket")
        self.assertEqual(parsed_uri.key_id, "mydir/my:key")
        self.assertEqual(parsed_uri.access_id, None)
        self.assertEqual(parsed_uri.access_secret, None)

    def test_s3_uri_with_at_symbol_in_key_name0(self):
        """ Correctly parse the s3 url if there is an @ symbol (and colon) in the key or dir """
        parsed_uri = smart_open_lib._parse_uri("s3://mybucket/mydir:my@key")
        self.assertEqual(parsed_uri.scheme, "s3")
        self.assertEqual(parsed_uri.bucket_id, "mybucket")
        self.assertEqual(parsed_uri.key_id, "mydir:my@key")
        self.assertEqual(parsed_uri.access_id, None)
        self.assertEqual(parsed_uri.access_secret, None)

    def test_s3_uri_with_at_symbol_in_key_name1(self):
        """ Correctly parse the s3 url if there is an @ symbol (and colon) in the key or dir """
        parsed_uri = smart_open_lib._parse_uri("s3://mybucket/my:dir@my/key")
        self.assertEqual(parsed_uri.scheme, "s3")
        self.assertEqual(parsed_uri.bucket_id, "mybucket")
        self.assertEqual(parsed_uri.key_id, "my:dir@my/key")
        self.assertEqual(parsed_uri.access_id, None)
        self.assertEqual(parsed_uri.access_secret, None)

    def test_s3_uri_contains_question_mark(self):
        parsed_uri = smart_open_lib._parse_uri("s3://mybucket/mydir/mykey?param")
        self.assertEqual(parsed_uri.scheme, "s3")
        self.assertEqual(parsed_uri.bucket_id, "mybucket")
        self.assertEqual(parsed_uri.key_id, "mydir/mykey?param")

    def test_host_and_port(self):
        as_string = 's3u://user:secret@host:1234@mybucket/mykey.txt'
        uri = smart_open_lib._parse_uri(as_string)
        self.assertEqual(uri.scheme, "s3u")
        self.assertEqual(uri.bucket_id, "mybucket")
        self.assertEqual(uri.key_id, "mykey.txt")
        self.assertEqual(uri.access_id, "user")
        self.assertEqual(uri.access_secret, "secret")
        self.assertEqual(uri.host, "host")
        self.assertEqual(uri.port, 1234)

    def test_invalid_port(self):
        as_string = 's3u://user:secret@host:port@mybucket/mykey.txt'
        self.assertRaises(ValueError, smart_open_lib._parse_uri, as_string)

    def test_invalid_port2(self):
        as_string = 's3u://user:secret@host:port:foo@mybucket/mykey.txt'
        self.assertRaises(ValueError, smart_open_lib._parse_uri, as_string)

    def test_leading_slash_local_file(self):
        path = "/home/misha/hello.txt"
        uri = smart_open_lib._parse_uri(path)
        self.assertEqual(uri.scheme, "file")
        self.assertEqual(uri.uri_path, path)

        uri = smart_open_lib._parse_uri('//' + path)
        self.assertEqual(uri.scheme, "file")
        self.assertEqual(uri.uri_path, '//' + path)

    def test_ssh(self):
        as_string = 'ssh://user@host:1234/path/to/file'
        uri = smart_open_lib._parse_uri(as_string)
        self.assertEqual(uri.scheme, 'ssh')
        self.assertEqual(uri.uri_path, '/path/to/file')
        self.assertEqual(uri.user, 'user')
        self.assertEqual(uri.host, 'host')
        self.assertEqual(uri.port, 1234)
        self.assertEqual(uri.password, None)

    def test_ssh_with_pass(self):
        as_string = 'ssh://user:pass@host:1234/path/to/file'
        uri = smart_open_lib._parse_uri(as_string)
        self.assertEqual(uri.scheme, 'ssh')
        self.assertEqual(uri.uri_path, '/path/to/file')
        self.assertEqual(uri.user, 'user')
        self.assertEqual(uri.host, 'host')
        self.assertEqual(uri.port, 1234)
        self.assertEqual(uri.password, 'pass')

    def test_scp(self):
        as_string = 'scp://user@host:/path/to/file'
        uri = smart_open_lib._parse_uri(as_string)
        self.assertEqual(uri.scheme, 'scp')
        self.assertEqual(uri.uri_path, '/path/to/file')
        self.assertEqual(uri.user, 'user')
        self.assertEqual(uri.host, 'host')
        self.assertEqual(uri.port, None)
        self.assertEqual(uri.password, None)

    def test_scp_with_pass(self):
        as_string = 'scp://user:pass@host:/path/to/file'
        uri = smart_open_lib._parse_uri(as_string)
        self.assertEqual(uri.scheme, 'scp')
        self.assertEqual(uri.uri_path, '/path/to/file')
        self.assertEqual(uri.user, 'user')
        self.assertEqual(uri.host, 'host')
        self.assertEqual(uri.port, None)
        self.assertEqual(uri.password, 'pass')

    def test_sftp(self):
        as_string = 'sftp://host/path/to/file'
        uri = smart_open_lib._parse_uri(as_string)
        self.assertEqual(uri.scheme, 'sftp')
        self.assertEqual(uri.uri_path, '/path/to/file')
        self.assertEqual(uri.user, None)
        self.assertEqual(uri.host, 'host')
        self.assertEqual(uri.port, None)
        self.assertEqual(uri.password, None)

    def test_sftp_with_user_and_pass(self):
        as_string = 'sftp://user:pass@host:2222/path/to/file'
        uri = smart_open_lib._parse_uri(as_string)
        self.assertEqual(uri.scheme, 'sftp')
        self.assertEqual(uri.uri_path, '/path/to/file')
        self.assertEqual(uri.user, 'user')
        self.assertEqual(uri.host, 'host')
        self.assertEqual(uri.port, 2222)
        self.assertEqual(uri.password, 'pass')

    def test_ssh_complex_password_with_colon(self):
        as_string = 'sftp://user:some:complex@password$$@host:2222/path/to/file'
        uri = smart_open_lib._parse_uri(as_string)
        self.assertEqual(uri.password, 'some:complex@password$$')

    def test_gs_uri(self):
        """Do GCS URIs parse correctly?"""
        # correct uri without credentials
        parsed_uri = smart_open_lib._parse_uri("gs://mybucket/myblob")
        self.assertEqual(parsed_uri.scheme, "gs")
        self.assertEqual(parsed_uri.bucket_id, "mybucket")
        self.assertEqual(parsed_uri.blob_id, "myblob")

    def test_gs_uri_contains_slash(self):
        parsed_uri = smart_open_lib._parse_uri("gs://mybucket/mydir/myblob")
        self.assertEqual(parsed_uri.scheme, "gs")
        self.assertEqual(parsed_uri.bucket_id, "mybucket")
        self.assertEqual(parsed_uri.blob_id, "mydir/myblob")

    def test_gs_uri_contains_question_mark(self):
        parsed_uri = smart_open_lib._parse_uri("gs://mybucket/mydir/myblob?param")
        self.assertEqual(parsed_uri.scheme, "gs")
        self.assertEqual(parsed_uri.bucket_id, "mybucket")
        self.assertEqual(parsed_uri.blob_id, "mydir/myblob?param")

    def test_azure_blob_uri(self):
        """Do Azure Blob URIs parse correctly?"""
        # correct uri without credentials
        parsed_uri = smart_open_lib._parse_uri("azure://mycontainer/myblob")
        self.assertEqual(parsed_uri.scheme, "azure")
        self.assertEqual(parsed_uri.container_id, "mycontainer")
        self.assertEqual(parsed_uri.blob_id, "myblob")

    def test_azure_blob_uri_root_container(self):
        parsed_uri = smart_open_lib._parse_uri("azure://myblob")
        self.assertEqual(parsed_uri.scheme, "azure")
        self.assertEqual(parsed_uri.container_id, "$root")
        self.assertEqual(parsed_uri.blob_id, "myblob")

    def test_azure_blob_uri_contains_slash(self):
        parsed_uri = smart_open_lib._parse_uri("azure://mycontainer/mydir/myblob")
        self.assertEqual(parsed_uri.scheme, "azure")
        self.assertEqual(parsed_uri.container_id, "mycontainer")
        self.assertEqual(parsed_uri.blob_id, "mydir/myblob")

    def test_pathlib_monkeypatch(self):
        from smart_open.smart_open_lib import pathlib

        assert pathlib.Path.open != smart_open.open

        with patch_pathlib():
            assert pathlib.Path.open == smart_open.open

        assert pathlib.Path.open != smart_open.open

        obj = patch_pathlib()
        assert pathlib.Path.open == smart_open.open

        _patch_pathlib(obj.old_impl)
        assert pathlib.Path.open != smart_open.open

    def test_pathlib_monkeypatch_read_gz(self):
        from smart_open.smart_open_lib import pathlib

        path = pathlib.Path(CURR_DIR) / 'test_data' / 'crime-and-punishment.txt.gz'

        # Check that standard implementation can't work with gzip
        with path.open("r") as infile:
            with self.assertRaises(Exception):
                lines = infile.readlines()

        # Check that our implementation works with gzip
        obj = patch_pathlib()
        try:
            with path.open("r", encoding='utf-8') as infile:
                lines = infile.readlines()
            self.assertEqual(len(lines), 3)
        finally:
            _patch_pathlib(obj.old_impl)


class SmartOpenHttpTest(unittest.TestCase):
    """
    Test reading from HTTP connections in various ways.

    """
    @mock.patch('smart_open.ssh.open', return_value=open(__file__))
    def test_read_ssh(self, mock_open):
        """Is SSH line iterator called correctly?"""
        obj = smart_open.open(
            "ssh://ubuntu:pass@ip_address:1022/some/path/lines.txt",
            mode='rb',
            transport_params=dict(hello='world'),
        )
        obj.__iter__()
        mock_open.assert_called_with(
            '/some/path/lines.txt',
            'rb',
            host='ip_address',
            user='ubuntu',
            password='pass',
            port=1022,
        )

    @responses.activate
    def test_http_read(self):
        """Does http read method work correctly"""
        responses.add(responses.GET, "http://127.0.0.1/index.html",
                      body='line1\nline2')
        smart_open_object = smart_open.open("http://127.0.0.1/index.html", 'rb')
        self.assertEqual(smart_open_object.read().decode("utf-8"), "line1\nline2")

    @responses.activate
    def test_https_readline(self):
        """Does https readline method work correctly"""
        responses.add(responses.GET, "https://127.0.0.1/index.html",
                      body=u'line1\u2028still line1\nline2')
        smart_open_object = smart_open.open("https://127.0.0.1/index.html", 'rb')
        self.assertEqual(smart_open_object.readline().decode("utf-8"), u"line1\u2028still line1\n")
        smart_open_object = smart_open.open("https://127.0.0.1/index.html", 'r', encoding='utf-8')
        self.assertEqual(smart_open_object.readline(), u"line1\u2028still line1\n")

    @responses.activate
    def test_http_pass(self):
        """Does http authentication work correctly"""
        responses.add(responses.GET, "http://127.0.0.1/index.html",
                      body='line1\nline2')
        tp = dict(user='me', password='pass')
        smart_open.open("http://127.0.0.1/index.html", transport_params=tp)
        self.assertEqual(len(responses.calls), 1)
        actual_request = responses.calls[0].request
        self.assertTrue('Authorization' in actual_request.headers)
        self.assertTrue(actual_request.headers['Authorization'].startswith('Basic '))

    @responses.activate
    def test_http_cert(self):
        """Does cert parameter get passed to requests"""
        responses.add(responses.GET, "http://127.0.0.1/index.html",
                      body='line1\nline2')
        cert_path = '/path/to/my/cert.pem'
        tp = dict(cert=cert_path)
        smart_open.open("http://127.0.0.1/index.html", transport_params=tp)
        self.assertEqual(len(responses.calls), 1)
        actual_request = responses.calls[0].request
        self.assertEqual(cert_path, actual_request.req_kwargs['cert'])

    @responses.activate
    def _test_compressed_http(self, suffix, query):
        """Can open <suffix> via http?"""
        assert suffix in ('.gz', '.bz2')

        raw_data = b'Hello World Compressed.' * 10000
        compressed_data = gzip_compress(raw_data) if suffix == '.gz' else bz2.compress(raw_data)

        responses.add(responses.GET, 'http://127.0.0.1/data' + suffix, body=compressed_data)
        url = 'http://127.0.0.1/data%s%s' % (suffix, '?some_param=some_val' if query else '')
        smart_open_object = smart_open.open(url, 'rb')

        assert smart_open_object.read() == raw_data

    def test_http_gz(self):
        """Can open gzip via http?"""
        self._test_compressed_http(".gz", False)

    def test_http_bz2(self):
        """Can open bzip2 via http?"""
        self._test_compressed_http(".bz2", False)

    def test_http_gz_query(self):
        """Can open gzip via http with a query appended to URI?"""
        self._test_compressed_http(".gz", True)

    def test_http_bz2_query(self):
        """Can open bzip2 via http with a query appended to URI?"""
        self._test_compressed_http(".bz2", True)


class RealFileSystemTests(unittest.TestCase):
    """Tests that touch the file system via temporary files."""

    def setUp(self):
        with named_temporary_file(prefix='test', delete=False) as fout:
            fout.write(SAMPLE_BYTES)
            self.temp_file = fout.name

    def tearDown(self):
        os.unlink(self.temp_file)

    def test_rt(self):
        with smart_open.open(self.temp_file, 'rt') as fin:
            data = fin.read()
        self.assertEqual(data, SAMPLE_TEXT)

    def test_wt(self):
        #
        # The file already contains SAMPLE_TEXT, so write something different.
        #
        text = 'nippon budokan'
        with smart_open.open(self.temp_file, 'wt') as fout:
            fout.write(text)

        with smart_open.open(self.temp_file, 'rt') as fin:
            data = fin.read()
        self.assertEqual(data, text)

    def test_ab(self):
        with smart_open.open(self.temp_file, 'ab') as fout:
            fout.write(SAMPLE_BYTES)
        with smart_open.open(self.temp_file, 'rb') as fin:
            data = fin.read()
        self.assertEqual(data, SAMPLE_BYTES * 2)

    def test_aplus(self):
        with smart_open.open(self.temp_file, 'a+') as fout:
            fout.write(SAMPLE_TEXT)
        with smart_open.open(self.temp_file, 'rt') as fin:
            text = fin.read()
        self.assertEqual(text, SAMPLE_TEXT * 2)

    def test_at(self):
        with smart_open.open(self.temp_file, 'at') as fout:
            fout.write(SAMPLE_TEXT)
        with smart_open.open(self.temp_file, 'rt') as fin:
            text = fin.read()
        self.assertEqual(text, SAMPLE_TEXT * 2)

    def test_atplus(self):
        with smart_open.open(self.temp_file, 'at+') as fout:
            fout.write(SAMPLE_TEXT)
        with smart_open.open(self.temp_file, 'rt') as fin:
            text = fin.read()
        self.assertEqual(text, SAMPLE_TEXT * 2)


class CompressionRealFileSystemTests(RealFileSystemTests):
    """Same as RealFileSystemTests but with a compressed file."""

    def setUp(self):
        with named_temporary_file(prefix='test', suffix='.zst', delete=False) as fout:
            self.temp_file = fout.name
        with smart_open.open(self.temp_file, 'wb') as fout:
            fout.write(SAMPLE_BYTES)

    def test_aplus(self):
        pass  # transparent (de)compression unsupported for mode 'ab+'

    def test_atplus(self):
        pass  # transparent (de)compression unsupported for mode 'ab+'


#
# What exactly to patch here differs on _how_ we're opening the file.
# See the _shortcut_open function for details.
#
_IO_OPEN = 'io.open'
_BUILTIN_OPEN = 'smart_open.smart_open_lib._builtin_open'


class SmartOpenReadTest(unittest.TestCase):
    """
    Test reading from files under various schemes.

    """

    def test_shortcut(self):
        fpath = os.path.join(CURR_DIR, 'test_data/crime-and-punishment.txt')
        with mock.patch('smart_open.smart_open_lib._builtin_open') as mock_open:
            smart_open.open(fpath, 'r').read()
        mock_open.assert_called_with(fpath, 'r', buffering=-1)

    def test_open_binary(self):
        fpath = os.path.join(CURR_DIR, 'test_data/cp852.tsv.txt')
        with open(fpath, 'rb') as fin:
            expected = fin.read()
        with smart_open.open(fpath, 'rb') as fin:
            actual = fin.read()
        self.assertEqual(expected, actual)

    def test_open_with_keywords(self):
        """This test captures Issue #142."""
        fpath = os.path.join(CURR_DIR, 'test_data/cp852.tsv.txt')
        with open(fpath, 'r', encoding='cp852') as fin:
            expected = fin.read()
        with smart_open.open(fpath, encoding='cp852') as fin:
            actual = fin.read()
        self.assertEqual(expected, actual)

    def test_open_with_keywords_explicit_r(self):
        fpath = os.path.join(CURR_DIR, 'test_data/cp852.tsv.txt')
        with open(fpath, 'r', encoding='cp852') as fin:
            expected = fin.read()
        with smart_open.open(fpath, mode='r', encoding='cp852') as fin:
            actual = fin.read()
        self.assertEqual(expected, actual)

    def test_open_and_read_pathlib_path(self):
        """If ``pathlib.Path`` is available we should be able to open and read."""
        from smart_open.smart_open_lib import pathlib

        fpath = os.path.join(CURR_DIR, 'test_data/cp852.tsv.txt')
        with open(fpath, 'rb') as fin:
            expected = fin.read().decode('cp852')
        with smart_open.open(pathlib.Path(fpath), mode='r', encoding='cp852', newline='') as fin:
            actual = fin.read()
        self.assertEqual(expected, actual)

    @mock_s3
    def test_read_never_returns_none(self):
        """read should never return None."""
        s3 = _resource('s3')
        s3.create_bucket(Bucket='mybucket')

        test_string = u"ветер по морю гуляет..."
        with smart_open.open("s3://mybucket/mykey", "wb") as fout:
            fout.write(test_string.encode('utf8'))

        r = smart_open.open("s3://mybucket/mykey", "rb")
        self.assertEqual(r.read(), test_string.encode("utf-8"))
        self.assertEqual(r.read(), b"")
        self.assertEqual(r.read(), b"")

    @mock_s3
    def test_read_newline_none(self):
        """Does newline open() parameter for reading work according to
           https://docs.python.org/3/library/functions.html#open-newline-parameter
        """
        _resource('s3').create_bucket(Bucket='mybucket')
        # Unicode line separator and various others must never split lines
        test_file = u"line\u2028 LF\nline\x1c CR\rline\x85 CRLF\r\nlast line"
        with smart_open.open("s3://mybucket/mykey", "wb") as fout:
            fout.write(test_file.encode("utf-8"))
        # No newline parameter means newline=None i.e. universal newline mode with all
        # line endings translated to '\n'
        with smart_open.open("s3://mybucket/mykey", "r", encoding='utf-8') as fin:
            self.assertEqual(list(fin), [
                u"line\u2028 LF\n",
                u"line\x1c CR\n",
                u"line\x85 CRLF\n",
                u"last line"
            ])

    @mock_s3
    def test_read_newline_empty(self):
        _resource('s3').create_bucket(Bucket='mybucket')
        test_file = u"line\u2028 LF\nline\x1c CR\rline\x85 CRLF\r\nlast line"
        with smart_open.open("s3://mybucket/mykey", "wb") as fout:
            fout.write(test_file.encode("utf-8"))
        # If newline='' universal newline mode is enabled but line separators are not replaced
        with smart_open.open("s3://mybucket/mykey", "r", encoding='utf-8', newline='') as fin:
            self.assertEqual(list(fin), [
                u"line\u2028 LF\n",
                u"line\x1c CR\r",
                u"line\x85 CRLF\r\n",
                u"last line"
            ])

    @mock_s3
    def test_read_newline_cr(self):
        _resource('s3').create_bucket(Bucket='mybucket')
        test_file = u"line\u2028 LF\nline\x1c CR\rline\x85 CRLF\r\nlast line"
        with smart_open.open("s3://mybucket/mykey", "wb") as fout:
            fout.write(test_file.encode("utf-8"))
        # If newline='\r' only CR splits lines
        with smart_open.open("s3://mybucket/mykey", "r", encoding='utf-8', newline='\r') as fin:
            self.assertEqual(list(fin), [
                u"line\u2028 LF\nline\x1c CR\r",
                u"line\x85 CRLF\r",
                u"\nlast line"
            ])

    @mock_s3
    def test_read_newline_lf(self):
        _resource('s3').create_bucket(Bucket='mybucket')
        test_file = u"line\u2028 LF\nline\x1c CR\rline\x85 CRLF\r\nlast line"
        with smart_open.open("s3://mybucket/mykey", "wb") as fout:
            fout.write(test_file.encode("utf-8"))
        # If newline='\n' only LF splits lines
        with smart_open.open("s3://mybucket/mykey", "r", encoding='utf-8', newline='\n') as fin:
            self.assertEqual(list(fin), [
                u"line\u2028 LF\n",
                u"line\x1c CR\rline\x85 CRLF\r\n",
                u"last line"
            ])

    @mock_s3
    def test_read_newline_crlf(self):
        _resource('s3').create_bucket(Bucket='mybucket')
        test_file = u"line\u2028 LF\nline\x1c CR\rline\x85 CRLF\r\nlast line"
        with smart_open.open("s3://mybucket/mykey", "wb") as fout:
            fout.write(test_file.encode("utf-8"))
        # If newline='\r\n' only CRLF splits lines
        with smart_open.open("s3://mybucket/mykey", "r", encoding='utf-8', newline='\r\n') as fin:
            self.assertEqual(list(fin), [
                u"line\u2028 LF\nline\x1c CR\rline\x85 CRLF\r\n",
                u"last line"
            ])

    @mock_s3
    def test_read_newline_slurp(self):
        _resource('s3').create_bucket(Bucket='mybucket')
        test_file = u"line\u2028 LF\nline\x1c CR\rline\x85 CRLF\r\nlast line"
        with smart_open.open("s3://mybucket/mykey", "wb") as fout:
            fout.write(test_file.encode("utf-8"))
        # Even reading the whole file with read() must replace newlines
        with smart_open.open("s3://mybucket/mykey", "r", encoding='utf-8', newline=None) as fin:
            self.assertEqual(
                fin.read(),
                u"line\u2028 LF\nline\x1c CR\nline\x85 CRLF\nlast line"
            )

    @mock_s3
    def test_read_newline_binary(self):
        _resource('s3').create_bucket(Bucket='mybucket')
        test_file = u"line\u2028 LF\nline\x1c CR\rline\x85 CRLF\r\nlast line"
