"""Unit tests for io.py."""

from __future__ import print_function

from __future__ import unicode_literals

import os

import sys

import time

import array

import threading

import random

import unittest

from itertools import chain, cycle

from test import test_support

import codecs

import io  # The module under test

class MockRawIO(io.RawIOBase):

    def __init__(self, read_stack=()):

        self._read_stack = list(read_stack)

        self._write_stack = []

    def read(self, n=None):

        try:

            return self._read_stack.pop(0)

        except:

            return b""

    def write(self, b):

        self._write_stack.append(b[:])

        return len(b)

    def writable(self):

        return True

    def fileno(self):

        return 42

    def readable(self):

        return True

    def seekable(self):

        return True

    def seek(self, pos, whence):

        pass

    def tell(self):

        return 42

class MockFileIO(io.BytesIO):

    def __init__(self, data):

        self.read_history = []

        io.BytesIO.__init__(self, data)

    def read(self, n=None):

        res = io.BytesIO.read(self, n)

        self.read_history.append(None if res is None else len(res))

        return res

class MockNonBlockWriterIO(io.RawIOBase):

    def __init__(self, blocking_script):

        self._blocking_script = list(blocking_script)

        self._write_stack = []

    def write(self, b):

        self._write_stack.append(b[:])

        n = self._blocking_script.pop(0)

        if (n < 0):

            raise io.BlockingIOError(0, "test blocking", -n)

        else:

            return n

    def writable(self):

        return True

