felixm
90944948f5
TUI for orchestrating AI coding harness sessions (Claude Code, Codex, Kiro, etc.) via tmux. Click CLI bootstraps a Textual TUI over ProjectService/SessionService backed by SQLite, spawning harness sessions as tmux windows through TmuxManager. Includes recent fixes: - Visible Tab focus highlight on dialog OK/Cancel buttons - Auto-select first project on launch - Auto-select first session + per-project session-selection memory - tmux new-window targets an explicit free index, fixing "index N in use" failures (broken spawn/attach in attached sessions) Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
810 lines
30 KiB
Python
810 lines
30 KiB
Python
import pytest
|
|
from unittest.mock import AsyncMock
|
|
|
|
from hqt.tmux.runner import TmuxRunner
|
|
from hqt.tmux.manager import TmuxManager, SpawnRequest, SpawnResult
|
|
|
|
|
|
@pytest.fixture
|
|
def runner():
|
|
r = TmuxRunner(session_name="hqt-main")
|
|
r._exec = AsyncMock(return_value=(0, "", ""))
|
|
return r
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_new_window(runner):
|
|
runner._exec.side_effect = [
|
|
(0, "0\n", ""), # list-windows -F '#{window_index}' (next-index query)
|
|
(0, "@1\n", ""), # new-window -P -F
|
|
(0, "", ""), # set-option (combined: remain-on-exit + allow-rename + automatic-rename)
|
|
(0, "", ""), # respawn-pane
|
|
]
|
|
result = await runner.new_window("hqt-1", "/tmp", "kiro-cli chat")
|
|
assert result == "@1"
|
|
# Step 2: combined set-option call with ";" separators
|
|
set_call = runner._exec.call_args_list[2].args
|
|
assert set_call == (
|
|
"set-option", "-t", "hqt-main:=hqt-1", "remain-on-exit", "on",
|
|
";", "set-option", "-t", "hqt-main:=hqt-1", "allow-rename", "off",
|
|
";", "set-option", "-t", "hqt-main:=hqt-1", "automatic-rename", "off",
|
|
)
|
|
# respawn-pane runs the actual command
|
|
calls = runner._exec.call_args_list
|
|
respawn_args = calls[3].args
|
|
assert "respawn-pane" in respawn_args
|
|
assert "kiro-cli chat" in respawn_args
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_new_window_targets_explicit_free_index(runner):
|
|
"""Regression: new-window must target an explicit free index, not the bare
|
|
session name.
|
|
|
|
With a session-only target (`-t hqt`), an *attached* tmux resolves the target
|
|
to the session's current/active window and tries to create the new window at
|
|
that window's index — which is always occupied — failing with
|
|
'create window failed: index N in use'. That broke both spawning new sessions
|
|
and respawning dead ones (so attach failed). Computing max(existing index)+1
|
|
and targeting `session:idx` sidesteps the resolution entirely."""
|
|
runner._exec.side_effect = [
|
|
(0, "0\n1\n2\n", ""), # list-windows -F '#{window_index}' (existing: 0,1,2)
|
|
(0, "@9\n", ""), # new-window -P -F -> window_id
|
|
(0, "", ""), # set-option
|
|
(0, "", ""), # respawn-pane
|
|
]
|
|
window_id = await runner.new_window("hqt-9", "/tmp", "claude")
|
|
assert window_id == "@9"
|
|
|
|
new_win_args = runner._exec.call_args_list[1].args
|
|
assert "new-window" in new_win_args
|
|
# Explicit free index = max(0,1,2)+1 = 3, targeted as "<session>:3".
|
|
assert "hqt-main:3" in new_win_args, f"expected explicit free index target, got {new_win_args}"
|
|
# Must NOT use the bare session name as the new-window target.
|
|
ti = new_win_args.index("-t")
|
|
assert new_win_args[ti + 1] == "hqt-main:3"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_has_window_true(runner):
|
|
runner._exec.return_value = (0, "hqt-1\nhqt-2\n", "")
|
|
assert await runner.has_window("hqt-1") is True
|
|
assert runner._exec.call_count == 1
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_has_window_false(runner):
|
|
runner._exec.return_value = (0, "hqt-2\n", "")
|
|
assert await runner.has_window("hqt-1") is False
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_is_pane_dead_true(runner):
|
|
runner._exec.return_value = (0, "1\n", "")
|
|
assert await runner.is_pane_dead("hqt-1") is True
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_is_pane_dead_false(runner):
|
|
runner._exec.return_value = (0, "0\n", "")
|
|
assert await runner.is_pane_dead("hqt-1") is False
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_select_window(runner):
|
|
await runner.select_window("hqt-1")
|
|
runner._exec.assert_called_with("select-window", "-t", "hqt-main:=hqt-1")
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_respawn_pane(runner):
|
|
await runner.respawn_pane("hqt-1", "kiro-cli chat", "/tmp")
|
|
runner._exec.assert_called_with(
|
|
"respawn-pane", "-t", "hqt-main:=hqt-1", "-k", "-c", "/tmp", "kiro-cli chat",
|
|
)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_list_windows(runner):
|
|
runner._exec.return_value = (0, "hqt\nhqt-1\nhqt-2\n", "")
|
|
result = await runner.list_windows()
|
|
assert result == ["hqt", "hqt-1", "hqt-2"]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_manager_spawn(runner):
|
|
runner._exec.side_effect = [
|
|
(0, "0\n", ""), # list-windows (next-index query)
|
|
(0, "@3\n", ""), # new-window -P -F
|
|
(0, "", ""), # set-option
|
|
(0, "", ""), # respawn-pane
|
|
]
|
|
runner.verify_window_alive = AsyncMock(return_value=(True, ""))
|
|
mgr = TmuxManager(runner)
|
|
req = SpawnRequest(window_name="hqt-1", command=["kiro-cli", "chat"], cwd="/home")
|
|
result = await mgr.spawn(req)
|
|
assert result.ok is True
|
|
assert result.window_id == "@3"
|
|
# new-window call must NOT have trailing command
|
|
new_win_args = runner._exec.call_args_list[1].args
|
|
assert "new-window" in new_win_args
|
|
assert "kiro-cli chat" not in new_win_args
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_manager_attach_alive(runner):
|
|
"""Attach to alive window just selects it."""
|
|
mgr = TmuxManager(runner)
|
|
runner._exec.reset_mock()
|
|
runner._exec.return_value = (0, "hqt-1\n", "")
|
|
result = await mgr.attach("hqt-1")
|
|
assert result is True
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_manager_is_alive_true(runner):
|
|
"""Window exists and pane running."""
|
|
# has_window needs list-windows to return the name (one _exec call)
|
|
# is_pane_dead needs list-panes to return "0"
|
|
runner._exec.side_effect = [
|
|
(0, "hqt-1\n", ""), # has_window
|
|
(0, "0\n", ""), # is_pane_dead
|
|
]
|
|
mgr = TmuxManager(runner)
|
|
assert await mgr.is_alive("hqt-1") is True
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_manager_is_alive_false_dead_pane(runner):
|
|
"""Window exists but pane dead."""
|
|
runner._exec.side_effect = [
|
|
(0, "hqt-1\n", ""), # has_window
|
|
(0, "1\n", ""), # is_pane_dead
|
|
]
|
|
mgr = TmuxManager(runner)
|
|
assert await mgr.is_alive("hqt-1") is False
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_manager_is_alive_false_no_window(runner):
|
|
"""Window doesn't exist at all."""
|
|
runner._exec.side_effect = [
|
|
(0, "hqt-2\n", ""), # has_window — hqt-1 not in output
|
|
]
|
|
mgr = TmuxManager(runner)
|
|
assert await mgr.is_alive("hqt-1") is False
|
|
|
|
|
|
def test_target_exact_match(runner):
|
|
"""_target() must use =prefix to force exact-match-only resolution in tmux."""
|
|
assert runner._target("hqt-1") == "hqt-main:=hqt-1"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_has_window_false_single_exec(runner):
|
|
"""has_window returns False with a single exec when name is absent."""
|
|
runner._exec.return_value = (0, "hqt-2\n", "")
|
|
assert await runner.has_window("hqt-1") is False
|
|
assert runner._exec.call_count == 1
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Task 2: Safe spawn sequence — new_window race-free redesign
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_new_window_race_free_sequence(runner):
|
|
"""new_window issues: new-window (no command), set-option, respawn-pane — in order."""
|
|
runner._exec.side_effect = [
|
|
(0, "0\n", ""), # list-windows (next-index query)
|
|
(0, "@5\n", ""), # new-window -P -F -> window_id
|
|
(0, "", ""), # set-option
|
|
(0, "", ""), # respawn-pane
|
|
]
|
|
window_id = await runner.new_window("hqt-1", "/tmp", "kiro-cli chat")
|
|
assert window_id == "@5"
|
|
|
|
calls = runner._exec.call_args_list
|
|
assert len(calls) == 4
|
|
|
|
# Step 1: new-window without a trailing command; must have -P and -F
|
|
new_win_args = calls[1].args
|
|
assert "new-window" in new_win_args
|
|
assert "-P" in new_win_args
|
|
assert "#{window_id}" in new_win_args
|
|
# Must NOT have the command as the final positional arg (no trailing command)
|
|
assert new_win_args[-1] != "kiro-cli chat"
|
|
assert "kiro-cli chat" not in new_win_args
|
|
|
|
# Step 2: combined set-option call with ";" separators for all three options
|
|
set_args = calls[2].args
|
|
assert set_args == (
|
|
"set-option", "-t", "hqt-main:=hqt-1", "remain-on-exit", "on",
|
|
";", "set-option", "-t", "hqt-main:=hqt-1", "allow-rename", "off",
|
|
";", "set-option", "-t", "hqt-main:=hqt-1", "automatic-rename", "off",
|
|
)
|
|
|
|
# Step 3: respawn-pane runs the command
|
|
resp_args = calls[3].args
|
|
assert "respawn-pane" in resp_args
|
|
assert "kiro-cli chat" in resp_args
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_new_window_with_env(runner):
|
|
"""new_window passes -e KEY=VALUE flags on respawn-pane (step 3), NOT on new-window (step 1)."""
|
|
runner._exec.side_effect = [
|
|
(0, "0\n", ""), # list-windows (next-index query)
|
|
(0, "@7\n", ""),
|
|
(0, "", ""),
|
|
(0, "", ""),
|
|
]
|
|
window_id = await runner.new_window("hqt-2", "/tmp", "claude", env={"FOO": "bar", "X": "1"})
|
|
assert window_id == "@7"
|
|
|
|
calls_list = runner._exec.call_args_list
|
|
|
|
# Step 1 (new-window) must NOT carry -e flags — they're useless there
|
|
new_win_args = calls_list[1].args
|
|
assert "-e" not in new_win_args
|
|
assert "FOO=bar" not in new_win_args
|
|
assert "X=1" not in new_win_args
|
|
|
|
# Step 3 (respawn-pane) must carry -e KEY=VALUE pairs
|
|
respawn_args = calls_list[3].args
|
|
assert "respawn-pane" in respawn_args
|
|
assert "-e" in respawn_args
|
|
flat = list(respawn_args)
|
|
assert "FOO=bar" in flat
|
|
assert "X=1" in flat
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_new_window_set_option_failure_aborts(runner):
|
|
"""If set-option fails, new_window kills the orphaned window and returns None."""
|
|
runner._exec.side_effect = [
|
|
(0, "0\n", ""), # list-windows (next-index query)
|
|
(0, "@3\n", ""), # new-window succeeds
|
|
(1, "", "some tmux error"), # set-option fails
|
|
(0, "", ""), # kill-window cleanup
|
|
]
|
|
result = await runner.new_window("hqt-3", "/tmp", "claude")
|
|
assert result is None
|
|
# kill-window must have been called to clean up the orphaned shell
|
|
assert runner._exec.call_count == 4
|
|
kill_args = runner._exec.call_args_list[3].args
|
|
assert "kill-window" in kill_args
|
|
assert "hqt-main:=hqt-3" in kill_args
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_new_window_respawn_failure_kills_window(runner):
|
|
"""If respawn-pane fails, new_window kills the orphaned window and returns None."""
|
|
runner._exec.side_effect = [
|
|
(0, "0\n", ""), # list-windows (next-index query)
|
|
(0, "@4\n", ""), # new-window succeeds
|
|
(0, "", ""), # set-option succeeds
|
|
(1, "", "respawn error"), # respawn-pane fails
|
|
(0, "", ""), # kill-window cleanup
|
|
]
|
|
result = await runner.new_window("hqt-4", "/tmp", "claude")
|
|
assert result is None
|
|
assert runner._exec.call_count == 5
|
|
kill_args = runner._exec.call_args_list[4].args
|
|
assert "kill-window" in kill_args
|
|
assert "hqt-main:=hqt-4" in kill_args
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_new_window_returns_window_id(runner):
|
|
"""new_window returns the window_id string on full success."""
|
|
runner._exec.side_effect = [
|
|
(0, "0\n", ""), # list-windows (next-index query)
|
|
(0, "@42\n", ""),
|
|
(0, "", ""),
|
|
(0, "", ""),
|
|
]
|
|
result = await runner.new_window("hqt-5", "/var", "sleep 30")
|
|
assert result == "@42"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_new_window_returns_none_on_new_window_failure(runner):
|
|
"""new_window returns None when the initial new-window command fails."""
|
|
runner._exec.side_effect = [
|
|
(0, "0\n", ""), # list-windows (next-index query)
|
|
(1, "", "session not found"), # new-window fails
|
|
]
|
|
result = await runner.new_window("hqt-x", "/tmp", "sleep 1")
|
|
assert result is None
|
|
assert runner._exec.call_count == 2
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Task 2: verify_window_alive
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_verify_window_alive_alive(runner):
|
|
"""Pane stays alive through timeout → (True, '')."""
|
|
# is_pane_dead returns False (pane alive) on every poll
|
|
runner._exec.return_value = (0, "0\n", "")
|
|
ok, text = await runner.verify_window_alive("hqt-1", timeout=0.1, interval=0.05)
|
|
assert ok is True
|
|
assert text == ""
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_verify_window_alive_dead_with_text(runner):
|
|
"""Pane dies within timeout → (False, captured text)."""
|
|
runner._exec.side_effect = [
|
|
(0, "1\n", ""), # is_pane_dead → dead
|
|
(0, "command not found\n", ""), # capture_pane
|
|
]
|
|
ok, text = await runner.verify_window_alive("hqt-1", timeout=0.5, interval=0.05)
|
|
assert ok is False
|
|
assert "command not found" in text
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_verify_window_alive_dead_empty_text(runner):
|
|
"""Pane dies with empty capture → (False, '') still False."""
|
|
runner._exec.side_effect = [
|
|
(0, "1\n", ""), # is_pane_dead
|
|
(0, "", ""), # capture_pane empty
|
|
]
|
|
ok, text = await runner.verify_window_alive("hqt-1", timeout=0.5, interval=0.05)
|
|
assert ok is False
|
|
assert text == ""
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_verify_window_alive_dies_during_last_interval(runner):
|
|
"""Pane alive throughout polling loop but dies during the final sleep → (False, text)."""
|
|
# All in-loop polls return alive (0), final post-loop check returns dead (1)
|
|
runner._exec.side_effect = [
|
|
(0, "0\n", ""), # is_pane_dead (in-loop) → alive
|
|
(0, "1\n", ""), # is_pane_dead (post-loop final check) → dead
|
|
(0, "died at last\n", ""), # capture_pane
|
|
]
|
|
ok, text = await runner.verify_window_alive("hqt-1", timeout=0.05, interval=0.1)
|
|
assert ok is False
|
|
assert "died at last" in text
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Task 2: SpawnResult + manager.spawn changes
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_spawn_result_ok(runner):
|
|
"""spawn returns SpawnResult(ok=True, window_id=...) on success."""
|
|
runner._exec.side_effect = [
|
|
(0, "0\n", ""), # list-windows (next-index query)
|
|
(0, "@10\n", ""), # new-window
|
|
(0, "", ""), # set-option
|
|
(0, "", ""), # respawn-pane
|
|
]
|
|
runner.verify_window_alive = AsyncMock(return_value=(True, ""))
|
|
mgr = TmuxManager(runner)
|
|
req = SpawnRequest(window_name="hqt-1", command=["sleep", "30"], cwd="/tmp")
|
|
result = await mgr.spawn(req)
|
|
assert isinstance(result, SpawnResult)
|
|
assert result.ok is True
|
|
assert result.window_id == "@10"
|
|
assert result.error == ""
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_spawn_result_failure_verification(runner):
|
|
"""spawn returns SpawnResult(ok=False, error=...) when pane dies immediately."""
|
|
runner._exec.side_effect = [
|
|
(0, "0\n", ""), # list-windows (next-index query)
|
|
(0, "@11\n", ""), # new-window
|
|
(0, "", ""), # set-option
|
|
(0, "", ""), # respawn-pane
|
|
]
|
|
runner.verify_window_alive = AsyncMock(return_value=(False, "no such binary\n"))
|
|
mgr = TmuxManager(runner)
|
|
req = SpawnRequest(window_name="hqt-1", command=["bad-cmd"], cwd="/tmp")
|
|
result = await mgr.spawn(req)
|
|
assert result.ok is False
|
|
assert "no such binary" in result.error
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_spawn_shlex_join(runner):
|
|
"""spawn uses shlex.join so args with spaces are properly quoted."""
|
|
runner._exec.side_effect = [
|
|
(0, "0\n", ""), # list-windows (next-index query)
|
|
(0, "@99\n", ""),
|
|
(0, "", ""),
|
|
(0, "", ""),
|
|
]
|
|
runner.verify_window_alive = AsyncMock(return_value=(True, ""))
|
|
mgr = TmuxManager(runner)
|
|
req = SpawnRequest(window_name="hqt-1", command=["my cmd", "arg with space"], cwd="/tmp")
|
|
await mgr.spawn(req)
|
|
|
|
# The respawn-pane call (now the 4th _exec call) should have the shlex-joined string
|
|
respawn_call = runner._exec.call_args_list[3].args
|
|
assert "respawn-pane" in respawn_call
|
|
joined_cmd = respawn_call[-1]
|
|
# shlex.join of ["my cmd", "arg with space"] → "'my cmd' 'arg with space'"
|
|
import shlex
|
|
expected = shlex.join(["my cmd", "arg with space"])
|
|
assert joined_cmd == expected
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_spawn_passes_env(runner):
|
|
"""spawn passes env dict through to respawn-pane (step 3), NOT to new-window (step 1)."""
|
|
runner._exec.side_effect = [
|
|
(0, "0\n", ""), # list-windows (next-index query)
|
|
(0, "@20\n", ""),
|
|
(0, "", ""),
|
|
(0, "", ""),
|
|
]
|
|
runner.verify_window_alive = AsyncMock(return_value=(True, ""))
|
|
mgr = TmuxManager(runner)
|
|
req = SpawnRequest(window_name="hqt-env", command=["bash"], cwd="/tmp", env={"MY_VAR": "hello"})
|
|
result = await mgr.spawn(req)
|
|
assert result.ok is True
|
|
|
|
# new-window (step 1) must NOT carry -e
|
|
new_win_args = runner._exec.call_args_list[1].args
|
|
assert "-e" not in new_win_args
|
|
assert "MY_VAR=hello" not in new_win_args
|
|
|
|
# respawn-pane (step 3) must carry -e MY_VAR=hello
|
|
respawn_args = runner._exec.call_args_list[3].args
|
|
assert "respawn-pane" in respawn_args
|
|
assert "-e" in respawn_args
|
|
assert "MY_VAR=hello" in respawn_args
|
|
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_respawn_pane_with_env(runner):
|
|
"""respawn_pane passes -e KEY=VALUE flags when env is provided."""
|
|
await runner.respawn_pane("hqt-1", "claude", "/tmp", env={"MYKEY": "myval", "A": "b"})
|
|
call_args = runner._exec.call_args_list[0].args
|
|
assert "respawn-pane" in call_args
|
|
assert "-e" in call_args
|
|
assert "MYKEY=myval" in call_args
|
|
assert "A=b" in call_args
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_respawn_pane_no_env(runner):
|
|
"""respawn_pane without env emits no -e flags (backward compat)."""
|
|
await runner.respawn_pane("hqt-1", "claude", "/tmp")
|
|
call_args = runner._exec.call_args_list[0].args
|
|
assert "respawn-pane" in call_args
|
|
assert "-e" not in call_args
|
|
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_spawn_verify_dead_empty_text_returns_fallback_error(runner):
|
|
"""When verify_window_alive returns (False, ''), spawn returns the fallback error message."""
|
|
runner._exec.side_effect = [
|
|
(0, "0\n", ""), # list-windows (next-index query)
|
|
(0, "@30\n", ""), # new-window
|
|
(0, "", ""), # set-option
|
|
(0, "", ""), # respawn-pane
|
|
]
|
|
# Empty pane text — verify_window_alive signals immediate death with no output
|
|
runner.verify_window_alive = AsyncMock(return_value=(False, ""))
|
|
mgr = TmuxManager(runner)
|
|
req = SpawnRequest(window_name="hqt-dead", command=["gone-cmd"], cwd="/tmp")
|
|
result = await mgr.spawn(req)
|
|
assert result.ok is False
|
|
assert "pane died immediately" in result.error
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Task 3: respawn_verified
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_respawn_verified_success_path(runner):
|
|
"""respawn_verified: window exists, pane stays alive → SpawnResult(ok=True)."""
|
|
runner.verify_window_alive = AsyncMock(return_value=(True, ""))
|
|
runner._exec.side_effect = [
|
|
(0, "hqt-1\n", ""), # has_window → True
|
|
(0, "", ""), # respawn-pane
|
|
]
|
|
mgr = TmuxManager(runner)
|
|
result = await mgr.respawn_verified("hqt-1", ["claude", "--resume", "abc"], "/tmp")
|
|
assert result.ok is True
|
|
assert result.error == ""
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_respawn_verified_pane_dies_after_respawn(runner):
|
|
"""respawn_verified: window exists, pane dies immediately → SpawnResult(ok=False, error=pane text)."""
|
|
runner.verify_window_alive = AsyncMock(return_value=(False, "no such transcript\n"))
|
|
runner._exec.side_effect = [
|
|
(0, "hqt-1\n", ""), # has_window → True
|
|
(0, "", ""), # respawn-pane
|
|
]
|
|
mgr = TmuxManager(runner)
|
|
result = await mgr.respawn_verified("hqt-1", ["claude", "--resume", "abc"], "/tmp")
|
|
assert result.ok is False
|
|
assert "no such transcript" in result.error
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_respawn_verified_pane_dies_empty_text_fallback_message(runner):
|
|
"""respawn_verified: pane dies with no output → fallback error message."""
|
|
runner.verify_window_alive = AsyncMock(return_value=(False, ""))
|
|
runner._exec.side_effect = [
|
|
(0, "hqt-1\n", ""), # has_window → True
|
|
(0, "", ""), # respawn-pane
|
|
]
|
|
mgr = TmuxManager(runner)
|
|
result = await mgr.respawn_verified("hqt-1", ["claude", "--resume", "abc"], "/tmp")
|
|
assert result.ok is False
|
|
assert "pane died immediately" in result.error
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_respawn_verified_window_gone_branch(runner):
|
|
"""respawn_verified: window is gone → new_window + verify → SpawnResult(ok=True, window_id=...)."""
|
|
runner.verify_window_alive = AsyncMock(return_value=(True, ""))
|
|
runner._exec.side_effect = [
|
|
(0, "other\n", ""), # has_window → False
|
|
(0, "0\n", ""), # new_window: list-windows (next-index query)
|
|
(0, "@9\n", ""), # new-window step 1
|
|
(0, "", ""), # set-option step 2
|
|
(0, "", ""), # respawn-pane step 3
|
|
]
|
|
mgr = TmuxManager(runner)
|
|
result = await mgr.respawn_verified("hqt-gone", ["claude", "--resume", "abc"], "/tmp")
|
|
assert result.ok is True
|
|
assert result.window_id == "@9"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_respawn_verified_window_gone_new_window_fails(runner):
|
|
"""respawn_verified: window gone, new_window fails → SpawnResult(ok=False)."""
|
|
runner._exec.side_effect = [
|
|
(0, "other\n", ""), # has_window → False
|
|
(0, "0\n", ""), # new_window: list-windows (next-index query)
|
|
(1, "", "session not found"), # new-window fails
|
|
]
|
|
mgr = TmuxManager(runner)
|
|
result = await mgr.respawn_verified("hqt-gone", ["claude", "--resume", "abc"], "/tmp")
|
|
assert result.ok is False
|
|
assert result.error != ""
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_respawn_verified_forwards_env(runner):
|
|
"""respawn_verified passes env to respawn_pane when window exists."""
|
|
runner.verify_window_alive = AsyncMock(return_value=(True, ""))
|
|
runner._exec.side_effect = [
|
|
(0, "hqt-1\n", ""), # has_window → True
|
|
(0, "", ""), # respawn-pane
|
|
]
|
|
mgr = TmuxManager(runner)
|
|
result = await mgr.respawn_verified(
|
|
"hqt-1", ["claude"], "/tmp", env={"MY_VAR": "abc"}
|
|
)
|
|
assert result.ok is True
|
|
resp_call = runner._exec.call_args_list[1].args
|
|
assert "respawn-pane" in resp_call
|
|
assert "-e" in resp_call
|
|
assert "MY_VAR=abc" in resp_call
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_respawn_verified_uses_shlex_join(runner):
|
|
"""respawn_verified uses shlex.join so args with spaces are properly quoted."""
|
|
import shlex
|
|
runner.verify_window_alive = AsyncMock(return_value=(True, ""))
|
|
runner._exec.side_effect = [
|
|
(0, "hqt-1\n", ""), # has_window → True
|
|
(0, "", ""), # respawn-pane
|
|
]
|
|
mgr = TmuxManager(runner)
|
|
await mgr.respawn_verified("hqt-1", ["cmd with space", "arg"], "/tmp")
|
|
|
|
expected_cmd = shlex.join(["cmd with space", "arg"])
|
|
resp_call = runner._exec.call_args_list[1].args
|
|
assert "respawn-pane" in resp_call
|
|
assert expected_cmd in resp_call
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_respawn_verified_window_gone_forwards_env(runner):
|
|
"""respawn_verified forwards env to new_window when window is gone."""
|
|
runner.verify_window_alive = AsyncMock(return_value=(True, ""))
|
|
runner._exec.side_effect = [
|
|
(0, "other-window\n", ""), # has_window → False
|
|
(0, "0\n", ""), # new_window: list-windows (next-index query)
|
|
(0, "@5\n", ""), # new-window step 1
|
|
(0, "", ""), # set-option step 2
|
|
(0, "", ""), # respawn-pane step 3
|
|
]
|
|
mgr = TmuxManager(runner)
|
|
result = await mgr.respawn_verified("hqt-gone", ["bash"], "/tmp", env={"GONE_VAR": "x"})
|
|
assert result.ok is True
|
|
|
|
# The respawn-pane (step 3 of new_window) should carry the env
|
|
resp_call = runner._exec.call_args_list[4].args
|
|
assert "respawn-pane" in resp_call
|
|
assert "-e" in resp_call
|
|
assert "GONE_VAR=x" in resp_call
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Task 5: Literal send-keys (-l)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_send_text_uses_literal_flag(runner):
|
|
"""send_text must pass -l and '--' before the text."""
|
|
await runner.send_text("hqt-1", "Enter; echo hi C-c")
|
|
runner._exec.assert_called_once_with(
|
|
"send-keys", "-t", "hqt-main:=hqt-1", "-l", "--", "Enter; echo hi C-c",
|
|
)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_send_enter_no_literal_flag(runner):
|
|
"""send_enter intentionally sends the Enter key name — must NOT have -l."""
|
|
await runner.send_enter("hqt-1")
|
|
runner._exec.assert_called_once_with(
|
|
"send-keys", "-t", "hqt-main:=hqt-1", "Enter",
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Task 7: list_windows_info + WindowInfo + capture_pane -J + send_text --
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_list_windows_info_alive_and_activity(runner):
|
|
"""list_windows_info parses 3-column output; alive=True when pane_dead=0."""
|
|
runner._exec.return_value = (0, "hqt-1\t0\t1000000\nhqt-2\t1\t999999\n", "")
|
|
from hqt.tmux.runner import WindowInfo
|
|
result = await runner.list_windows_info()
|
|
assert result["hqt-1"] == WindowInfo(alive=True, last_activity=1000000)
|
|
assert result["hqt-2"] == WindowInfo(alive=False, last_activity=999999)
|
|
runner._exec.assert_called_once_with(
|
|
"list-panes", "-s", "-t", "hqt-main", "-F",
|
|
"#{window_name}\t#{pane_dead}\t#{window_activity}",
|
|
)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_list_windows_info_multi_pane_alive_max_activity(runner):
|
|
"""Multi-pane: alive if any pane alive; last_activity = max across panes."""
|
|
runner._exec.return_value = (
|
|
0,
|
|
"hqt-1\t1\t1000\nhqt-1\t0\t2000\nhqt-2\t1\t500\nhqt-2\t1\t600\n",
|
|
"",
|
|
)
|
|
from hqt.tmux.runner import WindowInfo
|
|
result = await runner.list_windows_info()
|
|
assert result["hqt-1"] == WindowInfo(alive=True, last_activity=2000)
|
|
assert result["hqt-2"] == WindowInfo(alive=False, last_activity=600)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_list_windows_info_malformed_line_skipped(runner):
|
|
"""Lines that don't have 3 tab-separated columns are skipped silently."""
|
|
runner._exec.return_value = (0, "hqt-1\t0\t1000\nbad_line\nhqt-2\t1\t500\n", "")
|
|
result = await runner.list_windows_info()
|
|
assert "hqt-1" in result
|
|
assert "bad_line" not in result
|
|
assert "hqt-2" in result
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_list_windows_info_rc_nonzero_returns_empty(runner):
|
|
"""rc != 0 → empty dict."""
|
|
runner._exec.return_value = (1, "", "no server running")
|
|
result = await runner.list_windows_info()
|
|
assert result == {}
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_capture_pane_includes_J_flag(runner):
|
|
"""capture_pane must pass -J to join wrapped lines."""
|
|
runner._exec.return_value = (0, "some output\n", "")
|
|
await runner.capture_pane("hqt-1")
|
|
args = runner._exec.call_args.args
|
|
assert "-J" in args
|
|
assert "capture-pane" in args
|
|
assert "-p" in args
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_send_text_includes_double_dash(runner):
|
|
"""send_text must pass '--' before the text to prevent flag parsing."""
|
|
await runner.send_text("hqt-1", "-flag-looking text")
|
|
runner._exec.assert_called_once_with(
|
|
"send-keys", "-t", "hqt-main:=hqt-1", "-l", "--", "-flag-looking text",
|
|
)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_send_text_double_dash_with_normal_text(runner):
|
|
"""send_text includes '--' even for normal text (belt-and-suspenders)."""
|
|
await runner.send_text("hqt-1", "hello world")
|
|
runner._exec.assert_called_once_with(
|
|
"send-keys", "-t", "hqt-main:=hqt-1", "-l", "--", "hello world",
|
|
)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_poll_info_single_exec(runner):
|
|
"""poll_info calls _exec exactly once and returns WindowInfo per name."""
|
|
runner._exec.return_value = (0, "hqt-1\t0\t1234\nhqt-2\t1\t5678\n", "")
|
|
from hqt.tmux.runner import WindowInfo
|
|
mgr = TmuxManager(runner)
|
|
result = await mgr.poll_info(["hqt-1", "hqt-2", "hqt-missing"])
|
|
assert runner._exec.call_count == 1
|
|
assert result["hqt-1"] == WindowInfo(alive=True, last_activity=1234)
|
|
assert result["hqt-2"] == WindowInfo(alive=False, last_activity=5678)
|
|
# Missing window gets a dead placeholder
|
|
assert result["hqt-missing"].alive is False
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_poll_info_rc_nonzero_all_dead(runner):
|
|
"""poll_info rc!=0 → all names map to dead WindowInfo."""
|
|
runner._exec.return_value = (1, "", "error")
|
|
mgr = TmuxManager(runner)
|
|
result = await mgr.poll_info(["hqt-x"])
|
|
assert result["hqt-x"].alive is False
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_set_window_label_sets_option_and_format(runner):
|
|
"""set_window_label sets @hqt_label + a label-aware window-status-format,
|
|
targeting the window by NAME (identity preserved — no rename)."""
|
|
from hqt.tmux.runner import WINDOW_STATUS_FORMAT
|
|
|
|
await runner.set_window_label("hqt-1", "◐ myproj")
|
|
args = runner._exec.call_args.args
|
|
# Targets by name (exact-match prefix), never renames the window.
|
|
assert "hqt-main:=hqt-1" in args
|
|
assert "@hqt_label" in args
|
|
assert "◐ myproj" in args
|
|
# Both formats reference the label option so the bar shows it.
|
|
assert args.count(WINDOW_STATUS_FORMAT) == 2
|
|
assert "window-status-format" in args
|
|
assert "window-status-current-format" in args
|
|
# All in a single atomic tmux invocation.
|
|
assert runner._exec.call_count == 1
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_manager_set_window_label_delegates(runner):
|
|
"""TmuxManager.set_window_label forwards to the runner."""
|
|
mgr = TmuxManager(runner)
|
|
await mgr.set_window_label("hqt-2", "○ deadproj")
|
|
args = runner._exec.call_args.args
|
|
assert "hqt-main:=hqt-2" in args
|
|
assert "○ deadproj" in args
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_poll_info_empty_names(runner):
|
|
"""poll_info with empty names list returns empty dict with exactly one exec."""
|
|
runner._exec.return_value = (0, "", "")
|
|
mgr = TmuxManager(runner)
|
|
result = await mgr.poll_info([])
|
|
assert result == {}
|
|
assert runner._exec.call_count == 1
|