import errno
import itertools
import os
import select
import signal
import sys
import threading
import unittest
from functools import partial
from _colorize import ANSIColors
from test.support import force_color, os_helper, force_not_colorized_test_class
from test.support import is_android, is_apple_mobile, is_wasm32
from test.support import threading_helper

from unittest import TestCase
from unittest.mock import MagicMock, call, patch, ANY, Mock

from .support import handle_all_events, code_to_events, more_lines

try:
    from _pyrepl.console import Event
    from _pyrepl.unix_console import UnixConsole
except ImportError:
    pass

from _pyrepl.terminfo import _TERMINAL_CAPABILITIES

TERM_CAPABILITIES = _TERMINAL_CAPABILITIES["ansi"]


def unix_console(events, **kwargs):
    console = UnixConsole(term="xterm")
    console.get_event = MagicMock(side_effect=events)
    console.getpending = MagicMock(return_value=Event("key", ""))

    height = kwargs.get("height", 25)
    width = kwargs.get("width", 80)
    console.getheightwidth = MagicMock(side_effect=lambda: (height, width))
    console.wait = MagicMock()

    console.prepare()
    for key, val in kwargs.items():
        setattr(console, key, val)
    return console


handle_events_unix_console = partial(
    handle_all_events,
    prepare_console=unix_console,
)
handle_events_narrow_unix_console = partial(
    handle_all_events,
    prepare_console=partial(unix_console, width=5),
)
handle_events_short_unix_console = partial(
    handle_all_events,
    prepare_console=partial(unix_console, height=1),
)
handle_events_unix_console_height_3 = partial(
    handle_all_events, prepare_console=partial(unix_console, height=3)
)