class IOTest(unittest.TestCase):

    def tearDown(self):

        test_support.unlink(test_support.TESTFN)

    def write_ops(self, f):

        self.assertEqual(f.write(b"blah."), 5)

        self.assertEqual(f.seek(0), 0)

        self.assertEqual(f.write(b"Hello."), 6)

        self.assertEqual(f.tell(), 6)

        self.assertEqual(f.seek(-1, 1), 5)

        self.assertEqual(f.tell(), 5)

        self.assertEqual(f.write(bytearray(b" world\n\n\n")), 9)

        self.assertEqual(f.seek(0), 0)

        self.assertEqual(f.write(b"h"), 1)

        self.assertEqual(f.seek(-1, 2), 13)

        self.assertEqual(f.tell(), 13)

        self.assertEqual(f.truncate(12), 12)

        self.assertEqual(f.tell(), 12)

        self.assertRaises(TypeError, f.seek, 0.0)

    def read_ops(self, f, buffered=False):

        data = f.read(5)

        self.assertEqual(data, b"hello")

        data = bytearray(data)

        self.assertEqual(f.readinto(data), 5)

        self.assertEqual(data, b" worl")

        self.assertEqual(f.readinto(data), 2)

        self.assertEqual(len(data), 5)

        self.assertEqual(data[:2], b"d\n")

        self.assertEqual(f.seek(0), 0)

        self.assertEqual(f.read(20), b"hello world\n")

        self.assertEqual(f.read(1), b"")

        self.assertEqual(f.readinto(bytearray(b"x")), 0)

        self.assertEqual(f.seek(-6, 2), 6)

        self.assertEqual(f.read(5), b"world")

        self.assertEqual(f.read(0), b"")

        self.assertEqual(f.readinto(bytearray()), 0)

        self.assertEqual(f.seek(-6, 1), 5)

        self.assertEqual(f.read(5), b" worl")

        self.assertEqual(f.tell(), 10)

        self.assertRaises(TypeError, f.seek, 0.0)

        if buffered:

            f.seek(0)

            self.assertEqual(f.read(), b"hello world\n")

            f.seek(6)

            self.assertEqual(f.read(), b"world\n")

            self.assertEqual(f.read(), b"")

    LARGE = 2**31

    def large_file_ops(self, f):

        assert f.readable()

        assert f.writable()

        self.assertEqual(f.seek(self.LARGE), self.LARGE)

        self.assertEqual(f.tell(), self.LARGE)

        self.assertEqual(f.write(b"xxx"), 3)

        self.assertEqual(f.tell(), self.LARGE + 3)

        self.assertEqual(f.seek(-1, 1), self.LARGE + 2)

        self.assertEqual(f.truncate(), self.LARGE + 2)

        self.assertEqual(f.tell(), self.LARGE + 2)

        self.assertEqual(f.seek(0, 2), self.LARGE + 2)

        self.assertEqual(f.truncate(self.LARGE + 1), self.LARGE + 1)

        self.assertEqual(f.tell(), self.LARGE + 1)

        self.assertEqual(f.seek(0, 2), self.LARGE + 1)

        self.assertEqual(f.seek(-1, 2), self.LARGE)

        self.assertEqual(f.read(2), b"x")

    def test_raw_file_io(self):

        f = io.open(test_support.TESTFN, "wb", buffering=0)

        self.assertEqual(f.readable(), False)

        self.assertEqual(f.writable(), True)

        self.assertEqual(f.seekable(), True)

        self.write_ops(f)

        f.close()

        f = io.open(test_support.TESTFN, "rb", buffering=0)

        self.assertEqual(f.readable(), True)

        self.assertEqual(f.writable(), False)

        self.assertEqual(f.seekable(), True)

        self.read_ops(f)

        f.close()

    def test_buffered_file_io(self):

        f = io.open(test_support.TESTFN, "wb")

        self.assertEqual(f.readable(), False)

        self.assertEqual(f.writable(), True)

        self.assertEqual(f.seekable(), True)

        self.write_ops(f)

        f.close()

        f = io.open(test_support.TESTFN, "rb")

        self.assertEqual(f.readable(), True)

        self.assertEqual(f.writable(), False)

        self.assertEqual(f.seekable(), True)

        self.read_ops(f, True)

        f.close()

    def test_readline(self):

        f = io.open(test_support.TESTFN, "wb")

        f.write(b"abc\ndef\nxyzzy\nfoo")

        f.close()

        f = io.open(test_support.TESTFN, "rb")

        self.assertEqual(f.readline(), b"abc\n")

        self.assertEqual(f.readline(10), b"def\n")

        self.assertEqual(f.readline(2), b"xy")

        self.assertEqual(f.readline(4), b"zzy\n")

        self.assertEqual(f.readline(), b"foo")

        f.close()

    def test_raw_bytes_io(self):

        f = io.BytesIO()

        self.write_ops(f)

        data = f.getvalue()

        self.assertEqual(data, b"hello world\n")

        f = io.BytesIO(data)

        self.read_ops(f, True)

    def test_large_file_ops(self):

        # On Windows and Mac OSX this test comsumes large resources; It takes

        # a long time to build the >2GB file and takes >2GB of disk space

        # therefore the resource must be enabled to run this test.

        if sys.platform[:3] in ('win', 'os2') or sys.platform == 'darwin':

            if not test_support.is_resource_enabled("largefile"):

                print("\nTesting large file ops skipped on %s." % sys.platform,

                      file=sys.stderr)

                print("It requires %d bytes and a long time." % self.LARGE,

                      file=sys.stderr)

                print("Use 'regrtest.py -u largefile test_io' to run it.",

                      file=sys.stderr)

                return

        f = io.open(test_support.TESTFN, "w+b", 0)

        self.large_file_ops(f)

        f.close()

        f = io.open(test_support.TESTFN, "w+b")

        self.large_file_ops(f)

        f.close()

    def test_with_open(self):

        for bufsize in (0, 1, 100):

            f = None

            with open(test_support.TESTFN, "wb", bufsize) as f:

                f.write(b"xxx")

            self.assertEqual(f.closed, True)

            f = None

            try:

                with open(test_support.TESTFN, "wb", bufsize) as f:

                    1/0

            except ZeroDivisionError:

                self.assertEqual(f.closed, True)

            else:

                self.fail("1/0 didn't raise an exception")

    def test_destructor(self):

        record = []

        class MyFileIO(io.FileIO):

            def __del__(self):

                record.append(1)

                io.FileIO.__del__(self)

            def close(self):

                record.append(2)

                io.FileIO.close(self)

            def flush(self):

                record.append(3)

                io.FileIO.flush(self)

        f = MyFileIO(test_support.TESTFN, "w")

        f.write("xxx")

        del f

        self.assertEqual(record, [1, 2, 3])

    def test_close_flushes(self):

        f = io.open(test_support.TESTFN, "wb")

        f.write(b"xxx")

        f.close()

        f = io.open(test_support.TESTFN, "rb")

        self.assertEqual(f.read(), b"xxx")

        f.close()

    def XXXtest_array_writes(self):

        # XXX memory view not available yet

        a = array.array('i', range(10))

        n = len(memoryview(a))

        f = io.open(test_support.TESTFN, "wb", 0)

        self.assertEqual(f.write(a), n)

        f.close()

        f = io.open(test_support.TESTFN, "wb")

        self.assertEqual(f.write(a), n)

        f.close()

    def test_closefd(self):

        self.assertRaises(ValueError, io.open, test_support.TESTFN, 'w',

                          closefd=False)

