"""Exercise the installed dashboard PTY reattachment contract without model calls."""

from __future__ import annotations

import importlib.util
import os
from pathlib import Path
import signal
from types import SimpleNamespace
import unittest
from unittest.mock import Mock, patch


ROOT = Path(os.environ.get("HERMES_SOURCE_ROOT", "/opt/hermes"))


def load_source(name: str):
    """Load one installed upstream module without starting the dashboard."""
    spec = importlib.util.spec_from_file_location(name, ROOT / "hermes_cli" / f"{name}.py")
    module = importlib.util.module_from_spec(spec)
    spec.loader.exec_module(module)
    return module


bridge_module = load_source("pty_bridge")
session_module = load_source("pty_session")


class DashboardRedrawTests(unittest.TestCase):
    """Only the opted-in, living dashboard child receives a redraw signal."""

    def test_spawn_requires_both_dashboard_and_alternate_mode(self):
        for env, expected in (
            ({}, False),
            ({"HERMES_TUI_DASHBOARD": "1"}, False),
            ({"HERMES_TUI_INLINE": "0"}, False),
            ({"HERMES_TUI_DASHBOARD": "1", "HERMES_TUI_INLINE": "1"}, False),
            ({"HERMES_TUI_DASHBOARD": "1", "HERMES_TUI_INLINE": "0"}, True),
        ):
            with self.subTest(env=env):
                proc = SimpleNamespace(fd=12, pid=321, isalive=lambda: True)
                spawn = Mock(return_value=proc)
                backend = SimpleNamespace(PtyProcess=SimpleNamespace(spawn=spawn))
                with patch.object(bridge_module, "ptyprocess", backend), patch.object(
                    bridge_module, "_PTY_AVAILABLE", True
                ), patch.object(bridge_module.os, "kill") as kill, patch.object(
                    bridge_module.os, "write"
                ) as write:
                    bridge = bridge_module.PtyBridge.spawn(["node", "entry.js"], env=env)
                    bridge.redraw_after_attach()
                    self.assertEqual(kill.call_count, int(expected))
                    if expected:
                        kill.assert_called_once_with(321, signal.SIGCONT)
                    write.assert_not_called()

    def test_dead_or_closed_children_are_not_signalled(self):
        for alive, closed in ((False, False), (True, True)):
            proc = SimpleNamespace(fd=12, pid=321, isalive=lambda: alive)
            bridge = bridge_module.PtyBridge(proc)
            bridge._dashboard_redraw_on_attach = True
            bridge._closed = closed
            with patch.object(bridge_module.os, "kill") as kill:
                bridge.redraw_after_attach()
                kill.assert_not_called()

    def test_child_exit_race_does_not_break_attachment(self):
        proc = SimpleNamespace(fd=12, pid=321, isalive=lambda: True)
        bridge = bridge_module.PtyBridge(proc)
        bridge._dashboard_redraw_on_attach = True
        with patch.object(bridge_module.os, "kill", side_effect=ProcessLookupError):
            bridge.redraw_after_attach()


class ReplayTests(unittest.IsolatedAsyncioTestCase):
    """Reattachment restores the terminal after bounded replay completes."""

    async def test_truncated_replay_is_followed_by_redraw_without_input(self):
        events = []
        bridge = SimpleNamespace(redraw_after_attach=lambda: events.append("redraw"))
        session = session_module.PtySession("test", bridge, buffer_cap=8, read_timeout=.1)

        class Socket:
            async def send_bytes(self, data):
                events.append(data)

            async def close(self, code):
                events.append(code)

        first = Socket()
        await session.attach(first)
        self.assertEqual(events, [])
        session.detach(first)
        session.buffer.append(b"\x1b[?1049h" + b"x" * 64)
        self.assertTrue(session.buffer.truncated)
        self.assertEqual(session.buffer.snapshot(), b"x" * 8)
        second = Socket()
        await session.attach(second)
        self.assertEqual(events, [b"x" * 8, "redraw"])
        self.assertTrue(session.attached)
        # A superseded socket must not detach the active viewer.
        session.detach(first)
        self.assertTrue(session.attached)

    async def test_legacy_bridges_still_reattach(self):
        session = session_module.PtySession("test", object(), buffer_cap=8, read_timeout=.1)
        first = SimpleNamespace()
        await session.attach(first)
        session.detach(first)
        await session.attach(SimpleNamespace())
        self.assertTrue(session.attached)


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