2198 lines
78 KiB
Python
2198 lines
78 KiB
Python
import asyncio
|
|
from unittest.mock import AsyncMock, MagicMock
|
|
|
|
import pytest
|
|
from hqt.tui.app import HqtApp
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def mock_services(tmp_path, monkeypatch):
|
|
"""Mock DB and services so the TUI can mount without real DB/tmux."""
|
|
from hqt.config import Settings
|
|
|
|
settings = Settings(db_path=tmp_path / "test.db")
|
|
monkeypatch.setattr("hqt.tui.app.get_settings", lambda: settings)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_app_mounts():
|
|
app = HqtApp()
|
|
async with app.run_test():
|
|
assert app.title == "HQ Terminal"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_no_header_bar():
|
|
"""The 'HQ Terminal' Header bar is removed for a cleaner, chrome-free top row."""
|
|
from textual.widgets import Header
|
|
|
|
app = HqtApp()
|
|
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():
|
|
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
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
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)):
|
|
from hqt.db.models import Project, Harness, Session
|
|
|
|
seed_db = app._db_factory()
|
|
proj = Project(name="test", path="/tmp/test")
|
|
seed_db.add(proj)
|
|
seed_db.flush()
|
|
harness = seed_db.query(Harness).first()
|
|
sess = Session(
|
|
project_id=proj.id,
|
|
harness_id=harness.id,
|
|
tmux_session_name="hqt-test-dup",
|
|
archived=False,
|
|
)
|
|
seed_db.add(sess)
|
|
seed_db.commit()
|
|
proj_id = proj.id
|
|
seed_db.close()
|
|
|
|
app._selected_project_id = proj_id
|
|
# Call refresh multiple times rapidly (simulates poll timer overlap)
|
|
await app._refresh_sessions()
|
|
await app._refresh_sessions()
|
|
await app._refresh_sessions()
|
|
|
|
# Wait for a poll timer cycle
|
|
await asyncio.sleep(4)
|
|
assert app.is_running
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_session_selection_preserved_across_refresh():
|
|
"""Selection must survive poll-driven refreshes."""
|
|
app = HqtApp()
|
|
async with app.run_test(size=(120, 40)) as pilot:
|
|
from hqt.db.models import Project, Harness, Session
|
|
from hqt.tui.widgets.session_list import SessionList
|
|
|
|
seed_db = app._db_factory()
|
|
proj = Project(name="test", path="/tmp/test-sel")
|
|
seed_db.add(proj)
|
|
seed_db.flush()
|
|
harness = seed_db.query(Harness).first()
|
|
sess = Session(
|
|
project_id=proj.id,
|
|
harness_id=harness.id,
|
|
tmux_session_name="hqt-test-sel",
|
|
archived=False,
|
|
)
|
|
seed_db.add(sess)
|
|
seed_db.commit()
|
|
proj_id = proj.id
|
|
sess_id = sess.id
|
|
seed_db.close()
|
|
|
|
app._selected_project_id = proj_id
|
|
await app._refresh_sessions()
|
|
await pilot.pause()
|
|
|
|
sl = app.query_one(SessionList)
|
|
lv = sl.query_one("#session-list")
|
|
lv.focus()
|
|
await pilot.press("down")
|
|
await pilot.pause()
|
|
|
|
assert sl.get_selected_session_id() == sess_id
|
|
|
|
# Refresh should preserve selection
|
|
await app._refresh_sessions()
|
|
await pilot.pause()
|
|
assert sl.get_selected_session_id() == sess_id
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Task 7: SessionList widget renders status in label
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
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()
|
|
session.id = session_id
|
|
session.nickname = name
|
|
session.tmux_session_name = f"hqt-{session_id}"
|
|
session.harness = harness
|
|
return SessionInfo(session=session, alive=alive, status=status)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_session_list_label_includes_status():
|
|
"""SessionList rows must include status text in their label."""
|
|
app = HqtApp()
|
|
async with app.run_test(size=(120, 40)) as pilot:
|
|
from hqt.tui.widgets.session_list import SessionList
|
|
from textual.widgets import Label
|
|
|
|
sl = app.query_one(SessionList)
|
|
infos = [
|
|
_make_session_info(1, "mywork", "claude", "working", True),
|
|
_make_session_info(2, "idle-sess", "codex", "idle", True),
|
|
_make_session_info(3, "dead-sess", "kiro", "dead", False),
|
|
]
|
|
await sl.refresh_sessions(infos)
|
|
await pilot.pause()
|
|
|
|
lv = sl.query_one("#session-list")
|
|
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}"
|
|
)
|
|
|
|
# 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}"
|
|
)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_session_list_waiting_symbol():
|
|
"""SessionList: 'waiting' status uses ◉ symbol."""
|
|
app = HqtApp()
|
|
async with app.run_test(size=(120, 40)) as pilot:
|
|
from hqt.tui.widgets.session_list import SessionList
|
|
from textual.widgets import Label
|
|
|
|
sl = app.query_one(SessionList)
|
|
infos = [_make_session_info(1, "waiting-sess", "claude", "waiting", True)]
|
|
await sl.refresh_sessions(infos)
|
|
await pilot.pause()
|
|
|
|
lv = sl.query_one("#session-list")
|
|
label_texts = [str(child.query_one(Label).render()) for child in lv.children]
|
|
assert any("waiting" in t for t in label_texts)
|
|
assert any("◉" in t for t in label_texts)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Task 8: Stop session binding + action
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_stop_session_action_calls_service_and_refreshes():
|
|
"""action_stop_session calls stop_session on the service then refreshes."""
|
|
app = HqtApp()
|
|
async with app.run_test(size=(120, 40)) as pilot:
|
|
from hqt.db.models import Project, Harness, Session
|
|
from hqt.tui.widgets.session_list import SessionList
|
|
|
|
seed_db = app._db_factory()
|
|
proj = Project(name="stop-proj", path="/tmp/stop-proj")
|
|
seed_db.add(proj)
|
|
seed_db.flush()
|
|
harness = seed_db.query(Harness).first()
|
|
sess = Session(
|
|
project_id=proj.id,
|
|
harness_id=harness.id,
|
|
tmux_session_name="hqt-stop-test",
|
|
archived=False,
|
|
)
|
|
seed_db.add(sess)
|
|
seed_db.commit()
|
|
proj_id = proj.id
|
|
sess_id = sess.id
|
|
seed_db.close()
|
|
|
|
app._selected_project_id = proj_id
|
|
await app._refresh_sessions()
|
|
await pilot.pause()
|
|
|
|
# Select the session row
|
|
sl = app.query_one(SessionList)
|
|
lv = sl.query_one("#session-list")
|
|
lv.focus()
|
|
await pilot.press("down")
|
|
await pilot.pause()
|
|
assert sl.get_selected_session_id() == sess_id
|
|
|
|
# Replace service with a mock that records calls
|
|
mock_service = MagicMock()
|
|
mock_service.stop_session = AsyncMock()
|
|
mock_service.list_sessions = AsyncMock(return_value=[])
|
|
mock_service.sync_window_labels = AsyncMock(return_value=[])
|
|
app._session_service = mock_service
|
|
|
|
# Invoke the action
|
|
await app.run_action("stop_session")
|
|
await pilot.pause()
|
|
|
|
mock_service.stop_session.assert_awaited_once_with(sess_id)
|
|
mock_service.list_sessions.assert_awaited()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_stop_session_binding_exists():
|
|
"""'s' key must be declared in BINDINGS."""
|
|
bindings = {b.key: b for b in HqtApp.BINDINGS}
|
|
assert "s" in bindings
|
|
assert bindings["s"].action == "stop_session"
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Task 8: new_session on_dismiss — sequential create then refresh (Fix 2)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_new_session_on_dismiss_sequential_create_then_refresh():
|
|
"""on_dismiss path: create_session completes before list_sessions is called.
|
|
|
|
Exercises the real action_new_session / on_dismiss path in app.py so that
|
|
reverting to two separate workers would break this test.
|
|
"""
|
|
app = HqtApp()
|
|
async with app.run_test(size=(120, 40)) as pilot:
|
|
from hqt.db.models import Project
|
|
|
|
seed_db = app._db_factory()
|
|
proj = Project(name="seq-proj", path="/tmp/seq-proj")
|
|
seed_db.add(proj)
|
|
seed_db.flush()
|
|
seed_db.commit()
|
|
proj_id = proj.id
|
|
seed_db.close()
|
|
app._selected_project_id = proj_id
|
|
|
|
call_order: list[str] = []
|
|
|
|
async def fake_create(
|
|
project_id,
|
|
harness_name,
|
|
nickname,
|
|
model,
|
|
worktree_branch=None,
|
|
sandbox=None,
|
|
):
|
|
# Brief yield so that, if two separate workers were used, the list
|
|
# worker would be able to start and record "list" before this
|
|
# completes — exposing the regression.
|
|
await asyncio.sleep(0.05)
|
|
call_order.append("create")
|
|
from hqt.sessions.service import CreateSessionResult
|
|
|
|
return CreateSessionResult(
|
|
session=MagicMock(), spawn_ok=True, spawn_error=""
|
|
)
|
|
|
|
async def fake_list(project_id, archived=False):
|
|
call_order.append("list")
|
|
return []
|
|
|
|
mock_service = MagicMock()
|
|
mock_service.create_session = AsyncMock(side_effect=fake_create)
|
|
mock_service.list_sessions = AsyncMock(side_effect=fake_list)
|
|
mock_service.sync_window_labels = AsyncMock(return_value=[])
|
|
app._session_service = mock_service
|
|
|
|
# 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 app.workers.wait_for_complete()
|
|
await pilot.pause()
|
|
|
|
# The NewSessionScreen is now on top; dismiss it with a valid result tuple.
|
|
app.screen.dismiss(("claude", "mynick", None, None, None))
|
|
await pilot.pause()
|
|
|
|
# Wait for the worker spawned by on_dismiss to finish deterministically.
|
|
await app.workers.wait_for_complete()
|
|
await pilot.pause()
|
|
|
|
assert call_order == ["create", "list"], (
|
|
f"Expected create before list, got: {call_order}"
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Fix 3: TUI notifies with error when spawn fails
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_new_session_on_dismiss_notifies_on_spawn_failure():
|
|
"""on_dismiss: failing spawn → app.notify called with severity='error' containing error text."""
|
|
from unittest.mock import patch, AsyncMock, MagicMock
|
|
from hqt.sessions.service import CreateSessionResult
|
|
from hqt.db.models import Session as DBSession, Harness
|
|
|
|
app = HqtApp()
|
|
async with app.run_test(size=(120, 40)) as pilot:
|
|
from hqt.db.models import Project
|
|
|
|
seed_db = app._db_factory()
|
|
proj = Project(name="fail-proj", path="/tmp/fail-proj")
|
|
seed_db.add(proj)
|
|
seed_db.flush()
|
|
seed_db.commit()
|
|
proj_id = proj.id
|
|
app._selected_project_id = proj_id
|
|
|
|
# Build a minimal fake Session object
|
|
harness_row = seed_db.query(Harness).first()
|
|
fake_sess = DBSession(
|
|
project_id=proj_id,
|
|
harness_id=harness_row.id,
|
|
tmux_session_name="hqt-fail",
|
|
archived=False,
|
|
)
|
|
seed_db.add(fake_sess)
|
|
seed_db.flush()
|
|
seed_db.expunge(fake_sess)
|
|
seed_db.close()
|
|
|
|
fail_result = CreateSessionResult(
|
|
session=fake_sess,
|
|
spawn_ok=False,
|
|
spawn_error="bash: claude: command not found\nsome extra line",
|
|
)
|
|
|
|
mock_service = MagicMock()
|
|
mock_service.create_session = AsyncMock(return_value=fail_result)
|
|
mock_service.list_sessions = AsyncMock(return_value=[])
|
|
mock_service.sync_window_labels = AsyncMock(return_value=[])
|
|
app._session_service = mock_service
|
|
|
|
notify_calls = []
|
|
|
|
def capture_notify(msg, **kwargs):
|
|
notify_calls.append((msg, kwargs))
|
|
|
|
with patch("hqt.tui.app.discover_harnesses", return_value={"claude": object()}):
|
|
with patch.object(app, "notify", side_effect=capture_notify):
|
|
app.action_new_session()
|
|
await app.workers.wait_for_complete()
|
|
await pilot.pause()
|
|
|
|
app.screen.dismiss(("claude", "mynick", None, None, None))
|
|
await pilot.pause()
|
|
await app.workers.wait_for_complete()
|
|
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}"
|
|
)
|
|
# 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}"
|
|
)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_new_session_on_dismiss_no_notify_on_success():
|
|
"""on_dismiss: successful spawn → no error notify."""
|
|
from unittest.mock import patch, AsyncMock, MagicMock
|
|
from hqt.sessions.service import CreateSessionResult
|
|
from hqt.db.models import Session as DBSession, Harness
|
|
|
|
app = HqtApp()
|
|
async with app.run_test(size=(120, 40)) as pilot:
|
|
from hqt.db.models import Project
|
|
|
|
seed_db = app._db_factory()
|
|
proj = Project(name="ok-proj", path="/tmp/ok-proj")
|
|
seed_db.add(proj)
|
|
seed_db.flush()
|
|
seed_db.commit()
|
|
proj_id = proj.id
|
|
app._selected_project_id = proj_id
|
|
|
|
harness_row = seed_db.query(Harness).first()
|
|
fake_sess = DBSession(
|
|
project_id=proj_id,
|
|
harness_id=harness_row.id,
|
|
tmux_session_name="hqt-ok",
|
|
archived=False,
|
|
)
|
|
seed_db.add(fake_sess)
|
|
seed_db.flush()
|
|
seed_db.expunge(fake_sess)
|
|
seed_db.close()
|
|
|
|
ok_result = CreateSessionResult(
|
|
session=fake_sess, spawn_ok=True, spawn_error=""
|
|
)
|
|
|
|
mock_service = MagicMock()
|
|
mock_service.create_session = AsyncMock(return_value=ok_result)
|
|
mock_service.list_sessions = AsyncMock(return_value=[])
|
|
mock_service.sync_window_labels = AsyncMock(return_value=[])
|
|
app._session_service = mock_service
|
|
|
|
notify_calls = []
|
|
|
|
def capture_notify(msg, **kwargs):
|
|
notify_calls.append((msg, kwargs))
|
|
|
|
with patch("hqt.tui.app.discover_harnesses", return_value={"claude": object()}):
|
|
with patch.object(app, "notify", side_effect=capture_notify):
|
|
app.action_new_session()
|
|
await app.workers.wait_for_complete()
|
|
await pilot.pause()
|
|
|
|
app.screen.dismiss(("claude", "mynick", None, None, None))
|
|
await pilot.pause()
|
|
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}"
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# NewSessionScreen: keyboard selection of a harness
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_attach_enter_binding_disabled_in_modal():
|
|
"""Root cause: the app binds Enter -> attach_session with priority=True, so
|
|
it fires app-wide — including inside modal dialogs, where it swallowed Enter
|
|
before the dialog's widgets could see it. check_action must disable the
|
|
action while a modal screen is active so Enter falls through to the dialog."""
|
|
app = HqtApp()
|
|
async with app.run_test(size=(120, 40)) as pilot:
|
|
from hqt.tui.screens.new_session import NewSessionScreen
|
|
|
|
# On the main screen the attach binding is live.
|
|
assert app.check_action("attach_session", ()) is True
|
|
|
|
app.push_screen(NewSessionScreen(["claude", "codex"], is_git_repo=True))
|
|
await pilot.pause()
|
|
|
|
# With a modal on top, the binding is disabled so Enter reaches the modal.
|
|
assert app.check_action("attach_session", ()) is False
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_new_session_enter_selects_highlighted_harness():
|
|
"""Keyboard: open the dropdown, arrow to the 2nd harness, Enter confirms it.
|
|
|
|
Regression: two bugs made this mouse-only. (1) The app's priority Enter
|
|
binding intercepted Enter inside the modal (fixed via check_action). (2)
|
|
Select defaulted to allow_blank=True, prepending a phantom blank option so
|
|
Down/Enter landed one row off. With both fixed, Down then Enter selects the
|
|
harness the user navigated to.
|
|
"""
|
|
app = HqtApp()
|
|
async with app.run_test(size=(120, 40)) as pilot:
|
|
from hqt.tui.screens.new_session import NewSessionScreen
|
|
from textual.widgets import Select
|
|
|
|
app.push_screen(NewSessionScreen(["claude", "codex", "kiro"], is_git_repo=True))
|
|
await pilot.pause()
|
|
sel = app.screen.query_one("#harness-select", Select)
|
|
sel.focus()
|
|
await pilot.pause()
|
|
|
|
await pilot.press("enter") # open the dropdown
|
|
await pilot.pause()
|
|
await pilot.press("down") # move to the second harness
|
|
await pilot.pause()
|
|
await pilot.press("enter") # confirm the highlighted harness
|
|
await pilot.pause()
|
|
|
|
assert sel.value == "codex", f"expected 'codex', got {sel.value!r}"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_new_session_enter_without_navigation_selects_first_harness():
|
|
"""Pressing Enter to open then Enter to confirm selects the first harness,
|
|
not a blank — so the dialog yields a usable result with two keystrokes."""
|
|
app = HqtApp()
|
|
async with app.run_test(size=(120, 40)) as pilot:
|
|
from hqt.tui.screens.new_session import NewSessionScreen
|
|
from textual.widgets import Select
|
|
|
|
app.push_screen(NewSessionScreen(["claude", "codex", "kiro"], is_git_repo=True))
|
|
await pilot.pause()
|
|
sel = app.screen.query_one("#harness-select", Select)
|
|
sel.focus()
|
|
await pilot.pause()
|
|
|
|
await pilot.press("enter")
|
|
await pilot.pause()
|
|
await pilot.press("enter")
|
|
await pilot.pause()
|
|
|
|
assert sel.value == "claude", f"expected 'claude', got {sel.value!r}"
|
|
assert sel.value != Select.BLANK
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_new_session_enter_in_input_submits_dialog():
|
|
"""Pressing Enter in a text field submits the dialog with the current
|
|
harness selection — completing the flow from the keyboard, no mouse."""
|
|
app = HqtApp()
|
|
async with app.run_test(size=(120, 40)) as pilot:
|
|
from hqt.tui.screens.new_session import NewSessionScreen
|
|
from textual.widgets import Input
|
|
|
|
results: list = []
|
|
app.push_screen(
|
|
NewSessionScreen(["claude", "codex"], is_git_repo=True), results.append
|
|
)
|
|
await pilot.pause()
|
|
|
|
nick = app.screen.query_one("#nickname-input", Input)
|
|
nick.focus()
|
|
nick.value = "mywork"
|
|
await pilot.pause()
|
|
await pilot.press("enter")
|
|
await pilot.pause()
|
|
|
|
assert results == [("claude", "mywork", None, None, None)], f"got {results}"
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# ProjectFormScreen: shared add/edit form
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_project_form_prefills_initial_values():
|
|
app = HqtApp()
|
|
async with app.run_test(size=(120, 40)) as pilot:
|
|
from hqt.tui.screens.project_form import ProjectFormScreen
|
|
from textual.widgets import Input
|
|
|
|
app.push_screen(
|
|
ProjectFormScreen(
|
|
title="Edit Project",
|
|
initial_name="myproj",
|
|
initial_path="/tmp/myproj",
|
|
)
|
|
)
|
|
await pilot.pause()
|
|
assert app.screen.query_one("#name-input", Input).value == "myproj"
|
|
assert app.screen.query_one("#path-input", Input).value == "/tmp/myproj"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_project_form_add_mode_defaults_name_to_basename():
|
|
app = HqtApp()
|
|
async with app.run_test(size=(120, 40)) as pilot:
|
|
from hqt.tui.screens.project_form import ProjectFormScreen
|
|
from textual.widgets import Button, Input
|
|
|
|
results = []
|
|
app.push_screen(ProjectFormScreen(), results.append)
|
|
await pilot.pause()
|
|
app.screen.query_one("#path-input", Input).value = "/tmp/somerepo"
|
|
app.screen.query_one("#ok-btn", Button).press()
|
|
await pilot.pause()
|
|
assert results == [("somerepo", "/tmp/somerepo")]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_project_list_get_selected_project_id(tmp_path):
|
|
app = HqtApp()
|
|
async with app.run_test(size=(120, 40)) as pilot:
|
|
from hqt.tui.widgets.project_list import ProjectList
|
|
|
|
p = app._project_service.create("selproj", str(tmp_path))
|
|
app._load_projects()
|
|
await app.workers.wait_for_complete()
|
|
await pilot.pause()
|
|
|
|
pl = app.query_one(ProjectList)
|
|
lv = pl.query_one("#project-list")
|
|
lv.focus()
|
|
await pilot.press("down")
|
|
await pilot.pause()
|
|
|
|
assert pl.get_selected_project_id() == p.id
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Edit project: binding + action
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_edit_project_binding_exists():
|
|
bindings = {b.key: b for b in HqtApp.BINDINGS}
|
|
assert "e" in bindings
|
|
assert bindings["e"].action == "edit_project"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_edit_project_no_selection_warns():
|
|
app = HqtApp()
|
|
async with app.run_test(size=(120, 40)) as pilot:
|
|
from unittest.mock import patch
|
|
|
|
notify_calls = []
|
|
with patch.object(
|
|
app, "notify", side_effect=lambda msg, **kw: notify_calls.append((msg, kw))
|
|
):
|
|
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}"
|
|
)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_edit_project_opens_prefilled_form_and_updates(tmp_path):
|
|
app = HqtApp()
|
|
async with app.run_test(size=(120, 40)) as pilot:
|
|
from pathlib import Path
|
|
from textual.widgets import Input
|
|
from hqt.tui.screens.project_form import ProjectFormScreen
|
|
from hqt.tui.widgets.project_list import ProjectList
|
|
|
|
old_dir = tmp_path / "oldpath"
|
|
old_dir.mkdir()
|
|
new_dir = tmp_path / "newpath"
|
|
new_dir.mkdir()
|
|
old_path = str(old_dir)
|
|
new_path = str(new_dir)
|
|
|
|
p = app._project_service.create("oldname", old_path)
|
|
app._load_projects()
|
|
await app.workers.wait_for_complete()
|
|
await pilot.pause()
|
|
|
|
pl = app.query_one(ProjectList)
|
|
pl.query_one("#project-list").focus()
|
|
await pilot.press("down")
|
|
await pilot.pause()
|
|
assert pl.get_selected_project_id() == p.id
|
|
|
|
await app.run_action("edit_project")
|
|
await pilot.pause()
|
|
|
|
assert isinstance(app.screen, ProjectFormScreen)
|
|
assert app.screen.query_one("#name-input", Input).value == "oldname"
|
|
assert app.screen.query_one("#path-input", Input).value == str(
|
|
Path(old_path).resolve()
|
|
)
|
|
|
|
app.screen.dismiss(("newname", new_path))
|
|
await pilot.pause()
|
|
|
|
refreshed = app._project_service.get(p.id)
|
|
assert refreshed.name == "newname"
|
|
assert refreshed.path == str(Path(new_path).resolve())
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_edit_project_duplicate_path_notifies_error(tmp_path):
|
|
app = HqtApp()
|
|
async with app.run_test(size=(120, 40)) as pilot:
|
|
from pathlib import Path
|
|
from unittest.mock import patch
|
|
from hqt.tui.widgets.project_list import ProjectList
|
|
|
|
first_dir = tmp_path / "first"
|
|
first_dir.mkdir()
|
|
second_dir = tmp_path / "second"
|
|
second_dir.mkdir()
|
|
first_path = str(first_dir)
|
|
second_path = str(second_dir)
|
|
|
|
app._project_service.create("first", first_path)
|
|
p2 = app._project_service.create("second", second_path)
|
|
app._load_projects()
|
|
await app.workers.wait_for_complete()
|
|
await pilot.pause()
|
|
|
|
pl = app.query_one(ProjectList)
|
|
pl.query_one("#project-list").focus()
|
|
# Two items: press down twice to land on the second project
|
|
await pilot.press("down")
|
|
await pilot.press("down")
|
|
await pilot.pause()
|
|
assert pl.get_selected_project_id() == p2.id
|
|
|
|
notify_calls = []
|
|
with patch.object(
|
|
app, "notify", side_effect=lambda msg, **kw: notify_calls.append((msg, kw))
|
|
):
|
|
await app.run_action("edit_project")
|
|
await pilot.pause()
|
|
# Move "second" onto "first"'s path to trigger the duplicate error.
|
|
app.screen.dismiss(("second", first_path))
|
|
await pilot.pause()
|
|
|
|
assert any(
|
|
kw.get("severity") == "error" and "already uses path" in msg
|
|
for msg, kw in notify_calls
|
|
), f"Expected duplicate-path error notification, got: {notify_calls}"
|
|
# DB unchanged
|
|
assert app._project_service.get(p2.id).path == str(Path(second_path).resolve())
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Catppuccin Frappé theme: full definition
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_frappe_theme_fully_applied():
|
|
app = HqtApp()
|
|
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["footer-background"] == "#51576d" # Surface1
|
|
assert variables["input-cursor-background"] == "#f2d5cf" # Rosewater
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_primary_accent_resolves_to_blue_in_css():
|
|
"""Regression: a $primary-driven CSS rule must resolve to Frappé Blue in
|
|
actual rendered styles. The theme variable being correct is not enough —
|
|
if the theme is applied before it is registered, CSS caches a stale
|
|
$primary (Mauve) even while get_css_variables() reports Blue."""
|
|
app = HqtApp()
|
|
async with app.run_test(size=(120, 40)) as pilot:
|
|
from hqt.tui.widgets.project_list import ProjectList
|
|
|
|
await pilot.pause()
|
|
pl = app.query_one(ProjectList)
|
|
# styles.tcss sets `border-right: solid $primary` on ProjectList.
|
|
_border_style, color = pl.styles.border_right
|
|
assert color.hex == "#8CAAEE", f"expected Frappé Blue, got {color.hex}"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_list_panels_use_dark_background_not_surface():
|
|
"""The ListView bodies must use the dark $background, not Textual's default
|
|
ListView $surface (Surface0 #414559). With $surface they paint the whole
|
|
panel a light grey-blue that clashes with the Catppuccin background."""
|
|
app = HqtApp()
|
|
async with app.run_test(size=(120, 40)) as pilot:
|
|
from hqt.tui.widgets.project_list import ProjectList
|
|
from hqt.tui.widgets.session_list import SessionList
|
|
|
|
await pilot.pause()
|
|
pbg = app.query_one(ProjectList).query_one("#project-list").styles.background
|
|
sbg = app.query_one(SessionList).query_one("#session-list").styles.background
|
|
assert pbg.hex == "#292C3C", f"project list bg {pbg.hex}, expected $background"
|
|
assert sbg.hex == "#292C3C", f"session list bg {sbg.hex}, expected $background"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_project_form_dialog_styled():
|
|
app = HqtApp()
|
|
async with app.run_test(size=(120, 40)) as pilot:
|
|
from hqt.tui.screens.project_form import ProjectFormScreen
|
|
|
|
app.push_screen(ProjectFormScreen())
|
|
await pilot.pause()
|
|
dialog = app.screen.query_one("#project-form-dialog")
|
|
assert dialog.styles.width.value == 60
|
|
assert dialog.styles.border_top[0] == "solid"
|
|
# $surface (Frappé Surface0) — the fill that makes the dialog opaque
|
|
# against the modal backdrop; guards against silent removal.
|
|
assert dialog.styles.background.hex == "#414559"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_dialog_buttons_are_flat_and_side_by_side():
|
|
app = HqtApp()
|
|
async with app.run_test(size=(120, 40)) as pilot:
|
|
from hqt.tui.screens.project_form import ProjectFormScreen
|
|
from textual.widgets import Button
|
|
|
|
app.push_screen(ProjectFormScreen())
|
|
await pilot.pause()
|
|
ok = app.screen.query_one("#ok-btn", Button)
|
|
cancel = app.screen.query_one("#cancel-btn", Button)
|
|
# Flat: a single row, not the default 3-row beveled block.
|
|
assert ok.styles.height.value == 1
|
|
assert cancel.styles.height.value == 1
|
|
# Primary stays filled Blue; Cancel is a ghost (no fill).
|
|
assert ok.styles.background.hex == "#8CAAEE"
|
|
assert ok.styles.color.hex == "#292C3C" # $background, dark text on blue
|
|
assert cancel.styles.background.a == 0 # transparent
|
|
# Laid out on one row, not stacked vertically.
|
|
assert ok.region.y == cancel.region.y
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Dialog button focus must be visibly distinct (Tab-cycling)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_dialog_button_focus_is_visibly_highlighted():
|
|
"""Tabbing onto OK/Cancel must show a clear highlight, not just bold text.
|
|
|
|
The buttons drop Textual's default border + reverse focus styling, so a
|
|
focused button was nearly indistinguishable from an unfocused one. A focused
|
|
button now gets a Peach ($accent) fill with dark text — overriding each
|
|
button's id-level background so the highlight is visible on both OK and
|
|
Cancel."""
|
|
app = HqtApp()
|
|
async with app.run_test(size=(120, 40)) as pilot:
|
|
from hqt.tui.screens.project_form import ProjectFormScreen
|
|
from textual.widgets import Button
|
|
|
|
app.push_screen(ProjectFormScreen())
|
|
await pilot.pause()
|
|
|
|
# Unfocused: OK is blue-dominant, Cancel is transparent.
|
|
for btn_id in ("#ok-btn", "#cancel-btn"):
|
|
btn = app.screen.query_one(btn_id, Button)
|
|
btn.focus()
|
|
await pilot.pause()
|
|
bg = btn.styles.background
|
|
# $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 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"
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Auto-select first project on start; first session + per-project memory
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_first_project_selected_on_load(tmp_path):
|
|
"""If a project exists, the first is selected on load with no key press."""
|
|
app = HqtApp()
|
|
async with app.run_test(size=(120, 40)) as pilot:
|
|
from hqt.tui.widgets.project_list import ProjectList
|
|
|
|
d1 = tmp_path / "p1-auto"
|
|
d1.mkdir()
|
|
d2 = tmp_path / "p2-auto"
|
|
d2.mkdir()
|
|
app._project_service.create("p1", str(d1))
|
|
app._project_service.create("p2", str(d2))
|
|
app._load_projects()
|
|
await app.workers.wait_for_complete()
|
|
await pilot.pause()
|
|
|
|
pl = app.query_one(ProjectList)
|
|
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
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_first_session_selected_and_remembered_per_project():
|
|
"""First session is auto-selected per project; switching projects restores
|
|
each project's previously selected session."""
|
|
app = HqtApp()
|
|
async with app.run_test(size=(120, 40)) as pilot:
|
|
from hqt.db.models import Project, Harness, Session
|
|
from hqt.tui.widgets.project_list import ProjectSelected
|
|
from hqt.tui.widgets.session_list import SessionList
|
|
|
|
seed_db = app._db_factory()
|
|
h = seed_db.query(Harness).first()
|
|
pa = Project(name="A", path="/tmp/proj-A")
|
|
pb = Project(name="B", path="/tmp/proj-B")
|
|
seed_db.add_all([pa, pb])
|
|
seed_db.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,
|
|
)
|
|
seed_db.add_all([a1, a2, b1])
|
|
seed_db.commit()
|
|
pa_id, pb_id = pa.id, pb.id
|
|
a1_id, a2_id, b1_id = a1.id, a2.id, b1.id
|
|
seed_db.close()
|
|
|
|
sl = app.query_one(SessionList)
|
|
|
|
# Enter project A: first session auto-selected.
|
|
app.on_project_selected(ProjectSelected(pa_id))
|
|
await app.workers.wait_for_complete()
|
|
await pilot.pause()
|
|
assert sl.get_selected_session_id() == a1_id
|
|
|
|
# Move selection to the second session in A.
|
|
lv = sl.query_one("#session-list")
|
|
lv.focus()
|
|
await pilot.press("down")
|
|
await pilot.pause()
|
|
assert sl.get_selected_session_id() == a2_id
|
|
|
|
# Switch to project B: its first session is auto-selected.
|
|
app.on_project_selected(ProjectSelected(pb_id))
|
|
await app.workers.wait_for_complete()
|
|
await pilot.pause()
|
|
assert sl.get_selected_session_id() == b1_id
|
|
|
|
# Switch back to A: the previously selected session (a2) is restored.
|
|
app.on_project_selected(ProjectSelected(pa_id))
|
|
await app.workers.wait_for_complete()
|
|
await pilot.pause()
|
|
assert sl.get_selected_session_id() == a2_id
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Status symbol colors + harness-name markup escape
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_format_session_text_colors_and_escapes():
|
|
from hqt.tui.widgets.session_list import format_session_text
|
|
|
|
text = format_session_text("mywork", "claude", "working", "◐")
|
|
assert text.startswith("[#a6d189]◐[/]") # Green symbol
|
|
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
|
|
|
|
|
|
def _make_scheduled_prompt(due_at):
|
|
prompt = MagicMock()
|
|
prompt.due_at = due_at
|
|
prompt.status = "pending"
|
|
return prompt
|
|
|
|
|
|
def test_format_session_text_shows_scheduled_prompt_countdown():
|
|
from datetime import datetime, timedelta
|
|
from hqt.tui.widgets.session_list import format_session_text
|
|
|
|
now = datetime(2026, 6, 11, 14, 0, 0)
|
|
scheduled = _make_scheduled_prompt(now + timedelta(minutes=12, seconds=1))
|
|
|
|
text = format_session_text(
|
|
"mywork",
|
|
"claude",
|
|
"idle",
|
|
"●",
|
|
scheduled_prompt=scheduled,
|
|
now=now,
|
|
)
|
|
|
|
assert "@ 13m" in text
|
|
|
|
|
|
def test_format_session_text_shows_scheduled_prompt_clock_time():
|
|
from datetime import datetime, timedelta
|
|
from hqt.tui.widgets.session_list import format_session_text
|
|
|
|
now = datetime(2026, 6, 11, 14, 0, 0)
|
|
scheduled = _make_scheduled_prompt(now + timedelta(hours=2, minutes=30))
|
|
|
|
text = format_session_text(
|
|
"mywork",
|
|
"claude",
|
|
"idle",
|
|
"●",
|
|
scheduled_prompt=scheduled,
|
|
now=now,
|
|
)
|
|
|
|
assert "@ 16:30" in text
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_session_list_renders_harness_name_brackets():
|
|
"""Regression: '[claude]' must be visible, not swallowed as a markup tag."""
|
|
app = HqtApp()
|
|
async with app.run_test(size=(120, 40)) as pilot:
|
|
from textual.widgets import Label
|
|
from hqt.tui.widgets.session_list import SessionList
|
|
|
|
sl = app.query_one(SessionList)
|
|
infos = [_make_session_info(1, "mywork", "claude", "working", True)]
|
|
await sl.refresh_sessions(infos)
|
|
await pilot.pause()
|
|
|
|
lv = sl.query_one("#session-list")
|
|
texts = [str(child.query_one(Label).render()) for child in lv.children]
|
|
assert any("[claude]" in t for t in texts), f"harness name missing: {texts}"
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# SchedulePromptScreen: parse + submit
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_parse_delay_accepts_compact_seconds_minutes_hours():
|
|
from datetime import timedelta
|
|
|
|
from hqt.tui.screens.schedule_prompt import parse_delay
|
|
|
|
assert parse_delay("90s") == timedelta(seconds=90)
|
|
assert parse_delay("30m") == timedelta(minutes=30)
|
|
assert parse_delay("2h") == timedelta(hours=2)
|
|
assert parse_delay("1h30m") == timedelta(hours=1, minutes=30)
|
|
|
|
|
|
@pytest.mark.parametrize("value", ["", "0m", "abc", "1d", "1h 30m"])
|
|
def test_parse_delay_rejects_invalid_values(value):
|
|
from hqt.tui.screens.schedule_prompt import parse_delay
|
|
|
|
with pytest.raises(ValueError):
|
|
parse_delay(value)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.parametrize("submit_action", ["enter", "ok"])
|
|
async def test_schedule_prompt_screen_submits_prompt_and_delay(submit_action):
|
|
from datetime import timedelta
|
|
|
|
from hqt.tui.screens.schedule_prompt import SchedulePromptScreen
|
|
from textual.widgets import Button, Input
|
|
|
|
app = HqtApp()
|
|
async with app.run_test(size=(120, 40)) as pilot:
|
|
results = []
|
|
app.push_screen(SchedulePromptScreen(), results.append)
|
|
await pilot.pause()
|
|
|
|
prompt = app.screen.query_one("#prompt-input", Input)
|
|
prompt.value = "continue work"
|
|
app.screen.query_one("#delay-input", Input).value = "1h30m"
|
|
prompt.focus()
|
|
await pilot.pause()
|
|
if submit_action == "enter":
|
|
await pilot.press("enter")
|
|
else:
|
|
app.screen.query_one("#ok-btn", Button).press()
|
|
await pilot.pause()
|
|
|
|
assert results == [("continue work", timedelta(hours=1, minutes=30))]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_schedule_prompt_screen_invalid_delay_stays_open_with_error():
|
|
from hqt.tui.screens.schedule_prompt import SchedulePromptScreen
|
|
from textual.widgets import Button, Input, Label
|
|
|
|
app = HqtApp()
|
|
async with app.run_test(size=(120, 40)) as pilot:
|
|
results = []
|
|
app.push_screen(SchedulePromptScreen(), results.append)
|
|
await pilot.pause()
|
|
|
|
app.screen.query_one("#prompt-input", Input).value = "continue work"
|
|
app.screen.query_one("#delay-input", Input).value = "1h 30m"
|
|
app.screen.query_one("#ok-btn", Button).press()
|
|
await pilot.pause()
|
|
|
|
assert isinstance(app.screen, SchedulePromptScreen)
|
|
assert results == []
|
|
error = app.screen.query_one("#schedule-error", Label)
|
|
assert "Use a delay like" in str(error.render())
|
|
assert app.screen.focused == app.screen.query_one("#delay-input", Input)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_schedule_prompt_screen_empty_prompt_stays_open_with_error():
|
|
from hqt.tui.screens.schedule_prompt import SchedulePromptScreen
|
|
from textual.widgets import Button, Input, Label
|
|
|
|
app = HqtApp()
|
|
async with app.run_test(size=(120, 40)) as pilot:
|
|
results = []
|
|
app.push_screen(SchedulePromptScreen(), results.append)
|
|
await pilot.pause()
|
|
|
|
app.screen.query_one("#prompt-input", Input).value = " "
|
|
app.screen.query_one("#delay-input", Input).value = "30m"
|
|
app.screen.query_one("#ok-btn", Button).press()
|
|
await pilot.pause()
|
|
|
|
assert isinstance(app.screen, SchedulePromptScreen)
|
|
assert results == []
|
|
error = app.screen.query_one("#schedule-error", Label)
|
|
assert "Prompt cannot be empty" in str(error.render())
|
|
assert app.screen.focused == app.screen.query_one("#prompt-input", Input)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_schedule_prompt_screen_cancel_returns_none():
|
|
from hqt.tui.screens.schedule_prompt import SchedulePromptScreen
|
|
from textual.widgets import Button
|
|
|
|
app = HqtApp()
|
|
async with app.run_test(size=(120, 40)) as pilot:
|
|
results = []
|
|
app.push_screen(SchedulePromptScreen(), results.append)
|
|
await pilot.pause()
|
|
|
|
app.screen.query_one("#cancel-btn", Button).press()
|
|
await pilot.pause()
|
|
|
|
assert results == [None]
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# RenameSessionScreen: prefill + submit
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_rename_screen_prefills_current_nickname():
|
|
app = HqtApp()
|
|
async with app.run_test(size=(120, 40)) as pilot:
|
|
from hqt.tui.screens.rename_session import RenameSessionScreen
|
|
from textual.widgets import Input
|
|
|
|
app.push_screen(RenameSessionScreen(initial_nickname="mywork"))
|
|
await pilot.pause()
|
|
assert app.screen.query_one("#nickname-input", Input).value == "mywork"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_rename_screen_ok_returns_stripped_value():
|
|
app = HqtApp()
|
|
async with app.run_test(size=(120, 40)) as pilot:
|
|
from hqt.tui.screens.rename_session import RenameSessionScreen
|
|
from textual.widgets import Button, Input
|
|
|
|
results = []
|
|
app.push_screen(RenameSessionScreen(initial_nickname=""), results.append)
|
|
await pilot.pause()
|
|
app.screen.query_one("#nickname-input", Input).value = " renamed "
|
|
app.screen.query_one("#ok-btn", Button).press()
|
|
await pilot.pause()
|
|
assert results == ["renamed"]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_rename_screen_cancel_returns_none():
|
|
app = HqtApp()
|
|
async with app.run_test(size=(120, 40)) as pilot:
|
|
from hqt.tui.screens.rename_session import RenameSessionScreen
|
|
from textual.widgets import Button
|
|
|
|
results = []
|
|
app.push_screen(RenameSessionScreen(initial_nickname="x"), results.append)
|
|
await pilot.pause()
|
|
app.screen.query_one("#cancel-btn", Button).press()
|
|
await pilot.pause()
|
|
assert results == [None]
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Rename session: binding + action
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_rename_session_binding_exists():
|
|
bindings = {b.key: b for b in HqtApp.BINDINGS}
|
|
assert "r" in bindings
|
|
assert bindings["r"].action == "rename_session"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_rename_session_no_selection_warns():
|
|
app = HqtApp()
|
|
async with app.run_test(size=(120, 40)) as pilot:
|
|
from unittest.mock import patch
|
|
|
|
notify_calls = []
|
|
with patch.object(
|
|
app, "notify", side_effect=lambda msg, **kw: notify_calls.append((msg, kw))
|
|
):
|
|
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}"
|
|
)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_rename_session_opens_prefilled_and_updates():
|
|
app = HqtApp()
|
|
async with app.run_test(size=(120, 40)) as pilot:
|
|
from hqt.db.models import Project, Harness, Session
|
|
from hqt.tui.widgets.session_list import SessionList
|
|
from hqt.tui.screens.rename_session import RenameSessionScreen
|
|
from textual.widgets import Input
|
|
|
|
seed_db = app._db_factory()
|
|
proj = Project(name="ren-proj", path="/tmp/ren-proj")
|
|
seed_db.add(proj)
|
|
seed_db.flush()
|
|
harness = seed_db.query(Harness).first()
|
|
sess = Session(
|
|
project_id=proj.id,
|
|
harness_id=harness.id,
|
|
nickname="before",
|
|
tmux_session_name="hqt-ren",
|
|
archived=False,
|
|
)
|
|
seed_db.add(sess)
|
|
seed_db.commit()
|
|
proj_id = proj.id
|
|
sess_id = sess.id
|
|
seed_db.close()
|
|
|
|
app._selected_project_id = proj_id
|
|
await app._refresh_sessions()
|
|
await pilot.pause()
|
|
|
|
sl = app.query_one(SessionList)
|
|
sl.query_one("#session-list").focus()
|
|
await pilot.press("down")
|
|
await pilot.pause()
|
|
assert sl.get_selected_session_id() == sess_id
|
|
|
|
await app.run_action("rename_session")
|
|
await pilot.pause()
|
|
|
|
assert isinstance(app.screen, RenameSessionScreen)
|
|
assert app.screen.query_one("#nickname-input", Input).value == "before"
|
|
|
|
app.screen.dismiss("after")
|
|
await pilot.pause()
|
|
await app.workers.wait_for_complete()
|
|
await pilot.pause()
|
|
|
|
verify_db = app._db_factory()
|
|
assert verify_db.get(Session, sess_id).nickname == "after"
|
|
verify_db.close()
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Vim-style j/k navigation in the project and session lists
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_project_list_jk_navigation(tmp_path):
|
|
"""j/k move the highlight down/up in the project list like the arrow keys."""
|
|
app = HqtApp()
|
|
async with app.run_test(size=(120, 40)) as pilot:
|
|
from hqt.tui.widgets.project_list import ProjectList
|
|
|
|
d1 = tmp_path / "jk-first"
|
|
d1.mkdir()
|
|
d2 = tmp_path / "jk-second"
|
|
d2.mkdir()
|
|
p1 = app._project_service.create("first", str(d1))
|
|
p2 = app._project_service.create("second", str(d2))
|
|
app._load_projects()
|
|
await app.workers.wait_for_complete()
|
|
await pilot.pause()
|
|
|
|
pl = app.query_one(ProjectList)
|
|
pl.query_one("#project-list").focus()
|
|
await pilot.pause()
|
|
# j walks down to the second project...
|
|
await pilot.press("j")
|
|
await pilot.pause()
|
|
assert pl.get_selected_project_id() == p2.id
|
|
# ...and k walks back up to the first.
|
|
await pilot.press("k")
|
|
await pilot.pause()
|
|
assert pl.get_selected_project_id() == p1.id
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_session_list_jk_navigation():
|
|
"""j/k move the highlight down/up in the session list like the arrow keys."""
|
|
from hqt.db.models import Harness, Project, Session
|
|
from hqt.tui.widgets.session_list import SessionList
|
|
|
|
app = HqtApp()
|
|
async with app.run_test(size=(120, 40)) as pilot:
|
|
seed_db = app._db_factory()
|
|
proj = Project(name="jk-proj", path="/tmp/jk-proj")
|
|
seed_db.add(proj)
|
|
seed_db.flush()
|
|
harness = seed_db.query(Harness).first()
|
|
sessions = []
|
|
for n in range(2):
|
|
s = Session(
|
|
project_id=proj.id,
|
|
harness_id=harness.id,
|
|
nickname=f"sess-{n}",
|
|
tmux_session_name=f"hqt-jk-{n}",
|
|
archived=False,
|
|
)
|
|
seed_db.add(s)
|
|
sessions.append(s)
|
|
seed_db.commit()
|
|
proj_id = proj.id
|
|
session_ids = [s.id for s in sessions]
|
|
seed_db.close()
|
|
|
|
app._selected_project_id = proj_id
|
|
# restore_selection mirrors a real project switch, which highlights the
|
|
# first session — giving j/k a defined starting point.
|
|
await app._refresh_sessions(restore_selection=True)
|
|
await pilot.pause()
|
|
|
|
sl = app.query_one(SessionList)
|
|
sl.query_one("#session-list").focus()
|
|
await pilot.pause()
|
|
await pilot.press("j")
|
|
await pilot.pause()
|
|
assert sl.get_selected_session_id() == session_ids[1]
|
|
await pilot.press("k")
|
|
await pilot.pause()
|
|
assert sl.get_selected_session_id() == session_ids[0]
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Worktree session UI: NewSessionScreen worktree controls
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_new_session_worktree_checkbox_disabled_when_not_git_repo():
|
|
"""The worktree checkbox is disabled for non-git-repo projects."""
|
|
app = HqtApp()
|
|
async with app.run_test(size=(120, 40)) as pilot:
|
|
from hqt.tui.screens.new_session import NewSessionScreen
|
|
from textual.widgets import Checkbox
|
|
|
|
app.push_screen(NewSessionScreen(["claude"], is_git_repo=False))
|
|
await pilot.pause()
|
|
cb = app.screen.query_one("#worktree-checkbox", Checkbox)
|
|
assert cb.disabled is True
|
|
assert "not a git repo" in str(cb.label)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_new_session_worktree_unchecked_returns_none_branch():
|
|
"""With the checkbox unchecked, worktree_branch is None."""
|
|
app = HqtApp()
|
|
async with app.run_test(size=(120, 40)) as pilot:
|
|
from hqt.tui.screens.new_session import NewSessionScreen
|
|
from textual.widgets import Button, Input
|
|
|
|
results: list = []
|
|
app.push_screen(NewSessionScreen(["claude"], is_git_repo=True), results.append)
|
|
await pilot.pause()
|
|
app.screen.query_one("#nickname-input", Input).value = "mywork"
|
|
app.screen.query_one("#ok-btn", Button).press()
|
|
await pilot.pause()
|
|
assert results == [("claude", "mywork", None, None, None)], f"got {results}"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_new_session_worktree_checked_uses_typed_branch():
|
|
"""Checked checkbox + typed branch returns that branch."""
|
|
app = HqtApp()
|
|
async with app.run_test(size=(120, 40)) as pilot:
|
|
from hqt.tui.screens.new_session import NewSessionScreen
|
|
from textual.widgets import Button, Checkbox, Input
|
|
|
|
results: list = []
|
|
app.push_screen(NewSessionScreen(["claude"], is_git_repo=True), results.append)
|
|
await pilot.pause()
|
|
app.screen.query_one("#nickname-input", Input).value = "My Work"
|
|
app.screen.query_one("#worktree-checkbox", Checkbox).value = True
|
|
await pilot.pause()
|
|
# Branch input is revealed and prefilled from the nickname slug.
|
|
branch = app.screen.query_one("#branch-input", Input)
|
|
assert branch.display is True
|
|
assert branch.value == "my-work"
|
|
# User overrides with their own branch name.
|
|
branch.value = "feature-x"
|
|
app.screen.query_one("#ok-btn", Button).press()
|
|
await pilot.pause()
|
|
assert results == [("claude", "My Work", None, "feature-x", None)], (
|
|
f"got {results}"
|
|
)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_new_session_worktree_checked_blank_branch_slugifies_nickname():
|
|
"""Checked checkbox + blank branch derives the branch from the nickname slug."""
|
|
app = HqtApp()
|
|
async with app.run_test(size=(120, 40)) as pilot:
|
|
from hqt.tui.screens.new_session import NewSessionScreen
|
|
from textual.widgets import Button, Checkbox, Input
|
|
|
|
results: list = []
|
|
app.push_screen(NewSessionScreen(["claude"], is_git_repo=True), results.append)
|
|
await pilot.pause()
|
|
app.screen.query_one("#nickname-input", Input).value = "Cool Feature!"
|
|
cb = app.screen.query_one("#worktree-checkbox", Checkbox)
|
|
cb.value = True
|
|
await pilot.pause()
|
|
# Clear the prefilled branch so _submit must re-derive it from nickname.
|
|
app.screen.query_one("#branch-input", Input).value = ""
|
|
app.screen.query_one("#ok-btn", Button).press()
|
|
await pilot.pause()
|
|
assert results == [("claude", "Cool Feature!", None, "cool-feature", None)], (
|
|
f"got {results}"
|
|
)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_new_session_branch_input_hidden_until_checked():
|
|
"""The branch input starts hidden and toggles with the checkbox."""
|
|
app = HqtApp()
|
|
async with app.run_test(size=(120, 40)) as pilot:
|
|
from hqt.tui.screens.new_session import NewSessionScreen
|
|
from textual.widgets import Checkbox, Input
|
|
|
|
app.push_screen(NewSessionScreen(["claude"], is_git_repo=True))
|
|
await pilot.pause()
|
|
branch = app.screen.query_one("#branch-input", Input)
|
|
assert branch.display is False
|
|
cb = app.screen.query_one("#worktree-checkbox", Checkbox)
|
|
cb.value = True
|
|
await pilot.pause()
|
|
assert branch.display is True
|
|
cb.value = False
|
|
await pilot.pause()
|
|
assert branch.display is False
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Worktree session UI: ConfirmDeleteScreen
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_confirm_delete_checkbox_checked_by_default_when_no_warning():
|
|
"""No warning (clean worktree) → 'remove worktree' pre-checked → (True, True)."""
|
|
app = HqtApp()
|
|
async with app.run_test(size=(120, 40)) as pilot:
|
|
from hqt.tui.screens.confirm_delete import ConfirmDeleteScreen
|
|
from textual.widgets import Button, Checkbox
|
|
|
|
results: list = []
|
|
app.push_screen(ConfirmDeleteScreen("mysess"), results.append)
|
|
await pilot.pause()
|
|
assert app.screen.query_one("#remove-worktree-checkbox", Checkbox).value is True
|
|
app.screen.query_one("#ok-btn", Button).press()
|
|
await pilot.pause()
|
|
assert results == [(True, True)]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_confirm_delete_checkbox_unchecked_by_default_with_warning():
|
|
"""A warning (dirty/unmerged) → checkbox unchecked → (True, False) by default."""
|
|
app = HqtApp()
|
|
async with app.run_test(size=(120, 40)) as pilot:
|
|
from hqt.tui.screens.confirm_delete import ConfirmDeleteScreen
|
|
from textual.widgets import Button, Checkbox, Label
|
|
|
|
results: list = []
|
|
app.push_screen(
|
|
ConfirmDeleteScreen("mysess", warning="Work will be lost"),
|
|
results.append,
|
|
)
|
|
await pilot.pause()
|
|
assert (
|
|
app.screen.query_one("#remove-worktree-checkbox", Checkbox).value is False
|
|
)
|
|
warning = app.screen.query_one("#delete-warning", Label)
|
|
assert "Work will be lost" in str(warning.render())
|
|
app.screen.query_one("#ok-btn", Button).press()
|
|
await pilot.pause()
|
|
assert results == [(True, False)]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_confirm_delete_cancel_returns_none():
|
|
app = HqtApp()
|
|
async with app.run_test(size=(120, 40)) as pilot:
|
|
from hqt.tui.screens.confirm_delete import ConfirmDeleteScreen
|
|
from textual.widgets import Button
|
|
|
|
results: list = []
|
|
app.push_screen(ConfirmDeleteScreen("mysess"), results.append)
|
|
await pilot.pause()
|
|
app.screen.query_one("#cancel-btn", Button).press()
|
|
await pilot.pause()
|
|
assert results == [None]
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Worktree session UI: list marker + non-worktree delete unchanged
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_format_session_text_appends_worktree_branch_marker():
|
|
"""nickname != branch → show both nickname and branch around the ⎇ glyph."""
|
|
from hqt.tui.widgets.session_list import format_session_text
|
|
|
|
text = format_session_text("mywork", "claude", "idle", "●", "feature-x")
|
|
assert "mywork ⎇ feature-x" in text
|
|
|
|
|
|
def test_format_session_text_marker_shows_once_when_nickname_equals_branch():
|
|
"""nickname == branch (the common case) → ⎇ still appears, glyph once, no
|
|
duplicated branch name."""
|
|
from hqt.tui.widgets.session_list import format_session_text
|
|
|
|
text = format_session_text("feature-x", "claude", "idle", "●", "feature-x")
|
|
assert "⎇" in text
|
|
# Glyph appears exactly once, and the branch is not duplicated.
|
|
assert text.count("⎇") == 1
|
|
assert "⎇ feature-x" in text
|
|
assert "feature-x ⎇" not in text
|
|
|
|
|
|
def test_format_session_text_no_marker_for_plain_session():
|
|
from hqt.tui.widgets.session_list import format_session_text
|
|
|
|
text = format_session_text("plain", "claude", "idle", "●", None)
|
|
assert "⎇" not in text
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_delete_non_worktree_session_does_not_push_modal():
|
|
"""Plain (non-worktree) sessions delete immediately, with no confirm modal."""
|
|
from hqt.db.models import Harness, Project, Session
|
|
from hqt.tui.widgets.session_list import SessionList
|
|
from hqt.tui.screens.confirm_delete import ConfirmDeleteScreen
|
|
|
|
app = HqtApp()
|
|
async with app.run_test(size=(120, 40)) as pilot:
|
|
seed_db = app._db_factory()
|
|
proj = Project(name="del-proj", path="/tmp/del-proj")
|
|
seed_db.add(proj)
|
|
seed_db.flush()
|
|
harness = seed_db.query(Harness).first()
|
|
sess = Session(
|
|
project_id=proj.id,
|
|
harness_id=harness.id,
|
|
tmux_session_name="hqt-del",
|
|
archived=False,
|
|
)
|
|
seed_db.add(sess)
|
|
seed_db.commit()
|
|
proj_id = proj.id
|
|
sess_id = sess.id
|
|
seed_db.close()
|
|
|
|
app._selected_project_id = proj_id
|
|
await app._refresh_sessions()
|
|
await pilot.pause()
|
|
|
|
sl = app.query_one(SessionList)
|
|
sl.query_one("#session-list").focus()
|
|
await pilot.press("down")
|
|
await pilot.pause()
|
|
assert sl.get_selected_session_id() == sess_id
|
|
|
|
await app.run_action("delete_session")
|
|
await pilot.pause()
|
|
await app.workers.wait_for_complete()
|
|
await pilot.pause()
|
|
|
|
# No confirm modal was pushed; the session is gone from the DB.
|
|
assert not isinstance(app.screen, ConfirmDeleteScreen)
|
|
verify_db = app._db_factory()
|
|
assert verify_db.get(Session, sess_id) is None
|
|
verify_db.close()
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Task 6: Sandbox controls in NewSessionScreen
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_policy_from_disabled_returns_none():
|
|
from hqt.tui.screens.new_session import NewSessionScreen
|
|
|
|
assert NewSessionScreen._policy_from(False, "rw", True) is None
|
|
|
|
|
|
def test_policy_from_enabled_builds_policy():
|
|
from hqt.sandbox import SandboxPolicy
|
|
from hqt.tui.screens.new_session import NewSessionScreen
|
|
|
|
p = NewSessionScreen._policy_from(True, "ro", False)
|
|
assert p == SandboxPolicy(fs="ro", net=False)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_sandbox_switch_disabled_when_unavailable():
|
|
from hqt.tui.screens.new_session import NewSessionScreen
|
|
from textual.widgets import Switch
|
|
|
|
app = HqtApp()
|
|
async with app.run_test(size=(120, 40)) as pilot:
|
|
app.push_screen(
|
|
NewSessionScreen(["claude"], is_git_repo=False, sandbox_available=False)
|
|
)
|
|
await pilot.pause()
|
|
assert app.screen.query_one("#sandbox-switch", Switch).disabled is True
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_sandbox_switch_enabled_when_available():
|
|
from hqt.tui.screens.new_session import NewSessionScreen
|
|
from textual.widgets import Switch
|
|
|
|
app = HqtApp()
|
|
async with app.run_test(size=(120, 40)) as pilot:
|
|
app.push_screen(
|
|
NewSessionScreen(["claude"], is_git_repo=False, sandbox_available=True)
|
|
)
|
|
await pilot.pause()
|
|
assert app.screen.query_one("#sandbox-switch", Switch).disabled is False
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# delayed prompt: bindings, actions, poll dispatch
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_prompt_later_binding_exists():
|
|
bindings = {b.key: b for b in HqtApp.BINDINGS}
|
|
assert "p" in bindings
|
|
assert bindings["p"].action == "schedule_prompt"
|
|
assert "P" in bindings
|
|
assert bindings["P"].action == "cancel_scheduled_prompt"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_schedule_prompt_action_calls_service_and_refreshes():
|
|
from datetime import datetime
|
|
from unittest.mock import patch
|
|
from hqt.db.models import Harness, Project, Session
|
|
from hqt.tui.widgets.session_list import SessionList
|
|
|
|
app = HqtApp()
|
|
async with app.run_test(size=(120, 40)) as pilot:
|
|
seed_db = app._db_factory()
|
|
proj = Project(name="prompt-proj", path="/tmp/prompt-proj")
|
|
seed_db.add(proj)
|
|
seed_db.flush()
|
|
harness = seed_db.query(Harness).first()
|
|
sess = Session(
|
|
project_id=proj.id,
|
|
harness_id=harness.id,
|
|
tmux_session_name="hqt-prompt-test",
|
|
archived=False,
|
|
)
|
|
seed_db.add(sess)
|
|
seed_db.commit()
|
|
proj_id = proj.id
|
|
sess_id = sess.id
|
|
seed_db.close()
|
|
|
|
app._selected_project_id = proj_id
|
|
await app._refresh_sessions()
|
|
await pilot.pause()
|
|
app.query_one(SessionList).query_one("#session-list").focus()
|
|
await pilot.press("down")
|
|
await pilot.pause()
|
|
|
|
mock_service = MagicMock()
|
|
mock_service.schedule_prompt = MagicMock()
|
|
mock_service.list_sessions = AsyncMock(return_value=[])
|
|
mock_service.sync_window_labels = AsyncMock(return_value=[])
|
|
mock_service.dispatch_due_prompts = AsyncMock(return_value=[])
|
|
app._session_service = mock_service
|
|
|
|
notify_calls = []
|
|
|
|
def capture_notify(message, **kwargs):
|
|
notify_calls.append((message, kwargs))
|
|
|
|
with patch.object(app, "notify", side_effect=capture_notify):
|
|
app.action_schedule_prompt()
|
|
await pilot.pause()
|
|
app.screen.dismiss(
|
|
("continue", __import__("datetime").timedelta(minutes=30))
|
|
)
|
|
await pilot.pause()
|
|
await app.workers.wait_for_complete()
|
|
|
|
assert mock_service.schedule_prompt.call_count == 1
|
|
args = mock_service.schedule_prompt.call_args.args
|
|
assert args[0] == sess_id
|
|
assert args[1] == "continue"
|
|
assert isinstance(args[2], datetime)
|
|
mock_service.list_sessions.assert_awaited()
|
|
assert any("Prompt scheduled" in call[0] for call in notify_calls)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_cancel_scheduled_prompt_action_calls_service_and_refreshes():
|
|
from hqt.db.models import Harness, Project, Session
|
|
from hqt.tui.widgets.session_list import SessionList
|
|
|
|
app = HqtApp()
|
|
async with app.run_test(size=(120, 40)) as pilot:
|
|
seed_db = app._db_factory()
|
|
proj = Project(name="cancel-proj", path="/tmp/cancel-proj")
|
|
seed_db.add(proj)
|
|
seed_db.flush()
|
|
harness = seed_db.query(Harness).first()
|
|
sess = Session(
|
|
project_id=proj.id,
|
|
harness_id=harness.id,
|
|
tmux_session_name="hqt-cancel-test",
|
|
archived=False,
|
|
)
|
|
seed_db.add(sess)
|
|
seed_db.commit()
|
|
proj_id = proj.id
|
|
sess_id = sess.id
|
|
seed_db.close()
|
|
|
|
app._selected_project_id = proj_id
|
|
await app._refresh_sessions()
|
|
await pilot.pause()
|
|
app.query_one(SessionList).query_one("#session-list").focus()
|
|
await pilot.press("down")
|
|
await pilot.pause()
|
|
|
|
mock_service = MagicMock()
|
|
mock_service.cancel_scheduled_prompt = MagicMock()
|
|
mock_service.list_sessions = AsyncMock(return_value=[])
|
|
mock_service.sync_window_labels = AsyncMock(return_value=[])
|
|
mock_service.dispatch_due_prompts = AsyncMock(return_value=[])
|
|
app._session_service = mock_service
|
|
|
|
await app.run_action("cancel_scheduled_prompt")
|
|
await pilot.pause()
|
|
await app.workers.wait_for_complete()
|
|
|
|
mock_service.cancel_scheduled_prompt.assert_called_once_with(sess_id)
|
|
mock_service.list_sessions.assert_awaited()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_poll_sessions_dispatches_due_prompts_and_notifies():
|
|
app = HqtApp()
|
|
async with app.run_test(size=(120, 40)):
|
|
result = MagicMock()
|
|
result.ok = True
|
|
result.window_name = "hqt-1"
|
|
result.error = ""
|
|
|
|
mock_service = MagicMock()
|
|
mock_service.dispatch_due_prompts = AsyncMock(return_value=[result])
|
|
mock_service.sync_window_labels = AsyncMock(return_value=[])
|
|
app._session_service = mock_service
|
|
|
|
notify_calls = []
|
|
|
|
def capture_notify(message, **kwargs):
|
|
notify_calls.append((message, kwargs))
|
|
|
|
from unittest.mock import patch
|
|
|
|
with patch.object(app, "notify", side_effect=capture_notify):
|
|
await app._poll_sessions()
|
|
|
|
mock_service.dispatch_due_prompts.assert_awaited_once()
|
|
assert any("Prompt sent" in call[0] for call in notify_calls)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_sandbox_options_hidden_until_switch_on():
|
|
"""The fs/net rows are sub-options of the sandbox switch: hidden until it is
|
|
turned on, shown when on, hidden again when turned back off."""
|
|
app = HqtApp()
|
|
async with app.run_test(size=(120, 40)) as pilot:
|
|
from hqt.tui.screens.new_session import NewSessionScreen
|
|
from textual.widgets import Switch
|
|
|
|
app.push_screen(
|
|
NewSessionScreen(["claude"], is_git_repo=True, sandbox_available=True)
|
|
)
|
|
await pilot.pause()
|
|
options = app.screen.query_one("#sandbox-options")
|
|
assert options.display is False
|
|
|
|
switch = app.screen.query_one("#sandbox-switch", Switch)
|
|
switch.value = True
|
|
await pilot.pause()
|
|
assert options.display is True
|
|
|
|
switch.value = False
|
|
await pilot.pause()
|
|
assert options.display is False
|
|
|
|
|
|
def test_dialog_focus_highlight_uses_accent():
|
|
"""Every focusable control in the New Session dialog shares the Peach accent
|
|
focus border (consistent with the dialog buttons), not Textual's default
|
|
Lavender. Guards against the scoped :focus rule being dropped."""
|
|
from pathlib import Path
|
|
import hqt.tui
|
|
|
|
css = (Path(hqt.tui.__file__).parent / "styles.tcss").read_text()
|
|
for selector in (
|
|
"#new-session-dialog Switch:focus",
|
|
"#new-session-dialog Checkbox:focus",
|
|
"#new-session-dialog Select:focus",
|
|
"#new-session-dialog Input:focus",
|
|
):
|
|
assert selector in css, f"missing focus rule: {selector}"
|
|
assert "border: tall $accent;" in css
|
|
|
|
|
|
def test_new_session_has_jk_focus_bindings():
|
|
"""j -> focus_next, k -> focus_previous, declared on the screen."""
|
|
from hqt.tui.screens.new_session import NewSessionScreen
|
|
|
|
actions = {b.key: b.action for b in NewSessionScreen.BINDINGS}
|
|
assert actions.get("j") == "focus_next"
|
|
assert actions.get("k") == "focus_previous"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_jk_moves_focus_between_controls():
|
|
"""On a non-text control, j moves focus to another control."""
|
|
app = HqtApp()
|
|
async with app.run_test(size=(120, 40)) as pilot:
|
|
from hqt.tui.screens.new_session import NewSessionScreen
|
|
from textual.widgets import Switch
|
|
|
|
app.push_screen(
|
|
NewSessionScreen(["claude"], is_git_repo=True, sandbox_available=True)
|
|
)
|
|
await pilot.pause()
|
|
switch = app.screen.query_one("#sandbox-switch", Switch)
|
|
switch.focus()
|
|
await pilot.pause()
|
|
assert app.focused is switch
|
|
|
|
await pilot.press("j")
|
|
await pilot.pause()
|
|
assert app.focused is not switch
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_jk_typed_into_input_is_literal():
|
|
"""Inside a text input, j/k type literally and do not move focus."""
|
|
app = HqtApp()
|
|
async with app.run_test(size=(120, 40)) as pilot:
|
|
from hqt.tui.screens.new_session import NewSessionScreen
|
|
from textual.widgets import Input
|
|
|
|
app.push_screen(NewSessionScreen(["claude"], is_git_repo=True))
|
|
await pilot.pause()
|
|
nickname = app.screen.query_one("#nickname-input", Input)
|
|
nickname.focus()
|
|
await pilot.pause()
|
|
|
|
await pilot.press("j", "k")
|
|
await pilot.pause()
|
|
assert nickname.value == "jk"
|
|
assert app.focused is nickname
|
|
|
|
|
|
def test_policy_is_none_when_sandbox_off_regardless_of_suboptions():
|
|
"""When sandboxing is off, the (now hidden) fs/net sub-option values are
|
|
discarded — the submitted policy is None no matter what they hold."""
|
|
from hqt.tui.screens.new_session import NewSessionScreen
|
|
|
|
assert NewSessionScreen._policy_from(False, "ro", False) is None
|
|
assert NewSessionScreen._policy_from(False, "rw", True) is None
|
|
|
|
|
|
def test_policy_reflects_suboptions_when_sandbox_on():
|
|
"""When sandboxing is on, the fs/net sub-option values flow into the policy."""
|
|
from hqt.tui.screens.new_session import NewSessionScreen
|
|
|
|
policy = NewSessionScreen._policy_from(True, "ro", False)
|
|
assert policy is not None
|
|
assert policy.fs == "ro"
|
|
assert policy.net is False
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Task 3: h/l column focus navigation
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_hl_focus_bindings_exist():
|
|
bindings = {b.key: b for b in HqtApp.BINDINGS}
|
|
assert bindings["h"].action == "focus_projects"
|
|
assert bindings["l"].action == "focus_sessions"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_h_focuses_projects_l_focuses_sessions(tmp_path):
|
|
app = HqtApp()
|
|
async with app.run_test(size=(120, 40)) as pilot:
|
|
d = tmp_path / "hl-proj"
|
|
d.mkdir()
|
|
app._project_service.create("hl", str(d))
|
|
app._load_projects()
|
|
await app.workers.wait_for_complete()
|
|
await pilot.pause()
|
|
|
|
await pilot.press("l")
|
|
await pilot.pause()
|
|
assert app.focused is app.query_one("#session-list")
|
|
|
|
await pilot.press("h")
|
|
await pilot.pause()
|
|
assert app.focused is app.query_one("#project-list")
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_session_list_header_reflects_archived_mode():
|
|
app = HqtApp()
|
|
async with app.run_test(size=(120, 40)) as pilot:
|
|
from hqt.tui.widgets.session_list import SessionList
|
|
from textual.widgets import Label
|
|
|
|
sl = app.query_one(SessionList)
|
|
header = sl.query_one("#session-header", Label)
|
|
assert str(header.render()) == "Sessions"
|
|
|
|
sl.set_mode(archived=True)
|
|
await pilot.pause()
|
|
assert "archived" in str(header.render()).lower()
|
|
|
|
sl.set_mode(archived=False)
|
|
await pilot.pause()
|
|
assert str(header.render()) == "Sessions"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_archive_session_binding_exists():
|
|
bindings = {b.key: b for b in HqtApp.BINDINGS}
|
|
assert bindings["x"].action == "archive_session"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_archive_session_action_calls_service_and_refreshes():
|
|
app = HqtApp()
|
|
async with app.run_test(size=(120, 40)) as pilot:
|
|
from hqt.db.models import Project, Harness, Session
|
|
from hqt.tui.widgets.session_list import SessionList
|
|
|
|
seed_db = app._db_factory()
|
|
proj = Project(name="arch-proj", path="/tmp/arch-proj")
|
|
seed_db.add(proj)
|
|
seed_db.flush()
|
|
harness = seed_db.query(Harness).first()
|
|
sess = Session(
|
|
project_id=proj.id,
|
|
harness_id=harness.id,
|
|
tmux_session_name="hqt-arch-action",
|
|
archived=False,
|
|
)
|
|
seed_db.add(sess)
|
|
seed_db.commit()
|
|
proj_id = proj.id
|
|
sess_id = sess.id
|
|
seed_db.close()
|
|
|
|
app._selected_project_id = proj_id
|
|
await app._refresh_sessions()
|
|
await pilot.pause()
|
|
|
|
sl = app.query_one(SessionList)
|
|
sl.query_one("#session-list").focus()
|
|
await pilot.press("down")
|
|
await pilot.pause()
|
|
assert sl.get_selected_session_id() == sess_id
|
|
|
|
mock_service = MagicMock()
|
|
mock_service.archive_session = AsyncMock()
|
|
mock_service.list_sessions = AsyncMock(return_value=[])
|
|
mock_service.sync_window_labels = AsyncMock(return_value=[])
|
|
app._session_service = mock_service
|
|
|
|
await app.run_action("archive_session")
|
|
await pilot.pause()
|
|
|
|
mock_service.archive_session.assert_awaited_once_with(sess_id)
|
|
mock_service.list_sessions.assert_awaited()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_toggle_archived_binding_exists():
|
|
bindings = {b.key: b for b in HqtApp.BINDINGS}
|
|
assert bindings["A"].action == "toggle_archived"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_toggle_archived_switches_list_and_header():
|
|
app = HqtApp()
|
|
async with app.run_test(size=(120, 40)) as pilot:
|
|
from hqt.tui.widgets.session_list import SessionList
|
|
from textual.widgets import Label
|
|
|
|
proj_id = 42
|
|
app._selected_project_id = proj_id
|
|
|
|
mock_service = MagicMock()
|
|
mock_service.list_sessions = AsyncMock(return_value=[])
|
|
mock_service.sync_window_labels = AsyncMock(return_value=[])
|
|
app._session_service = mock_service
|
|
|
|
assert app._show_archived is False
|
|
|
|
await app.run_action("toggle_archived")
|
|
await pilot.pause()
|
|
|
|
assert app._show_archived is True
|
|
mock_service.list_sessions.assert_awaited_with(proj_id, archived=True)
|
|
header = app.query_one(SessionList).query_one("#session-header", Label)
|
|
assert "archived" in str(header.render()).lower()
|
|
|
|
await app.run_action("toggle_archived")
|
|
await pilot.pause()
|
|
|
|
assert app._show_archived is False
|
|
mock_service.list_sessions.assert_awaited_with(proj_id, archived=False)
|
|
assert str(header.render()) == "Sessions"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_unarchive_binding_exists():
|
|
bindings = {b.key: b for b in HqtApp.BINDINGS}
|
|
assert bindings["u"].action == "unarchive_session"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_unarchive_action_calls_service_and_refreshes():
|
|
app = HqtApp()
|
|
async with app.run_test(size=(120, 40)) as pilot:
|
|
from hqt.db.models import Project, Harness, Session
|
|
from hqt.tui.widgets.session_list import SessionList
|
|
|
|
seed_db = app._db_factory()
|
|
proj = Project(name="restore-proj", path="/tmp/restore-proj")
|
|
seed_db.add(proj)
|
|
seed_db.flush()
|
|
harness = seed_db.query(Harness).first()
|
|
sess = Session(
|
|
project_id=proj.id,
|
|
harness_id=harness.id,
|
|
tmux_session_name="hqt-restore",
|
|
archived=True,
|
|
)
|
|
seed_db.add(sess)
|
|
seed_db.commit()
|
|
proj_id = proj.id
|
|
sess_id = sess.id
|
|
seed_db.close()
|
|
|
|
app._selected_project_id = proj_id
|
|
app._show_archived = True
|
|
await app._refresh_sessions()
|
|
await pilot.pause()
|
|
|
|
sl = app.query_one(SessionList)
|
|
sl.query_one("#session-list").focus()
|
|
await pilot.press("down")
|
|
await pilot.pause()
|
|
assert sl.get_selected_session_id() == sess_id
|
|
|
|
mock_service = MagicMock()
|
|
mock_service.unarchive_session = AsyncMock()
|
|
mock_service.list_sessions = AsyncMock(return_value=[])
|
|
mock_service.sync_window_labels = AsyncMock(return_value=[])
|
|
app._session_service = mock_service
|
|
|
|
await app.run_action("unarchive_session")
|
|
await pilot.pause()
|
|
|
|
mock_service.unarchive_session.assert_awaited_once_with(sess_id)
|
|
mock_service.list_sessions.assert_awaited()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_archive_restore_actions_gated_by_view():
|
|
app = HqtApp()
|
|
async with app.run_test(size=(120, 40)):
|
|
# Active view (default): archive offered, restore hidden.
|
|
assert app.check_action("archive_session", ()) is True
|
|
assert app.check_action("unarchive_session", ()) is None
|
|
|
|
# Archived view: restore offered, archive hidden.
|
|
app._show_archived = True
|
|
assert app.check_action("archive_session", ()) is None
|
|
assert app.check_action("unarchive_session", ()) is True
|