@unittest.skipIf(sys.platform == "win32", "No Unix event queue on Windows")
@patch(
    "_pyrepl.terminfo.tparm",
    lambda s, *args: s + b":" + b",".join(str(i).encode() for i in args),
)
@patch(
    "termios.tcgetattr",
    lambda _: [
        27394,
        3,
        19200,
        536872399,
        38400,
        38400,
        [
            b"\x04",
            b"\xff",
            b"\xff",
            b"\x7f",
            b"\x17",
            b"\x15",
            b"\x12",
            b"\x00",
            b"\x03",
            b"\x1c",
            b"\x1a",
            b"\x19",
            b"\x11",
            b"\x13",
            b"\x16",
            b"\x0f",
            b"\x01",
            b"\x00",
            b"\x14",
            b"\x00",
        ],
    ],
)
@patch("termios.tcsetattr", lambda a, b, c: None)
@patch("os.write")
@force_not_colorized_test_class
class TestConsole(TestCase):
    @staticmethod
    def _prepare_reader_with_prompts(console, **kwargs):
        from _pyrepl.readline import ReadlineAlikeReader, ReadlineConfig

        config = ReadlineConfig(
            readline_completer=kwargs.pop("readline_completer", None)
        )
        reader = ReadlineAlikeReader(console=console, config=config)
        reader.paste_mode = False
        for key, val in kwargs.items():
            setattr(reader, key, val)
        return reader

    def test_colorized_multiline_typing_does_not_redraw_previous_line(self, _os_write):
        def prepare_reader_with_prompts(console, **kwargs):
            reader = self._prepare_reader_with_prompts(console, **kwargs)
            reader.more_lines = partial(more_lines, namespace=None)
            return reader

        with force_color(True):
            events = itertools.chain(
                code_to_events("def foo():"),
                [Event(evt="key", data="\n", raw=bytearray(b"\n"))],
                code_to_events("x = 1"),
                [Event(evt="key", data="\n", raw=bytearray(b"\n"))],
                code_to_events("y"),
            )
            _, con = handle_all_events(
                events,
                prepare_console=unix_console,
                prepare_reader=prepare_reader_with_prompts,
            )
            con.restore()

        self.assertNotIn(
            call(ANY, b" \x1b[0m    x \x1b[0m=\x1b[0m "),
            _os_write.mock_calls,
        )
        self.assertIn(call(ANY, b"y"), _os_write.mock_calls)

    def test_no_newline(self, _os_write):
        code = "1"
        events = code_to_events(code)
        _, con = handle_events_unix_console(events)
        self.assertNotIn(call(ANY, b'\n'), _os_write.mock_calls)
        con.restore()

    def test_reset_on_finish(self, _os_write):
        # gh-152068: finish() must emit the ANSI reset sequence so any
        # active color does not leak past the prompt.
        code = "1"
        events = code_to_events(code)
        _, con = handle_events_unix_console(events)
        con.finish()
        _os_write.assert_any_call(ANY, ANSIColors.RESET.encode(con.encoding))
        con.restore()

    def test_reset_on_restore(self, _os_write):
        # gh-152068: restore() must emit the ANSI reset sequence.
        code = "1"
        events = code_to_events(code)
        _, con = handle_events_unix_console(events)
        con.restore()
        _os_write.assert_any_call(ANY, ANSIColors.RESET.encode(con.encoding))

    def test_newline(self, _os_write):
        code = "\n"
        events = code_to_events(code)
        _, con = handle_events_unix_console(events)
        _os_write.assert_any_call(ANY, b"\n")
        con.restore()

    def test_simple_addition(self, _os_write):
        code = "12+34"
        events = code_to_events(code)
        _, con = handle_events_unix_console(events)
        _os_write.assert_any_call(ANY, b"1")
        _os_write.assert_any_call(ANY, b"2")
        _os_write.assert_any_call(ANY, b"+")
        _os_write.assert_any_call(ANY, b"3")
        _os_write.assert_any_call(ANY, b"4")
        con.restore()

    def test_wrap(self, _os_write):
        code = "12+34"
        events = code_to_events(code)
        _, con = handle_events_narrow_unix_console(events)
        _os_write.assert_any_call(ANY, b"1")
        _os_write.assert_any_call(ANY, b"2")
        _os_write.assert_any_call(ANY, b"+")
        _os_write.assert_any_call(ANY, b"3")
        _os_write.assert_any_call(ANY, b"\\")
        _os_write.assert_any_call(ANY, b"\n")
        _os_write.assert_any_call(ANY, b"4")
        con.restore()

    def test_cursor_left(self, _os_write):
        code = "1"
        events = itertools.chain(
            code_to_events(code),
            [Event(evt="key", data="left", raw=bytearray(b"\x1bOD"))],
        )
        _, con = handle_events_unix_console(events)
        _os_write.assert_any_call(ANY, TERM_CAPABILITIES["cub"] + b":1")
        con.restore()

    def test_cursor_left_right(self, _os_write):
        code = "1"
        events = itertools.chain(
            code_to_events(code),
            [
                Event(evt="key", data="left", raw=bytearray(b"\x1bOD")),
                Event(evt="key", data="right", raw=bytearray(b"\x1bOC")),
            ],
        )
        _, con = handle_events_unix_console(events)
        _os_write.assert_any_call(ANY, TERM_CAPABILITIES["cub"] + b":1")
        _os_write.assert_any_call(ANY, TERM_CAPABILITIES["cuf"] + b":1")
        con.restore()

    def test_cursor_up(self, _os_write):
        code = "1\n2+3"
        events = itertools.chain(
            code_to_events(code),
            [Event(evt="key", data="up", raw=bytearray(b"\x1bOA"))],
        )
        _, con = handle_events_unix_console(events)
        _os_write.assert_any_call(ANY, TERM_CAPABILITIES["cuu"] + b":1")
        con.restore()

    def test_cursor_up_down(self, _os_write):
        code = "1\n2+3"
        events = itertools.chain(
            code_to_events(code),
            [
                Event(evt="key", data="up", raw=bytearray(b"\x1bOA")),
                Event(evt="key", data="down", raw=bytearray(b"\x1bOB")),
            ],
        )
        _, con = handle_events_unix_console(events)
        _os_write.assert_any_call(ANY, TERM_CAPABILITIES["cuu"] + b":1")
        _os_write.assert_any_call(ANY, TERM_CAPABILITIES["cud"] + b":1")
        con.restore()

    @unittest.expectedFailureIf(sys.platform == "android", "TODO: RUSTPYTHON; TypeError: signal handler must be signal.SIG_IGN, signal.SIG_DFL, or a callable object")
    def test_cursor_back_write(self, _os_write):
        events = itertools.chain(
            code_to_events("1"),
            [Event(evt="key", data="left", raw=bytearray(b"\x1bOD"))],
            code_to_events("2"),
        )
        _, con = handle_events_unix_console(events)
        _os_write.assert_any_call(ANY, b"1")
        _os_write.assert_any_call(ANY, TERM_CAPABILITIES["cub"] + b":1")
        _os_write.assert_any_call(ANY, b"2")
        con.restore()

    def test_multiline_function_move_up_short_terminal(self, _os_write):
        # fmt: off
        code = (
            "def f():\n"
            "  foo"
        )
        # fmt: on

        events = itertools.chain(
            code_to_events(code),
            [
                Event(evt="key", data="up", raw=bytearray(b"\x1bOA")),
                Event(evt="scroll", data=None),
            ],
        )
        _, con = handle_events_short_unix_console(events)
        _os_write.assert_any_call(ANY, TERM_CAPABILITIES["ri"] + b":")
        con.restore()

    def test_multiline_function_move_up_down_short_terminal(self, _os_write):
        # fmt: off
        code = (
            "def f():\n"
            "  foo"
        )
        # fmt: on

        events = itertools.chain(
            code_to_events(code),
            [
                Event(evt="key", data="up", raw=bytearray(b"\x1bOA")),
                Event(evt="scroll", data=None),
                Event(evt="key", data="down", raw=bytearray(b"\x1bOB")),
                Event(evt="scroll", data=None),
            ],
        )
        _, con = handle_events_short_unix_console(events)
        _os_write.assert_any_call(ANY, TERM_CAPABILITIES["ri"] + b":")
        _os_write.assert_any_call(ANY, TERM_CAPABILITIES["ind"] + b":")
        con.restore()

    def test_resize_bigger_on_multiline_function(self, _os_write):
        # fmt: off
        code = (
            "def f():\n"
            "  foo"
        )
        # fmt: on

        events = itertools.chain(code_to_events(code))
        reader, console = handle_events_short_unix_console(events)

        console.getheightwidth = MagicMock(side_effect=lambda: (2, 80))

        def same_reader(_):
            return reader

        def same_console(events):
            console.get_event = MagicMock(side_effect=events)
            return console

        _, con = handle_all_events(
            [Event(evt="resize", data=None)],
            prepare_reader=same_reader,
            prepare_console=same_console,
        )
        _os_write.assert_has_calls(
            [
                call(ANY, TERM_CAPABILITIES["ri"] + b":"),
                call(ANY, TERM_CAPABILITIES["cup"] + b":0,0"),
                call(ANY, b"def f():"),
            ]
        )
        console.restore()
        con.restore()

    def test_resize_smaller_on_multiline_function(self, _os_write):
        # fmt: off
        code = (
            "def f():\n"
            "  foo"
        )
        # fmt: on

        events = itertools.chain(code_to_events(code))
        reader, console = handle_events_unix_console_height_3(events)

        console.getheightwidth = MagicMock(side_effect=lambda: (1, 80))

        def same_reader(_):
            return reader

        def same_console(events):
            console.get_event = MagicMock(side_effect=events)
            return console

        _, con = handle_all_events(
            [Event(evt="resize", data=None)],
            prepare_reader=same_reader,
            prepare_console=same_console,
        )
        _os_write.assert_has_calls(
            [
                call(ANY, TERM_CAPABILITIES["ind"] + b":"),
                call(ANY, TERM_CAPABILITIES["cup"] + b":0,0"),
                call(ANY, b"  foo"),
            ]
        )
        console.restore()
        con.restore()

    def test_getheightwidth_with_invalid_environ(self, _os_write):
        # gh-128636
        console = UnixConsole(term="xterm")
        with os_helper.EnvironmentVarGuard() as env:
            env["LINES"] = ""
            self.assertIsInstance(console.getheightwidth(), tuple)
            env["COLUMNS"] = ""
            self.assertIsInstance(console.getheightwidth(), tuple)
            os.environ = []
            self.assertIsInstance(console.getheightwidth(), tuple)

    @unittest.skipUnless(sys.platform == "darwin", "requires macOS")
    def test_restore_with_invalid_environ_on_macos(self, _os_write):
        # gh-128636 for macOS
        console = UnixConsole(term="xterm")
        with os_helper.EnvironmentVarGuard():
            os.environ = []
            console.prepare()  # needed to call restore()
            console.restore()  # this should succeed

    @threading_helper.reap_threads
    @threading_helper.requires_working_threading()
    def test_restore_in_thread(self, _os_write):
        # gh-139391: ensure that console.restore() silently suppresses
        # exceptions when calling signal.signal() from a non-main thread.
        console = unix_console([])
        console.old_sigwinch = signal.SIG_DFL
        thread = threading.Thread(target=console.restore)
        thread.start()
        thread.join()  # this should not raise


