gitea-admin/herdr-session-switch
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Simplify session manager interaction model

Browse Source
This commit is contained in:
gitea-admin
2026-09-03 14:42:25 +00:00
parent 58e68b26de
commit 36bee161cb
4 changed files with 1010 additions and 489 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/test_selector.py
+345 -85
View File
@@ -37,6 +37,28 @@ def wait_for_action(selector: SessionSelector) -> None:
raise AssertionError("selector action did not finish")
class FakeScreen:
def __init__(self, height: int = 12, width: int = 80) -> None:
self.height = height
self.width = width
self.writes: list[tuple[int, int, str]] = []
def erase(self) -> None:
self.writes.clear()
def getmaxyx(self) -> tuple[int, int]:
return self.height, self.width
def addstr(self, y: int, x: int, value: str, _attribute: int) -> None:
self.writes.append((y, x, value))
def move(self, _y: int, _x: int) -> None:
pass
def refresh(self) -> None:
pass
class SelectorCatalogTests(unittest.TestCase):
def test_terminal_display_width_counts_cjk_and_combining_characters(self) -> None:
self.assertEqual(terminal_display_width("Rename: work"), 12)
@@ -86,17 +108,17 @@ class SelectorCatalogTests(unittest.TestCase):
local_sessions=["default", "draft"],
)
by_key = {row.key: row for row in rows}
self.assertEqual(by_key["window:@1"].main, "▶ L default")
self.assertEqual(by_key["window:@2"].main, "× R orangepi/work")
self.assertEqual(by_key["window:@1"].main, "▶ 本机")
self.assertEqual(by_key["window:@2"].main, "× orangepi / work")
self.assertIn("remote:orangepi:default", by_key)
self.assertIn("remote:orangepi:old", by_key)
self.assertTrue(by_key["remote:orangepi:old"].deletable)
self.assertIn("d/D 删除", by_key["remote:orangepi:old"].detail)
self.assertEqual(by_key["remote:orangepi:old"].detail, "已停止")
self.assertFalse(by_key["remote:orangepi:default"].deletable)
self.assertIn("new:orangepi", by_key)
self.assertEqual(by_key["local:draft"].main, "○ draft")
self.assertEqual(by_key["close-active"].window_id, "@1")
self.assertIn("manage-sessions", by_key)
self.assertIn("new-session", by_key)
self.assertEqual(by_key["local:draft"].main, "○ 本机 / draft")
self.assertNotIn("close-active", by_key)
self.assertNotIn("manage-sessions", by_key)
def test_loading_error_and_filter_rows(self) -> None:
entry = RemoteEntry("prod")
@@ -110,7 +132,9 @@ class SelectorCatalogTests(unittest.TestCase):
)
self.assertTrue(any(row.kind == "retry" for row in failed))
retry = next(row for row in failed if row.kind == "retry")
self.assertIn("d/D 移除记录", retry.main)
self.assertEqual(retry.main, "! prod")
self.assertEqual(retry.detail, "无法连接")
self.assertEqual(retry.error, "offline")
matches = filter_rows(failed, "offline")
self.assertEqual([row.kind for row in matches], ["retry"])
@@ -143,55 +167,61 @@ class SelectorCatalogTests(unittest.TestCase):
row = next(item for item in rows if item.key == "window:@2")
self.assertTrue(row.deletable)
self.assertEqual(row.window_id, "@2")
self.assertIn("同时关闭 Tab", row.detail)
self.assertEqual(row.detail, "已停止")
default = next(
item for item in rows if item.key == "remote:orangepi:default"
)
self.assertFalse(default.deletable)
self.assertIn("Herdr 保留 default", default.detail)
self.assertIn("default 保留项", default.detail)
def test_active_remote_session_has_close_and_stop_action(self) -> None:
local = CatalogWindow(
"@1",
"local:default",
"L default",
False,
False,
kind="local",
session="default",
)
remote = CatalogWindow(
"@2",
"remote:orangepi:work",
"R orangepi/work",
False,
True,
kind="remote",
target="orangepi",
session="work",
)
rows = build_main_rows(
[local, remote],
[RemoteEntry("orangepi")],
{},
{"orangepi"},
)
stop = next(row for row in rows if row.kind == "stop-session")
self.assertTrue(stop.selectable)
self.assertEqual(stop.window_id, "@2")
self.assertIn("任务会中断", stop.detail)
with tempfile.TemporaryDirectory() as directory:
registry = RemoteRegistry(Path(directory) / "remotes.json")
registry.upsert(RemoteEntry("orangepi"))
local = CatalogWindow(
"@1",
"local:default",
"L default",
False,
False,
kind="local",
session="default",
)
remote = CatalogWindow(
"@2",
"remote:orangepi:work",
"R orangepi/work",
False,
True,
kind="remote",
target="orangepi",
session="work",
)
selector = SessionSelector(
[local, remote],
registry,
str(Path(directory) / "missing-ssh"),
)
self.addCleanup(selector.pool.close)
selector.results["orangepi"] = RemoteQueryResult(
RemoteEntry("orangepi"),
sessions=(RemoteSession("work", True),),
)
last_tab_rows = build_main_rows(
[remote],
[RemoteEntry("orangepi")],
{},
{"orangepi"},
)
last_stop = next(
row for row in last_tab_rows if row.kind == "stop-session"
)
self.assertTrue(last_stop.selectable)
self.assertIn("Tab 将保留", last_stop.detail)
main_rows = selector._rows() # type: ignore[attr-defined]
self.assertFalse(any(row.kind == "stop-session" for row in main_rows))
remote_row = next(row for row in main_rows if row.key == "window:@2")
selector._show_actions(remote_row) # type: ignore[attr-defined]
actions = selector._rows() # type: ignore[attr-defined]
stop = next(row for row in actions if row.kind == "action-stop")
self.assertEqual(stop.window_id, "@2")
self.assertIn("中断", stop.detail)
self.assertTrue(any(row.kind == "action-close" for row in actions))
selector.windows = [remote]
actions = selector._rows() # type: ignore[attr-defined]
self.assertFalse(any(row.kind == "action-close" for row in actions))
self.assertTrue(any(row.kind == "action-stop" for row in actions))
def test_windows_remote_migrates_to_ssh_transport(self) -> None:
window = CatalogWindow(
@@ -250,8 +280,9 @@ class SelectorCatalogTests(unittest.TestCase):
{"windows": RemoteQueryResult(entry)},
set(),
)
new_row = next(item for item in rows if item.kind == "new")
self.assertEqual(new_row.remote_transport, "ssh")
empty_row = next(item for item in rows if item.kind == "remote-empty")
self.assertEqual(empty_row.remote_transport, "ssh")
self.assertTrue(any(item.kind == "new-session" for item in rows))
def test_windows_new_session_keeps_the_row_transport(self) -> None:
with tempfile.TemporaryDirectory() as directory:
@@ -265,10 +296,16 @@ class SelectorCatalogTests(unittest.TestCase):
sessions=(RemoteSession("default", True),),
platform="windows",
)
new_menu = next(
item
for item in selector._rows() # type: ignore[attr-defined]
if item.kind == "new-session"
)
selector._activate(new_menu) # type: ignore[attr-defined]
new_row = next(
item
for item in selector._rows() # type: ignore[attr-defined]
if item.kind == "new"
if item.kind == "action-new"
)
selector._activate(new_row) # type: ignore[attr-defined]
result = selector._finish_new("work") # type: ignore[attr-defined]
@@ -279,7 +316,182 @@ class SelectorCatalogTests(unittest.TestCase):
class SelectorActionTests(unittest.TestCase):
def test_unreachable_remote_can_be_removed_with_delete_key(self) -> None:
def test_search_is_explicit_and_escape_preserves_selection(self) -> None:
with tempfile.TemporaryDirectory() as directory:
window = CatalogWindow(
"@1",
"local:default",
"L default",
False,
True,
kind="local",
session="default",
default_label="L default",
)
selector = SessionSelector(
[window],
RemoteRegistry(Path(directory) / "remotes.json"),
"/missing/ssh",
)
self.addCleanup(selector.pool.close)
rows = selector._rows() # type: ignore[attr-defined]
selector.selected_key = "window:@1"
selector._handle_key("d", rows) # type: ignore[attr-defined]
self.assertEqual(selector.view, "main")
self.assertEqual(selector.query, "")
selector._handle_key("/", rows) # type: ignore[attr-defined]
self.assertEqual(selector.view, "search")
selector._handle_key("本", rows) # type: ignore[attr-defined]
self.assertEqual(selector.query, "本")
selector.selected_key = "window:@1"
selector._handle_key("\x1b", rows) # type: ignore[attr-defined]
self.assertEqual(selector.view, "main")
self.assertEqual(selector.query, "")
self.assertEqual(selector.selected_key, "window:@1")
def test_space_opens_actions_and_escape_returns_to_same_session(self) -> None:
with tempfile.TemporaryDirectory() as directory:
selector = SessionSelector(
[
CatalogWindow(
"@1",
"local:default",
"L default",
False,
True,
kind="local",
session="default",
)
],
RemoteRegistry(Path(directory) / "remotes.json"),
"/missing/ssh",
)
self.addCleanup(selector.pool.close)
rows = selector._rows() # type: ignore[attr-defined]
selector.selected_key = "window:@1"
selector._handle_key(" ", rows) # type: ignore[attr-defined]
self.assertEqual(selector.view, "actions")
self.assertEqual(selector.context_row.key, "window:@1")
selector._handle_key( # type: ignore[attr-defined]
"\x1b",
selector._rows(), # type: ignore[attr-defined]
)
self.assertEqual(selector.view, "main")
self.assertEqual(selector.selected_key, "window:@1")
def test_confirmation_defaults_to_cancel_and_escape_returns_to_actions(self) -> None:
with tempfile.TemporaryDirectory() as directory:
registry = RemoteRegistry(Path(directory) / "remotes.json")
entry = RemoteEntry("offline-host")
registry.upsert(entry)
selector = SessionSelector([], registry, "/missing/ssh")
self.addCleanup(selector.pool.close)
selector.results[entry.target] = RemoteQueryResult(entry, error="offline")
source = next(
row
for row in selector._rows() # type: ignore[attr-defined]
if row.kind == "retry"
)
selector._show_actions(source) # type: ignore[attr-defined]
remove = next(
row
for row in selector._rows() # type: ignore[attr-defined]
if row.kind == "action-remove"
)
selector._activate(remove) # type: ignore[attr-defined]
self.assertEqual(selector.view, "confirm")
self.assertEqual(selector.selected_key, "confirm-no")
selector._handle_key( # type: ignore[attr-defined]
"\x1b",
selector._rows(), # type: ignore[attr-defined]
)
self.assertEqual(selector.view, "actions")
self.assertEqual(selector.context_row.key, source.key)
def test_close_tab_action_updates_list_without_closing_popup(self) -> None:
with tempfile.TemporaryDirectory() as directory:
first = CatalogWindow(
"@1", "local:default", "L default", False, True,
kind="local", session="default",
)
second = CatalogWindow(
"@2", "local:work", "L work", False, False,
kind="local", session="work",
)
closed: list[str] = []
def close_handler(window_id: str) -> list[CatalogWindow]:
closed.append(window_id)
return [first]
selector = SessionSelector(
[first, second],
RemoteRegistry(Path(directory) / "remotes.json"),
"/missing/ssh",
close_window_handler=close_handler,
)
self.addCleanup(selector.pool.close)
source = next(
row
for row in selector._rows() # type: ignore[attr-defined]
if row.key == "window:@2"
)
selector._show_actions(source) # type: ignore[attr-defined]
close = next(
row
for row in selector._rows() # type: ignore[attr-defined]
if row.kind == "action-close"
)
result = selector._activate(close) # type: ignore[attr-defined]
self.assertIsNone(result)
self.assertEqual(closed, ["@2"])
self.assertEqual(selector.view, "main")
self.assertEqual(selector.windows, [first])
self.assertIn("已关闭 Hub Tab", selector.message)
def test_narrow_layout_prioritizes_name_and_row_menu_is_clickable(self) -> None:
with tempfile.TemporaryDirectory() as directory:
selector = SessionSelector(
[
CatalogWindow(
"@1",
"local:important-work",
"L important-work",
False,
False,
kind="local",
session="important-work",
)
],
RemoteRegistry(Path(directory) / "remotes.json"),
"/missing/ssh",
)
self.addCleanup(selector.pool.close)
rows = selector._rows() # type: ignore[attr-defined]
selector._normalize_selection(rows) # type: ignore[attr-defined]
screen = FakeScreen(width=32)
with patch("herdr_hub.selector.curses.curs_set"):
selector._draw(screen, rows) # type: ignore[attr-defined]
self.assertFalse(any("已打开" in value for _, _, value in screen.writes))
self.assertTrue(any("本机 / important" in value for _, _, value in screen.writes))
button_y, (button_start, _, source) = next(
iter(selector._row_action_buttons.items()) # type: ignore[attr-defined]
)
with patch(
"herdr_hub.selector.curses.getmouse",
return_value=(0, button_start, button_y, 0, curses.BUTTON1_RELEASED),
):
selector._handle_mouse(rows) # type: ignore[attr-defined]
self.assertEqual(selector.view, "actions")
self.assertEqual(selector.context_row.key, source.key)
def test_unreachable_remote_can_be_removed_from_its_action_menu(self) -> None:
with tempfile.TemporaryDirectory() as directory:
registry = RemoteRegistry(Path(directory) / "remotes.json")
entry = RemoteEntry("tang1@10.8.0.7")
@@ -295,12 +507,19 @@ class SelectorActionTests(unittest.TestCase):
rows = selector._rows() # type: ignore[attr-defined]
retry = next(item for item in rows if item.kind == "retry")
selector.selected_key = retry.key
selector._handle_key("D", rows) # type: ignore[attr-defined]
selector._handle_key(" ", rows) # type: ignore[attr-defined]
self.assertEqual(selector.view, "confirm-remove")
self.assertEqual(selector.view, "actions")
actions = selector._rows() # type: ignore[attr-defined]
remove = next(item for item in actions if item.kind == "action-remove")
selector._activate(remove) # type: ignore[attr-defined]
self.assertEqual(selector.view, "confirm")
self.assertEqual(selector.confirm_kind, "remove")
self.assertEqual(selector.input_target, entry.target)
confirm_rows = selector._rows() # type: ignore[attr-defined]
selector._handle_key("d", confirm_rows) # type: ignore[attr-defined]
confirm = next(item for item in confirm_rows if item.kind == "confirm-yes")
selector._activate(confirm) # type: ignore[attr-defined]
self.assertEqual(selector.view, "main")
self.assertEqual(registry.load(), [])
@@ -353,20 +572,27 @@ class SelectorActionTests(unittest.TestCase):
sessions=(RemoteSession("work", True),),
)
remote_row = next(
row
for row in selector._rows() # type: ignore[attr-defined]
if row.key == "window:@2"
)
selector._show_actions(remote_row) # type: ignore[attr-defined]
stop_row = next(
row
for row in selector._rows() # type: ignore[attr-defined]
if row.kind == "stop-session"
if row.kind == "action-stop"
)
selector._activate(stop_row) # type: ignore[attr-defined]
self.assertEqual(selector.view, "confirm-stop-session")
self.assertEqual(selector.view, "confirm")
self.assertEqual(selector.confirm_kind, "stop")
confirm = next(
row
for row in selector._rows() # type: ignore[attr-defined]
if row.kind == "stop-session-yes"
)
selector._activate(confirm) # type: ignore[attr-defined]
self.assertEqual(selector.view, "working-stop-session")
self.assertEqual(selector.view, "working")
wait_for_action(selector)
self.assertEqual(stopped, [("@2", "orangepi", "work")])
@@ -403,17 +629,30 @@ class SelectorActionTests(unittest.TestCase):
row for row in rows if row.key == "remote:orangepi:default"
)
selector.selected_key = running.key
selector._handle_key("d", rows) # type: ignore[attr-defined]
self.assertEqual(selector.view, "main")
selector._handle_key(" ", rows) # type: ignore[attr-defined]
self.assertEqual(selector.view, "actions")
self.assertFalse(
any(
item.kind == "action-delete"
for item in selector._rows() # type: ignore[attr-defined]
)
)
self.assertEqual(deleted, [])
self.assertEqual(selector.query, "")
self.assertIn("default", selector.message)
selector._go_back() # type: ignore[attr-defined]
selector.query = ""
rows = selector._rows() # type: ignore[attr-defined]
old = next(row for row in rows if row.key == "remote:orangepi:old")
selector.selected_key = old.key
selector._handle_key("D", rows) # type: ignore[attr-defined]
self.assertEqual(selector.view, "confirm-delete-session")
selector._handle_key(" ", rows) # type: ignore[attr-defined]
delete = next(
item
for item in selector._rows() # type: ignore[attr-defined]
if item.kind == "action-delete"
)
selector._activate(delete) # type: ignore[attr-defined]
self.assertEqual(selector.view, "confirm")
self.assertEqual(selector.confirm_kind, "delete")
confirm = next(
row
for row in selector._rows() # type: ignore[attr-defined]
@@ -427,7 +666,7 @@ class SelectorActionTests(unittest.TestCase):
selector._handle_mouse( # type: ignore[attr-defined]
selector._rows() # type: ignore[attr-defined]
)
self.assertEqual(selector.view, "working-delete-session")
self.assertEqual(selector.view, "working")
wait_for_action(selector)
self.assertEqual(deleted, [("orangepi", "old", None)])
@@ -490,8 +729,15 @@ class SelectorActionTests(unittest.TestCase):
rows = selector._rows() # type: ignore[attr-defined]
row = next(item for item in rows if item.key == "window:@2")
selector.selected_key = row.key
selector._handle_key("d", rows) # type: ignore[attr-defined]
self.assertEqual(selector.view, "confirm-delete-session")
selector._handle_key(" ", rows) # type: ignore[attr-defined]
delete = next(
item
for item in selector._rows() # type: ignore[attr-defined]
if item.kind == "action-delete"
)
selector._activate(delete) # type: ignore[attr-defined]
self.assertEqual(selector.view, "confirm")
self.assertEqual(selector.confirm_kind, "delete")
confirm = next(
item
for item in selector._rows() # type: ignore[attr-defined]
@@ -545,7 +791,7 @@ class SelectorActionTests(unittest.TestCase):
self.assertEqual(window_result, SelectorResult.window("@2", dead=True))
close_result = selector._activate( # type: ignore[attr-defined]
PickerRow("close", "close", "close", window_id="@2")
PickerRow("close", "action-close", "close", window_id="@2")
)
self.assertEqual(close_result, SelectorResult.close("@2"))
@@ -572,20 +818,20 @@ class SelectorActionTests(unittest.TestCase):
aliases={"local:default": "日常工作"},
)
self.addCleanup(selector.pool.close)
selector._activate( # type: ignore[attr-defined]
next(
row
for row in selector._rows() # type: ignore[attr-defined]
if row.kind == "manage-sessions"
)
session_row = next(
row
for row in selector._rows() # type: ignore[attr-defined]
if row.key == "window:@3"
)
selector._show_actions(session_row) # type: ignore[attr-defined]
rename_row = next(
row
for row in selector._rows() # type: ignore[attr-defined]
if row.kind == "rename"
if row.kind == "action-rename"
)
selector._activate(rename_row) # type: ignore[attr-defined]
self.assertEqual(selector.view, "input-rename")
self.assertEqual(selector.view, "input")
self.assertEqual(selector.input_kind, "rename")
self.assertEqual(selector.input_value, "日常工作")
self.assertEqual(
selector._finish_rename(" 新名称 "), # type: ignore[attr-defined]
@@ -605,24 +851,30 @@ class SelectorActionTests(unittest.TestCase):
rename_handler=lambda key, alias: renamed.append((key, alias)),
)
self.addCleanup(live_selector.pool.close)
live_selector._show_manage_sessions() # type: ignore[attr-defined]
live_session_row = next(
row
for row in live_selector._rows() # type: ignore[attr-defined]
if row.key == "window:@3"
)
live_selector._show_actions(live_session_row) # type: ignore[attr-defined]
live_selector._activate( # type: ignore[attr-defined]
next(
row
for row in live_selector._rows() # type: ignore[attr-defined]
if row.kind == "rename"
if row.kind == "action-rename"
)
)
self.assertIsNone(
live_selector._finish_rename("连续改名") # type: ignore[attr-defined]
)
self.assertEqual(renamed, [("local:default", "连续改名")])
self.assertEqual(live_selector.view, "manage-sessions")
self.assertEqual(live_selector.selected_key, "rename:local:default")
self.assertEqual(live_selector.view, "main")
self.assertEqual(live_selector.selected_key, "window:@3")
self.assertEqual(live_selector.aliases["local:default"], "连续改名")
self.assertIn("已保存", live_selector.message)
selector.view = "input-new"
selector.view = "input"
selector.input_kind = "new"
selector.input_target = "orangepi"
new_result = selector._finish_new("work") # type: ignore[attr-defined]
self.assertEqual(
@@ -658,8 +910,16 @@ class SelectorActionTests(unittest.TestCase):
output += os.read(descriptor, 4096)
self.assertIn(b"Sessions", output)
# Empty registries render "+ 添加远端" as the first selectable
# row at terminal row 5. A single SGR mouse click must activate it.
# row. It opens one small connection-mode menu.
os.write(descriptor, b"\x1b[<0;5;5M\x1b[<0;5;5m")
mode_menu = b""
mode_deadline = time.monotonic() + 2
while time.monotonic() < mode_deadline and b"Linux / macOS" not in mode_menu:
readable, _, _ = select.select([descriptor], [], [], 0.1)
if readable:
mode_menu += os.read(descriptor, 4096)
self.assertIn(b"Linux / macOS", mode_menu)
os.write(descriptor, b"\x1b[<0;5;6M\x1b[<0;5;6m")
opened_input = b""
input_deadline = time.monotonic() + 2
while time.monotonic() < input_deadline and b"SSH target" not in opened_input:
@@ -667,7 +927,7 @@ class SelectorActionTests(unittest.TestCase):
if readable:
opened_input += os.read(descriptor, 4096)
self.assertIn(b"SSH target", opened_input)
os.write(descriptor, b"\x1b\x1b")
os.write(descriptor, b"\x1b\x1b\x1b")
while time.monotonic() < deadline:
completed, wait_status = os.waitpid(pid, os.WNOHANG)
if completed == pid: