#!/usr/bin/env python3
from __future__ import annotations

import fcntl
import os
import pty
import pwd
import select
import signal
import struct
import subprocess
import sys
import tempfile
import termios
import time
import unittest
from unittest.mock import patch

_LIB = os.path.join(os.path.dirname(os.path.abspath(__file__)), "..", "lib")
if _LIB not in sys.path:
    sys.path.insert(0, _LIB)

import seat_tmux_mediator as mediator


class SeatTmuxAttachPtyTests(unittest.TestCase):
    def test_live_io_resize_and_detach_through_mediator_exec(self) -> None:
        with tempfile.TemporaryDirectory() as tmp:
            socket_path = os.path.join(tmp, "tmux.sock")
            pane = os.path.join(tmp, "pane.py")
            with open(pane, "w", encoding="utf-8") as output:
                output.write(
                    "#!/usr/bin/env python3\n"
                    "import os, signal, sys\n"
                    "def resized(_signal, _frame):\n"
                    "    size = os.get_terminal_size()\n"
                    "    print(f'{size.lines} {size.columns}\\nRESIZED', flush=True)\n"
                    "signal.signal(signal.SIGWINCH, resized)\n"
                    "print('READY', flush=True)\n"
                    "line = sys.stdin.readline().rstrip('\\n')\n"
                    "print(f'INPUT:{line}', flush=True)\n"
                    "while True:\n"
                    "    signal.pause()\n"
                )
            os.chmod(pane, 0o755)
            subprocess.run(["tmux", "-S", socket_path, "new-session", "-d", "-s", "main", pane], check=True)
            binding = mediator.bind_socket_ancestry(socket_path)
            pid, master = pty.fork()
            if pid == 0:
                account = pwd.getpwuid(os.getuid())
                with patch("seat_tmux_mediator.pwd.getpwnam", return_value=account), patch(
                    "seat_tmux_mediator.os.setgroups"
                ), patch("seat_tmux_mediator.os.setgid"), patch("seat_tmux_mediator.os.setuid"):
                    mediator.exec_tmux_attach(account.pw_name, binding, ["attach-session", "-t", "main"])
                os._exit(91)
            binding.close()
            transcript = bytearray()
            try:
                self._read_until(master, transcript, b"READY", 5)
                os.write(master, b"hello-authority\n")
                self._read_until(master, transcript, b"INPUT:hello-authority", 5)
                fcntl.ioctl(master, termios.TIOCSWINSZ, struct.pack("HHHH", 41, 109, 0, 0))
                os.kill(pid, signal.SIGWINCH)
                self._read_until(master, transcript, b"40 109", 5)
                self._read_until(master, transcript, b"RESIZED", 5)
                os.write(master, b"\x02d")
                deadline = time.monotonic() + 5
                while time.monotonic() < deadline:
                    done, status = os.waitpid(pid, os.WNOHANG)
                    if done:
                        self.assertTrue(os.WIFEXITED(status))
                        self.assertEqual(os.WEXITSTATUS(status), 0)
                        break
                    time.sleep(0.02)
                else:
                    self.fail("attach did not detach")
                alive = subprocess.run(["tmux", "-S", socket_path, "has-session", "-t", "main"], check=False)
                self.assertEqual(alive.returncode, 0)
            finally:
                try:
                    os.close(master)
                except OSError:
                    pass
                try:
                    os.kill(pid, signal.SIGKILL)
                except ProcessLookupError:
                    pass
                try:
                    os.waitpid(pid, 0)
                except ChildProcessError:
                    pass
                subprocess.run(["tmux", "-S", socket_path, "kill-server"], check=False,
                               stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)

    @staticmethod
    def _read_until(fd: int, transcript: bytearray, needle: bytes, timeout: float) -> None:
        deadline = time.monotonic() + timeout
        while needle not in transcript:
            remaining = deadline - time.monotonic()
            if remaining <= 0:
                raise AssertionError(f"missing {needle!r} in {bytes(transcript)!r}")
            ready, _, _ = select.select([fd], [], [], remaining)
            if ready:
                try:
                    transcript.extend(os.read(fd, 4096))
                except OSError as error:
                    if error.errno != 5:
                        raise
                    raise AssertionError(f"PTY closed before {needle!r}: {bytes(transcript)!r}") from error


if __name__ == "__main__":
    unittest.main()