@unittest.skipIf(sys.platform == "win32", "No Unix console on Windows")
class TestUnixConsoleEIOHandling(TestCase):

    @patch('_pyrepl.unix_console.tcsetattr')
    @patch('_pyrepl.unix_console.tcgetattr')
    def test_eio_error_handling_in_restore(self, mock_tcgetattr, mock_tcsetattr):

        import termios
        mock_termios = Mock()
        mock_termios.iflag = 0
        mock_termios.oflag = 0
        mock_termios.cflag = 0
        mock_termios.lflag = 0
        mock_termios.cc = [0] * 32
        mock_termios.copy.return_value = mock_termios
        mock_tcgetattr.return_value = mock_termios

        console = UnixConsole(term="xterm")
        console.prepare()

        mock_tcsetattr.side_effect = termios.error(errno.EIO, "Input/output error")

        # EIO error should be handled gracefully in restore()
        console.restore()


try:
    import pty
    import termios as _termios
except ImportError:
    pty = None


@unittest.skipIf(sys.platform == "win32", "No Unix console on Windows")
@unittest.skipUnless(pty, "requires pty")
@unittest.skipIf(is_android or is_apple_mobile or is_wasm32,
                 "pty is not available on this platform")
class TestUnixConsoleInputHook(TestCase):
    # gh-152907: the console must restore cooked output (OPOST) around
    # input-hook calls, then re-enter raw mode.

    def test_input_hook_output_is_cooked(self):
        master_fd, slave_fd = pty.openpty()
        self.addCleanup(os.close, master_fd)

        # tcsetattr(TCSADRAIN) blocks on some platforms (e.g. macOS) while the
        # master still holds unread output, so empty it before each mode switch.
        def drain():
            out = b""
            while select.select([master_fd], [], [], 0)[0]:
                try:
                    data = os.read(master_fd, 4096)
                except OSError:
                    break
                if not data:
                    break
                out += data
            return out

        # Start from a cooked terminal so there are saved flags to restore.
        attr = _termios.tcgetattr(slave_fd)
        attr[1] |= _termios.OPOST | _termios.ONLCR
        _termios.tcsetattr(slave_fd, _termios.TCSANOW, attr)

        console = UnixConsole(slave_fd, slave_fd, term="xterm")
        console.prepare()
        try:
            drain()  # discard prepare()'s own setup sequences
            # pyrepl's own rendering runs with OPOST cleared.
            self.assertFalse(_termios.tcgetattr(slave_fd)[1] & _termios.OPOST)

            observed = {}

            def fake_hook():
                observed["oflag"] = _termios.tcgetattr(slave_fd)[1]
                os.write(slave_fd, b"line1\nline2\n")
                observed["output"] = drain()
                return 0

            with patch("_pyrepl.unix_console.posix") as mock_posix:
                mock_posix._is_inputhook_installed.return_value = True
                mock_posix._inputhook.side_effect = fake_hook
                hook = console.input_hook
                self.assertIsNotNone(hook)
                self.assertEqual(hook(), 0)

            # The hook ran with cooked output (OPOST on)...
            self.assertTrue(observed["oflag"] & _termios.OPOST)
            # ...and raw mode was restored afterwards.
            self.assertFalse(_termios.tcgetattr(slave_fd)[1] & _termios.OPOST)
            # The tty translated the hook's bare '\n' into '\r\n'.
            self.assertEqual(observed["output"], b"line1\r\nline2\r\n")
        finally:
            # restore() writes and only then switches modes, so there is no
            # point left to drain from here; keep the master empty elsewhere.
            stop = threading.Event()

            def pump():
                while not stop.is_set():
                    if select.select([master_fd], [], [], 0.05)[0]:
                        try:
                            if not os.read(master_fd, 4096):
                                break
                        except OSError:
                            break

            pump_thread = threading.Thread(target=pump)
            pump_thread.start()
            try:
                console.restore()
            finally:
                stop.set()
                pump_thread.join()
                os.close(slave_fd)