class MemorySeekTestMixin:

    def testInit(self):

        buf = self.buftype("1234567890")

        bytesIo = self.ioclass(buf)

    def testRead(self):

        buf = self.buftype("1234567890")

        bytesIo = self.ioclass(buf)

        self.assertEquals(buf[:1], bytesIo.read(1))

        self.assertEquals(buf[1:5], bytesIo.read(4))

        self.assertEquals(buf[5:], bytesIo.read(900))

        self.assertEquals(self.EOF, bytesIo.read())

    def testReadNoArgs(self):

        buf = self.buftype("1234567890")

        bytesIo = self.ioclass(buf)

        self.assertEquals(buf, bytesIo.read())

        self.assertEquals(self.EOF, bytesIo.read())

    def testSeek(self):

        buf = self.buftype("1234567890")

        bytesIo = self.ioclass(buf)

        bytesIo.read(5)

        bytesIo.seek(0)

        self.assertEquals(buf, bytesIo.read())

        bytesIo.seek(3)

        self.assertEquals(buf[3:], bytesIo.read())

        self.assertRaises(TypeError, bytesIo.seek, 0.0)

    def testTell(self):

        buf = self.buftype("1234567890")

        bytesIo = self.ioclass(buf)

        self.assertEquals(0, bytesIo.tell())

        bytesIo.seek(5)

        self.assertEquals(5, bytesIo.tell())

        bytesIo.seek(10000)

        self.assertEquals(10000, bytesIo.tell())

class BytesIOTest(MemorySeekTestMixin, unittest.TestCase):

    @staticmethod

    def buftype(s):

        return s.encode("utf-8")

    ioclass = io.BytesIO

    EOF = b""

class StringIOTest(MemorySeekTestMixin, unittest.TestCase):

    buftype = str

    ioclass = io.StringIO

    EOF = ""

class BufferedReaderTest(unittest.TestCase):

    def testRead(self):

        rawio = MockRawIO((b"abc", b"d", b"efg"))

        bufio = io.BufferedReader(rawio)

        self.assertEquals(b"abcdef", bufio.read(6))

    def testBuffering(self):

        data = b"abcdefghi"

        dlen = len(data)

        tests = [

            [ 100, [ 3, 1, 4, 8 ], [ dlen, 0 ] ],

            [ 100, [ 3, 3, 3],     [ dlen ]    ],

            [   4, [ 1, 2, 4, 2 ], [ 4, 4, 1 ] ],

        ]

        for bufsize, buf_read_sizes, raw_read_sizes in tests:

            rawio = MockFileIO(data)

            bufio = io.BufferedReader(rawio, buffer_size=bufsize)

            pos = 0

            for nbytes in buf_read_sizes:

                self.assertEquals(bufio.read(nbytes), data[pos:pos+nbytes])

                pos += nbytes

            self.assertEquals(rawio.read_history, raw_read_sizes)

    def testReadNonBlocking(self):

        # Inject some None's in there to simulate EWOULDBLOCK

        rawio = MockRawIO((b"abc", b"d", None, b"efg", None, None))

        bufio = io.BufferedReader(rawio)

        self.assertEquals(b"abcd", bufio.read(6))

        self.assertEquals(b"e", bufio.read(1))

        self.assertEquals(b"fg", bufio.read())

        self.assert_(None is bufio.read())

        self.assertEquals(b"", bufio.read())

    def testReadToEof(self):

        rawio = MockRawIO((b"abc", b"d", b"efg"))

        bufio = io.BufferedReader(rawio)

        self.assertEquals(b"abcdefg", bufio.read(9000))

    def testReadNoArgs(self):

        rawio = MockRawIO((b"abc", b"d", b"efg"))

        bufio = io.BufferedReader(rawio)

        self.assertEquals(b"abcdefg", bufio.read())

    def testFileno(self):

        rawio = MockRawIO((b"abc", b"d", b"efg"))

        bufio = io.BufferedReader(rawio)

        self.assertEquals(42, bufio.fileno())

    def testFilenoNoFileno(self):

        # XXX will we always have fileno() function? If so, kill

        # this test. Else, write it.

        pass

    def testThreads(self):

        try:

            # Write out many bytes with exactly the same number of 0's,

            # 1's... 255's. This will help us check that concurrent reading

            # doesn't duplicate or forget contents.

            N = 1000

            l = range(256) * N

            random.shuffle(l)

            s = bytes(bytearray(l))

            with io.open(test_support.TESTFN, "wb") as f:

                f.write(s)

            with io.open(test_support.TESTFN, "rb", buffering=0) as raw:

                bufio = io.BufferedReader(raw, 8)

                errors = []

                results = []

                def f():

                    try:

                        # Intra-buffer read then buffer-flushing read

                        for n in cycle([1, 19]):

                            s = bufio.read(n)

                            if not s:

                                break

                            # list.append() is atomic

                            results.append(s)

                    except Exception as e:

                        errors.append(e)

                        raise

                threads = [threading.Thread(target=f) for x in range(20)]

                for t in threads:

                    t.start()

                time.sleep(0.02) # yield

                for t in threads:

                    t.join()

                self.assertFalse(errors,

                    "the following exceptions were caught: %r" % errors)

                s = b''.join(results)

                for i in range(256):

                    c = bytes(bytearray([i]))

                    self.assertEqual(s.count(c), N)

        finally:

            test_support.unlink(test_support.TESTFN)

