'''
Tests for fileinput module.
Nick Mathewson
'''
import os
import sys
import re
import fileinput
import collections
import builtins
import tempfile
import unittest

try:
    import bz2
except ImportError:
    bz2 = None
try:
    import gzip
except ImportError:
    gzip = None

from io import BytesIO, StringIO
from fileinput import FileInput, hook_encoded
from pathlib import Path

from test.support import verbose, TESTFN, check_warnings
from test.support import unlink as safe_unlink
from test import support
from unittest import mock


# The fileinput module has 2 interfaces: the FileInput class which does
# all the work, and a few functions (input, etc.) that use a global _state
# variable.

class BaseTests:
    # Write a content (str or bytes) to temp file, and return the
    # temp file's name.
    def writeTmp(self, content, *, mode='w'):  # opening in text mode is the default
        fd, name = tempfile.mkstemp()
        self.addCleanup(support.unlink, name)
        with open(fd, mode) as f:
            f.write(content)
        return name

class LineReader:

    def __init__(self):
        self._linesread = []

    @property
    def linesread(self):
        try:
            return self._linesread[:]
        finally:
            self._linesread = []

    def openhook(self, filename, mode):
        self.it = iter(filename.splitlines(True))
        return self

    def readline(self, size=None):
        line = next(self.it, '')
        self._linesread.append(line)
        return line

    def readlines(self, hint=-1):
        lines = []
        size = 0
        while True:
            line = self.readline()
            if not line:
                return lines
            lines.append(line)
            size += len(line)
            if size >= hint:
                return lines

    def close(self):
        pass

class BufferSizesTests(BaseTests, unittest.TestCase):
    def test_buffer_sizes(self):

        t1 = self.writeTmp(''.join("Line %s of file 1\n" % (i+1) for i in range(15)))
        t2 = self.writeTmp(''.join("Line %s of file 2\n" % (i+1) for i in range(10)))
        t3 = self.writeTmp(''.join("Line %s of file 3\n" % (i+1) for i in range(5)))
        t4 = self.writeTmp(''.join("Line %s of file 4\n" % (i+1) for i in range(1)))

        pat = re.compile(r'LINE (\d+) OF FILE (\d+)')

        if verbose:
            print('1. Simple iteration')
        fi = FileInput(files=(t1, t2, t3, t4))
        lines = list(fi)
        fi.close()
        self.assertEqual(len(lines), 31)
        self.assertEqual(lines[4], 'Line 5 of file 1\n')
        self.assertEqual(lines[30], 'Line 1 of file 4\n')
        self.assertEqual(fi.lineno(), 31)
        self.assertEqual(fi.filename(), t4)

        if verbose:
            print('2. Status variables')
        fi = FileInput(files=(t1, t2, t3, t4))
        s = "x"
        while s and s != 'Line 6 of file 2\n':
            s = fi.readline()
        self.assertEqual(fi.filename(), t2)
        self.assertEqual(fi.lineno(), 21)
        self.assertEqual(fi.filelineno(), 6)
        self.assertFalse(fi.isfirstline())
        self.assertFalse(fi.isstdin())

        if verbose:
            print('3. Nextfile')
        fi.nextfile()
        self.assertEqual(fi.readline(), 'Line 1 of file 3\n')
        self.assertEqual(fi.lineno(), 22)
        fi.close()

        if verbose:
            print('4. Stdin')
        fi = FileInput(files=(t1, t2, t3, t4, '-'))
        savestdin = sys.stdin
        try:
            sys.stdin = StringIO("Line 1 of stdin\nLine 2 of stdin\n")
            lines = list(fi)
            self.assertEqual(len(lines), 33)
            self.assertEqual(lines[32], 'Line 2 of stdin\n')
            self.assertEqual(fi.filename(), '<stdin>')
            fi.nextfile()
        finally:
            sys.stdin = savestdin

        if verbose:
            print('5. Boundary conditions')
        fi = FileInput(files=(t1, t2, t3, t4))
        self.assertEqual(fi.lineno(), 0)
        self.assertEqual(fi.filename(), None)
        fi.nextfile()
        self.assertEqual(fi.lineno(), 0)
        self.assertEqual(fi.filename(), None)

        if verbose:
            print('6. Inplace')
        savestdout = sys.stdout
        try:
            fi = FileInput(files=(t1, t2, t3, t4), inplace=1)
            for line in fi:
                line = line[:-1].upper()
                print(line)
            fi.close()
        finally:
            sys.stdout = savestdout

        fi = FileInput(files=(t1, t2, t3, t4))
        for line in fi:
            self.assertEqual(line[-1], '\n')
            m = pat.match(line[:-1])
            self.assertNotEqual(m, None)
            self.assertEqual(int(m.group(1)), fi.filelineno())
        fi.close()

