tmux-amq/tests/test_ptytap_integration.py

import os
import shutil
import sys
import time
from pathlib import Path
from uuid import uuid4

import pytest

from tamq.tmux import LaunchPlan, TmuxManager


@pytest.mark.skipif(shutil.which("tmux") is None, reason="tmux is not installed")
def test_tap_propagates_terminal_size_resize_and_raw_mouse_input(tmp_path, monkeypatch):
    repo = tmp_path / "activity-core"
    repo.mkdir()
    fixture = tmp_path / "terminal_fixture.py"
    fixture.write_text(
        "import os, signal, tty\n"
        "def size(prefix):\n"
        "    value = os.get_terminal_size(0)\n"
        "    print(f'{prefix} {value.lines}x{value.columns}', flush=True)\n"
        "def resized(signum, frame): size('RESIZE')\n"
        "signal.signal(signal.SIGWINCH, resized)\n"
        "size('INITIAL')\n"
        "tty.setraw(0)\n"
        "print('READY', flush=True)\n"
        "data = b''\n"
        "while not data.endswith(b'M'):\n"
        "    data += os.read(0, 1024)\n"
        "print(f'INPUT {data.hex()}', flush=True)\n"
        "while True: signal.pause()\n",
        encoding="utf-8",
    )
    project_src = str(Path(__file__).resolve().parents[1] / "src")
    existing_pythonpath = os.environ.get("PYTHONPATH")
    monkeypatch.setenv(
        "PYTHONPATH",
        project_src if not existing_pythonpath else f"{project_src}:{existing_pythonpath}",
    )
    monkeypatch.setenv("TAMQ_STATE_DIR", str(tmp_path / "state"))

    socket_name = f"tamq-pty-{os.getpid()}-{uuid4().hex[:8]}"
    session = f"tamq-pty-{uuid4().hex[:8]}"
    manager = TmuxManager(
        session,
        tmux_command=("tmux", "-L", socket_name),
        tamq_command=(sys.executable, "-m", "tamq.cli"),
    )
    plan = LaunchPlan(
        ("activity-core",),
        {"activity-core": str(repo)},
        (sys.executable, str(fixture)),
    )

    def capture() -> str:
        return manager._run("capture-pane", "-p", "-t", f"{session}:activity-core")

    def wait_for(marker: str) -> str:
        deadline = time.monotonic() + 5
        output = ""
        while time.monotonic() < deadline:
            output = capture()
            if marker in output:
                return output
            time.sleep(0.05)
        pytest.fail(f"timed out waiting for {marker!r}; pane contained:\n{output}")

    try:
        manager.ensure_plan(plan, tap=True)
        output = wait_for("READY")
        dimensions = manager._run(
            "display-message",
            "-p",
            "-t",
            f"{session}:activity-core",
            "#{pane_height}x#{pane_width}",
        )
        assert f"INITIAL {dimensions}" in output

        mouse_sequence = "\x1b[<64;10;5M"
        manager._run(
            "send-keys", "-t", f"{session}:activity-core", "-l", "--", mouse_sequence
        )
        output = wait_for(f"INPUT {mouse_sequence.encode().hex()}")
        assert f"INPUT {mouse_sequence.encode().hex()}" in output

        manager._run("resize-window", "-t", session, "-x", "100", "-y", "40")
        resized = manager._run(
            "display-message",
            "-p",
            "-t",
            f"{session}:activity-core",
            "#{pane_height}x#{pane_width}",
        )
        output = wait_for(f"RESIZE {resized}")
        assert f"RESIZE {resized}" in output
    finally:
        manager._run("kill-server", check=False)