286 lines
9.8 KiB
Python
286 lines
9.8 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)
|
||
|
||
|
||
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"
|