gitea-adminandClaude Fable 5.1
907813867a
configure() ran one tmux client per option and per key binding, and refresh_labels() one or two per window, on every start/refresh/open: a no-op refresh with three windows spawned 45 tmux processes. On a phone each spawn is noticeable. Add run_batch(), which joins commands with ';' separators (escaping arguments that end in ';', which tmux would otherwise split) and replays them one by one if the batch fails so errors stay attributable. Use it for configure(), window creation, endpoint replacement and label refreshes. list-windows now also reports the stored short label so unchanged labels are skipped, and managed_windows() no longer probes has-session first. The same no-op refresh now spawns 9 processes. Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
434 lines
17 KiB
Python
434 lines
17 KiB
Python
from __future__ import annotations
|
||
|
||
from concurrent.futures import ThreadPoolExecutor
|
||
import os
|
||
from pathlib import Path
|
||
import shutil
|
||
import tempfile
|
||
import time
|
||
import unittest
|
||
import uuid
|
||
from unittest.mock import patch
|
||
|
||
from herdr_hub.cli import (
|
||
BOOTSTRAP_WINDOW_NAME,
|
||
Endpoint,
|
||
HubError,
|
||
TmuxHub,
|
||
WINDOW_SHORT_LABEL_OPTION,
|
||
build_parser,
|
||
command_control,
|
||
)
|
||
|
||
|
||
@unittest.skipUnless(shutil.which("tmux"), "tmux is required")
|
||
class TmuxIntegrationTests(unittest.TestCase):
|
||
def test_verified_handoff_reconnects_only_originating_client(self) -> None:
|
||
self.hub.ensure_server()
|
||
original = self.hub.create_window(Endpoint.local("first"), str(self.fake_herdr))
|
||
other = self.hub.create_window(Endpoint.local("second"), str(self.fake_herdr))
|
||
self.hub.select_window(original.window_id)
|
||
def pid(window):
|
||
return self.hub.run(["display-message", "-p", "-t", window.window_id, "#{pane_pid}"]).stdout.strip()
|
||
original_pid, other_pid = pid(original), pid(other)
|
||
def close_popup(run):
|
||
self.hub.select_window(other.window_id)
|
||
run.__self__.reconnect_requested = True
|
||
return 0
|
||
with patch("herdr_hub.cli.make_hub", return_value=self.hub), patch(
|
||
"herdr_hub.cli.resolve_herdr", return_value=str(self.fake_herdr)
|
||
), patch("herdr_hub.cli.curses.wrapper", side_effect=close_popup):
|
||
self.assertEqual(command_control(build_parser().parse_args(["control"])), 0)
|
||
self.assertNotEqual(pid(original), original_pid)
|
||
self.assertEqual(pid(other), other_pid)
|
||
|
||
def setUp(self) -> None:
|
||
self.temporary_directory = tempfile.TemporaryDirectory()
|
||
self.addCleanup(self.temporary_directory.cleanup)
|
||
root = Path(self.temporary_directory.name)
|
||
self.fake_herdr = root / "herdr"
|
||
self.fake_herdr.write_text(
|
||
"#!/bin/sh\n"
|
||
"if [ \"${1:-}\" = \"--session\" ] && [ \"${2:-}\" = \"dead\" ]; then\n"
|
||
" printf '%s\\n' 'fake startup failure' >&2\n"
|
||
" exit 7\n"
|
||
"fi\n"
|
||
"exec sleep 30\n",
|
||
encoding="utf-8",
|
||
)
|
||
self.fake_herdr.chmod(0o755)
|
||
self.fake_ssh = root / "ssh"
|
||
self.fake_ssh.write_text("#!/bin/sh\nexec sleep 30\n", encoding="utf-8")
|
||
self.fake_ssh.chmod(0o755)
|
||
self.socket_name = f"herdr-hub-test-{os.getpid()}-{uuid.uuid4().hex[:8]}"
|
||
self.socket_path = Path(
|
||
os.environ.get("TMUX_TMPDIR", f"/tmp/tmux-{os.getuid()}")
|
||
) / self.socket_name
|
||
self.hub = TmuxHub(
|
||
tmux_bin=shutil.which("tmux") or "tmux",
|
||
ssh_bin=str(self.fake_ssh),
|
||
socket_name=self.socket_name,
|
||
session_name="hub",
|
||
config_file="/dev/null",
|
||
script_path="/dev/null",
|
||
)
|
||
self.addCleanup(self._stop_hub)
|
||
|
||
def _stop_hub(self) -> None:
|
||
try:
|
||
self.hub.stop()
|
||
except Exception:
|
||
pass
|
||
self.socket_path.unlink(missing_ok=True)
|
||
|
||
def test_control_uses_existing_prefix_and_active_window(self) -> None:
|
||
self.hub.ensure_server()
|
||
binding = self.hub.run(["list-keys", "-T", "prefix", "h"]).stdout
|
||
self.assertIn("display-popup", binding)
|
||
self.assertIn("control", binding)
|
||
self.assertNotIn("--window", binding)
|
||
self.assertNotIn("#{window_id}", binding)
|
||
old_binding = self.hub.run(
|
||
["list-keys", "-T", "root", "C-g"], check=False
|
||
)
|
||
self.assertNotEqual(old_binding.returncode, 0)
|
||
self.assertEqual(self.hub.run(["show-options", "-gv", "prefix"]).stdout.strip(), "C-a")
|
||
self.assertEqual(self.hub.run(["show-options", "-gv", "prefix2"]).stdout.strip(), "None")
|
||
self.assertIn("choose", self.hub.run(["list-keys", "-T", "prefix", "g"]).stdout)
|
||
|
||
def test_help_popup_replaces_status_shortcut_list(self) -> None:
|
||
self.hub.ensure_server()
|
||
binding = self.hub.run(["list-keys", "-T", "prefix", "?"]).stdout
|
||
self.assertIn("display-popup", binding)
|
||
self.assertIn(" keys", binding)
|
||
status_right = self.hub.run(
|
||
["show-options", "-gv", "status-right"]
|
||
).stdout
|
||
self.assertIn("C-a ?", status_right)
|
||
self.assertNotIn("choose", status_right)
|
||
self.assertNotIn("refresh", status_right)
|
||
self.assertNotIn("C-a R", status_right)
|
||
|
||
def test_configures_isolated_server_and_adds_without_duplicates(self) -> None:
|
||
self.assertTrue(self.hub.ensure_server())
|
||
endpoints = [Endpoint.local(), Endpoint.remote("orangepi")]
|
||
added = self.hub.add_missing(endpoints, str(self.fake_herdr))
|
||
self.assertEqual(len(added), 2)
|
||
self.assertEqual(self.hub.add_missing(endpoints, str(self.fake_herdr)), [])
|
||
|
||
windows = self.hub.managed_windows()
|
||
self.assertEqual(
|
||
{window.key for window in windows},
|
||
{"local:default", "remote:orangepi:default"},
|
||
)
|
||
remote_window = next(
|
||
window for window in windows if window.key == "remote:orangepi:default"
|
||
)
|
||
self.assertEqual(remote_window.endpoint, Endpoint.remote("orangepi"))
|
||
all_names = self.hub.run(
|
||
["list-windows", "-t", "hub", "-F", "#{window_name}"]
|
||
).stdout.splitlines()
|
||
self.assertNotIn(BOOTSTRAP_WINDOW_NAME, all_names)
|
||
|
||
prefix = self.hub.run(["show-options", "-gv", "prefix"]).stdout.strip()
|
||
status = self.hub.run(["show-options", "-gv", "status"]).stdout.strip()
|
||
position = self.hub.run(
|
||
["show-options", "-gv", "status-position"]
|
||
).stdout.strip()
|
||
mouse = self.hub.run(["show-options", "-gv", "mouse"]).stdout.strip()
|
||
clipboard = self.hub.run(
|
||
["show-options", "-sv", "set-clipboard"]
|
||
).stdout.strip()
|
||
status_right = self.hub.run(
|
||
["show-options", "-gv", "status-right"]
|
||
).stdout.strip()
|
||
expanded_status_left = self.hub.run(
|
||
["display-message", "-p", "#{E:status-left}"]
|
||
).stdout.strip()
|
||
dead_format = self.hub.run(
|
||
["show-options", "-wgv", "remain-on-exit-format"]
|
||
).stdout.strip()
|
||
inactive_window_format = self.hub.run(
|
||
["show-options", "-wgv", "window-status-format"]
|
||
).stdout.strip()
|
||
current_window_format = self.hub.run(
|
||
["show-options", "-wgv", "window-status-current-format"]
|
||
).stdout.strip()
|
||
status_left_binding = self.hub.run(
|
||
["list-keys", "-T", "root", "MouseDown1StatusLeft"]
|
||
).stdout.strip()
|
||
close_binding = self.hub.run(
|
||
["list-keys", "-T", "prefix", "q"]
|
||
).stdout.strip()
|
||
legacy_close_binding = self.hub.run(
|
||
["list-keys", "-T", "prefix", "&"], check=False
|
||
)
|
||
uppercase_close_binding = self.hub.run(
|
||
["list-keys", "-T", "prefix", "Q"], check=False
|
||
)
|
||
self.assertEqual(prefix, "C-a")
|
||
self.assertEqual(status, "on")
|
||
self.assertEqual(position, "top")
|
||
self.assertEqual(mouse, "on")
|
||
self.assertEqual(clipboard, "on")
|
||
self.assertIn("C-a ?", status_right)
|
||
self.assertTrue(expanded_status_left.startswith("#[bold,fg=colour81]"))
|
||
self.assertNotIn("C-a R", status_right)
|
||
self.assertIn("连接已断开", dead_format)
|
||
self.assertIn("Ctrl+A R", dead_format)
|
||
self.assertIn("client_width", inactive_window_format)
|
||
self.assertIn(WINDOW_SHORT_LABEL_OPTION, inactive_window_format)
|
||
self.assertNotIn("#I", inactive_window_format)
|
||
self.assertIn("client_width", current_window_format)
|
||
self.assertIn(WINDOW_SHORT_LABEL_OPTION, current_window_format)
|
||
self.assertIn("display-popup", status_left_binding)
|
||
self.assertIn("-EE", status_left_binding)
|
||
self.assertIn("confirm-before", close_binding)
|
||
self.assertIn(" close --window ", close_binding)
|
||
self.assertNotEqual(legacy_close_binding.returncode, 0)
|
||
self.assertNotEqual(uppercase_close_binding.returncode, 0)
|
||
|
||
local_short_label = self.hub.run(
|
||
[
|
||
"show-options",
|
||
"-wv",
|
||
"-t",
|
||
next(window.window_id for window in windows if window.key == "local:default"),
|
||
WINDOW_SHORT_LABEL_OPTION,
|
||
]
|
||
).stdout.strip()
|
||
remote_short_label = self.hub.run(
|
||
[
|
||
"show-options",
|
||
"-wv",
|
||
"-t",
|
||
remote_window.window_id,
|
||
WINDOW_SHORT_LABEL_OPTION,
|
||
]
|
||
).stdout.strip()
|
||
rendered_local_short_label = self.hub.run(
|
||
[
|
||
"display-message",
|
||
"-p",
|
||
"-t",
|
||
next(window.window_id for window in windows if window.key == "local:default"),
|
||
"#{=12:@herdr_hub_short_label}",
|
||
]
|
||
).stdout.strip()
|
||
self.assertEqual(local_short_label, "local")
|
||
self.assertEqual(remote_short_label, "orangepi")
|
||
self.assertEqual(rendered_local_short_label, "local")
|
||
|
||
def test_dismissed_window_stays_hidden_until_explicitly_reopened(self) -> None:
|
||
self.hub.ensure_server()
|
||
endpoints = [Endpoint.local(), Endpoint.local("work")]
|
||
self.hub.add_missing(endpoints, str(self.fake_herdr))
|
||
|
||
default_window = self.hub.window_for_key("local:default")
|
||
assert default_window is not None
|
||
self.hub.dismiss_window(default_window.window_id)
|
||
self.assertEqual(self.hub.dismissed_keys(), {"local:default"})
|
||
self.assertIsNone(self.hub.window_for_key("local:default"))
|
||
self.assertEqual(self.hub.add_missing(endpoints, str(self.fake_herdr)), [])
|
||
|
||
reopened = self.hub.activate_endpoint(Endpoint.local(), str(self.fake_herdr))
|
||
self.assertEqual(reopened.key, "local:default")
|
||
self.assertEqual(self.hub.dismissed_keys(), set())
|
||
|
||
work_window = self.hub.window_for_key("local:work")
|
||
assert work_window is not None
|
||
self.hub.dismiss_window(work_window.window_id)
|
||
with self.assertRaises(HubError):
|
||
self.hub.dismiss_window(reopened.window_id)
|
||
|
||
def test_dead_window_is_retained_and_can_be_respawned(self) -> None:
|
||
self.hub.ensure_server()
|
||
window = self.hub.create_window(Endpoint.local("dead"), str(self.fake_herdr))
|
||
self.hub.remove_bootstrap()
|
||
|
||
deadline = time.monotonic() + 3
|
||
dead = False
|
||
while time.monotonic() < deadline:
|
||
current = self.hub.window_for_key("local:dead")
|
||
dead = current.dead if current is not None else False
|
||
if dead:
|
||
break
|
||
time.sleep(0.05)
|
||
self.assertTrue(dead)
|
||
|
||
rendered_status = self.hub.run(
|
||
[
|
||
"display-message",
|
||
"-p",
|
||
"-t",
|
||
window.window_id,
|
||
"#{?pane_dead,× #{window_name} [断开],● #{window_name}}",
|
||
]
|
||
).stdout.strip()
|
||
self.assertEqual(rendered_status, f"× {window.name} [断开]")
|
||
|
||
self.hub.reconnect(window.window_id)
|
||
self.assertIsNotNone(self.hub.window_for_key("local:dead"))
|
||
|
||
def test_startup_verification_reports_immediate_process_failure(self) -> None:
|
||
self.hub.ensure_server()
|
||
with self.assertRaisesRegex(HubError, "fake startup failure"):
|
||
self.hub.activate_endpoint(
|
||
Endpoint.local("dead"),
|
||
str(self.fake_herdr),
|
||
verify_startup=True,
|
||
)
|
||
|
||
failed = self.hub.window_for_key("local:dead")
|
||
self.assertIsNotNone(failed)
|
||
assert failed is not None
|
||
self.assertTrue(failed.dead)
|
||
|
||
def test_alias_is_applied_immediately_and_can_be_reset(self) -> None:
|
||
self.hub.aliases["remote:orangepi:work"] = "客厅开发"
|
||
self.hub.ensure_server()
|
||
window = self.hub.create_window(
|
||
Endpoint.remote("orangepi", "work"),
|
||
str(self.fake_herdr),
|
||
)
|
||
self.hub.remove_bootstrap()
|
||
|
||
current = self.hub.window_for_key(window.key)
|
||
assert current is not None
|
||
self.assertEqual(current.name, "客厅开发")
|
||
short_label = self.hub.run(
|
||
[
|
||
"show-options",
|
||
"-wv",
|
||
"-t",
|
||
window.window_id,
|
||
WINDOW_SHORT_LABEL_OPTION,
|
||
]
|
||
).stdout.strip()
|
||
self.assertEqual(short_label, "客厅开发")
|
||
|
||
renamed = self.hub.apply_alias(window.key, "临时名称")
|
||
assert renamed is not None
|
||
self.assertEqual(renamed.name, "临时名称")
|
||
# tmux parses a leading '-' as a flag unless the name is guarded; a
|
||
# stored name like this must also survive the label refresh that every
|
||
# later start/refresh performs.
|
||
dashed = self.hub.apply_alias(window.key, "-dash")
|
||
assert dashed is not None
|
||
self.assertEqual(dashed.name, "-dash")
|
||
self.hub.ensure_server()
|
||
refreshed = self.hub.window_for_key(window.key)
|
||
assert refreshed is not None
|
||
self.assertEqual(refreshed.name, "-dash")
|
||
restored = self.hub.apply_alias(window.key, "")
|
||
assert restored is not None
|
||
self.assertEqual(restored.name, "R orangepi/work")
|
||
|
||
def test_failed_native_window_can_migrate_to_windows_ssh_transport(self) -> None:
|
||
self.hub.ensure_server()
|
||
native = self.hub.create_window(
|
||
Endpoint.remote("windows"),
|
||
str(self.fake_herdr),
|
||
)
|
||
self.hub.remove_bootstrap()
|
||
|
||
migrated = self.hub.activate_endpoint(
|
||
Endpoint.remote("windows", remote_transport="ssh"),
|
||
str(self.fake_herdr),
|
||
)
|
||
self.assertEqual(migrated.window_id, native.window_id)
|
||
assert migrated.endpoint is not None
|
||
self.assertEqual(migrated.endpoint.remote_transport, "ssh")
|
||
command = self.hub.run(
|
||
[
|
||
"show-options",
|
||
"-wv",
|
||
"-t",
|
||
native.window_id,
|
||
"@herdr_hub_command",
|
||
]
|
||
).stdout.strip()
|
||
self.assertIn(str(self.fake_ssh), command)
|
||
self.assertIn("-tt", command)
|
||
|
||
def test_reconfiguration_batches_tmux_calls_and_escapes_semicolons(self) -> None:
|
||
calls: list[list[str]] = []
|
||
real_run = self.hub._run # type: ignore[attr-defined]
|
||
|
||
def counting_run(command, **keywords): # type: ignore[no-untyped-def]
|
||
calls.append(list(command))
|
||
return real_run(command, **keywords)
|
||
|
||
self.hub._run = counting_run # type: ignore[attr-defined]
|
||
self.hub.ensure_server()
|
||
for name in ("a", "b", "c"):
|
||
self.hub.create_window(Endpoint.local(name), str(self.fake_herdr))
|
||
self.hub.remove_bootstrap()
|
||
|
||
# A no-op refresh (nothing new discovered, labels unchanged) used to
|
||
# spawn one tmux client per option and per window label.
|
||
calls.clear()
|
||
self.hub.ensure_server()
|
||
self.hub.add_missing([], str(self.fake_herdr))
|
||
self.hub.managed_windows()
|
||
self.assertLessEqual(len(calls), 10, calls)
|
||
self.assertEqual(self.hub.run(["show-options", "-gv", "prefix"]).stdout.strip(), "C-a")
|
||
self.assertIn("display-popup", self.hub.run(["list-keys", "-T", "prefix", "g"]).stdout)
|
||
|
||
# tmux treats any argument ending in ';' as a command separator.
|
||
self.hub.run_batch(
|
||
[
|
||
["set-option", "-g", "@herdr_hub_test_one", "value;"],
|
||
["set-option", "-g", "@herdr_hub_test_two", "other"],
|
||
]
|
||
)
|
||
self.assertEqual(
|
||
self.hub.run(["show-options", "-gv", "@herdr_hub_test_one"]).stdout.strip(),
|
||
"value;",
|
||
)
|
||
self.assertEqual(
|
||
self.hub.run(["show-options", "-gv", "@herdr_hub_test_two"]).stdout.strip(),
|
||
"other",
|
||
)
|
||
# A failing command is attributed precisely instead of silently
|
||
# skipping the rest of the batch.
|
||
with self.assertRaisesRegex(HubError, "no-such-option"):
|
||
self.hub.run_batch(
|
||
[
|
||
["set-option", "-g", "@herdr_hub_test_three", "set"],
|
||
["set-option", "-g", "no-such-option", "1"],
|
||
]
|
||
)
|
||
self.assertEqual(
|
||
self.hub.run(["show-options", "-gv", "@herdr_hub_test_three"]).stdout.strip(),
|
||
"set",
|
||
)
|
||
|
||
def test_concurrent_open_of_the_same_endpoint_creates_one_window(self) -> None:
|
||
self.hub.ensure_server()
|
||
second_hub = TmuxHub(
|
||
tmux_bin=shutil.which("tmux") or "tmux",
|
||
ssh_bin=str(self.fake_ssh),
|
||
socket_name=self.socket_name,
|
||
session_name="hub",
|
||
config_file="/dev/null",
|
||
script_path="/dev/null",
|
||
)
|
||
endpoint = Endpoint.local("concurrent")
|
||
with ThreadPoolExecutor(max_workers=2) as executor:
|
||
windows = list(
|
||
executor.map(
|
||
lambda hub: hub.activate_endpoint(endpoint, str(self.fake_herdr)),
|
||
[self.hub, second_hub],
|
||
)
|
||
)
|
||
self.assertEqual(windows[0].window_id, windows[1].window_id)
|
||
self.assertEqual(
|
||
[window.key for window in self.hub.managed_windows()],
|
||
["local:concurrent"],
|
||
)
|
||
|
||
|
||
if __name__ == "__main__":
|
||
unittest.main()
|