Zhengshou Lai
88170bdf57
Bare -r/--resume now opens a numbered picker merging jsonl (source of truth) with the xiaohe hosted-session index, deduped by cli_session_id and sorted by real content activity (not jsonl mtime, which the xiaohe projection re-touches). -l gains a Src column (xiaohe/cli); -r <id> and -c stay passthrough. Adds read-only xiaohe_sessions data layer + tests.
91 lines
3.3 KiB
Python
91 lines
3.3 KiB
Python
"""Tests for the read-only xiaohe hosted-session data layer."""
|
|
|
|
import sqlite3
|
|
from pathlib import Path
|
|
|
|
from myagents.xiaohe_sessions import session_index, workspace_ids_for_cwd
|
|
|
|
|
|
def _make_workspaces_db(path: Path, rows: list[tuple[str, str | None]]) -> None:
|
|
conn = sqlite3.connect(path)
|
|
conn.execute(
|
|
"CREATE TABLE workspaces (workspace_id TEXT, workspace_path TEXT)"
|
|
)
|
|
conn.executemany("INSERT INTO workspaces VALUES (?, ?)", rows)
|
|
conn.commit()
|
|
conn.close()
|
|
|
|
|
|
def _make_sessions_db(path: Path, rows: list[tuple]) -> None:
|
|
conn = sqlite3.connect(path)
|
|
conn.execute(
|
|
"CREATE TABLE sessions ("
|
|
"session_id TEXT PRIMARY KEY, workspace_id TEXT, cli_session_id TEXT, "
|
|
"title TEXT, status TEXT, updated_at REAL)"
|
|
)
|
|
conn.executemany("INSERT INTO sessions VALUES (?, ?, ?, ?, ?, ?)", rows)
|
|
conn.commit()
|
|
conn.close()
|
|
|
|
|
|
class TestWorkspaceIdsForCwd:
|
|
def test_returns_matching_bind_workspaces(self, tmp_path: Path) -> None:
|
|
"""Bind workspaces whose path resolves to the cwd are returned."""
|
|
db = tmp_path / "workspaces.db"
|
|
cwd = tmp_path / "ws"
|
|
cwd.mkdir()
|
|
_make_workspaces_db(
|
|
db,
|
|
[
|
|
("primary", str(cwd)),
|
|
("bot-f", str(cwd)),
|
|
("other", str(tmp_path / "elsewhere")),
|
|
("session-ws", None),
|
|
],
|
|
)
|
|
result = workspace_ids_for_cwd(cwd, workspaces_db=db)
|
|
assert result == ["primary", "bot-f"]
|
|
|
|
def test_no_match_returns_empty(self, tmp_path: Path) -> None:
|
|
db = tmp_path / "workspaces.db"
|
|
cwd = tmp_path / "ws"
|
|
_make_workspaces_db(db, [("primary", str(tmp_path / "other"))])
|
|
assert workspace_ids_for_cwd(cwd, workspaces_db=db) == []
|
|
|
|
def test_missing_store_returns_empty(self, tmp_path: Path) -> None:
|
|
assert workspace_ids_for_cwd(tmp_path / "ws") == []
|
|
|
|
|
|
class TestSessionIndex:
|
|
def test_maps_cli_ids_for_matching_workspaces(self, tmp_path: Path) -> None:
|
|
db = tmp_path / "sessions-claude.db"
|
|
_make_sessions_db(
|
|
db,
|
|
[
|
|
("s1", "primary", "c1", "Title A", "ready", 1000.0),
|
|
("s2", "primary", "c2", None, "error", 2000.0),
|
|
("s3", "other", "c3", "Title B", "ready", 3000.0),
|
|
],
|
|
)
|
|
index = session_index(["primary"], sessions_db=db)
|
|
assert set(index) == {"c1", "c2"}
|
|
assert index["c1"].xiaohe_session_id == "s1"
|
|
assert index["c1"].status == "ready"
|
|
assert index["c2"].title is None
|
|
assert index["c2"].updated_at == 2000.0
|
|
|
|
def test_skips_null_cli_session_id(self, tmp_path: Path) -> None:
|
|
db = tmp_path / "sessions-claude.db"
|
|
_make_sessions_db(db, [("s1", "primary", None, "T", "ready", 1.0)])
|
|
assert session_index(["primary"], sessions_db=db) == {}
|
|
|
|
def test_empty_workspace_ids_skips_query(self, tmp_path: Path) -> None:
|
|
db = tmp_path / "sessions-claude.db"
|
|
_make_sessions_db(db, [("s1", "primary", "c1", "T", "ready", 1.0)])
|
|
assert session_index([], sessions_db=db) == {}
|
|
|
|
def test_missing_store_returns_empty(self, tmp_path: Path) -> None:
|
|
assert (
|
|
session_index(["primary"], sessions_db=tmp_path / "nope.db") == {}
|
|
)
|