class UnconditionallyRaise:
    def __init__(self, exception_type):
        self.exception_type = exception_type
        self.invoked = False
    def __call__(self, *args, **kwargs):
        self.invoked = True
        raise self.exception_type()

class FileInputTests(BaseTests, unittest.TestCase):

    def test_zero_byte_files(self):
        t1 = self.writeTmp("")
        t2 = self.writeTmp("")
        t3 = self.writeTmp("The only line there is.\n")
        t4 = self.writeTmp("")
        fi = FileInput(files=(t1, t2, t3, t4))

        line = fi.readline()
        self.assertEqual(line, 'The only line there is.\n')
        self.assertEqual(fi.lineno(), 1)
        self.assertEqual(fi.filelineno(), 1)
        self.assertEqual(fi.filename(), t3)

        line = fi.readline()
        self.assertFalse(line)
        self.assertEqual(fi.lineno(), 1)
        self.assertEqual(fi.filelineno(), 0)
        self.assertEqual(fi.filename(), t4)
        fi.close()

    def test_files_that_dont_end_with_newline(self):
        t1 = self.writeTmp("A\nB\nC")
        t2 = self.writeTmp("D\nE\nF")
        fi = FileInput(files=(t1, t2))
        lines = list(fi)
        self.assertEqual(lines, ["A\n", "B\n", "C", "D\n", "E\n", "F"])
        self.assertEqual(fi.filelineno(), 3)
        self.assertEqual(fi.lineno(), 6)

