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
from tamq.store import Store


@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"),
        command_dir=tmp_path / "commands",
    )
    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)


@pytest.mark.skipif(shutil.which("tmux") is None, reason="tmux is not installed")
def test_real_pty_attributes_worker_output_and_suppresses_operator_echo(
    tmp_path, monkeypatch
):
    source = tmp_path / "source"
    source.mkdir()
    fixture = tmp_path / "duplex_fixture.py"
    fixture.write_text(
        "import sys\n"
        "print('READY', flush=True)\n"
        "for line in sys.stdin:\n"
        "    print(line.rstrip('\\r\\n'), flush=True)\n"
        "    print('\\x1b[5;1H• tO:target: worker reply\\x1b[6;1H', flush=True)\n",
        encoding="utf-8",
    )
    bin_dir = tmp_path / "bin"
    bin_dir.mkdir()
    gita = bin_dir / "gita"
    gita.write_text(
        "#!/bin/sh\n"
        "if [ \"$1\" = ls ]; then printf 'source target\\n'; exit 0; fi\n"
        "exit 2\n",
        encoding="utf-8",
    )
    gita.chmod(0o755)
    project_src = str(Path(__file__).resolve().parents[1] / "src")
    monkeypatch.setenv("PYTHONPATH", project_src)
    monkeypatch.setenv("PATH", f"{bin_dir}:{os.environ['PATH']}")
    state = tmp_path / "state"
    monkeypatch.setenv("TAMQ_STATE_DIR", str(state))
    socket_name = f"tamq-duplex-{uuid4().hex[:8]}"
    session = f"tamq-duplex-{uuid4().hex[:8]}"
    manager = TmuxManager(
        session,
        tmux_command=("tmux", "-L", socket_name),
        tamq_command=(sys.executable, "-m", "tamq.cli"),
        command_dir=tmp_path / "commands",
    )
    plan = LaunchPlan(
        ("source",), {"source": str(source)}, (sys.executable, str(fixture))
    )
    store = None
    try:
        endpoint = manager.ensure_plan(plan, tap=True, tap_limits=(8, 1024, 32768))
        deadline = time.monotonic() + 5
        while time.monotonic() < deadline:
            if "READY" in manager._run(
                "capture-pane", "-p", "-t", f"{session}:source"
            ):
                break
            time.sleep(0.05)
        manager._run(
            "send-keys",
            "-t",
            f"{session}:source",
            "-l",
            "--",
            "To:target: operator request",
        )
        manager._run("send-keys", "-t", f"{session}:source", "Enter")
        store = Store(state / "tamq.sqlite3")
        deadline = time.monotonic() + 5
        rows = []
        while time.monotonic() < deadline:
            rows = store.list()
            if len(rows) >= 2:
                break
            time.sleep(0.05)
        assert [(row["body"], row["provenance"]) for row in rows] == [
            ("operator request", "operator_input"),
            ("worker reply", "worker_output"),
        ]
        assert all(row["endpoint_id"] == endpoint.instance_id for row in rows)
    finally:
        if store is not None:
            store.close()
        manager._run("kill-server", check=False)