class BufferedWriterTest(unittest.TestCase):

    def testWrite(self):

        # Write to the buffered IO but don't overflow the buffer.

        writer = MockRawIO()

        bufio = io.BufferedWriter(writer, 8)

        bufio.write(b"abc")

        self.assertFalse(writer._write_stack)

    def testWriteOverflow(self):

        writer = MockRawIO()

        bufio = io.BufferedWriter(writer, 8)

        bufio.write(b"abc")

        bufio.write(b"defghijkl")

        self.assertEquals(b"abcdefghijkl", writer._write_stack[0])

    def testWriteNonBlocking(self):

        raw = MockNonBlockWriterIO((9, 2, 22, -6, 10, 12, 12))

        bufio = io.BufferedWriter(raw, 8, 16)

        bufio.write(b"asdf")

        bufio.write(b"asdfa")

        self.assertEquals(b"asdfasdfa", raw._write_stack[0])

        bufio.write(b"asdfasdfasdf")

        self.assertEquals(b"asdfasdfasdf", raw._write_stack[1])

        bufio.write(b"asdfasdfasdf")

        self.assertEquals(b"dfasdfasdf", raw._write_stack[2])

        self.assertEquals(b"asdfasdfasdf", raw._write_stack[3])

        bufio.write(b"asdfasdfasdf")

        # XXX I don't like this test. It relies too heavily on how the

        # algorithm actually works, which we might change. Refactor

        # later.

    def testFileno(self):

        rawio = MockRawIO((b"abc", b"d", b"efg"))

        bufio = io.BufferedWriter(rawio)

        self.assertEquals(42, bufio.fileno())

    def testFlush(self):

        writer = MockRawIO()

        bufio = io.BufferedWriter(writer, 8)

        bufio.write(b"abc")

        bufio.flush()

        self.assertEquals(b"abc", writer._write_stack[0])

    def testThreads(self):

        # BufferedWriter should not raise exceptions or crash

        # when called from multiple threads.

        try:

            # We use a real file object because it allows us to

            # exercise situations where the GIL is released before

            # writing the buffer to the raw streams. This is in addition

            # to concurrency issues due to switching threads in the middle

            # of Python code.

            with io.open(test_support.TESTFN, "wb", buffering=0) as raw:

                bufio = io.BufferedWriter(raw, 8)

                errors = []

                def f():

                    try:

                        # Write enough bytes to flush the buffer

                        s = b"a" * 19

                        for i in range(50):

                            bufio.write(s)

                    except Exception as e:

                        errors.append(e)

                        raise

                threads = [threading.Thread(target=f) for x in range(20)]

                for t in threads:

                    t.start()

                time.sleep(0.02) # yield

                for t in threads:

                    t.join()

                self.assertFalse(errors,

                    "the following exceptions were caught: %r" % errors)

        finally:

            test_support.unlink(test_support.TESTFN)

class BufferedRWPairTest(unittest.TestCase):