1521 lines
55 KiB
Python
1521 lines
55 KiB
Python
import pytest
|
||
from pathlib import Path
|
||
from unittest.mock import AsyncMock, MagicMock, patch
|
||
|
||
from sqlalchemy import create_engine
|
||
from sqlalchemy.orm import sessionmaker
|
||
from sqlalchemy.pool import StaticPool
|
||
|
||
from hqt.db.models import Base, Harness, Project
|
||
from hqt.errors import ServiceError
|
||
from hqt.sessions.service import SessionInfo, SessionService
|
||
from hqt.tmux.manager import SpawnResult, TmuxManager
|
||
|
||
|
||
@pytest.fixture
|
||
def factory():
|
||
engine = create_engine(
|
||
"sqlite://",
|
||
connect_args={"check_same_thread": False},
|
||
poolclass=StaticPool,
|
||
)
|
||
Base.metadata.create_all(engine)
|
||
return sessionmaker(bind=engine, expire_on_commit=False)
|
||
|
||
|
||
@pytest.fixture
|
||
def db(factory):
|
||
# Seeding/assertion handle onto the same in-memory DB the service uses.
|
||
session = factory()
|
||
session.add(Harness(name="claude-code", display_name="Claude Code"))
|
||
session.add(Project(name="myproj", path="/tmp/myproj"))
|
||
session.commit()
|
||
yield session
|
||
session.close()
|
||
|
||
|
||
@pytest.fixture
|
||
def tmux():
|
||
m = MagicMock(spec=TmuxManager)
|
||
m.spawn = AsyncMock(return_value=SpawnResult(ok=True, window_id="@1", error=""))
|
||
m.kill = AsyncMock()
|
||
m.is_alive = AsyncMock(return_value=True)
|
||
m.poll_info = AsyncMock(return_value={})
|
||
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="")
|
||
)
|
||
return m
|
||
|
||
|
||
@pytest.fixture
|
||
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.parse_status = MagicMock(return_value=None)
|
||
return {"claude-code": h}
|
||
|
||
|
||
@pytest.fixture
|
||
def service(factory, db, tmux, harnesses):
|
||
# depends on db so the seed rows exist before the service runs
|
||
return SessionService(factory=factory, tmux=tmux, harnesses=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"
|
||
)
|
||
sess = result.session
|
||
assert sess.id is not None
|
||
assert sess.tmux_session_name == f"hqt-{sess.id}"
|
||
assert sess.harness_session_id == "sess-1"
|
||
tmux.spawn.assert_called_once()
|
||
harnesses["claude-code"].build_spawn_config.assert_called_once()
|
||
|
||
|
||
@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)
|
||
assert len(result) == 1
|
||
assert isinstance(result[0], SessionInfo)
|
||
assert result[0].alive is True
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_stop_session(service, tmux):
|
||
await service.create_session(project_id=1, harness_name="claude-code")
|
||
await service.stop_session(session_id=1)
|
||
tmux.kill.assert_called_once_with("hqt-1")
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_stop_session_no_kill_when_window_missing(service, db, tmux):
|
||
"""stop_session must NOT call kill when the window doesn't exist."""
|
||
await service.create_session(project_id=1, harness_name="claude-code")
|
||
tmux.window_exists = AsyncMock(return_value=False)
|
||
tmux.kill.reset_mock()
|
||
await service.stop_session(session_id=1)
|
||
tmux.kill.assert_not_called()
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
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"
|
||
)
|
||
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
|
||
# Error was logged and includes captured pane text
|
||
error_messages = [r.message for r in caplog.records if r.levelno == logging.ERROR]
|
||
assert any("command not found" in m for m in error_messages)
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
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")
|
||
)
|
||
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.attach = AsyncMock(return_value=True)
|
||
|
||
await service.attach_session(session_id=1)
|
||
|
||
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"}
|
||
)
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Task 3: fallback ladder + model preserved on resume
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
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")
|
||
)
|
||
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"),
|
||
)
|
||
)
|
||
|
||
tmux.is_alive = AsyncMock(return_value=False)
|
||
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)
|
||
|
||
assert result is True
|
||
tmux.attach.assert_called_once()
|
||
# Only one respawn_verified call (the resume rung), no fallback
|
||
assert tmux.respawn_verified.call_count == 1
|
||
first_cmd = tmux.respawn_verified.call_args_list[0].args[1]
|
||
assert "--resume" in first_cmd
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
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")
|
||
)
|
||
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.attach = AsyncMock(return_value=True)
|
||
|
||
result = await service.attach_session(session_id=1)
|
||
|
||
assert result is True
|
||
# Two respawn_verified calls: first resume, then spawn
|
||
assert tmux.respawn_verified.call_count == 2
|
||
first_cmd = tmux.respawn_verified.call_args_list[0].args[1]
|
||
second_cmd = tmux.respawn_verified.call_args_list[1].args[1]
|
||
assert "--resume" in first_cmd
|
||
assert "--session-id" in second_cmd
|
||
tmux.attach.assert_called_once()
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
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")
|
||
)
|
||
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.attach = AsyncMock(return_value=True)
|
||
|
||
result = await service.attach_session(session_id=1)
|
||
|
||
assert result is False
|
||
tmux.attach.assert_not_called()
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Task 4: captures_session_id flag + bounded async retry in create_session
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
@pytest.fixture
|
||
def harnesses_no_capture():
|
||
"""Configurator with captures_session_id=False (like claude-code)."""
|
||
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")
|
||
)
|
||
return {"claude-code": h}
|
||
|
||
|
||
@pytest.fixture
|
||
def harnesses_capture():
|
||
"""Configurator with captures_session_id=True (like codex)."""
|
||
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")
|
||
)
|
||
return {"claude-code": h}
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_create_session_no_capture_flag_never_calls_capture(
|
||
factory, db, tmux, harnesses_no_capture
|
||
):
|
||
"""When captures_session_id=False, capture_session_id is never called."""
|
||
service = SessionService(factory=factory, tmux=tmux, harnesses=harnesses_no_capture)
|
||
await service.create_session(project_id=1, harness_name="claude-code")
|
||
harnesses_no_capture["claude-code"].capture_session_id.assert_not_called()
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_create_session_capture_retries_until_success(
|
||
factory, 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",
|
||
]
|
||
|
||
sleep_calls = []
|
||
|
||
async def fake_sleep(t):
|
||
sleep_calls.append(t)
|
||
|
||
service = SessionService(factory=factory, tmux=tmux, harnesses=harnesses_capture)
|
||
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 == "captured-id-xyz"
|
||
# capture was called 3 times: first immediate, then twice after sleeps
|
||
assert harnesses_capture["claude-code"].capture_session_id.call_count == 3
|
||
assert len(sleep_calls) == 2
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_create_session_capture_all_none_keeps_placeholder(
|
||
factory, 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
|
||
|
||
harnesses_capture["claude-code"].capture_session_id.return_value = None
|
||
|
||
async def fake_sleep(t):
|
||
pass
|
||
|
||
service = SessionService(factory=factory, 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"
|
||
)
|
||
|
||
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
|
||
)
|
||
# Exhaustion must emit a 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(
|
||
factory, db, tmux, harnesses_capture
|
||
):
|
||
"""captures_session_id=True: first attempt succeeds → no sleeps needed."""
|
||
import hqt.sessions.service as svc_mod
|
||
|
||
harnesses_capture["claude-code"].capture_session_id.return_value = "immediate-id"
|
||
|
||
sleep_calls = []
|
||
|
||
async def fake_sleep(t):
|
||
sleep_calls.append(t)
|
||
|
||
service = SessionService(factory=factory, tmux=tmux, harnesses=harnesses_capture)
|
||
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 == "immediate-id"
|
||
assert len(sleep_calls) == 0 # no sleep needed — succeeded on first try
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Task 7: list_sessions status computation
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_list_sessions_dead_window(factory, db, tmux, harnesses):
|
||
"""list_sessions: dead window → status='dead', alive=False."""
|
||
from hqt.tmux.runner import WindowInfo
|
||
|
||
await SessionService(
|
||
factory=factory, tmux=tmux, harnesses=harnesses
|
||
).create_session(project_id=1, harness_name="claude-code")
|
||
tmux.poll_info.return_value = {"hqt-1": WindowInfo(alive=False, last_activity=0)}
|
||
service = SessionService(factory=factory, tmux=tmux, harnesses=harnesses)
|
||
result = await service.list_sessions(project_id=1)
|
||
assert len(result) == 1
|
||
assert result[0].alive is False
|
||
assert result[0].status == "dead"
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_list_sessions_missing_window_is_dead(factory, db, tmux, harnesses):
|
||
"""list_sessions: window not in poll_info result → status='dead', alive=False."""
|
||
await SessionService(
|
||
factory=factory, tmux=tmux, harnesses=harnesses
|
||
).create_session(project_id=1, harness_name="claude-code")
|
||
tmux.poll_info.return_value = {} # no windows at all
|
||
service = SessionService(factory=factory, tmux=tmux, harnesses=harnesses)
|
||
result = await service.list_sessions(project_id=1)
|
||
assert result[0].alive is False
|
||
assert result[0].status == "dead"
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_list_sessions_alive_parse_working(factory, db, tmux, harnesses):
|
||
"""Alive window + harness parse returns 'working' → status='working'."""
|
||
from hqt.tmux.runner import WindowInfo
|
||
|
||
await SessionService(
|
||
factory=factory, 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=1999)}
|
||
tmux.capture_pane = AsyncMock(return_value="Esc to interrupt\n")
|
||
harnesses["claude-code"].parse_status = MagicMock(return_value="working")
|
||
|
||
service = SessionService(factory=factory, tmux=tmux, harnesses=harnesses)
|
||
result = await service.list_sessions(project_id=1, now=now)
|
||
|
||
assert result[0].status == "working"
|
||
assert result[0].alive is True
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_list_sessions_alive_parse_none_recent_activity(
|
||
factory, db, tmux, harnesses
|
||
):
|
||
"""Alive window + parse_status None + recent activity → status='active'."""
|
||
from hqt.tmux.runner import WindowInfo
|
||
|
||
await SessionService(
|
||
factory=factory, 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.capture_pane = AsyncMock(return_value="some text")
|
||
harnesses["claude-code"].parse_status = MagicMock(return_value=None)
|
||
|
||
service = SessionService(factory=factory, tmux=tmux, harnesses=harnesses)
|
||
result = await service.list_sessions(project_id=1, now=now)
|
||
|
||
assert result[0].status == "active"
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_list_sessions_alive_parse_none_stale_activity(
|
||
factory, db, tmux, harnesses
|
||
):
|
||
"""Alive window + parse_status None + stale activity → status='idle'."""
|
||
from hqt.tmux.runner import WindowInfo
|
||
|
||
await SessionService(
|
||
factory=factory, 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.capture_pane = AsyncMock(return_value="some text")
|
||
harnesses["claude-code"].parse_status = MagicMock(return_value=None)
|
||
|
||
service = SessionService(factory=factory, tmux=tmux, harnesses=harnesses)
|
||
result = await service.list_sessions(project_id=1, now=now)
|
||
|
||
assert result[0].status == "idle"
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_list_sessions_capture_empty_falls_back_to_activity(
|
||
factory, db, tmux, harnesses
|
||
):
|
||
"""capture_pane returns empty text → activity-based fallback, no crash."""
|
||
from hqt.tmux.runner import WindowInfo
|
||
|
||
await SessionService(
|
||
factory=factory, 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=1999)}
|
||
tmux.capture_pane = AsyncMock(return_value="") # empty — treat as capture failure
|
||
harnesses["claude-code"].parse_status = MagicMock(return_value=None)
|
||
|
||
service = SessionService(factory=factory, tmux=tmux, harnesses=harnesses)
|
||
result = await service.list_sessions(project_id=1, now=now)
|
||
|
||
# Should still produce a valid status via activity
|
||
assert result[0].status in ("active", "idle")
|
||
assert result[0].alive is True
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_list_sessions_alive_parse_waiting(factory, db, tmux, harnesses):
|
||
"""Alive window + parse_status 'waiting' → status='waiting'."""
|
||
from hqt.tmux.runner import WindowInfo
|
||
|
||
await SessionService(
|
||
factory=factory, 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=1999)}
|
||
tmux.capture_pane = AsyncMock(return_value="Do you want to proceed?\n❯ 1. Yes\n")
|
||
harnesses["claude-code"].parse_status = MagicMock(return_value="waiting")
|
||
|
||
service = SessionService(factory=factory, tmux=tmux, harnesses=harnesses)
|
||
result = await service.list_sessions(project_id=1, now=now)
|
||
|
||
assert result[0].status == "waiting"
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# sync_window_labels: status-reflecting tmux window titles (session-wide)
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_sync_window_labels_sets_status_symbol(factory, db, tmux, harnesses):
|
||
"""sync_window_labels pushes a '<symbol> <nickname>' label per session."""
|
||
from hqt.tmux.runner import WindowInfo
|
||
|
||
service = SessionService(factory=factory, tmux=tmux, harnesses=harnesses)
|
||
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)}
|
||
tmux.capture_pane = AsyncMock(return_value="Esc to interrupt\n")
|
||
harnesses["claude-code"].parse_status = MagicMock(return_value="working")
|
||
tmux.set_window_label = AsyncMock()
|
||
|
||
infos = await service.sync_window_labels(now=now)
|
||
|
||
assert len(infos) == 1
|
||
assert infos[0].status == "working"
|
||
# Label uses the working glyph (◐) and the nickname.
|
||
tmux.set_window_label.assert_awaited_once_with("hqt-1", "◐ mywork")
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_sync_window_labels_dead_window_uses_dead_glyph(
|
||
factory, db, tmux, harnesses
|
||
):
|
||
"""A dead/missing window gets the ○ glyph and falls back to the window name."""
|
||
from hqt.tmux.runner import WindowInfo
|
||
|
||
service = SessionService(factory=factory, tmux=tmux, harnesses=harnesses)
|
||
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()
|
||
|
||
infos = await service.sync_window_labels()
|
||
|
||
assert infos[0].status == "dead"
|
||
# No nickname → falls back to the tmux window name.
|
||
tmux.set_window_label.assert_awaited_once_with("hqt-1", "○ hqt-1")
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_sync_window_labels_covers_all_projects(factory, db, tmux, harnesses):
|
||
"""sync_window_labels is session-wide: it labels sessions across all projects."""
|
||
from hqt.db.models import Project
|
||
from hqt.tmux.runner import WindowInfo
|
||
|
||
db.add(Project(name="proj2", path="/tmp/proj2"))
|
||
db.commit()
|
||
|
||
service = SessionService(factory=factory, tmux=tmux, harnesses=harnesses)
|
||
await service.create_session(project_id=1, harness_name="claude-code", nickname="a")
|
||
await service.create_session(project_id=2, harness_name="claude-code", nickname="b")
|
||
|
||
tmux.poll_info.return_value = {
|
||
"hqt-1": WindowInfo(alive=False, last_activity=0),
|
||
"hqt-2": WindowInfo(alive=False, last_activity=0),
|
||
}
|
||
tmux.set_window_label = AsyncMock()
|
||
|
||
infos = await service.sync_window_labels()
|
||
|
||
assert len(infos) == 2 # both projects' sessions processed
|
||
labelled = {c.args[0] for c in tmux.set_window_label.await_args_list}
|
||
assert labelled == {"hqt-1", "hqt-2"}
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Item 3: real-configurator integration-seam test
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_list_sessions_real_configurator_waiting(factory, db):
|
||
"""list_sessions with real ClaudeConfigurator + spec'd TmuxManager mock → status='waiting'.
|
||
|
||
Uses the REAL ClaudeConfigurator (not a mock) to close the decorative-fixture
|
||
gap and ensure parse_status is correctly wired end-to-end.
|
||
|
||
TDD story: before the capture_pane delegate is added to TmuxManager,
|
||
MagicMock(spec=TmuxManager).capture_pane raises AttributeError, which propagates
|
||
as an AttributeError from list_sessions. After the delegate is added the spec
|
||
allows the attribute and this test passes.
|
||
"""
|
||
from hqt.tmux.runner import WindowInfo
|
||
from hqt.tmux.manager import TmuxManager
|
||
from hqt.harnesses.configurators.claude import ClaudeConfigurator
|
||
|
||
# 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.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.attach = AsyncMock()
|
||
tmux_spec.window_exists = AsyncMock(return_value=True)
|
||
tmux_spec.respawn_verified = AsyncMock(
|
||
return_value=SpawnResult(ok=True, window_id=None, error="")
|
||
)
|
||
|
||
configurator = ClaudeConfigurator()
|
||
harnesses = {"claude-code": configurator}
|
||
service = SessionService(factory=factory, tmux=tmux_spec, harnesses=harnesses)
|
||
|
||
# Create a session so there is a row in the DB
|
||
await service.create_session(project_id=1, harness_name="claude-code")
|
||
|
||
now = 2000.0
|
||
window_name = "hqt-1"
|
||
tmux_spec.poll_info.return_value = {
|
||
window_name: WindowInfo(alive=True, last_activity=1999)
|
||
}
|
||
|
||
result = await service.list_sessions(project_id=1, now=now)
|
||
|
||
assert len(result) == 1
|
||
assert result[0].status == "waiting"
|
||
assert result[0].alive is True
|
||
tmux_spec.capture_pane.assert_called_once_with(window_name)
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Item 4: capture_pane NOT called for dead/missing windows
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_list_sessions_dead_window_no_capture(factory, db, tmux):
|
||
"""list_sessions: dead window → capture_pane must NOT be called."""
|
||
from hqt.tmux.runner import WindowInfo
|
||
from hqt.harnesses.configurators.claude import ClaudeConfigurator
|
||
|
||
harnesses = {"claude-code": ClaudeConfigurator()}
|
||
service = SessionService(factory=factory, tmux=tmux, harnesses=harnesses)
|
||
await service.create_session(project_id=1, harness_name="claude-code")
|
||
|
||
tmux.poll_info.return_value = {"hqt-1": WindowInfo(alive=False, last_activity=0)}
|
||
tmux.capture_pane.reset_mock()
|
||
|
||
await service.list_sessions(project_id=1)
|
||
|
||
tmux.capture_pane.assert_not_called()
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_list_sessions_missing_window_no_capture(factory, db, tmux):
|
||
"""list_sessions: window missing from poll_info → capture_pane must NOT be called."""
|
||
from hqt.harnesses.configurators.claude import ClaudeConfigurator
|
||
|
||
harnesses = {"claude-code": ClaudeConfigurator()}
|
||
service = SessionService(factory=factory, tmux=tmux, harnesses=harnesses)
|
||
await service.create_session(project_id=1, harness_name="claude-code")
|
||
|
||
tmux.poll_info.return_value = {} # window not present at all
|
||
tmux.capture_pane.reset_mock()
|
||
|
||
await service.list_sessions(project_id=1)
|
||
|
||
tmux.capture_pane.assert_not_called()
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Fix 1: _respawn_with_fallback rung-2 updates harness_session_id
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
@pytest.fixture
|
||
def harnesses_capture_fallback():
|
||
"""Configurator with captures_session_id=True for fallback ladder tests."""
|
||
h = MagicMock()
|
||
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")
|
||
)
|
||
return {"claude-code": h}
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_respawn_fallback_rung2_updates_harness_session_id(
|
||
factory, 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"
|
||
|
||
async def fake_sleep(t):
|
||
pass
|
||
|
||
service = SessionService(
|
||
factory=factory, 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")
|
||
|
||
sess_id = result.session.id
|
||
|
||
# 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"
|
||
|
||
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.attach = AsyncMock(return_value=True)
|
||
|
||
with patch.object(svc_mod, "_capture_sleep", fake_sleep):
|
||
ok = await service.attach_session(session_id=sess_id)
|
||
|
||
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_attach_late_capture_updates_id_before_resume(
|
||
factory, 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(
|
||
factory=factory, 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(
|
||
factory, 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"
|
||
|
||
async def fake_sleep(t):
|
||
pass
|
||
|
||
service = SessionService(
|
||
factory=factory, 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")
|
||
|
||
sess_id = result.session.id
|
||
old_id = result.session.harness_session_id
|
||
|
||
# After create, reset capture to always return None
|
||
harnesses_capture_fallback["claude-code"].capture_session_id.reset_mock()
|
||
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.attach = AsyncMock(return_value=True)
|
||
|
||
with caplog.at_level(logging.WARNING):
|
||
with patch.object(svc_mod, "_capture_sleep", fake_sleep):
|
||
ok = await service.attach_session(session_id=sess_id)
|
||
|
||
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
|
||
]
|
||
assert any("exhausted" in m for m in warning_messages)
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_respawn_fallback_rung2_no_capture_when_flag_false(factory, db, tmux):
|
||
"""Rung-2 success + captures_session_id=False → capture_session_id never called."""
|
||
h = MagicMock()
|
||
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")
|
||
)
|
||
harnesses_nc = {"claude-code": h}
|
||
|
||
service = SessionService(factory=factory, tmux=tmux, harnesses=harnesses_nc)
|
||
result = await service.create_session(project_id=1, harness_name="claude-code")
|
||
sess_id = result.session.id
|
||
|
||
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.attach = AsyncMock(return_value=True)
|
||
|
||
await service.attach_session(session_id=sess_id)
|
||
|
||
h.capture_session_id.assert_not_called()
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Fix 2 (Minor): capture NOT called when spawn_result.ok is False
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_create_session_no_capture_when_spawn_fails(
|
||
factory, db, tmux, harnesses_capture
|
||
):
|
||
"""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"
|
||
)
|
||
|
||
async def fake_sleep(t):
|
||
pass
|
||
|
||
service = SessionService(factory=factory, tmux=tmux, harnesses=harnesses_capture)
|
||
with patch.object(svc_mod, "_capture_sleep", fake_sleep):
|
||
result = await service.create_session(project_id=1, harness_name="claude-code")
|
||
|
||
harnesses_capture["claude-code"].capture_session_id.assert_not_called()
|
||
assert result.spawn_ok is False
|
||
assert "spawn failed" in result.spawn_error
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Fix 3: create_session returns CreateSessionResult
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_create_session_returns_result_dataclass(service, db, tmux, harnesses):
|
||
"""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"
|
||
)
|
||
assert isinstance(result, CreateSessionResult)
|
||
assert result.spawn_ok is True
|
||
assert result.spawn_error == ""
|
||
assert result.session.id is not None
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_create_session_result_propagates_spawn_failure(
|
||
factory, db, tmux, harnesses
|
||
):
|
||
"""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"
|
||
)
|
||
service = SessionService(factory=factory, tmux=tmux, harnesses=harnesses)
|
||
result = await service.create_session(project_id=1, harness_name="claude-code")
|
||
|
||
assert isinstance(result, CreateSessionResult)
|
||
assert result.spawn_ok is False
|
||
assert "command not found" in result.spawn_error
|
||
# Session row was still committed
|
||
assert result.session.id is not None
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# rename_session / get_session
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
def _seed_session(db, nickname=None):
|
||
"""Create a session for project 1 / harness 'claude-code' (seeded by the db fixture)."""
|
||
from hqt.db.models import Harness, Session
|
||
|
||
harness = db.query(Harness).filter_by(name="claude-code").first()
|
||
sess = Session(
|
||
project_id=1,
|
||
harness_id=harness.id,
|
||
nickname=nickname,
|
||
tmux_session_name="hqt-rename",
|
||
archived=False,
|
||
)
|
||
db.add(sess)
|
||
db.commit()
|
||
return sess
|
||
|
||
|
||
def test_rename_session_sets_nickname(service, db):
|
||
from hqt.db.models import Session
|
||
|
||
sess = _seed_session(db, nickname="old")
|
||
service.rename_session(sess.id, "new-name")
|
||
db.expire_all()
|
||
assert db.get(Session, sess.id).nickname == "new-name"
|
||
|
||
|
||
def test_rename_session_empty_clears_to_none(service, db):
|
||
from hqt.db.models import Session
|
||
|
||
sess = _seed_session(db, nickname="old")
|
||
service.rename_session(sess.id, "")
|
||
db.expire_all()
|
||
assert db.get(Session, sess.id).nickname is None
|
||
|
||
|
||
def test_rename_session_whitespace_clears_to_none(service, db):
|
||
from hqt.db.models import Session
|
||
|
||
sess = _seed_session(db, nickname="old")
|
||
service.rename_session(sess.id, " ")
|
||
db.expire_all()
|
||
assert db.get(Session, sess.id).nickname is None
|
||
|
||
|
||
def test_get_session_returns_row(service, db):
|
||
sess = _seed_session(db, nickname="x")
|
||
assert service.get_session(sess.id).id == sess.id
|
||
|
||
|
||
def test_get_session_missing_returns_none(service):
|
||
assert service.get_session(99999) is None
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Task 4: per-operation sessions, ServiceError, eager-load harness
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_create_session_unknown_project_raises(service):
|
||
with pytest.raises(ServiceError, match="no longer exists"):
|
||
await service.create_session(9999, "claude-code")
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_create_session_uninstalled_harness_raises(factory, db, tmux):
|
||
# Harness row exists in the DB but the binary is gone from PATH
|
||
# (self.harnesses has no configurator for it).
|
||
service = SessionService(factory=factory, tmux=tmux, harnesses={})
|
||
with pytest.raises(ServiceError, match="not installed"):
|
||
await service.create_session(1, "claude-code")
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_attach_session_missing_row_raises(service):
|
||
with pytest.raises(ServiceError, match="not found"):
|
||
await service.attach_session(9999)
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_list_sessions_rows_usable_after_return(service, db, tmux):
|
||
await service.create_session(1, "claude-code")
|
||
infos = await service.list_sessions(1)
|
||
# The widget reads session.harness.name on detached rows; selectinload
|
||
# must have eager-loaded the relationship.
|
||
assert infos[0].session.harness.name == "claude-code"
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_list_sessions_skips_capture_for_deleted_project(
|
||
factory, db, tmux, harnesses_capture
|
||
):
|
||
# A session whose project row was deleted must not crash the poller.
|
||
# Capture returns None at create time so the placeholder id is kept; the
|
||
# poller then re-enters the capture path and must SKIP on the missing
|
||
# project rather than raise.
|
||
import hqt.sessions.service as svc_mod
|
||
from hqt.db.models import Session
|
||
|
||
harnesses_capture["claude-code"].capture_session_id.return_value = None
|
||
|
||
async def fake_sleep(t):
|
||
pass
|
||
|
||
service = SessionService(factory=factory, tmux=tmux, harnesses=harnesses_capture)
|
||
with patch.object(svc_mod, "_capture_sleep", fake_sleep):
|
||
await service.create_session(1, "claude-code")
|
||
|
||
# The placeholder id must still be in place so the poller attempts capture.
|
||
db.expire_all()
|
||
fresh = db.query(Session).filter_by(project_id=1).first()
|
||
assert fresh.harness_session_id == "placeholder-id"
|
||
|
||
db.query(Project).delete()
|
||
db.commit()
|
||
|
||
# Capture would now find an id, but the project lookup must short-circuit
|
||
# before it is consulted; either way no ServiceError may escape.
|
||
harnesses_capture["claude-code"].capture_session_id.return_value = "would-capture"
|
||
infos = await service.list_sessions(1)
|
||
assert len(infos) == 1 # no ServiceError escaped
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Task 3: worktree-aware session lifecycle
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
WT_PATH = Path("/tmp/myproj/.worktrees/feature-x")
|
||
|
||
|
||
@pytest.fixture
|
||
def fake_worktree(monkeypatch):
|
||
"""Monkeypatch hqt.git.worktree functions with recording async fakes.
|
||
|
||
Returns a namespace exposing the recorded call args and the WorktreeState
|
||
class so tests can assert ordering and arguments.
|
||
"""
|
||
import hqt.git.worktree as wt_mod
|
||
|
||
class Recorder:
|
||
def __init__(self):
|
||
self.create_calls = []
|
||
self.add_calls = []
|
||
self.remove_calls = []
|
||
self.state_calls = []
|
||
self.create_path = WT_PATH
|
||
self.create_error: Exception | None = None
|
||
self.add_error: Exception | None = None
|
||
self.remove_error: Exception | None = None
|
||
self.state_result = wt_mod.WorktreeState(
|
||
exists=True, dirty=False, unique_commits=2
|
||
)
|
||
|
||
async def create_worktree(self, repo, branch):
|
||
self.create_calls.append((repo, branch))
|
||
if self.create_error is not None:
|
||
raise self.create_error
|
||
return self.create_path
|
||
|
||
async def add_worktree_for_branch(self, repo, branch):
|
||
self.add_calls.append((repo, branch))
|
||
if self.add_error is not None:
|
||
raise self.add_error
|
||
return self.create_path
|
||
|
||
async def remove_worktree(self, repo, path, branch, force):
|
||
self.remove_calls.append((repo, path, branch, force))
|
||
if self.remove_error is not None:
|
||
raise self.remove_error
|
||
|
||
async def worktree_state(self, repo, path, branch):
|
||
self.state_calls.append((repo, path, branch))
|
||
return self.state_result
|
||
|
||
rec = Recorder()
|
||
monkeypatch.setattr(wt_mod, "create_worktree", rec.create_worktree)
|
||
monkeypatch.setattr(wt_mod, "add_worktree_for_branch", rec.add_worktree_for_branch)
|
||
monkeypatch.setattr(wt_mod, "remove_worktree", rec.remove_worktree)
|
||
monkeypatch.setattr(wt_mod, "worktree_state", rec.worktree_state)
|
||
return rec
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_create_session_worktree_branch(
|
||
service, db, tmux, harnesses, fake_worktree
|
||
):
|
||
"""worktree_branch set → create_worktree called with (project_path, branch);
|
||
spawn cwd and capture use the worktree path; row stores both fields."""
|
||
harnesses["claude-code"].captures_session_id = True
|
||
harnesses["claude-code"].capture_session_id.return_value = "cap-id"
|
||
harnesses["claude-code"].build_spawn_config.return_value = MagicMock(
|
||
command=["claude"], env={}, cwd=WT_PATH
|
||
)
|
||
|
||
result = await service.create_session(
|
||
project_id=1, harness_name="claude-code", worktree_branch="feature-x"
|
||
)
|
||
|
||
# create_worktree called with project path + branch
|
||
assert fake_worktree.create_calls == [(Path("/tmp/myproj"), "feature-x")]
|
||
# spawn request cwd is the worktree path
|
||
spawn_req = tmux.spawn.call_args.args[0]
|
||
assert spawn_req.cwd == str(WT_PATH)
|
||
# build_spawn_config used the worktree path, not the project path
|
||
assert harnesses["claude-code"].build_spawn_config.call_args.args[0] == WT_PATH
|
||
# capture used the worktree path
|
||
assert harnesses["claude-code"].capture_session_id.call_args.args[0] == WT_PATH
|
||
# row carries both worktree fields
|
||
sess = result.session
|
||
assert sess.worktree_path == str(WT_PATH)
|
||
assert sess.worktree_branch == "feature-x"
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_create_session_worktree_blank_nickname_defaults_to_branch(
|
||
service, db, tmux, harnesses, fake_worktree
|
||
):
|
||
"""Blank nickname + worktree_branch → nickname defaults to branch name."""
|
||
result = await service.create_session(
|
||
project_id=1,
|
||
harness_name="claude-code",
|
||
nickname="",
|
||
worktree_branch="feature-x",
|
||
)
|
||
assert result.session.nickname == "feature-x"
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_create_session_worktree_explicit_nickname_preserved(
|
||
service, db, tmux, harnesses, fake_worktree
|
||
):
|
||
"""Explicit nickname + worktree_branch → nickname preserved."""
|
||
result = await service.create_session(
|
||
project_id=1,
|
||
harness_name="claude-code",
|
||
nickname="my-work",
|
||
worktree_branch="feature-x",
|
||
)
|
||
assert result.session.nickname == "my-work"
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_create_session_worktree_create_fails_no_row_no_spawn(
|
||
factory, db, tmux, harnesses, fake_worktree
|
||
):
|
||
"""create_worktree raising ServiceError → propagates, no Session row, no spawn."""
|
||
from hqt.db.models import Session
|
||
|
||
fake_worktree.create_error = ServiceError("git worktree add failed")
|
||
service = SessionService(factory=factory, tmux=tmux, harnesses=harnesses)
|
||
|
||
with pytest.raises(ServiceError, match="worktree add failed"):
|
||
await service.create_session(
|
||
project_id=1, harness_name="claude-code", worktree_branch="feature-x"
|
||
)
|
||
|
||
# No Session row committed
|
||
db.expire_all()
|
||
assert db.query(Session).count() == 0
|
||
# tmux spawn never called
|
||
tmux.spawn.assert_not_called()
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_create_session_without_worktree_branch_regression(
|
||
service, db, tmux, harnesses, fake_worktree
|
||
):
|
||
"""No worktree_branch → create_worktree not called, row worktree fields None."""
|
||
result = await service.create_session(project_id=1, harness_name="claude-code")
|
||
assert fake_worktree.create_calls == []
|
||
assert result.session.worktree_path is None
|
||
assert result.session.worktree_branch is None
|
||
|
||
|
||
def _seed_worktree_session(db, branch="feature-x", path=str(WT_PATH)):
|
||
"""Seed a worktree session for project 1 / harness claude-code."""
|
||
from hqt.db.models import Harness, Session
|
||
|
||
harness = db.query(Harness).filter_by(name="claude-code").first()
|
||
sess = Session(
|
||
project_id=1,
|
||
harness_id=harness.id,
|
||
tmux_session_name="hqt-wt",
|
||
harness_session_id="sess-1",
|
||
worktree_path=path,
|
||
worktree_branch=branch,
|
||
archived=False,
|
||
)
|
||
db.add(sess)
|
||
db.commit()
|
||
return sess
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_attach_session_recreates_missing_worktree(
|
||
factory, db, tmux, harnesses, fake_worktree
|
||
):
|
||
"""attach_session: worktree dir missing on disk → add_worktree_for_branch called."""
|
||
# Use a path that does not exist on disk.
|
||
missing = "/tmp/does-not-exist/.worktrees/feature-x"
|
||
sess = _seed_worktree_session(db, path=missing)
|
||
|
||
harnesses["claude-code"].build_resume_config = MagicMock(
|
||
return_value=MagicMock(command=["claude"], env={}, cwd=Path(missing))
|
||
)
|
||
tmux.is_alive = AsyncMock(return_value=False)
|
||
tmux.respawn_verified = AsyncMock(
|
||
return_value=SpawnResult(ok=True, window_id=None, error="")
|
||
)
|
||
tmux.attach = AsyncMock(return_value=True)
|
||
|
||
service = SessionService(factory=factory, tmux=tmux, harnesses=harnesses)
|
||
await service.attach_session(session_id=sess.id)
|
||
|
||
assert fake_worktree.add_calls == [(Path("/tmp/myproj"), "feature-x")]
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_attach_session_missing_worktree_recreate_fails(
|
||
factory, db, tmux, harnesses, fake_worktree
|
||
):
|
||
"""attach_session: recreate raises → ServiceError mentioning deleting the session."""
|
||
missing = "/tmp/does-not-exist/.worktrees/feature-x"
|
||
sess = _seed_worktree_session(db, path=missing)
|
||
fake_worktree.add_error = ServiceError("branch gone")
|
||
|
||
service = SessionService(factory=factory, tmux=tmux, harnesses=harnesses)
|
||
with pytest.raises(ServiceError, match="delete the session"):
|
||
await service.attach_session(session_id=sess.id)
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_attach_session_worktree_resume_uses_worktree_path(
|
||
factory, db, tmux, harnesses, fake_worktree
|
||
):
|
||
"""Resume config for a worktree session uses the worktree path as cwd."""
|
||
# WT_PATH parent exists (/tmp/myproj/.worktrees won't, but use a real existing dir)
|
||
import tempfile
|
||
|
||
with tempfile.TemporaryDirectory() as tmpdir:
|
||
sess = _seed_worktree_session(db, path=tmpdir)
|
||
|
||
resume_cfg = MagicMock(command=["claude", "--resume"], env={}, cwd=Path(tmpdir))
|
||
harnesses["claude-code"].build_resume_config = MagicMock(
|
||
return_value=resume_cfg
|
||
)
|
||
tmux.is_alive = AsyncMock(return_value=False)
|
||
tmux.respawn_verified = AsyncMock(
|
||
return_value=SpawnResult(ok=True, window_id=None, error="")
|
||
)
|
||
tmux.attach = AsyncMock(return_value=True)
|
||
|
||
service = SessionService(factory=factory, tmux=tmux, harnesses=harnesses)
|
||
await service.attach_session(session_id=sess.id)
|
||
|
||
# build_resume_config received the worktree path as the project path arg
|
||
assert harnesses["claude-code"].build_resume_config.call_args.args[0] == Path(
|
||
tmpdir
|
||
)
|
||
# since the worktree exists on disk, recreate must not have been attempted
|
||
assert fake_worktree.add_calls == []
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_delete_session_remove_worktree_true(
|
||
factory, db, tmux, harnesses, fake_worktree
|
||
):
|
||
"""delete_session(remove_worktree=True) calls remove_worktree and deletes the row."""
|
||
from hqt.db.models import Session
|
||
|
||
sess = _seed_worktree_session(db)
|
||
sess_id = sess.id
|
||
|
||
service = SessionService(factory=factory, tmux=tmux, harnesses=harnesses)
|
||
await service.delete_session(sess_id, remove_worktree=True, force=True)
|
||
|
||
assert fake_worktree.remove_calls == [
|
||
(Path("/tmp/myproj"), WT_PATH, "feature-x", True)
|
||
]
|
||
db.expire_all()
|
||
assert db.get(Session, sess_id) is None
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_delete_session_remove_worktree_raises_keeps_row(
|
||
factory, db, tmux, harnesses, fake_worktree
|
||
):
|
||
"""remove_worktree raising → ServiceError propagates, row still present."""
|
||
from hqt.db.models import Session
|
||
|
||
sess = _seed_worktree_session(db)
|
||
sess_id = sess.id
|
||
fake_worktree.remove_error = ServiceError("worktree dirty")
|
||
|
||
service = SessionService(factory=factory, tmux=tmux, harnesses=harnesses)
|
||
with pytest.raises(ServiceError, match="worktree dirty"):
|
||
await service.delete_session(sess_id, remove_worktree=True)
|
||
|
||
db.expire_all()
|
||
assert db.get(Session, sess_id) is not None
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_delete_session_remove_worktree_false_skips_removal(
|
||
factory, db, tmux, harnesses, fake_worktree
|
||
):
|
||
"""remove_worktree=False → remove_worktree not called, row deleted."""
|
||
from hqt.db.models import Session
|
||
|
||
sess = _seed_worktree_session(db)
|
||
sess_id = sess.id
|
||
|
||
service = SessionService(factory=factory, tmux=tmux, harnesses=harnesses)
|
||
await service.delete_session(sess_id, remove_worktree=False)
|
||
|
||
assert fake_worktree.remove_calls == []
|
||
db.expire_all()
|
||
assert db.get(Session, sess_id) is None
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_worktree_state_for_plain_session_returns_none(
|
||
service, db, tmux, harnesses, fake_worktree
|
||
):
|
||
"""worktree_state_for: plain session → None."""
|
||
sess = _seed_session(db)
|
||
assert await service.worktree_state_for(sess.id) is None
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_worktree_state_for_missing_id_returns_none(service, db, fake_worktree):
|
||
"""worktree_state_for: unknown session id → None."""
|
||
assert await service.worktree_state_for(99999) is None
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_worktree_state_for_worktree_session_returns_state(
|
||
factory, db, tmux, harnesses, fake_worktree
|
||
):
|
||
"""worktree_state_for: worktree session → returns the module's WorktreeState."""
|
||
sess = _seed_worktree_session(db)
|
||
service = SessionService(factory=factory, tmux=tmux, harnesses=harnesses)
|
||
|
||
state = await service.worktree_state_for(sess.id)
|
||
|
||
assert state is fake_worktree.state_result
|
||
assert fake_worktree.state_calls == [(Path("/tmp/myproj"), WT_PATH, "feature-x")]
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_repo_path_fallback_slash_branch_when_project_gone(
|
||
factory, db, tmux, harnesses, fake_worktree
|
||
):
|
||
"""When the project row is gone, the repo is derived from the worktree
|
||
layout even for slash-containing branch names (feature/foo).
|
||
|
||
create_worktree builds the path as <repo>/.worktrees/<branch>, so for a
|
||
branch like 'feature/foo' the path is <repo>/.worktrees/feature/foo. The
|
||
fallback must strip '.worktrees/feature/foo' and recover <repo>, not the
|
||
wrong parent.parent result (<repo>/.worktrees).
|
||
"""
|
||
from hqt.db.models import Session
|
||
|
||
repo = Path("/tmp/myproj")
|
||
branch = "feature/foo"
|
||
# Mirror how worktree.create_worktree builds the path.
|
||
wt_path = repo / ".worktrees" / branch
|
||
sess = _seed_worktree_session(db, branch=branch, path=str(wt_path))
|
||
sess_id = sess.id
|
||
|
||
# Delete the project row so the fallback derivation is exercised.
|
||
db.query(Project).delete()
|
||
db.commit()
|
||
db.expire_all()
|
||
# Session row outlives the project (no FK cascade in this seed).
|
||
assert db.get(Session, sess_id) is not None
|
||
|
||
service = SessionService(factory=factory, tmux=tmux, harnesses=harnesses)
|
||
await service.worktree_state_for(sess_id)
|
||
|
||
# The fallback must recover the real repo, not <repo>/.worktrees.
|
||
assert fake_worktree.state_calls == [(repo, wt_path, branch)]
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Task 3 (P2): serialize the spawn->capture window with a lock
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_create_session_holds_capture_lock_for_capturing_harness(
|
||
factory, db, tmux
|
||
):
|
||
observed = {}
|
||
|
||
def _capture(project_path, since):
|
||
# capture runs inside the spawn->capture critical section
|
||
observed["locked_during_capture"] = service._capture_lock.locked()
|
||
return "real-codex-id"
|
||
|
||
h = MagicMock()
|
||
h.captures_session_id = True
|
||
h.generate_session_id.return_value = "placeholder"
|
||
h.build_spawn_config.return_value = MagicMock(
|
||
command=["codex"], env={}, cwd=Path("/tmp/myproj")
|
||
)
|
||
h.capture_session_id.side_effect = _capture
|
||
service = SessionService(factory=factory, tmux=tmux, harnesses={"claude-code": h})
|
||
|
||
result = await service.create_session(project_id=1, harness_name="claude-code")
|
||
|
||
assert observed["locked_during_capture"] is True
|
||
assert result.session.harness_session_id == "real-codex-id"
|
||
# Lock is released after create_session returns.
|
||
assert service._capture_lock.locked() is False
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_create_session_no_lock_for_non_capturing_harness(service, db, tmux):
|
||
observed = {}
|
||
|
||
async def _spawn(req):
|
||
observed["locked_during_spawn"] = service._capture_lock.locked()
|
||
return SpawnResult(ok=True, window_id="@1", error="")
|
||
|
||
tmux.spawn.side_effect = _spawn
|
||
|
||
await service.create_session(project_id=1, harness_name="claude-code")
|
||
|
||
# The `service` fixture's harness has captures_session_id = False, so the
|
||
# spawn must NOT be serialized under the capture lock.
|
||
assert observed["locked_during_spawn"] is False
|