##     def test_unicode_filenames(self):
##         # XXX A unicode string is always returned by writeTmp.
##         #     So is this needed?
##         t1 = self.writeTmp("A\nB")
##         encoding = sys.getfilesystemencoding()
##         if encoding is None:
##             encoding = 'ascii'
##         fi = FileInput(files=str(t1, encoding))
##         lines = list(fi)
##         self.assertEqual(lines, ["A\n", "B"])

    def test_fileno(self):
        t1 = self.writeTmp("A\nB")
        t2 = self.writeTmp("C\nD")
        fi = FileInput(files=(t1, t2))
        self.assertEqual(fi.fileno(), -1)
        line = next(fi)
        self.assertNotEqual(fi.fileno(), -1)
        fi.nextfile()
        self.assertEqual(fi.fileno(), -1)
        line = list(fi)
        self.assertEqual(fi.fileno(), -1)

    def test_opening_mode(self):
        try:
            # invalid mode, should raise ValueError
            fi = FileInput(mode="w")
            self.fail("FileInput should reject invalid mode argument")
        except ValueError:
            pass
        # try opening in universal newline mode
        t1 = self.writeTmp(b"A\nB\r\nC\rD", mode="wb")
        with check_warnings(('', DeprecationWarning)):
            fi = FileInput(files=t1, mode="U")
        with check_warnings(('', DeprecationWarning)):
            lines = list(fi)
        self.assertEqual(lines, ["A\n", "B\n", "C\n", "D"])

    def test_stdin_binary_mode(self):
        with mock.patch('sys.stdin') as m_stdin:
            m_stdin.buffer = BytesIO(b'spam, bacon, sausage, and spam')
            fi = FileInput(files=['-'], mode='rb')
            lines = list(fi)
            self.assertEqual(lines, [b'spam, bacon, sausage, and spam'])

    def test_detached_stdin_binary_mode(self):
        orig_stdin = sys.stdin
        try:
            sys.stdin = BytesIO(b'spam, bacon, sausage, and spam')
            self.assertFalse(hasattr(sys.stdin, 'buffer'))
            fi = FileInput(files=['-'], mode='rb')
            lines = list(fi)
            self.assertEqual(lines, [b'spam, bacon, sausage, and spam'])
        finally:
            sys.stdin = orig_stdin

    def test_file_opening_hook(self):
        try:
            # cannot use openhook and inplace mode
            fi = FileInput(inplace=1, openhook=lambda f, m: None)
            self.fail("FileInput should raise if both inplace "
                             "and openhook arguments are given")
        except ValueError:
            pass
        try:
            fi = FileInput(openhook=1)
            self.fail("FileInput should check openhook for being callable")
        except ValueError:
            pass

        class CustomOpenHook:
            def __init__(self):
                self.invoked = False
            def __call__(self, *args):
                self.invoked = True
                return open(*args)

        t = self.writeTmp("\n")
        custom_open_hook = CustomOpenHook()
        with FileInput([t], openhook=custom_open_hook) as fi:
            fi.readline()
        self.assertTrue(custom_open_hook.invoked, "openhook not invoked")

    def test_readline(self):
        with open(TESTFN, 'wb') as f:
            f.write(b'A\nB\r\nC\r')
            # Fill TextIOWrapper buffer.
            f.write(b'123456789\n' * 1000)
            # Issue #20501: readline() shouldn't read whole file.
            f.write(b'\x80')
        self.addCleanup(safe_unlink, TESTFN)

        with FileInput(files=TESTFN,
                       openhook=hook_encoded('ascii')) as fi:
            try:
                self.assertEqual(fi.readline(), 'A\n')
                self.assertEqual(fi.readline(), 'B\n')
                self.assertEqual(fi.readline(), 'C\n')
            except UnicodeDecodeError:
                self.fail('Read to end of file')
            with self.assertRaises(UnicodeDecodeError):
                # Read to the end of file.
                list(fi)
            self.assertEqual(fi.readline(), '')
            self.assertEqual(fi.readline(), '')

    def test_readline_binary_mode(self):
        with open(TESTFN, 'wb') as f:
            f.write(b'A\nB\r\nC\rD')
        self.addCleanup(safe_unlink, TESTFN)

        with FileInput(files=TESTFN, mode='rb') as fi:
            self.assertEqual(fi.readline(), b'A\n')
            self.assertEqual(fi.readline(), b'B\r\n')
            self.assertEqual(fi.readline(), b'C\rD')
            # Read to the end of file.
            self.assertEqual(fi.readline(), b'')
            self.assertEqual(fi.readline(), b'')

    def test_inplace_binary_write_mode(self):
        temp_file = self.writeTmp(b'Initial text.', mode='wb')
        with FileInput(temp_file, mode='rb', inplace=True) as fobj:
            line = fobj.readline()
            self.assertEqual(line, b'Initial text.')
            # print() cannot be used with files opened in binary mode.
            sys.stdout.write(b'New line.')
        with open(temp_file, 'rb') as f:
            self.assertEqual(f.read(), b'New line.')

    def test_context_manager(self):
        t1 = self.writeTmp("A\nB\nC")
        t2 = self.writeTmp("D\nE\nF")
        with FileInput(files=(t1, t2)) as fi:
            lines = list(fi)
        self.assertEqual(lines, ["A\n", "B\n", "C", "D\n", "E\n", "F"])
        self.assertEqual(fi.filelineno(), 3)
        self.assertEqual(fi.lineno(), 6)
        self.assertEqual(fi._files, ())

    def test_close_on_exception(self):
        t1 = self.writeTmp("")