felixm/hqt
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Satisfy project quality gates

Browse Source
This commit is contained in:
Felix Martin
2026-06-10 18:23:52 -04:00
parent 9379e1523c
commit a7851e7007
25 changed files with 1055 additions and 375 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/test_db.py
-1
View File
@@ -1,6 +1,5 @@
from pathlib import Path
from sqlalchemy import inspect
from sqlalchemy.orm import Session as DbSession
from hqt.config import Settings
from hqt.db.engine import ensure_db, get_engine, get_session_factory
tests/test_harnesses.py
+56 -7
View File
@@ -1,6 +1,7 @@
import json
import os
import time
from datetime import UTC, datetime
from pathlib import Path
from unittest.mock import patch
@@ -73,8 +74,16 @@ def test_codex_resume_uses_uuid():
def test_codex_capture_session_id(tmp_path):
sessions_dir = tmp_path / ".codex" / "sessions" / "2026" / "06" / "09"
sessions_dir.mkdir(parents=True)
rollout = sessions_dir / "rollout-20260609-abcdef12-3456-7890-abcd-ef1234567890.jsonl"
meta = {"type": "session_meta", "payload": {"id": "abcdef12-3456-7890-abcd-ef1234567890", "cwd": "/projects/foo"}}
rollout = (
sessions_dir / "rollout-20260609-abcdef12-3456-7890-abcd-ef1234567890.jsonl"
)
meta = {
"type": "session_meta",
"payload": {
"id": "abcdef12-3456-7890-abcd-ef1234567890",
"cwd": "/projects/foo",
},
}
rollout.write_text(json.dumps(meta) + "\n")
since = time.time() - 60 # far enough back that freshly-written file qualifies
@@ -100,7 +109,9 @@ def test_codex_capture_session_id_wrong_cwd(tmp_path):
@patch("hqt.harnesses.registry.shutil.which")
def test_registry_discovers_harnesses(mock_which):
mock_which.side_effect = lambda x: "/usr/bin/" + x if x in ("claude", "codex") else None
mock_which.side_effect = lambda x: (
"/usr/bin/" + x if x in ("claude", "codex") else None
)
result = discover_harnesses()
assert "claude" in result
assert "codex" in result
@@ -120,6 +131,7 @@ def test_codex_captures_session_id_flag():
def test_base_captures_session_id_flag_false():
"""HarnessConfigurator base class default must be False."""
from hqt.harnesses.base import HarnessConfigurator
assert HarnessConfigurator.captures_session_id is False
@@ -128,8 +140,12 @@ def test_claude_captures_session_id_flag_false():
assert ClaudeConfigurator.captures_session_id is False
def _write_rollout(path: Path, session_id: str, cwd: str) -> None:
def _write_rollout(
path: Path, session_id: str, cwd: str, timestamp: str | None = None
) -> None:
meta = {"type": "session_meta", "payload": {"id": session_id, "cwd": cwd}}
if timestamp:
meta["payload"]["timestamp"] = timestamp
path.write_text(json.dumps(meta) + "\n")
@@ -168,8 +184,8 @@ def test_codex_capture_accepts_file_after_since(tmp_path):
assert result == "new-id"
def test_codex_capture_newest_post_since_cwd_match_wins(tmp_path):
"""Among multiple post-since rollouts, newest matching cwd wins over older matching."""
def test_codex_capture_first_post_since_cwd_match_wins(tmp_path):
"""Among multiple post-since rollouts, the one started nearest since wins."""
sessions_dir = tmp_path / ".codex" / "sessions" / "dir"
sessions_dir.mkdir(parents=True)
@@ -186,7 +202,39 @@ def test_codex_capture_newest_post_since_cwd_match_wins(tmp_path):
c = CodexConfigurator()
with patch("hqt.harnesses.configurators.codex.Path.home", return_value=tmp_path):
result = c.capture_session_id(Path("/projects/foo"), since)
assert result == "newer-id"
assert result == "older-id"
def test_codex_capture_prefers_started_nearest_since_over_newest_file(tmp_path):
"""Late capture should match the Codex session spawned at since, not a later same-cwd session."""
sessions_dir = tmp_path / ".codex" / "sessions" / "dir"
sessions_dir.mkdir(parents=True)
target = sessions_dir / "rollout-target.jsonl"
unrelated_later = sessions_dir / "rollout-later.jsonl"
since = datetime(2026, 6, 10, 21, 9, 15, tzinfo=UTC).timestamp()
_write_rollout(
target,
"target-id",
"/projects/foo",
timestamp="2026-06-10T21:09:16.000Z",
)
_write_rollout(
unrelated_later,
"later-id",
"/projects/foo",
timestamp="2026-06-10T21:45:45.000Z",
)
now = time.time()
os.utime(target, (now - 60, now - 60))
os.utime(unrelated_later, (now, now))
c = CodexConfigurator()
with patch("hqt.harnesses.configurators.codex.Path.home", return_value=tmp_path):
result = c.capture_session_id(Path("/projects/foo"), since)
assert result == "target-id"
def test_codex_capture_skips_non_matching_cwd_picks_older_matching(tmp_path):
@@ -273,6 +321,7 @@ CODEX_IDLE_SCREEN = """\
def test_base_parse_status_returns_none():
"""Base HarnessConfigurator.parse_status always returns None."""
from hqt.harnesses.configurators.kiro import KiroConfigurator
k = KiroConfigurator()
assert k.parse_status("anything") is None
assert k.parse_status("") is None
tests/test_integration.py
+10 -4
View File
@@ -1,5 +1,4 @@
import pytest
from pathlib import Path
from unittest.mock import AsyncMock, MagicMock
from sqlalchemy import create_engine
@@ -47,14 +46,19 @@ async def test_full_lifecycle(setup):
assert project.id == 1
# Create session
create_result = await sess_svc.create_session(project_id=project.id, harness_name="claude-code", nickname="e2e")
create_result = await sess_svc.create_session(
project_id=project.id, harness_name="claude-code", nickname="e2e"
)
session = create_result.session
assert session.id is not None
tmux.spawn.assert_called_once()
# List sessions - alive
from hqt.tmux.runner import WindowInfo
tmux.poll_info.return_value = {session.tmux_session_name: WindowInfo(alive=True, last_activity=0)}
tmux.poll_info.return_value = {
session.tmux_session_name: WindowInfo(alive=True, last_activity=0)
}
infos = await sess_svc.list_sessions(project_id=project.id)
assert len(infos) == 1
assert infos[0].alive is True
@@ -64,7 +68,9 @@ async def test_full_lifecycle(setup):
tmux.kill.assert_called_once_with(session.tmux_session_name)
# List sessions - dead
tmux.poll_info.return_value = {session.tmux_session_name: WindowInfo(alive=False, last_activity=0)}
tmux.poll_info.return_value = {
session.tmux_session_name: WindowInfo(alive=False, last_activity=0)
}
infos = await sess_svc.list_sessions(project_id=project.id)
assert len(infos) == 1
assert infos[0].alive is False
tests/test_sessions.py
+230 -66
View File
@@ -33,7 +33,9 @@ def tmux():
m.capture_pane = AsyncMock(return_value="")
m.attach = AsyncMock()
m.window_exists = AsyncMock(return_value=True)
m.respawn_verified = AsyncMock(return_value=SpawnResult(ok=True, window_id=None, error=""))
m.respawn_verified = AsyncMock(
return_value=SpawnResult(ok=True, window_id=None, error="")
)
return m
@@ -42,7 +44,9 @@ def harnesses():
h = MagicMock()
h.captures_session_id = False
h.generate_session_id.return_value = "sess-1"
h.build_spawn_config.return_value = MagicMock(command=["claude"], env={}, cwd=Path("/tmp/myproj"))
h.build_spawn_config.return_value = MagicMock(
command=["claude"], env={}, cwd=Path("/tmp/myproj")
)
h.parse_status = MagicMock(return_value=None)
return {"claude-code": h}
@@ -54,7 +58,9 @@ def service(db, tmux, harnesses):
@pytest.mark.asyncio
async def test_create_session(service, db, tmux, harnesses):
result = await service.create_session(project_id=1, harness_name="claude-code", nickname="test")
result = await service.create_session(
project_id=1, harness_name="claude-code", nickname="test"
)
sess = result.session
assert sess.id is not None
assert sess.tmux_session_name == f"hqt-{sess.id}"
@@ -66,6 +72,7 @@ async def test_create_session(service, db, tmux, harnesses):
@pytest.mark.asyncio
async def test_list_sessions(service, db, tmux):
from hqt.tmux.runner import WindowInfo
await service.create_session(project_id=1, harness_name="claude-code")
tmux.poll_info.return_value = {"hqt-1": WindowInfo(alive=True, last_activity=1000)}
result = await service.list_sessions(project_id=1)
@@ -95,7 +102,10 @@ async def test_stop_session_no_kill_when_window_missing(service, db, tmux):
async def test_create_session_logs_failed_spawn(service, db, tmux, caplog):
"""create_session logs error including pane text when spawn fails, but still commits."""
import logging
tmux.spawn.return_value = SpawnResult(ok=False, window_id=None, error="bash: command not found")
tmux.spawn.return_value = SpawnResult(
ok=False, window_id=None, error="bash: command not found"
)
with caplog.at_level(logging.ERROR):
result = await service.create_session(project_id=1, harness_name="claude-code")
assert result.session.id is not None # row was committed
@@ -108,16 +118,21 @@ async def test_create_session_logs_failed_spawn(service, db, tmux, caplog):
async def test_attach_session_passes_env(service, db, tmux, harnesses):
"""attach_session passes cfg.env to tmux.respawn_verified when the window needs respawning."""
from unittest.mock import AsyncMock
# Create a session first
await service.create_session(project_id=1, harness_name="claude-code")
# Set up harness to return a config with env
env_cfg = MagicMock(command=["claude"], env={"SESSION_VAR": "abc"}, cwd=Path("/tmp/myproj"))
env_cfg = MagicMock(
command=["claude"], env={"SESSION_VAR": "abc"}, cwd=Path("/tmp/myproj")
)
harnesses["claude-code"].build_resume_config = MagicMock(return_value=env_cfg)
# Simulate window is not alive → respawn path
tmux.is_alive = AsyncMock(return_value=False)
tmux.respawn_verified = AsyncMock(return_value=SpawnResult(ok=True, window_id=None, error=""))
tmux.respawn_verified = AsyncMock(
return_value=SpawnResult(ok=True, window_id=None, error="")
)
tmux.attach = AsyncMock(return_value=True)
await service.attach_session(session_id=1)
@@ -125,8 +140,9 @@ async def test_attach_session_passes_env(service, db, tmux, harnesses):
tmux.respawn_verified.assert_called()
first_call = tmux.respawn_verified.call_args_list[0]
# env must be forwarded (passed as keyword or positional)
assert first_call.kwargs.get("env") == {"SESSION_VAR": "abc"} or \
(len(first_call.args) >= 4 and first_call.args[3] == {"SESSION_VAR": "abc"})
assert first_call.kwargs.get("env") == {"SESSION_VAR": "abc"} or (
len(first_call.args) >= 4 and first_call.args[3] == {"SESSION_VAR": "abc"}
)
# ---------------------------------------------------------------------------
@@ -135,18 +151,28 @@ async def test_attach_session_passes_env(service, db, tmux, harnesses):
@pytest.mark.asyncio
async def test_attach_session_dead_resume_ok_attaches_once(service, db, tmux, harnesses):
async def test_attach_session_dead_resume_ok_attaches_once(
service, db, tmux, harnesses
):
"""attach_session: resume ok → attach called once, no fallback spawn attempted."""
await service.create_session(project_id=1, harness_name="claude-code")
resume_cfg = MagicMock(command=["claude", "--resume", "sess-1"], env={}, cwd=Path("/tmp/myproj"))
resume_cfg = MagicMock(
command=["claude", "--resume", "sess-1"], env={}, cwd=Path("/tmp/myproj")
)
harnesses["claude-code"].build_resume_config = MagicMock(return_value=resume_cfg)
harnesses["claude-code"].build_spawn_config = MagicMock(
return_value=MagicMock(command=["claude", "--session-id", "sess-1"], env={}, cwd=Path("/tmp/myproj"))
return_value=MagicMock(
command=["claude", "--session-id", "sess-1"],
env={},
cwd=Path("/tmp/myproj"),
)
)
tmux.is_alive = AsyncMock(return_value=False)
tmux.respawn_verified = AsyncMock(return_value=SpawnResult(ok=True, window_id=None, error=""))
tmux.respawn_verified = AsyncMock(
return_value=SpawnResult(ok=True, window_id=None, error="")
)
tmux.attach = AsyncMock(return_value=True)
result = await service.attach_session(session_id=1)
@@ -160,20 +186,30 @@ async def test_attach_session_dead_resume_ok_attaches_once(service, db, tmux, ha
@pytest.mark.asyncio
async def test_attach_session_dead_resume_fails_fallback_spawn(service, db, tmux, harnesses):
async def test_attach_session_dead_resume_fails_fallback_spawn(
service, db, tmux, harnesses
):
"""attach_session: resume dies → fallback spawn with session-id, attach called once."""
await service.create_session(project_id=1, harness_name="claude-code")
resume_cfg = MagicMock(command=["claude", "--resume", "sess-1"], env={}, cwd=Path("/tmp/myproj"))
spawn_cfg = MagicMock(command=["claude", "--session-id", "sess-1"], env={}, cwd=Path("/tmp/myproj"))
resume_cfg = MagicMock(
command=["claude", "--resume", "sess-1"], env={}, cwd=Path("/tmp/myproj")
)
spawn_cfg = MagicMock(
command=["claude", "--session-id", "sess-1"], env={}, cwd=Path("/tmp/myproj")
)
harnesses["claude-code"].build_resume_config = MagicMock(return_value=resume_cfg)
harnesses["claude-code"].build_spawn_config = MagicMock(return_value=spawn_cfg)
tmux.is_alive = AsyncMock(return_value=False)
tmux.respawn_verified = AsyncMock(side_effect=[
SpawnResult(ok=False, window_id=None, error="no transcript"), # resume fails
SpawnResult(ok=True, window_id="@2", error=""), # fresh spawn ok
])
tmux.respawn_verified = AsyncMock(
side_effect=[
SpawnResult(
ok=False, window_id=None, error="no transcript"
), # resume fails
SpawnResult(ok=True, window_id="@2", error=""), # fresh spawn ok
]
)
tmux.attach = AsyncMock(return_value=True)
result = await service.attach_session(session_id=1)
@@ -189,17 +225,25 @@ async def test_attach_session_dead_resume_fails_fallback_spawn(service, db, tmux
@pytest.mark.asyncio
async def test_attach_session_both_rungs_fail_returns_false(service, db, tmux, harnesses):
async def test_attach_session_both_rungs_fail_returns_false(
service, db, tmux, harnesses
):
"""attach_session: both resume and spawn fail → False, no attach."""
await service.create_session(project_id=1, harness_name="claude-code")
resume_cfg = MagicMock(command=["claude", "--resume", "sess-1"], env={}, cwd=Path("/tmp/myproj"))
spawn_cfg = MagicMock(command=["claude", "--session-id", "sess-1"], env={}, cwd=Path("/tmp/myproj"))
resume_cfg = MagicMock(
command=["claude", "--resume", "sess-1"], env={}, cwd=Path("/tmp/myproj")
)
spawn_cfg = MagicMock(
command=["claude", "--session-id", "sess-1"], env={}, cwd=Path("/tmp/myproj")
)
harnesses["claude-code"].build_resume_config = MagicMock(return_value=resume_cfg)
harnesses["claude-code"].build_spawn_config = MagicMock(return_value=spawn_cfg)
tmux.is_alive = AsyncMock(return_value=False)
tmux.respawn_verified = AsyncMock(return_value=SpawnResult(ok=False, window_id=None, error="fatal"))
tmux.respawn_verified = AsyncMock(
return_value=SpawnResult(ok=False, window_id=None, error="fatal")
)
tmux.attach = AsyncMock(return_value=True)
result = await service.attach_session(session_id=1)
@@ -219,7 +263,9 @@ def harnesses_no_capture():
h = MagicMock()
h.captures_session_id = False
h.generate_session_id.return_value = "sess-nc"
h.build_spawn_config.return_value = MagicMock(command=["claude"], env={}, cwd=Path("/tmp/myproj"))
h.build_spawn_config.return_value = MagicMock(
command=["claude"], env={}, cwd=Path("/tmp/myproj")
)
return {"claude-code": h}
@@ -229,12 +275,16 @@ def harnesses_capture():
h = MagicMock()
h.captures_session_id = True
h.generate_session_id.return_value = "placeholder-id"
h.build_spawn_config.return_value = MagicMock(command=["codex"], env={}, cwd=Path("/tmp/myproj"))
h.build_spawn_config.return_value = MagicMock(
command=["codex"], env={}, cwd=Path("/tmp/myproj")
)
return {"claude-code": h}
@pytest.mark.asyncio
async def test_create_session_no_capture_flag_never_calls_capture(db, tmux, harnesses_no_capture):
async def test_create_session_no_capture_flag_never_calls_capture(
db, tmux, harnesses_no_capture
):
"""When captures_session_id=False, capture_session_id is never called."""
service = SessionService(db=db, tmux=tmux, harnesses=harnesses_no_capture)
await service.create_session(project_id=1, harness_name="claude-code")
@@ -242,11 +292,17 @@ async def test_create_session_no_capture_flag_never_calls_capture(db, tmux, harn
@pytest.mark.asyncio
async def test_create_session_capture_retries_until_success(db, tmux, harnesses_capture):
async def test_create_session_capture_retries_until_success(
db, tmux, harnesses_capture
):
"""captures_session_id=True: returns None twice then an id → retries and stores captured id."""
import hqt.sessions.service as svc_mod
harnesses_capture["claude-code"].capture_session_id.side_effect = [None, None, "captured-id-xyz"]
harnesses_capture["claude-code"].capture_session_id.side_effect = [
None,
None,
"captured-id-xyz",
]
sleep_calls = []
@@ -264,7 +320,9 @@ async def test_create_session_capture_retries_until_success(db, tmux, harnesses_
@pytest.mark.asyncio
async def test_create_session_capture_all_none_keeps_placeholder(db, tmux, harnesses_capture, caplog):
async def test_create_session_capture_all_none_keeps_placeholder(
db, tmux, harnesses_capture, caplog
):
"""All capture attempts return None → placeholder id kept, no crash, warning logged."""
import logging
import hqt.sessions.service as svc_mod
@@ -277,18 +335,27 @@ async def test_create_session_capture_all_none_keeps_placeholder(db, tmux, harne
service = SessionService(db=db, tmux=tmux, harnesses=harnesses_capture)
with caplog.at_level(logging.WARNING):
with patch.object(svc_mod, "_capture_sleep", fake_sleep):
result = await service.create_session(project_id=1, harness_name="claude-code")
result = await service.create_session(
project_id=1, harness_name="claude-code"
)
assert result.session.harness_session_id == "placeholder-id"
# Should have attempted up to 6 times (CAPTURE_MAX_ATTEMPTS)
assert harnesses_capture["claude-code"].capture_session_id.call_count == svc_mod.CAPTURE_MAX_ATTEMPTS
assert (
harnesses_capture["claude-code"].capture_session_id.call_count
== svc_mod.CAPTURE_MAX_ATTEMPTS
)
# Exhaustion must emit a warning
warning_messages = [r.message for r in caplog.records if r.levelno == logging.WARNING]
warning_messages = [
r.message for r in caplog.records if r.levelno == logging.WARNING
]
assert any("exhausted" in m for m in warning_messages)
@pytest.mark.asyncio
async def test_create_session_capture_first_attempt_succeeds(db, tmux, harnesses_capture):
async def test_create_session_capture_first_attempt_succeeds(
db, tmux, harnesses_capture
):
"""captures_session_id=True: first attempt succeeds → no sleeps needed."""
import hqt.sessions.service as svc_mod
@@ -316,6 +383,7 @@ async def test_create_session_capture_first_attempt_succeeds(db, tmux, harnesses
async def test_list_sessions_dead_window(db, tmux, harnesses):
"""list_sessions: dead window → status='dead', alive=False."""
from hqt.tmux.runner import WindowInfo
await SessionService(db=db, tmux=tmux, harnesses=harnesses).create_session(
project_id=1, harness_name="claude-code"
)
@@ -344,6 +412,7 @@ async def test_list_sessions_missing_window_is_dead(db, tmux, harnesses):
async def test_list_sessions_alive_parse_working(db, tmux, harnesses):
"""Alive window + harness parse returns 'working' → status='working'."""
from hqt.tmux.runner import WindowInfo
await SessionService(db=db, tmux=tmux, harnesses=harnesses).create_session(
project_id=1, harness_name="claude-code"
)
@@ -364,12 +433,15 @@ async def test_list_sessions_alive_parse_working(db, tmux, harnesses):
async def test_list_sessions_alive_parse_none_recent_activity(db, tmux, harnesses):
"""Alive window + parse_status None + recent activity → status='active'."""
from hqt.tmux.runner import WindowInfo
await SessionService(db=db, tmux=tmux, harnesses=harnesses).create_session(
project_id=1, harness_name="claude-code"
)
now = 2000.0
tmux.poll_info.return_value = {"hqt-1": WindowInfo(alive=True, last_activity=1998)} # 2s ago
tmux.poll_info.return_value = {
"hqt-1": WindowInfo(alive=True, last_activity=1998)
} # 2s ago
tmux.capture_pane = AsyncMock(return_value="some text")
harnesses["claude-code"].parse_status = MagicMock(return_value=None)
@@ -383,12 +455,15 @@ async def test_list_sessions_alive_parse_none_recent_activity(db, tmux, harnesse
async def test_list_sessions_alive_parse_none_stale_activity(db, tmux, harnesses):
"""Alive window + parse_status None + stale activity → status='idle'."""
from hqt.tmux.runner import WindowInfo
await SessionService(db=db, tmux=tmux, harnesses=harnesses).create_session(
project_id=1, harness_name="claude-code"
)
now = 2000.0
tmux.poll_info.return_value = {"hqt-1": WindowInfo(alive=True, last_activity=1990)} # 10s ago
tmux.poll_info.return_value = {
"hqt-1": WindowInfo(alive=True, last_activity=1990)
} # 10s ago
tmux.capture_pane = AsyncMock(return_value="some text")
harnesses["claude-code"].parse_status = MagicMock(return_value=None)
@@ -402,6 +477,7 @@ async def test_list_sessions_alive_parse_none_stale_activity(db, tmux, harnesses
async def test_list_sessions_capture_empty_falls_back_to_activity(db, tmux, harnesses):
"""capture_pane returns empty text → activity-based fallback, no crash."""
from hqt.tmux.runner import WindowInfo
await SessionService(db=db, tmux=tmux, harnesses=harnesses).create_session(
project_id=1, harness_name="claude-code"
)
@@ -423,6 +499,7 @@ async def test_list_sessions_capture_empty_falls_back_to_activity(db, tmux, harn
async def test_list_sessions_alive_parse_waiting(db, tmux, harnesses):
"""Alive window + parse_status 'waiting' → status='waiting'."""
from hqt.tmux.runner import WindowInfo
await SessionService(db=db, tmux=tmux, harnesses=harnesses).create_session(
project_id=1, harness_name="claude-code"
)
@@ -449,7 +526,9 @@ async def test_sync_window_labels_sets_status_symbol(db, tmux, harnesses):
from hqt.tmux.runner import WindowInfo
service = SessionService(db=db, tmux=tmux, harnesses=harnesses)
await service.create_session(project_id=1, harness_name="claude-code", nickname="mywork")
await service.create_session(
project_id=1, harness_name="claude-code", nickname="mywork"
)
now = 2000.0
tmux.poll_info.return_value = {"hqt-1": WindowInfo(alive=True, last_activity=1999)}
@@ -471,7 +550,9 @@ async def test_sync_window_labels_dead_window_uses_dead_glyph(db, tmux, harnesse
from hqt.tmux.runner import WindowInfo
service = SessionService(db=db, tmux=tmux, harnesses=harnesses)
await service.create_session(project_id=1, harness_name="claude-code") # no nickname
await service.create_session(
project_id=1, harness_name="claude-code"
) # no nickname
tmux.poll_info.return_value = {"hqt-1": WindowInfo(alive=False, last_activity=0)}
tmux.set_window_label = AsyncMock()
@@ -533,16 +614,22 @@ async def test_list_sessions_real_configurator_waiting(db):
# spec=TmuxManager: only methods that exist on TmuxManager are accessible.
# capture_pane MUST be on TmuxManager (via the delegate) for the spec to allow it.
tmux_spec = MagicMock(spec=TmuxManager)
tmux_spec.spawn = AsyncMock(return_value=SpawnResult(ok=True, window_id="@1", error=""))
tmux_spec.spawn = AsyncMock(
return_value=SpawnResult(ok=True, window_id="@1", error="")
)
tmux_spec.kill = AsyncMock()
tmux_spec.is_alive = AsyncMock(return_value=True)
tmux_spec.poll_info = AsyncMock(return_value={})
# capture_pane returns realistic "waiting" pane text; the spec must allow this
# attribute — fails with AttributeError before TmuxManager.capture_pane delegate exists.
tmux_spec.capture_pane = AsyncMock(return_value="Do you want to proceed?\n❯ 1. Yes\n")
tmux_spec.capture_pane = AsyncMock(
return_value="Do you want to proceed?\n❯ 1. Yes\n"
)
tmux_spec.attach = AsyncMock()
tmux_spec.window_exists = AsyncMock(return_value=True)
tmux_spec.respawn_verified = AsyncMock(return_value=SpawnResult(ok=True, window_id=None, error=""))
tmux_spec.respawn_verified = AsyncMock(
return_value=SpawnResult(ok=True, window_id=None, error="")
)
configurator = ClaudeConfigurator()
harnesses = {"claude-code": configurator}
@@ -553,7 +640,9 @@ async def test_list_sessions_real_configurator_waiting(db):
now = 2000.0
window_name = "hqt-1"
tmux_spec.poll_info.return_value = {window_name: WindowInfo(alive=True, last_activity=1999)}
tmux_spec.poll_info.return_value = {
window_name: WindowInfo(alive=True, last_activity=1999)
}
result = await service.list_sessions(project_id=1, now=now)
@@ -615,17 +704,25 @@ def harnesses_capture_fallback():
h.name = "claude-code"
h.captures_session_id = True
h.generate_session_id.return_value = "placeholder-id"
h.build_spawn_config.return_value = MagicMock(command=["codex"], env={}, cwd=Path("/tmp/myproj"))
h.build_resume_config.return_value = MagicMock(command=["codex", "resume", "placeholder-id"], env={}, cwd=Path("/tmp/myproj"))
h.build_spawn_config.return_value = MagicMock(
command=["codex"], env={}, cwd=Path("/tmp/myproj")
)
h.build_resume_config.return_value = MagicMock(
command=["codex", "resume", "placeholder-id"], env={}, cwd=Path("/tmp/myproj")
)
return {"claude-code": h}
@pytest.mark.asyncio
async def test_respawn_fallback_rung2_updates_harness_session_id(db, tmux, harnesses_capture_fallback):
async def test_respawn_fallback_rung2_updates_harness_session_id(
db, tmux, harnesses_capture_fallback
):
"""Rung-2 success + captures_session_id=True + capture returns new id → sess.harness_session_id updated and committed."""
import hqt.sessions.service as svc_mod
harnesses_capture_fallback["claude-code"].capture_session_id.return_value = "new-codex-id"
harnesses_capture_fallback[
"claude-code"
].capture_session_id.return_value = "new-codex-id"
async def fake_sleep(t):
pass
@@ -638,13 +735,19 @@ async def test_respawn_fallback_rung2_updates_harness_session_id(db, tmux, harne
# Simulate rung-1 resume fails, rung-2 fresh spawn succeeds
harnesses_capture_fallback["claude-code"].capture_session_id.reset_mock()
harnesses_capture_fallback["claude-code"].capture_session_id.return_value = "new-codex-id-after-rung2"
harnesses_capture_fallback[
"claude-code"
].capture_session_id.return_value = "new-codex-id-after-rung2"
tmux.is_alive = AsyncMock(return_value=False)
tmux.respawn_verified = AsyncMock(side_effect=[
SpawnResult(ok=False, window_id=None, error="no transcript"), # rung-1 resume fails
SpawnResult(ok=True, window_id="@2", error=""), # rung-2 fresh spawn ok
])
tmux.respawn_verified = AsyncMock(
side_effect=[
SpawnResult(
ok=False, window_id=None, error="no transcript"
), # rung-1 resume fails
SpawnResult(ok=True, window_id="@2", error=""), # rung-2 fresh spawn ok
]
)
tmux.attach = AsyncMock(return_value=True)
with patch.object(svc_mod, "_capture_sleep", fake_sleep):
@@ -653,17 +756,61 @@ async def test_respawn_fallback_rung2_updates_harness_session_id(db, tmux, harne
assert ok is True
# harness_session_id must now be the newly captured id
from hqt.db.models import Session as DBSession
updated = db.get(DBSession, sess_id)
assert updated.harness_session_id == "new-codex-id-after-rung2"
@pytest.mark.asyncio
async def test_respawn_fallback_rung2_capture_exhausted_keeps_old_id(db, tmux, harnesses_capture_fallback, caplog):
async def test_attach_late_capture_updates_id_before_resume(
db, tmux, harnesses_capture_fallback
):
"""If create-time Codex capture missed, attach captures the real id before resuming."""
import hqt.sessions.service as svc_mod
harness = harnesses_capture_fallback["claude-code"]
harness.capture_session_id.side_effect = [None] * svc_mod.CAPTURE_MAX_ATTEMPTS + [
"late-codex-id"
]
async def fake_sleep(t):
pass
service = SessionService(db=db, tmux=tmux, harnesses=harnesses_capture_fallback)
with patch.object(svc_mod, "_capture_sleep", fake_sleep):
result = await service.create_session(project_id=1, harness_name="claude-code")
assert result.session.harness_session_id == "placeholder-id"
tmux.is_alive = AsyncMock(return_value=False)
tmux.respawn_verified = AsyncMock(
return_value=SpawnResult(ok=True, window_id="@2", error="")
)
tmux.attach = AsyncMock(return_value=True)
ok = await service.attach_session(session_id=result.session.id)
assert ok is True
harness.build_resume_config.assert_called_with(
Path("/tmp/myproj"), "late-codex-id", result.session.model
)
from hqt.db.models import Session as DBSession
updated = db.get(DBSession, result.session.id)
assert updated.harness_session_id == "late-codex-id"
@pytest.mark.asyncio
async def test_respawn_fallback_rung2_capture_exhausted_keeps_old_id(
db, tmux, harnesses_capture_fallback, caplog
):
"""Rung-2 success + capture exhausted → old id kept + warning logged."""
import logging
import hqt.sessions.service as svc_mod
harnesses_capture_fallback["claude-code"].capture_session_id.return_value = "placeholder-id"
harnesses_capture_fallback[
"claude-code"
].capture_session_id.return_value = "placeholder-id"
async def fake_sleep(t):
pass
@@ -680,10 +827,12 @@ async def test_respawn_fallback_rung2_capture_exhausted_keeps_old_id(db, tmux, h
harnesses_capture_fallback["claude-code"].capture_session_id.return_value = None
tmux.is_alive = AsyncMock(return_value=False)
tmux.respawn_verified = AsyncMock(side_effect=[
SpawnResult(ok=False, window_id=None, error="no transcript"),
SpawnResult(ok=True, window_id="@2", error=""),
])
tmux.respawn_verified = AsyncMock(
side_effect=[
SpawnResult(ok=False, window_id=None, error="no transcript"),
SpawnResult(ok=True, window_id="@2", error=""),
]
)
tmux.attach = AsyncMock(return_value=True)
with caplog.at_level(logging.WARNING):
@@ -692,11 +841,14 @@ async def test_respawn_fallback_rung2_capture_exhausted_keeps_old_id(db, tmux, h
assert ok is True
from hqt.db.models import Session as DBSession
updated = db.get(DBSession, sess_id)
# Old id preserved
assert updated.harness_session_id == old_id
# Warning emitted
warning_messages = [r.message for r in caplog.records if r.levelno == logging.WARNING]
warning_messages = [
r.message for r in caplog.records if r.levelno == logging.WARNING
]
assert any("exhausted" in m for m in warning_messages)
@@ -707,8 +859,12 @@ async def test_respawn_fallback_rung2_no_capture_when_flag_false(db, tmux):
h.name = "claude-code"
h.captures_session_id = False
h.generate_session_id.return_value = "sess-nc"
h.build_spawn_config.return_value = MagicMock(command=["claude"], env={}, cwd=Path("/tmp/myproj"))
h.build_resume_config.return_value = MagicMock(command=["claude", "--resume", "sess-nc"], env={}, cwd=Path("/tmp/myproj"))
h.build_spawn_config.return_value = MagicMock(
command=["claude"], env={}, cwd=Path("/tmp/myproj")
)
h.build_resume_config.return_value = MagicMock(
command=["claude", "--resume", "sess-nc"], env={}, cwd=Path("/tmp/myproj")
)
harnesses_nc = {"claude-code": h}
service = SessionService(db=db, tmux=tmux, harnesses=harnesses_nc)
@@ -718,10 +874,12 @@ async def test_respawn_fallback_rung2_no_capture_when_flag_false(db, tmux):
h.capture_session_id.reset_mock()
tmux.is_alive = AsyncMock(return_value=False)
tmux.respawn_verified = AsyncMock(side_effect=[
SpawnResult(ok=False, window_id=None, error="no transcript"),
SpawnResult(ok=True, window_id="@2", error=""),
])
tmux.respawn_verified = AsyncMock(
side_effect=[
SpawnResult(ok=False, window_id=None, error="no transcript"),
SpawnResult(ok=True, window_id="@2", error=""),
]
)
tmux.attach = AsyncMock(return_value=True)
await service.attach_session(session_id=sess_id)
@@ -739,7 +897,9 @@ async def test_create_session_no_capture_when_spawn_fails(db, tmux, harnesses_ca
"""create_session: spawn fails → capture_session_id must NOT be called."""
import hqt.sessions.service as svc_mod
tmux.spawn.return_value = SpawnResult(ok=False, window_id=None, error="spawn failed")
tmux.spawn.return_value = SpawnResult(
ok=False, window_id=None, error="spawn failed"
)
async def fake_sleep(t):
pass
@@ -763,7 +923,9 @@ async def test_create_session_returns_result_dataclass(service, db, tmux, harnes
"""create_session returns a CreateSessionResult with spawn_ok=True on success."""
from hqt.sessions.service import CreateSessionResult
result = await service.create_session(project_id=1, harness_name="claude-code", nickname="test")
result = await service.create_session(
project_id=1, harness_name="claude-code", nickname="test"
)
assert isinstance(result, CreateSessionResult)
assert result.spawn_ok is True
assert result.spawn_error == ""
@@ -775,7 +937,9 @@ async def test_create_session_result_propagates_spawn_failure(db, tmux, harnesse
"""create_session result reflects spawn failure."""
from hqt.sessions.service import CreateSessionResult
tmux.spawn.return_value = SpawnResult(ok=False, window_id=None, error="command not found: claude")
tmux.spawn.return_value = SpawnResult(
ok=False, window_id=None, error="command not found: claude"
)
service = SessionService(db=db, tmux=tmux, harnesses=harnesses)
result = await service.create_session(project_id=1, harness_name="claude-code")
tests/test_skills.py
+3 -1
View File
@@ -6,7 +6,9 @@ from hqt.skills.registry import SkillRegistry
def test_discover_skills(tmp_path: Path):
skill_dir = tmp_path / "my-skill"
skill_dir.mkdir()
(skill_dir / "SKILL.md").write_text("---\nname: my-skill\ndescription: A test skill\n---\nSkill content here")
(skill_dir / "SKILL.md").write_text(
"---\nname: my-skill\ndescription: A test skill\n---\nSkill content here"
)
registry = SkillRegistry(tmp_path)
skills = registry.discover()
tests/test_tmux.py
+168 -88
View File
@@ -15,10 +15,14 @@ def runner():
@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, "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
(
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"
@@ -26,9 +30,23 @@ async def test_new_window(runner):
# options, then the per-window Frappé theme options (see new_window).
set_call = runner._exec.call_args_list[2].args
assert set_call[:17] == (
"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",
"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",
)
assert "window-status-current-style" in set_call
# respawn-pane runs the actual command
@@ -51,9 +69,9 @@ async def test_new_window_targets_explicit_free_index(runner):
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
(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"
@@ -61,7 +79,9 @@ async def test_new_window_targets_explicit_free_index(runner):
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}"
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"
@@ -102,7 +122,13 @@ async def test_select_window(runner):
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",
"respawn-pane",
"-t",
"hqt-main:=hqt-1",
"-k",
"-c",
"/tmp",
"kiro-cli chat",
)
@@ -116,10 +142,10 @@ async def test_list_windows(runner):
@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
(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)
@@ -150,7 +176,7 @@ async def test_manager_is_alive_true(runner):
# 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
(0, "0\n", ""), # is_pane_dead
]
mgr = TmuxManager(runner)
assert await mgr.is_alive("hqt-1") is True
@@ -161,7 +187,7 @@ 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
(0, "1\n", ""), # is_pane_dead
]
mgr = TmuxManager(runner)
assert await mgr.is_alive("hqt-1") is False
@@ -199,10 +225,10 @@ async def test_has_window_false_single_exec(runner):
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
(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"
@@ -226,9 +252,23 @@ async def test_new_window_race_free_sequence(runner):
set_args = calls[2].args
assert set_args[:17] == (
"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",
"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",
)
# Window-scoped theme options painted on this window, current-style = peach.
assert "window-status-current-style" in set_args
@@ -249,12 +289,14 @@ async def test_new_window_race_free_sequence(runner):
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, "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"})
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
@@ -278,10 +320,10 @@ async def test_new_window_with_env(runner):
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
(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
(0, "", ""), # kill-window cleanup
]
result = await runner.new_window("hqt-3", "/tmp", "claude")
assert result is None
@@ -296,11 +338,11 @@ async def test_new_window_set_option_failure_aborts(runner):
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
(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
@@ -314,7 +356,7 @@ async def test_new_window_respawn_failure_kills_window(runner):
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, "0\n", ""), # list-windows (next-index query)
(0, "@42\n", ""),
(0, "", ""),
(0, "", ""),
@@ -327,7 +369,7 @@ async def test_new_window_returns_window_id(runner):
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)
(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")
@@ -354,8 +396,8 @@ async def test_verify_window_alive_alive(runner):
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
(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
@@ -366,8 +408,8 @@ async def test_verify_window_alive_dead_with_text(runner):
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
(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
@@ -379,9 +421,9 @@ 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
(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
@@ -397,10 +439,10 @@ async def test_verify_window_alive_dies_during_last_interval(runner):
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
(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)
@@ -416,10 +458,10 @@ async def test_spawn_result_ok(runner):
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
(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)
@@ -433,14 +475,16 @@ async def test_spawn_result_failure_verification(runner):
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, "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")
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
@@ -449,6 +493,7 @@ async def test_spawn_shlex_join(runner):
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
@@ -457,14 +502,16 @@ async def test_spawn_shlex_join(runner):
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, "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"})
req = SpawnRequest(
window_name="hqt-env", command=["bash"], cwd="/tmp", env={"MY_VAR": "hello"}
)
result = await mgr.spawn(req)
assert result.ok is True
@@ -480,11 +527,12 @@ async def test_spawn_passes_env(runner):
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"})
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
@@ -501,15 +549,14 @@ async def test_respawn_pane_no_env(runner):
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
(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, ""))
@@ -531,7 +578,7 @@ async def test_respawn_verified_success_path(runner):
runner.verify_window_alive = AsyncMock(return_value=(True, ""))
runner._exec.side_effect = [
(0, "hqt-1\n", ""), # has_window → True
(0, "", ""), # respawn-pane
(0, "", ""), # respawn-pane
]
mgr = TmuxManager(runner)
result = await mgr.respawn_verified("hqt-1", ["claude", "--resume", "abc"], "/tmp")
@@ -545,7 +592,7 @@ async def test_respawn_verified_pane_dies_after_respawn(runner):
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
(0, "", ""), # respawn-pane
]
mgr = TmuxManager(runner)
result = await mgr.respawn_verified("hqt-1", ["claude", "--resume", "abc"], "/tmp")
@@ -559,7 +606,7 @@ async def test_respawn_verified_pane_dies_empty_text_fallback_message(runner):
runner.verify_window_alive = AsyncMock(return_value=(False, ""))
runner._exec.side_effect = [
(0, "hqt-1\n", ""), # has_window → True
(0, "", ""), # respawn-pane
(0, "", ""), # respawn-pane
]
mgr = TmuxManager(runner)
result = await mgr.respawn_verified("hqt-1", ["claude", "--resume", "abc"], "/tmp")
@@ -572,14 +619,16 @@ 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
(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")
result = await mgr.respawn_verified(
"hqt-gone", ["claude", "--resume", "abc"], "/tmp"
)
assert result.ok is True
assert result.window_id == "@9"
@@ -588,12 +637,14 @@ async def test_respawn_verified_window_gone_branch(runner):
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
(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")
result = await mgr.respawn_verified(
"hqt-gone", ["claude", "--resume", "abc"], "/tmp"
)
assert result.ok is False
assert result.error != ""
@@ -604,7 +655,7 @@ async def test_respawn_verified_forwards_env(runner):
runner.verify_window_alive = AsyncMock(return_value=(True, ""))
runner._exec.side_effect = [
(0, "hqt-1\n", ""), # has_window → True
(0, "", ""), # respawn-pane
(0, "", ""), # respawn-pane
]
mgr = TmuxManager(runner)
result = await mgr.respawn_verified(
@@ -621,10 +672,11 @@ async def test_respawn_verified_forwards_env(runner):
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
(0, "", ""), # respawn-pane
]
mgr = TmuxManager(runner)
await mgr.respawn_verified("hqt-1", ["cmd with space", "arg"], "/tmp")
@@ -641,13 +693,15 @@ async def test_respawn_verified_window_gone_forwards_env(runner):
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
(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"})
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
@@ -667,7 +721,12 @@ 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",
"send-keys",
"-t",
"hqt-main:=hqt-1",
"-l",
"--",
"Enter; echo hi C-c",
)
@@ -676,7 +735,10 @@ 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",
"send-keys",
"-t",
"hqt-main:=hqt-1",
"Enter",
)
@@ -690,11 +752,16 @@ 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",
"list-panes",
"-s",
"-t",
"hqt-main",
"-F",
"#{window_name}\t#{pane_dead}\t#{window_activity}",
)
@@ -708,6 +775,7 @@ async def test_list_windows_info_multi_pane_alive_max_activity(runner):
"",
)
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)
@@ -747,7 +815,12 @@ 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",
"send-keys",
"-t",
"hqt-main:=hqt-1",
"-l",
"--",
"-flag-looking text",
)
@@ -756,7 +829,12 @@ 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",
"send-keys",
"-t",
"hqt-main:=hqt-1",
"-l",
"--",
"hello world",
)
@@ -765,6 +843,7 @@ 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
@@ -881,7 +960,8 @@ async def test_apply_theme_paints_window_options_on_every_window(runner):
for window in ("⌂ HQT", "hqt-1", "hqt-2"):
target = f"hqt-main:={window}"
idxs = [
i for i, a in enumerate(args)
i
for i, a in enumerate(args)
if a == "window-status-current-style" and args[i - 1] == target
]
assert idxs, f"no current-style set for {window}"
tests/test_tui.py
+99 -44
View File
@@ -17,7 +17,7 @@ def mock_services(tmp_path, monkeypatch):
@pytest.mark.asyncio
async def test_app_mounts():
app = HqtApp()
async with app.run_test() as pilot:
async with app.run_test():
assert app.title == "HQ Terminal"
@@ -27,16 +27,17 @@ async def test_no_header_bar():
from textual.widgets import Header
app = HqtApp()
async with app.run_test() as pilot:
async with app.run_test():
assert len(app.query(Header)) == 0
@pytest.mark.asyncio
async def test_app_has_project_and_session_panels():
app = HqtApp()
async with app.run_test() as pilot:
async with app.run_test():
from hqt.tui.widgets.project_list import ProjectList
from hqt.tui.widgets.session_list import SessionList
assert app.query_one(ProjectList) is not None
assert app.query_one(SessionList) is not None
@@ -45,7 +46,7 @@ async def test_app_has_project_and_session_panels():
async def test_session_refresh_no_duplicate_ids():
"""Regression test: rapid session refreshes must not crash with DuplicateIds."""
app = HqtApp()
async with app.run_test(size=(120, 40)) as pilot:
async with app.run_test(size=(120, 40)):
from hqt.db.models import Project, Harness, Session
proj = Project(name="test", path="/tmp/test")
@@ -116,9 +117,12 @@ async def test_session_selection_preserved_across_refresh():
# ---------------------------------------------------------------------------
def _make_session_info(session_id: int, name: str, harness_name: str, status: str, alive: bool):
def _make_session_info(
session_id: int, name: str, harness_name: str, status: str, alive: bool
):
"""Build a minimal SessionInfo-like object for widget testing."""
from hqt.sessions.service import SessionInfo
harness = MagicMock()
harness.name = harness_name
session = MagicMock()
@@ -150,13 +154,23 @@ async def test_session_list_label_includes_status():
label_texts = [str(child.query_one(Label).render()) for child in lv.children]
# Every label must contain the status word
assert any("working" in t for t in label_texts), f"Expected 'working' in labels: {label_texts}"
assert any("idle" in t for t in label_texts), f"Expected 'idle' in labels: {label_texts}"
assert any("dead" in t for t in label_texts), f"Expected 'dead' in labels: {label_texts}"
assert any("working" in t for t in label_texts), (
f"Expected 'working' in labels: {label_texts}"
)
assert any("idle" in t for t in label_texts), (
f"Expected 'idle' in labels: {label_texts}"
)
assert any("dead" in t for t in label_texts), (
f"Expected 'dead' in labels: {label_texts}"
)
# Symbols: working → ◐, idle → ●, dead → ○
assert any("◐" in t for t in label_texts), f"Expected '◐' in labels: {label_texts}"
assert any("○" in t for t in label_texts), f"Expected '○' in labels: {label_texts}"
assert any("◐" in t for t in label_texts), (
f"Expected '◐' in labels: {label_texts}"
)
assert any("○" in t for t in label_texts), (
f"Expected '○' in labels: {label_texts}"
)
@pytest.mark.asyncio
@@ -270,7 +284,10 @@ async def test_new_session_on_dismiss_sequential_create_then_refresh():
await asyncio.sleep(0.05)
call_order.append("create")
from hqt.sessions.service import CreateSessionResult
return CreateSessionResult(session=MagicMock(), spawn_ok=True, spawn_error="")
return CreateSessionResult(
session=MagicMock(), spawn_ok=True, spawn_error=""
)
async def fake_list(project_id):
call_order.append("list")
@@ -284,6 +301,7 @@ async def test_new_session_on_dismiss_sequential_create_then_refresh():
# Patch discover_harnesses so the early-return guard doesn't fire.
from unittest.mock import patch
with patch("hqt.tui.app.discover_harnesses", return_value={"claude": object()}):
app.action_new_session()
await pilot.pause()
@@ -296,7 +314,9 @@ async def test_new_session_on_dismiss_sequential_create_then_refresh():
await app.workers.wait_for_complete()
await pilot.pause()
assert call_order == ["create", "list"], f"Expected create before list, got: {call_order}"
assert call_order == ["create", "list"], (
f"Expected create before list, got: {call_order}"
)
# ---------------------------------------------------------------------------
@@ -360,11 +380,17 @@ async def test_new_session_on_dismiss_notifies_on_spawn_failure():
await pilot.pause()
# Should have notified with severity="error"
error_notifs = [(m, kw) for m, kw in notify_calls if kw.get("severity") == "error"]
assert len(error_notifs) >= 1, f"Expected error notification, got: {notify_calls}"
error_notifs = [
(m, kw) for m, kw in notify_calls if kw.get("severity") == "error"
]
assert len(error_notifs) >= 1, (
f"Expected error notification, got: {notify_calls}"
)
# Message should contain the first line of the error
msg_text = error_notifs[0][0]
assert "command not found" in msg_text or "claude" in msg_text, f"Unexpected message: {msg_text}"
assert "command not found" in msg_text or "claude" in msg_text, (
f"Unexpected message: {msg_text}"
)
@pytest.mark.asyncio
@@ -394,7 +420,9 @@ async def test_new_session_on_dismiss_no_notify_on_success():
app._db_session.add(fake_sess)
app._db_session.flush()
ok_result = CreateSessionResult(session=fake_sess, spawn_ok=True, spawn_error="")
ok_result = CreateSessionResult(
session=fake_sess, spawn_ok=True, spawn_error=""
)
mock_service = MagicMock()
mock_service.create_session = AsyncMock(return_value=ok_result)
@@ -417,8 +445,12 @@ async def test_new_session_on_dismiss_no_notify_on_success():
await app.workers.wait_for_complete()
await pilot.pause()
error_notifs = [(m, kw) for m, kw in notify_calls if kw.get("severity") == "error"]
assert len(error_notifs) == 0, f"Expected no error notification, got: {error_notifs}"
error_notifs = [
(m, kw) for m, kw in notify_calls if kw.get("severity") == "error"
]
assert len(error_notifs) == 0, (
f"Expected no error notification, got: {error_notifs}"
)
# ---------------------------------------------------------------------------
@@ -469,7 +501,7 @@ async def test_new_session_enter_selects_highlighted_harness():
await pilot.press("enter") # open the dropdown
await pilot.pause()
await pilot.press("down") # move to the second harness
await pilot.press("down") # move to the second harness
await pilot.pause()
await pilot.press("enter") # confirm the highlighted harness
await pilot.pause()
@@ -609,9 +641,9 @@ async def test_edit_project_no_selection_warns():
await app.run_action("edit_project")
await pilot.pause()
assert any(
kw.get("severity") == "warning" for _, kw in notify_calls
), f"Expected warning notification, got: {notify_calls}"
assert any(kw.get("severity") == "warning" for _, kw in notify_calls), (
f"Expected warning notification, got: {notify_calls}"
)
@pytest.mark.asyncio
@@ -694,14 +726,14 @@ async def test_edit_project_duplicate_path_notifies_error():
@pytest.mark.asyncio
async def test_frappe_theme_fully_applied():
app = HqtApp()
async with app.run_test() as pilot:
async with app.run_test():
assert app.current_theme.name == "catppuccin-frappe"
variables = app.get_css_variables()
# Spot-check: explicit values, not Textual auto-derivations
assert variables["primary"] == "#8CAAEE" # Blue
assert variables["background"] == "#292C3C" # Mantle
assert variables["surface"] == "#414559" # Surface0
assert variables["border"] == "#babbf1" # Lavender
assert variables["primary"] == "#8CAAEE" # Blue
assert variables["background"] == "#292C3C" # Mantle
assert variables["surface"] == "#414559" # Surface0
assert variables["border"] == "#babbf1" # Lavender
assert variables["footer-background"] == "#51576d" # Surface1
assert variables["input-cursor-background"] == "#f2d5cf" # Rosewater
@@ -809,10 +841,14 @@ async def test_dialog_button_focus_is_visibly_highlighted():
# $accent (Frappé Peach #ef9f76) fill — peach is red-dominant, so the
# focused button is unmistakably distinct from the blue/ghost defaults.
# (Textual adds a small focus background-tint, so allow ±2 per channel.)
assert abs(bg.r - 0xEF) <= 2 and abs(bg.g - 0x9F) <= 2 and abs(bg.b - 0x76) <= 2, (
f"{btn_id} focus background {bg.hex}, expected ~Peach #EF9F76"
assert (
abs(bg.r - 0xEF) <= 2
and abs(bg.g - 0x9F) <= 2
and abs(bg.b - 0x76) <= 2
), f"{btn_id} focus background {bg.hex}, expected ~Peach #EF9F76"
assert bg.r > bg.b, (
f"{btn_id} focus highlight should be peach, not blue ({bg.hex})"
)
assert bg.r > bg.b, f"{btn_id} focus highlight should be peach, not blue ({bg.hex})"
assert btn.styles.color.hex == "#292C3C", (
f"{btn_id} focus text color {btn.styles.color.hex}, expected dark"
)
@@ -837,7 +873,9 @@ async def test_first_project_selected_on_load():
await pilot.pause()
pl = app.query_one(ProjectList)
first_id = pl.query_one("#project-list").children[0].data
first_item_id = pl.query_one("#project-list").children[0].id
assert first_item_id is not None
first_id = int(first_item_id.removeprefix("proj-"))
assert pl.get_selected_project_id() == first_id
assert app._selected_project_id == first_id
@@ -857,9 +895,24 @@ async def test_first_session_selected_and_remembered_per_project():
pb = Project(name="B", path="/tmp/proj-B")
app._db_session.add_all([pa, pb])
app._db_session.flush()
a1 = Session(project_id=pa.id, harness_id=h.id, tmux_session_name="hqt-a1", archived=False)
a2 = Session(project_id=pa.id, harness_id=h.id, tmux_session_name="hqt-a2", archived=False)
b1 = Session(project_id=pb.id, harness_id=h.id, tmux_session_name="hqt-b1", archived=False)
a1 = Session(
project_id=pa.id,
harness_id=h.id,
tmux_session_name="hqt-a1",
archived=False,
)
a2 = Session(
project_id=pa.id,
harness_id=h.id,
tmux_session_name="hqt-a2",
archived=False,
)
b1 = Session(
project_id=pb.id,
harness_id=h.id,
tmux_session_name="hqt-b1",
archived=False,
)
app._db_session.add_all([a1, a2, b1])
app._db_session.commit()
@@ -901,17 +954,21 @@ def test_format_session_text_colors_and_escapes():
text = format_session_text("mywork", "claude", "working", "◐")
assert text.startswith("[#a6d189]◐[/]") # Green symbol
assert "\\[claude]" in text # escaped, so brackets render
assert "\\[claude]" in text # escaped, so brackets render
assert "working" in text
def test_format_session_text_status_colors():
from hqt.tui.widgets.session_list import format_session_text
assert format_session_text("x", "h", "waiting", "◉").startswith("[#e5c890]") # Yellow
assert format_session_text("x", "h", "idle", "●").startswith("[#81c8be]") # Teal
assert format_session_text("x", "h", "active", "●").startswith("[#81c8be]") # Teal
assert format_session_text("x", "h", "dead", "○").startswith("[#737994]") # Overlay0
assert format_session_text("x", "h", "waiting", "◉").startswith(
"[#e5c890]"
) # Yellow
assert format_session_text("x", "h", "idle", "●").startswith("[#81c8be]") # Teal
assert format_session_text("x", "h", "active", "●").startswith("[#81c8be]") # Teal
assert format_session_text("x", "h", "dead", "○").startswith(
"[#737994]"
) # Overlay0
@pytest.mark.asyncio
@@ -1005,9 +1062,9 @@ async def test_rename_session_no_selection_warns():
await app.run_action("rename_session")
await pilot.pause()
assert any(
kw.get("severity") == "warning" for _, kw in notify_calls
), f"Expected warning notification, got: {notify_calls}"
assert any(kw.get("severity") == "warning" for _, kw in notify_calls), (
f"Expected warning notification, got: {notify_calls}"
)
@pytest.mark.asyncio
@@ -1128,5 +1185,3 @@ async def test_session_list_jk_navigation():
await pilot.press("k")
await pilot.pause()
assert sl.get_selected_session_id() == sessions[0].id