Merge branch 'tui-nav-archiving'
TUI h/l column navigation and recoverable session archiving. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
@@ -31,6 +31,62 @@ hqt
|
|||||||
| `Enter` | Attach to session (auto-resumes if not running) |
|
| `Enter` | Attach to session (auto-resumes if not running) |
|
||||||
| `r` | Rename session |
|
| `r` | Rename session |
|
||||||
| `s` | Stop session |
|
| `s` | Stop session |
|
||||||
|
| `h` / `l` | Focus the Projects / Sessions column |
|
||||||
|
| `x` | Archive selected session (instant; keeps the row, kills the window) |
|
||||||
|
| `A` | Toggle the Sessions column between active and archived |
|
||||||
|
| `u` | Restore (unarchive) the selected session — archived view only |
|
||||||
|
|
||||||
|
## tmux configuration
|
||||||
|
|
||||||
|
hqt runs inside tmux and tags each harness window with an `@hqt_label` user
|
||||||
|
option automatically — **it never edits your `~/.tmux.conf`**. The TUI itself
|
||||||
|
needs no tmux config to run. The bindings below are opt-in: add the ones you
|
||||||
|
want to your own `~/.tmux.conf` (`-n` = no prefix key).
|
||||||
|
|
||||||
|
### Tool palette — `Alt+p` (required for the palette feature)
|
||||||
|
|
||||||
|
Pops an fzf palette for the current session's project (nvim / lazygit / shell /
|
||||||
|
clone / reset-windows) from inside a harness window. Requires `hqt` on your
|
||||||
|
`PATH` (see Quick Start) and `fzf`. `run-shell` expands `#{window_name}` to a
|
||||||
|
concrete window name before `hqt` builds its `display-popup`:
|
||||||
|
|
||||||
|
```tmux
|
||||||
|
bind -n M-p run-shell -b "hqt palette '#{window_name}'"
|
||||||
|
```
|
||||||
|
|
||||||
|
If you run hqt from a checkout instead of installing it, point `uv run` at the
|
||||||
|
repo so the virtualenv self-heals when deps change:
|
||||||
|
|
||||||
|
```tmux
|
||||||
|
bind -n M-p run-shell -b "uv run --project /path/to/hq-term hqt palette '#{window_name}'"
|
||||||
|
```
|
||||||
|
|
||||||
|
### Session switcher — `Alt+o` (optional)
|
||||||
|
|
||||||
|
Pops an fzf switcher over whatever you're doing (works inside a harness too).
|
||||||
|
Rows are built from the `@hqt_label` hqt sets; `cut` takes the leading window
|
||||||
|
index and `select-window` jumps to it:
|
||||||
|
|
||||||
|
```tmux
|
||||||
|
bind -n M-o display-popup -E -w 50% -h 40% -T ' switch session ' \
|
||||||
|
"tmux list-windows -F '#{window_index} #{?@hqt_label,#{@hqt_label},#{window_name}}' \
|
||||||
|
| fzf --reverse --no-info --prompt='session ' \
|
||||||
|
| cut -d' ' -f1 \
|
||||||
|
| xargs -r tmux select-window -t"
|
||||||
|
```
|
||||||
|
|
||||||
|
### Scrollback passthrough — `PageUp` (optional)
|
||||||
|
|
||||||
|
The TUI uses tmux's alternate screen. This sends `PageUp` straight through to
|
||||||
|
full-screen apps (the TUI) while still entering copy-mode scrollback in normal
|
||||||
|
panes:
|
||||||
|
|
||||||
|
```tmux
|
||||||
|
bind -n PageUp if-shell -F '#{alternate_on}' 'send-keys PageUp' 'copy-mode -eu'
|
||||||
|
```
|
||||||
|
|
||||||
|
After editing `~/.tmux.conf`, reload it with `tmux source-file ~/.tmux.conf`
|
||||||
|
(or restart the server).
|
||||||
|
|
||||||
## Worktree-isolated sessions
|
## Worktree-isolated sessions
|
||||||
|
|
||||||
|
|||||||
@@ -826,6 +826,37 @@ class SessionService:
|
|||||||
db.delete(sess)
|
db.delete(sess)
|
||||||
db.commit()
|
db.commit()
|
||||||
|
|
||||||
|
async def archive_session(self, session_id: int) -> None:
|
||||||
|
"""Archive a session: kill its tmux window but keep the DB row.
|
||||||
|
|
||||||
|
A recoverable "remove from view" — the window/harness process is torn
|
||||||
|
down (freeing resources) but the row, ``harness_session_id``, and any
|
||||||
|
worktree are preserved so the session can be restored or referenced
|
||||||
|
later. No-op if the session does not exist.
|
||||||
|
"""
|
||||||
|
with self.factory() as db:
|
||||||
|
sess = db.get(Session, session_id)
|
||||||
|
if sess is None:
|
||||||
|
return
|
||||||
|
if await self.tmux.window_exists(sess.tmux_session_name):
|
||||||
|
await self.tmux.kill(sess.tmux_session_name)
|
||||||
|
sess.archived = True
|
||||||
|
db.commit()
|
||||||
|
|
||||||
|
async def unarchive_session(self, session_id: int) -> None:
|
||||||
|
"""Restore an archived session by clearing its archived flag.
|
||||||
|
|
||||||
|
The tmux window stays gone (the session reads as dead until resumed);
|
||||||
|
this only makes the row visible in the active list again. No-op if the
|
||||||
|
session does not exist.
|
||||||
|
"""
|
||||||
|
with self.factory() as db:
|
||||||
|
sess = db.get(Session, session_id)
|
||||||
|
if sess is None:
|
||||||
|
return
|
||||||
|
sess.archived = False
|
||||||
|
db.commit()
|
||||||
|
|
||||||
async def worktree_state_for(self, session_id: int) -> WorktreeState | None:
|
async def worktree_state_for(self, session_id: int) -> WorktreeState | None:
|
||||||
"""Return the WorktreeState for a session's worktree, or None.
|
"""Return the WorktreeState for a session's worktree, or None.
|
||||||
|
|
||||||
@@ -881,7 +912,7 @@ class SessionService:
|
|||||||
return {prompt.session_id: prompt for prompt in prompts}
|
return {prompt.session_id: prompt for prompt in prompts}
|
||||||
|
|
||||||
async def list_sessions(
|
async def list_sessions(
|
||||||
self, project_id: int, now: float | None = None
|
self, project_id: int, archived: bool = False, now: float | None = None
|
||||||
) -> list[SessionInfo]:
|
) -> list[SessionInfo]:
|
||||||
if now is None:
|
if now is None:
|
||||||
now = time.time()
|
now = time.time()
|
||||||
@@ -889,7 +920,7 @@ class SessionService:
|
|||||||
sessions = (
|
sessions = (
|
||||||
db.query(Session)
|
db.query(Session)
|
||||||
.options(selectinload(Session.harness))
|
.options(selectinload(Session.harness))
|
||||||
.filter_by(project_id=project_id, archived=False)
|
.filter_by(project_id=project_id, archived=archived)
|
||||||
.all()
|
.all()
|
||||||
)
|
)
|
||||||
names = [s.tmux_session_name for s in sessions]
|
names = [s.tmux_session_name for s in sessions]
|
||||||
|
|||||||
+51
-2
@@ -70,6 +70,11 @@ class HqtApp(App):
|
|||||||
Binding("a", "add_project", "Add Project"),
|
Binding("a", "add_project", "Add Project"),
|
||||||
Binding("e", "edit_project", "Edit Project"),
|
Binding("e", "edit_project", "Edit Project"),
|
||||||
Binding("tab", "toggle_focus", "Switch Panel"),
|
Binding("tab", "toggle_focus", "Switch Panel"),
|
||||||
|
Binding("h", "focus_projects", "Projects"),
|
||||||
|
Binding("l", "focus_sessions", "Sessions"),
|
||||||
|
Binding("x", "archive_session", "Archive"),
|
||||||
|
Binding("A", "toggle_archived", "Show Archived"),
|
||||||
|
Binding("u", "unarchive_session", "Restore"),
|
||||||
Binding("d", "delete_session", "Delete Session"),
|
Binding("d", "delete_session", "Delete Session"),
|
||||||
Binding("enter", "attach_session", "Attach", priority=True),
|
Binding("enter", "attach_session", "Attach", priority=True),
|
||||||
Binding("r", "rename_session", "Rename"),
|
Binding("r", "rename_session", "Rename"),
|
||||||
@@ -88,6 +93,9 @@ class HqtApp(App):
|
|||||||
self.register_theme(FRAPPE_THEME)
|
self.register_theme(FRAPPE_THEME)
|
||||||
self.theme = "catppuccin-frappe"
|
self.theme = "catppuccin-frappe"
|
||||||
self._selected_project_id: int | None = None
|
self._selected_project_id: int | None = None
|
||||||
|
# When True, the Sessions column shows this project's archived sessions
|
||||||
|
# instead of its active ones. Persists across project switches.
|
||||||
|
self._show_archived = False
|
||||||
# project_id -> last selected session_id, so switching projects restores
|
# project_id -> last selected session_id, so switching projects restores
|
||||||
# each project's previously highlighted session.
|
# each project's previously highlighted session.
|
||||||
self._selected_session_by_project: dict[int, int] = {}
|
self._selected_session_by_project: dict[int, int] = {}
|
||||||
@@ -178,7 +186,7 @@ class HqtApp(App):
|
|||||||
except ServiceError as err:
|
except ServiceError as err:
|
||||||
log.warning("Session poll failed: %s", err)
|
log.warning("Session poll failed: %s", err)
|
||||||
return
|
return
|
||||||
if self._selected_project_id is not None:
|
if self._selected_project_id is not None and not self._show_archived:
|
||||||
subset = [
|
subset = [
|
||||||
i for i in infos if i.session.project_id == self._selected_project_id
|
i for i in infos if i.session.project_id == self._selected_project_id
|
||||||
]
|
]
|
||||||
@@ -187,7 +195,9 @@ class HqtApp(App):
|
|||||||
async def _refresh_sessions(self, restore_selection: bool = False) -> None:
|
async def _refresh_sessions(self, restore_selection: bool = False) -> None:
|
||||||
if self._selected_project_id is None:
|
if self._selected_project_id is None:
|
||||||
return
|
return
|
||||||
infos = await self._sessions().list_sessions(self._selected_project_id)
|
infos = await self._sessions().list_sessions(
|
||||||
|
self._selected_project_id, archived=self._show_archived
|
||||||
|
)
|
||||||
sl = self.query_one(SessionList)
|
sl = self.query_one(SessionList)
|
||||||
if restore_selection:
|
if restore_selection:
|
||||||
# Project switch: restore this project's remembered session, or let the
|
# Project switch: restore this project's remembered session, or let the
|
||||||
@@ -213,6 +223,12 @@ class HqtApp(App):
|
|||||||
def action_toggle_focus(self) -> None:
|
def action_toggle_focus(self) -> None:
|
||||||
self.screen.focus_next()
|
self.screen.focus_next()
|
||||||
|
|
||||||
|
def action_focus_projects(self) -> None:
|
||||||
|
self.query_one("#project-list").focus()
|
||||||
|
|
||||||
|
def action_focus_sessions(self) -> None:
|
||||||
|
self.query_one("#session-list").focus()
|
||||||
|
|
||||||
def action_add_project(self) -> None:
|
def action_add_project(self) -> None:
|
||||||
def on_dismiss(result: tuple[str, str] | None) -> None:
|
def on_dismiss(result: tuple[str, str] | None) -> None:
|
||||||
if result:
|
if result:
|
||||||
@@ -334,6 +350,12 @@ class HqtApp(App):
|
|||||||
# the focused dialog widget.
|
# the focused dialog widget.
|
||||||
if action == "attach_session" and isinstance(self.screen, ModalScreen):
|
if action == "attach_session" and isinstance(self.screen, ModalScreen):
|
||||||
return False
|
return False
|
||||||
|
# Archive applies only to active sessions; restore only to archived ones.
|
||||||
|
# Hide the inapplicable one so the footer never advertises it.
|
||||||
|
if action == "archive_session" and self._show_archived:
|
||||||
|
return None
|
||||||
|
if action == "unarchive_session" and not self._show_archived:
|
||||||
|
return None
|
||||||
return True
|
return True
|
||||||
|
|
||||||
def action_attach_session(self) -> None:
|
def action_attach_session(self) -> None:
|
||||||
@@ -428,6 +450,33 @@ class HqtApp(App):
|
|||||||
|
|
||||||
self._run_service_worker(_do())
|
self._run_service_worker(_do())
|
||||||
|
|
||||||
|
def action_archive_session(self) -> None:
|
||||||
|
sid = self.query_one(SessionList).get_selected_session_id()
|
||||||
|
if not sid:
|
||||||
|
return
|
||||||
|
|
||||||
|
async def _do() -> None:
|
||||||
|
await self._sessions().archive_session(sid)
|
||||||
|
await self._refresh_sessions()
|
||||||
|
|
||||||
|
self._run_service_worker(_do())
|
||||||
|
|
||||||
|
def action_unarchive_session(self) -> None:
|
||||||
|
sid = self.query_one(SessionList).get_selected_session_id()
|
||||||
|
if not sid:
|
||||||
|
return
|
||||||
|
|
||||||
|
async def _do() -> None:
|
||||||
|
await self._sessions().unarchive_session(sid)
|
||||||
|
await self._refresh_sessions()
|
||||||
|
|
||||||
|
self._run_service_worker(_do())
|
||||||
|
|
||||||
|
def action_toggle_archived(self) -> None:
|
||||||
|
self._show_archived = not self._show_archived
|
||||||
|
self.query_one(SessionList).set_mode(archived=self._show_archived)
|
||||||
|
self._run_service_worker(self._refresh_sessions(restore_selection=True))
|
||||||
|
|
||||||
def action_schedule_prompt(self) -> None:
|
def action_schedule_prompt(self) -> None:
|
||||||
sid = self.query_one(SessionList).get_selected_session_id()
|
sid = self.query_one(SessionList).get_selected_session_id()
|
||||||
if sid is None:
|
if sid is None:
|
||||||
|
|||||||
@@ -72,6 +72,11 @@ class SessionList(Widget, can_focus=False):
|
|||||||
yield Label("Sessions", id="session-header")
|
yield Label("Sessions", id="session-header")
|
||||||
yield VimListView(id="session-list")
|
yield VimListView(id="session-list")
|
||||||
|
|
||||||
|
def set_mode(self, archived: bool) -> None:
|
||||||
|
"""Swap the column header to reflect whether archived sessions are shown."""
|
||||||
|
header = self.query_one("#session-header", Label)
|
||||||
|
header.update("Sessions (archived)" if archived else "Sessions")
|
||||||
|
|
||||||
async def refresh_sessions(
|
async def refresh_sessions(
|
||||||
self, session_infos: list, select_session_id=_UNSET
|
self, session_infos: list, select_session_id=_UNSET
|
||||||
) -> None:
|
) -> None:
|
||||||
|
|||||||
@@ -2150,3 +2150,63 @@ async def test_stop_session_for_window_kills_by_name(service, db, tmux):
|
|||||||
async def test_stop_session_for_window_unknown_window_raises(service):
|
async def test_stop_session_for_window_unknown_window_raises(service):
|
||||||
with pytest.raises(ServiceError):
|
with pytest.raises(ServiceError):
|
||||||
await service.stop_session_for_window("not-an-hqt-window")
|
await service.stop_session_for_window("not-an-hqt-window")
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.mark.asyncio
|
||||||
|
async def test_list_sessions_archived_filter(service, db, tmux):
|
||||||
|
h = db.query(Harness).filter_by(name="claude-code").first()
|
||||||
|
active = Session(
|
||||||
|
project_id=1, harness_id=h.id, tmux_session_name="hqt-active", archived=False
|
||||||
|
)
|
||||||
|
arch = Session(
|
||||||
|
project_id=1, harness_id=h.id, tmux_session_name="hqt-arch", archived=True
|
||||||
|
)
|
||||||
|
db.add_all([active, arch])
|
||||||
|
db.commit()
|
||||||
|
active_id, arch_id = active.id, arch.id
|
||||||
|
|
||||||
|
default = await service.list_sessions(project_id=1)
|
||||||
|
assert [i.session.id for i in default] == [active_id]
|
||||||
|
|
||||||
|
archived = await service.list_sessions(project_id=1, archived=True)
|
||||||
|
assert [i.session.id for i in archived] == [arch_id]
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.mark.asyncio
|
||||||
|
async def test_archive_session_kills_window_and_keeps_row(service, db, tmux):
|
||||||
|
await service.create_session(project_id=1, harness_name="claude-code")
|
||||||
|
await service.archive_session(1)
|
||||||
|
|
||||||
|
tmux.kill.assert_called_once_with("hqt-1")
|
||||||
|
db.expire_all()
|
||||||
|
row = db.get(Session, 1)
|
||||||
|
assert row is not None
|
||||||
|
assert row.archived is True
|
||||||
|
# Excluded from the default (active) listing.
|
||||||
|
assert await service.list_sessions(project_id=1) == []
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.mark.asyncio
|
||||||
|
async def test_archive_session_no_kill_when_window_missing(service, db, tmux):
|
||||||
|
await service.create_session(project_id=1, harness_name="claude-code")
|
||||||
|
tmux.window_exists = AsyncMock(return_value=False)
|
||||||
|
tmux.kill.reset_mock()
|
||||||
|
|
||||||
|
await service.archive_session(1)
|
||||||
|
|
||||||
|
tmux.kill.assert_not_called()
|
||||||
|
db.expire_all()
|
||||||
|
assert db.get(Session, 1).archived is True
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.mark.asyncio
|
||||||
|
async def test_unarchive_session_restores_to_active_list(service, db, tmux):
|
||||||
|
await service.create_session(project_id=1, harness_name="claude-code")
|
||||||
|
await service.archive_session(1)
|
||||||
|
|
||||||
|
await service.unarchive_session(1)
|
||||||
|
|
||||||
|
db.expire_all()
|
||||||
|
assert db.get(Session, 1).archived is False
|
||||||
|
active = await service.list_sessions(project_id=1)
|
||||||
|
assert [i.session.id for i in active] == [1]
|
||||||
|
|||||||
+213
-1
@@ -310,7 +310,7 @@ async def test_new_session_on_dismiss_sequential_create_then_refresh():
|
|||||||
session=MagicMock(), spawn_ok=True, spawn_error=""
|
session=MagicMock(), spawn_ok=True, spawn_error=""
|
||||||
)
|
)
|
||||||
|
|
||||||
async def fake_list(project_id):
|
async def fake_list(project_id, archived=False):
|
||||||
call_order.append("list")
|
call_order.append("list")
|
||||||
return []
|
return []
|
||||||
|
|
||||||
@@ -1983,3 +1983,215 @@ def test_policy_reflects_suboptions_when_sandbox_on():
|
|||||||
assert policy is not None
|
assert policy is not None
|
||||||
assert policy.fs == "ro"
|
assert policy.fs == "ro"
|
||||||
assert policy.net is False
|
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
|
||||||
|
|||||||
Reference in New Issue
Block a user