Zhengshou Lai
75a72a8b24
- _read_key 改按需读 CSI 序列（方向键读到终止符即返回），连按 \x1b[B 不再被吞进同一序列误判为 ESC 退出
- picker 全程隐藏光标（\x1b[?25l），退出恢复（\x1b[?25h），末尾不再突兀闪现光标
- 退出改用 Delete Line（\x1b[{n}M）删除块区域，替代逐行清空，不再留下 N 个空行
- tests: 新增快速连按方向键回归测试
358 lines
12 KiB
Python
358 lines
12 KiB
Python
"""Tests for unified claude session listing/resume (jsonl ∪ xiaohe index)."""
|
||
|
||
import copy
|
||
import json
|
||
import os
|
||
import re
|
||
import sys
|
||
from pathlib import Path
|
||
from types import SimpleNamespace
|
||
|
||
import pytest
|
||
|
||
import myagents.launcher as L
|
||
from myagents.xiaohe_sessions import XiaoheSession
|
||
|
||
|
||
def _slug(path: Path) -> str:
|
||
return re.sub(r"[^A-Za-z0-9]", "-", str(path.resolve()))
|
||
|
||
|
||
def _write_jsonl(
|
||
proj_dir: Path, cid: str, ts: str, *, ai_title: str | None = None
|
||
) -> Path:
|
||
"""Write a minimal Claude Code jsonl with a message timestamp + optional aiTitle."""
|
||
lines: list[dict[str, object]] = [
|
||
{"type": "mode", "mode": "normal", "sessionId": cid},
|
||
{
|
||
"type": "permission-mode",
|
||
"permissionMode": "bypassPermissions",
|
||
"sessionId": cid,
|
||
},
|
||
]
|
||
if ai_title:
|
||
lines.append(
|
||
{"type": "ai-title", "aiTitle": ai_title, "sessionId": cid}
|
||
)
|
||
lines.append(
|
||
{
|
||
"type": "user",
|
||
"sessionId": cid,
|
||
"message": {
|
||
"id": f"m-{cid}",
|
||
"role": "user",
|
||
"content": [{"type": "text", "text": f"first prompt {cid}"}],
|
||
"timestamp": ts,
|
||
},
|
||
}
|
||
)
|
||
path = proj_dir / f"{cid}.jsonl"
|
||
path.write_text(
|
||
"\n".join(json.dumps(line) for line in lines) + "\n", encoding="utf-8"
|
||
)
|
||
return path
|
||
|
||
|
||
@pytest.fixture
|
||
def claude_env(tmp_path: Path, monkeypatch):
|
||
"""Point the claude session root + xiaohe index at temp fixtures."""
|
||
projects = tmp_path / "projects"
|
||
cwd = tmp_path / "cwd"
|
||
cwd.mkdir()
|
||
proj_dir = projects / _slug(cwd)
|
||
proj_dir.mkdir(parents=True)
|
||
|
||
index: dict[str, XiaoheSession] = {}
|
||
|
||
def fake_workspace_ids(_chat_cwd):
|
||
return ["primary"]
|
||
|
||
def fake_session_index(workspace_ids):
|
||
return index
|
||
|
||
monkeypatch.setattr(L, "_BACKENDS", copy.deepcopy(L._BACKENDS))
|
||
L._BACKENDS["claude"]["sessions_root"] = lambda: projects
|
||
monkeypatch.setattr(L, "workspace_ids_for_cwd", fake_workspace_ids)
|
||
monkeypatch.setattr(L, "session_index", fake_session_index)
|
||
return SimpleNamespace(cwd=cwd, proj_dir=proj_dir, index=index)
|
||
|
||
|
||
class TestClaudeSessionRows:
|
||
def test_merges_jsonl_with_xiaohe_index(self, claude_env) -> None:
|
||
_write_jsonl(
|
||
claude_env.proj_dir,
|
||
"aaa",
|
||
"2026-08-15T04:00:00Z",
|
||
ai_title="Client A",
|
||
)
|
||
_write_jsonl(claude_env.proj_dir, "bbb", "2026-08-15T03:00:00Z")
|
||
claude_env.index["aaa"] = XiaoheSession(
|
||
cli_session_id="aaa",
|
||
xiaohe_session_id="x-1",
|
||
title="Client A",
|
||
status="ready",
|
||
updated_at=1000.0,
|
||
)
|
||
|
||
rows = L._claude_session_rows(claude_env.cwd)
|
||
by_id = {r.cli_session_id: r for r in rows}
|
||
|
||
assert by_id["aaa"].origin == "xiaohe"
|
||
assert by_id["aaa"].status == "ready"
|
||
assert by_id["aaa"].xiaohe_session_id == "x-1"
|
||
assert by_id["aaa"].title == "Client A"
|
||
assert by_id["bbb"].origin == "cli"
|
||
assert by_id["bbb"].status is None
|
||
|
||
def test_sorts_by_content_activity_desc(self, claude_env) -> None:
|
||
_write_jsonl(claude_env.proj_dir, "old", "2026-08-13T04:00:00Z")
|
||
_write_jsonl(claude_env.proj_dir, "mid", "2026-08-14T04:00:00Z")
|
||
_write_jsonl(claude_env.proj_dir, "new", "2026-08-15T04:00:00Z")
|
||
|
||
rows = L._claude_session_rows(claude_env.cwd)
|
||
assert [r.cli_session_id for r in rows] == ["new", "mid", "old"]
|
||
|
||
def test_db_only_session_is_client_only(self, claude_env) -> None:
|
||
_write_jsonl(claude_env.proj_dir, "aaa", "2026-08-15T04:00:00Z")
|
||
claude_env.index["nojsonl"] = XiaoheSession(
|
||
cli_session_id="nojsonl",
|
||
xiaohe_session_id="x-2",
|
||
title="Lost",
|
||
status="ready",
|
||
updated_at=2000.0,
|
||
)
|
||
|
||
rows = L._claude_session_rows(claude_env.cwd)
|
||
by_id = {r.cli_session_id: r for r in rows}
|
||
assert by_id["nojsonl"].jsonl_path is None
|
||
assert by_id["nojsonl"].origin == "xiaohe"
|
||
|
||
def test_dedup_by_cli_session_id(self, claude_env) -> None:
|
||
_write_jsonl(claude_env.proj_dir, "dup", "2026-08-15T04:00:00Z")
|
||
claude_env.index["dup"] = XiaoheSession(
|
||
cli_session_id="dup",
|
||
xiaohe_session_id="x-3",
|
||
title="Dup",
|
||
status="ready",
|
||
updated_at=1.0,
|
||
)
|
||
rows = L._claude_session_rows(claude_env.cwd)
|
||
ids = [r.cli_session_id for r in rows]
|
||
assert ids.count("dup") == 1
|
||
|
||
|
||
class TestResumePickerClaude:
|
||
def test_non_tty_prints_and_exits(self, claude_env, capsys) -> None:
|
||
_write_jsonl(claude_env.proj_dir, "aaa", "2026-08-15T04:00:00Z")
|
||
with pytest.raises(SystemExit) as exc:
|
||
L._resume_picker_claude(claude_env.cwd)
|
||
assert exc.value.code == 0
|
||
err = capsys.readouterr().err
|
||
assert "Not a terminal" in err
|
||
assert "resume directly with myclaude -r <id>" in err
|
||
|
||
def test_routing_select_launches(self, claude_env, monkeypatch) -> None:
|
||
"""Picking a session resumes via ``claude --resume <id>``."""
|
||
_write_jsonl(claude_env.proj_dir, "aaa", "2026-08-15T04:00:00Z")
|
||
launched: list[list[str]] = []
|
||
monkeypatch.setattr(sys.stdin, "isatty", lambda: True)
|
||
monkeypatch.setattr(L, "_pick_session", lambda rows, cwd: rows[0])
|
||
monkeypatch.setattr(
|
||
L, "_launch", lambda *_a, **_k: launched.append(_a[2])
|
||
)
|
||
L._resume_picker_claude(claude_env.cwd)
|
||
assert launched == [["--resume", "aaa"]]
|
||
|
||
def test_routing_quit_exits(self, claude_env, monkeypatch) -> None:
|
||
"""Quitting the picker exits 0 without launching."""
|
||
_write_jsonl(claude_env.proj_dir, "aaa", "2026-08-15T04:00:00Z")
|
||
launched: list[list[str]] = []
|
||
monkeypatch.setattr(sys.stdin, "isatty", lambda: True)
|
||
monkeypatch.setattr(L, "_pick_session", lambda rows, cwd: None)
|
||
monkeypatch.setattr(
|
||
L, "_launch", lambda *_a, **_k: launched.append(_a[2])
|
||
)
|
||
with pytest.raises(SystemExit) as exc:
|
||
L._resume_picker_claude(claude_env.cwd)
|
||
assert exc.value.code == 0
|
||
assert launched == []
|
||
|
||
|
||
def _make_rows(claude_env, n: int):
|
||
"""n sessions, s000 newest → rows[0] == s000."""
|
||
for i in range(n):
|
||
_write_jsonl(
|
||
claude_env.proj_dir,
|
||
f"s{i:03d}",
|
||
f"2026-08-15T{23 - i:02d}:00:00Z",
|
||
)
|
||
return L._claude_session_rows(claude_env.cwd)
|
||
|
||
|
||
def _ansi_strip(text: str) -> str:
|
||
return re.sub(r"\x1b\[[0-9;]*m", "", text)
|
||
|
||
|
||
class TestPickerBlock:
|
||
"""Fixed 80-col block rendering (header + rows + status), CJK-aware."""
|
||
|
||
def _rows(self, claude_env, n: int):
|
||
return _make_rows(claude_env, n)
|
||
|
||
def test_rows_are_single_line_within_80(self, claude_env) -> None:
|
||
rows = self._rows(claude_env, 3)
|
||
state = L._PickerState(rows=rows)
|
||
lines = L._picker_block(state, claude_env.cwd, None)
|
||
# header + 3 rows + status
|
||
assert len(lines) == 5
|
||
for line in lines[1:4]:
|
||
assert len(_ansi_strip(line)) <= 80
|
||
assert "Sessions in" in lines[0]
|
||
|
||
def test_selected_row_reversed(self, claude_env) -> None:
|
||
rows = self._rows(claude_env, 3)
|
||
state = L._PickerState(rows=rows, cursor=1)
|
||
lines = L._picker_block(state, claude_env.cwd, None)
|
||
assert "\x1b[7m" in lines[2]
|
||
assert "\x1b[7m" not in lines[1]
|
||
|
||
def test_title_truncated_with_ellipsis(self, claude_env) -> None:
|
||
# _make_rows writes title-less jsonl; inject a long CJK title directly.
|
||
row = L.ClaudeSessionRow(
|
||
cli_session_id="long",
|
||
title="很长的中文标题" * 10,
|
||
origin="cli",
|
||
status=None,
|
||
updated_at=1000.0,
|
||
jsonl_path=Path("x.jsonl"),
|
||
xiaohe_session_id=None,
|
||
)
|
||
state = L._PickerState(rows=[row])
|
||
lines = L._picker_block(state, claude_env.cwd, None)
|
||
plain = _ansi_strip(lines[1])
|
||
assert "…" in plain
|
||
assert len(plain) <= 80
|
||
|
||
def test_message_line_replaces_hint(self, claude_env) -> None:
|
||
rows = self._rows(claude_env, 3)
|
||
state = L._PickerState(rows=rows)
|
||
lines = L._picker_block(state, claude_env.cwd, "client-only: no jsonl")
|
||
assert "client-only" in lines[-1]
|
||
assert "↑↓" not in lines[-1]
|
||
|
||
def test_cjk_display_width(self) -> None:
|
||
assert L._disp_width("ab") == 2
|
||
assert L._disp_width("中文") == 4
|
||
assert L._disp_width("a中") == 3
|
||
|
||
|
||
class TestPickerAdvance:
|
||
"""Windowed picker state machine (no tty needed)."""
|
||
|
||
def test_enter_selects_latest(self, claude_env) -> None:
|
||
rows = _make_rows(claude_env, 3)
|
||
state = L._PickerState(rows=rows)
|
||
state, action, payload = L._picker_advance(state, "enter")
|
||
assert action == "select"
|
||
assert payload == rows[0]
|
||
|
||
def test_number_selects_row(self, claude_env) -> None:
|
||
rows = _make_rows(claude_env, 3)
|
||
state = L._PickerState(rows=rows)
|
||
state, action, payload = L._picker_advance(state, "2")
|
||
assert action == "none"
|
||
state, action, payload = L._picker_advance(state, "enter")
|
||
assert action == "select"
|
||
assert payload == rows[1]
|
||
|
||
def test_number_out_of_range(self, claude_env) -> None:
|
||
rows = _make_rows(claude_env, 3)
|
||
state = L._PickerState(rows=rows)
|
||
state, _, _ = L._picker_advance(state, "9")
|
||
state, action, payload = L._picker_advance(state, "enter")
|
||
assert action == "none"
|
||
assert isinstance(payload, str) and "range" in payload
|
||
|
||
def test_scroll_window_advances(self, claude_env) -> None:
|
||
rows = _make_rows(claude_env, 15)
|
||
state = L._PickerState(rows=rows)
|
||
for _ in range(10):
|
||
state, action, payload = L._picker_advance(state, "down")
|
||
assert action == "none"
|
||
# cursor hits window bottom, then the window scrolls
|
||
assert state.offset == 1
|
||
assert state.cursor == 9
|
||
|
||
def test_quit(self, claude_env) -> None:
|
||
rows = _make_rows(claude_env, 3)
|
||
state = L._PickerState(rows=rows)
|
||
state, action, payload = L._picker_advance(state, "q")
|
||
assert action == "quit"
|
||
assert payload is None
|
||
|
||
def test_db_only_row_not_selectable(self, claude_env) -> None:
|
||
_write_jsonl(claude_env.proj_dir, "aaa", "2026-08-15T04:00:00Z")
|
||
# hosted session with no jsonl on disk, dated newest (year 2033)
|
||
claude_env.index["nojsonl"] = XiaoheSession(
|
||
cli_session_id="nojsonl",
|
||
xiaohe_session_id="x",
|
||
title="Lost",
|
||
status="ready",
|
||
updated_at=2_000_000_000.0,
|
||
)
|
||
rows = L._claude_session_rows(claude_env.cwd)
|
||
assert rows[0].jsonl_path is None
|
||
state = L._PickerState(rows=rows)
|
||
state, action, payload = L._picker_advance(state, "enter")
|
||
assert action == "none"
|
||
assert isinstance(payload, str) and "client-only" in payload
|
||
|
||
|
||
def _read_key_from_bytes(data: bytes) -> str:
|
||
"""Feed bytes through a pipe into ``_read_key``."""
|
||
r, w = os.pipe()
|
||
try:
|
||
os.write(w, data)
|
||
return L._read_key(r)
|
||
finally:
|
||
os.close(r)
|
||
os.close(w)
|
||
|
||
|
||
class TestReadKey:
|
||
def test_enter(self) -> None:
|
||
assert _read_key_from_bytes(b"\r") == "enter"
|
||
|
||
def test_arrow_down(self) -> None:
|
||
assert _read_key_from_bytes(b"\x1b[B") == "down"
|
||
|
||
def test_arrow_up(self) -> None:
|
||
assert _read_key_from_bytes(b"\x1b[A") == "up"
|
||
|
||
def test_page_down(self) -> None:
|
||
assert _read_key_from_bytes(b"\x1b[6~") == "pgdown"
|
||
|
||
def test_q(self) -> None:
|
||
assert _read_key_from_bytes(b"q") == "q"
|
||
|
||
def test_digit(self) -> None:
|
||
assert _read_key_from_bytes(b"5") == "5"
|
||
|
||
def test_escape(self) -> None:
|
||
assert _read_key_from_bytes(b"\x1b") == "esc"
|
||
|
||
def test_rapid_arrows_not_merged(self) -> None:
|
||
"""Back-to-back arrow keys stay distinct (no bleed into one ESC)."""
|
||
r, w = os.pipe()
|
||
try:
|
||
# 4 rapid down presses back-to-back, exactly as the terminal sends them
|
||
os.write(w, b"\x1b[B" * 4)
|
||
keys = [L._read_key(r) for _ in range(4)]
|
||
assert keys == ["down"] * 4
|
||
# bare ESC still reads as its own key afterwards
|
||
os.write(w, b"\x1b")
|
||
assert L._read_key(r) == "esc"
|
||
finally:
|
||
os.close(r)
|
||
os.close(w)
|