import os import shutil import sys import time from pathlib import Path from uuid import uuid4 import pytest from tamq.broker import BrokerIdentity, InputBroker from tamq.control import ControlModeClient from tamq.store import Store from tamq.terminal import terminal_frame, write_terminal_output from tamq.tmux import LaunchPlan, TmuxManager @pytest.mark.skipif(shutil.which("tmux") is None, reason="tmux is not installed") def test_real_tmux_starts_two_repo_windows_and_reuses_them(tmp_path, monkeypatch): repo_a = tmp_path / "railiance-platform" repo_b = tmp_path / "activity-core" repo_a.mkdir() repo_b.mkdir() 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-pytest-{os.getpid()}-{uuid4().hex[:8]}" session = f"tamq-test-{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( ("railiance-platform", "activity-core"), {"railiance-platform": str(repo_a), "activity-core": str(repo_b)}, ("sh", "-c", "printf 'alpha-ready\\n'; exec sleep 30"), ) try: endpoint = manager.ensure_plan(plan, tap=True) assert endpoint.created_session is True assert endpoint.created_windows == ("railiance-platform", "activity-core") rows = manager._run( "list-windows", "-t", session, "-F", "#{window_name}|#{pane_current_path}|#{pane_pid}", ).splitlines() parsed = {name: (path, pid) for name, path, pid in (row.split("|", 2) for row in rows)} assert set(parsed) == {"railiance-platform", "activity-core"} assert parsed["railiance-platform"][0] == str(repo_a) assert parsed["activity-core"][0] == str(repo_b) deadline = time.monotonic() + 5 while time.monotonic() < deadline: captures = [ manager._run("capture-pane", "-p", "-t", f"{session}:{repo}") for repo in plan.repos ] if all("alpha-ready" in capture for capture in captures): break time.sleep(0.05) assert all("alpha-ready" in capture for capture in captures) reused = manager.ensure_plan(plan, tap=True) assert reused.created_session is False assert reused.created_windows == () reused_rows = manager._run( "list-windows", "-t", session, "-F", "#{window_name}|#{pane_current_path}|#{pane_pid}", ).splitlines() reused_parsed = {name: (path, pid) for name, path, pid in (row.split("|", 2) for row in reused_rows)} assert reused_parsed == parsed assert reused.instance_id == endpoint.instance_id store = Store(tmp_path / "state" / "tamq.sqlite3") control = ControlModeClient( session, tmux_command=("tmux", "-L", socket_name), ) try: message_id = store.add("local", "activity-core", "integration-message") control.start() delivered = InputBroker( store, BrokerIdentity(endpoint.instance_id, "railiance-platform"), ).deliver_pending( control, window_for_repo=lambda repo: f"{session}:{repo}", ) assert delivered == 1 assert store.list(state="injected")[0]["message_id"] == message_id deadline = time.monotonic() + 5 while time.monotonic() < deadline: capture = manager._run( "capture-pane", "-p", "-t", f"{session}:activity-core" ) if "#local: integration-message" in capture: break time.sleep(0.05) assert "#local: integration-message" in capture finally: control.close() store.close() finally: manager._run("kill-server", check=False) @pytest.mark.skipif(shutil.which("tmux") is None, reason="tmux is not installed") def test_real_tmux_output_preserves_partial_input_line_and_cursor(tmp_path): repo = tmp_path / "audit-core" repo.mkdir() socket_name = f"tamq-output-{os.getpid()}-{uuid4().hex[:8]}" session = f"tamq-output-{uuid4().hex[:8]}" manager = TmuxManager( session, tmux_command=("tmux", "-L", socket_name), tamq_command=(sys.executable, "-m", "tamq.cli"), command_dir=tmp_path / "commands", ) target = f"{session}:audit-core" plan = LaunchPlan(("audit-core",), {"audit-core": str(repo)}, ()) try: manager.ensure_plan(plan) deadline = time.monotonic() + 5 while time.monotonic() < deadline: current_command = manager._run( "display-message", "-p", "-t", target, "#{pane_current_command}" ) if current_command in {"bash", "sh", "zsh", "fish"}: break time.sleep(0.05) assert current_command in {"bash", "sh", "zsh", "fish"} manager._run( "send-keys", "-t", target, "-l", "--", "printf 'line-one\\nline-two\\n'" ) manager._run("send-keys", "-t", target, "Enter") deadline = time.monotonic() + 5 while time.monotonic() < deadline: initial_capture = manager._run("capture-pane", "-p", "-t", target) initial_lines = initial_capture.splitlines() if "line-one" in initial_lines and "line-two" in initial_lines: break time.sleep(0.05) assert "line-one" in initial_lines and "line-two" in initial_lines manager._run("send-keys", "-t", target, "-l", "--", "PARTIAL-INPUT") deadline = time.monotonic() + 5 while time.monotonic() < deadline: before_capture = manager._run("capture-pane", "-p", "-t", target) if "PARTIAL-INPUT" in before_capture: break time.sleep(0.05) assert "PARTIAL-INPUT" in before_capture control = ControlModeClient( session, tmux_command=("tmux", "-L", socket_name), ) before = control.pane_display(target) write_terminal_output( before.tty_path, terminal_frame( "flex-auth", "stable-message", "m-stable", cursor_y=before.cursor_y, pane_height=before.pane_height, alternate_on=before.alternate_on, ), ) deadline = time.monotonic() + 5 while time.monotonic() < deadline: after_capture = manager._run("capture-pane", "-p", "-t", target) if "#flex-auth: stable-message [m-stable]" in after_capture: break time.sleep(0.05) after = control.pane_display(target) assert "#flex-auth: stable-message [m-stable]" in after_capture before_input = next(line for line in before_capture.splitlines() if "PARTIAL-INPUT" in line) after_lines = after_capture.splitlines() assert any("PARTIAL-INPUT" in line for line in after_lines), repr(after_capture) after_input_index = next( index for index, line in enumerate(after_lines) if "PARTIAL-INPUT" in line ) assert after_lines[after_input_index] == before_input assert after_lines[after_input_index - 1] == "#flex-auth: stable-message [m-stable]" assert (after.cursor_x, after.cursor_y) == (before.cursor_x, before.cursor_y) finally: manager._run("kill-server", check=False)