gitea-adminandClaude Fable 5.1
ebbd5821fd
- Display names starting with '-' were parsed by tmux rename-window as flags. The alias was persisted before tmux rejected it, so every later start/refresh/open failed in refresh_labels until aliases.json was edited by hand. Guard rename-window with '--', apply the name before persisting it, and roll back the in-memory table on failure. - Start the Hub tmux server with -f /dev/null by default. Reading the user's ~/.tmux.conf let theme status-formats replace the Sessions bar and root-table bindings leak into Herdr; HERDR_HUB_TMUX_CONFIG still opts into a custom file. - An unreadable aliases.json blocked start/refresh/open/choose. Degrade to default names with a warning (stderr, or the popup message) and leave the file untouched for `doctor` to report. - `doctor` crashed with a traceback when a *_BIN override pointed at a missing executable; report it as a failure instead. - Refuse to run inside another tmux server before discovery and server creation, so no unattached Hub is left behind. Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
279 lines
10 KiB
Python
279 lines
10 KiB
Python
from __future__ import annotations
|
|
|
|
import argparse
|
|
import contextlib
|
|
import io
|
|
import json
|
|
import os
|
|
from pathlib import Path
|
|
import subprocess
|
|
import tempfile
|
|
import unittest
|
|
from unittest.mock import patch
|
|
|
|
from herdr_hub.cli import (
|
|
DEFAULT_SESSION_NAME,
|
|
DEFAULT_TMUX_CONFIG,
|
|
Endpoint,
|
|
HubError,
|
|
command_doctor,
|
|
command_start,
|
|
decode_cmdline,
|
|
dedupe_endpoints,
|
|
deserialize_endpoint,
|
|
discover_local_sessions,
|
|
discover_remote_processes,
|
|
make_hub,
|
|
parse_remote_argv,
|
|
serialize_endpoint,
|
|
)
|
|
|
|
|
|
class EndpointTests(unittest.TestCase):
|
|
def test_local_and_remote_commands(self) -> None:
|
|
self.assertEqual(Endpoint.local().command("herdr"), ["herdr"])
|
|
self.assertEqual(
|
|
Endpoint.local("work").command("/bin/herdr"),
|
|
["/bin/herdr", "--session", "work"],
|
|
)
|
|
self.assertEqual(
|
|
Endpoint.remote("orangepi", "work", "server").command("herdr"),
|
|
[
|
|
"herdr",
|
|
"--remote",
|
|
"orangepi",
|
|
"--remote-keybindings",
|
|
"server",
|
|
"--session",
|
|
"work",
|
|
],
|
|
)
|
|
self.assertEqual(
|
|
Endpoint.remote(
|
|
"ssh://me@windows:2222",
|
|
"work",
|
|
remote_transport="ssh",
|
|
).command("/bin/herdr", "/bin/ssh"),
|
|
[
|
|
"/bin/ssh",
|
|
"-tt",
|
|
"-o",
|
|
"ServerAliveInterval=15",
|
|
"-o",
|
|
"ServerAliveCountMax=3",
|
|
"-p",
|
|
"2222",
|
|
"-l",
|
|
"me",
|
|
"windows",
|
|
"herdr --session work",
|
|
],
|
|
)
|
|
|
|
def test_identity_ignores_remote_keybinding_mode(self) -> None:
|
|
local = Endpoint.remote("orangepi", remote_keybindings="local")
|
|
server = Endpoint.remote("orangepi", remote_keybindings="server")
|
|
self.assertEqual(local.key, server.key)
|
|
self.assertEqual(dedupe_endpoints([local, server]), [local])
|
|
self.assertEqual(
|
|
Endpoint.remote("orangepi").key,
|
|
Endpoint.remote("orangepi", remote_transport="ssh").key,
|
|
)
|
|
|
|
def test_short_labels_put_the_host_first_and_omit_default(self) -> None:
|
|
self.assertEqual(Endpoint.local().short_label, "local")
|
|
self.assertEqual(Endpoint.local("work").short_label, "local/work")
|
|
self.assertEqual(Endpoint.remote("orangepi").short_label, "orangepi")
|
|
self.assertEqual(Endpoint.remote("orangepi", "work").short_label, "orangepi/work")
|
|
|
|
def test_default_local_sorts_first(self) -> None:
|
|
endpoints = dedupe_endpoints(
|
|
[
|
|
Endpoint.remote("z"),
|
|
Endpoint.local("work"),
|
|
Endpoint.local(),
|
|
Endpoint.remote("a"),
|
|
]
|
|
)
|
|
self.assertEqual(
|
|
[endpoint.key for endpoint in endpoints],
|
|
["local:default", "local:work", "remote:a:default", "remote:z:default"],
|
|
)
|
|
|
|
def test_endpoint_metadata_and_legacy_command_round_trip(self) -> None:
|
|
endpoint = Endpoint.remote(
|
|
"ssh://me@host:2222",
|
|
"work",
|
|
"server",
|
|
"ssh",
|
|
)
|
|
self.assertEqual(deserialize_endpoint(serialize_endpoint(endpoint)), endpoint)
|
|
self.assertEqual(
|
|
deserialize_endpoint(
|
|
"",
|
|
"exec /usr/local/bin/herdr --remote orangepi --session work",
|
|
),
|
|
Endpoint.remote("orangepi", "work"),
|
|
)
|
|
|
|
|
|
class RemoteArgvTests(unittest.TestCase):
|
|
def test_separate_arguments(self) -> None:
|
|
endpoint = parse_remote_argv(
|
|
[
|
|
"/home/me/.local/bin/herdr",
|
|
"--remote",
|
|
"orangepi",
|
|
"--session",
|
|
"work",
|
|
"--remote-keybindings",
|
|
"server",
|
|
]
|
|
)
|
|
self.assertEqual(endpoint, Endpoint.remote("orangepi", "work", "server"))
|
|
|
|
def test_equal_arguments_and_arbitrary_order(self) -> None:
|
|
endpoint = parse_remote_argv(
|
|
["herdr", "--session=work", "--remote=ssh://me@host:2222"]
|
|
)
|
|
self.assertEqual(endpoint, Endpoint.remote("ssh://me@host:2222", "work"))
|
|
|
|
def test_default_session(self) -> None:
|
|
endpoint = parse_remote_argv(["herdr", "--remote", "orangepi"])
|
|
self.assertIsNotNone(endpoint)
|
|
assert endpoint is not None
|
|
self.assertEqual(endpoint.session, DEFAULT_SESSION_NAME)
|
|
|
|
def test_non_herdr_and_malformed_commands_are_ignored(self) -> None:
|
|
self.assertIsNone(parse_remote_argv(["bash", "-c", "herdr --remote host"]))
|
|
self.assertIsNone(parse_remote_argv(["herdr", "--remote"]))
|
|
self.assertIsNone(parse_remote_argv(["herdr", "--remote", "-bad"]))
|
|
|
|
def test_cmdline_is_nul_delimited(self) -> None:
|
|
self.assertEqual(
|
|
decode_cmdline(b"/bin/herdr\0--remote\0host name\0"),
|
|
["/bin/herdr", "--remote", "host name"],
|
|
)
|
|
|
|
|
|
class DiscoveryTests(unittest.TestCase):
|
|
def test_local_discovery_returns_only_running_sessions(self) -> None:
|
|
payload = {
|
|
"sessions": [
|
|
{"name": "default", "running": True},
|
|
{"name": "work", "running": True},
|
|
{"name": "old", "running": False},
|
|
]
|
|
}
|
|
|
|
def fake_run(*args, **kwargs): # type: ignore[no-untyped-def]
|
|
return subprocess.CompletedProcess(args[0], 0, json.dumps(payload), "")
|
|
|
|
report = discover_local_sessions("herdr", run=fake_run)
|
|
self.assertEqual(
|
|
[endpoint.key for endpoint in report.endpoints],
|
|
["local:default", "local:work"],
|
|
)
|
|
self.assertEqual(report.warnings, [])
|
|
|
|
def test_invalid_local_json_is_a_warning(self) -> None:
|
|
def fake_run(*args, **kwargs): # type: ignore[no-untyped-def]
|
|
return subprocess.CompletedProcess(args[0], 0, "not json", "")
|
|
|
|
report = discover_local_sessions("herdr", run=fake_run)
|
|
self.assertEqual(report.endpoints, [])
|
|
self.assertEqual(len(report.warnings), 1)
|
|
|
|
def test_proc_discovery_reads_exact_argv_and_deduplicates(self) -> None:
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
root = Path(directory)
|
|
commands = {
|
|
"101": ["/usr/bin/herdr", "--remote", "orangepi"],
|
|
"102": ["herdr", "--remote=orangepi"],
|
|
"103": ["herdr", "--remote", "prod", "--session", "work"],
|
|
"104": ["bash", "-c", "herdr --remote ignored"],
|
|
}
|
|
for process_id, argv in commands.items():
|
|
process = root / process_id
|
|
process.mkdir()
|
|
(process / "cmdline").write_bytes(
|
|
b"\0".join(os.fsencode(argument) for argument in argv) + b"\0"
|
|
)
|
|
|
|
report = discover_remote_processes(root, user_id=os.getuid())
|
|
|
|
self.assertEqual(
|
|
[endpoint.key for endpoint in report.endpoints],
|
|
["remote:orangepi:default", "remote:prod:work"],
|
|
)
|
|
|
|
|
|
class CliEnvironmentTests(unittest.TestCase):
|
|
def test_hub_ignores_user_tmux_conf_unless_overridden(self) -> None:
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
environment = {
|
|
"HERDR_HUB_TMUX_BIN": "/bin/true",
|
|
"HERDR_HUB_ALIASES_FILE": str(Path(directory) / "aliases.json"),
|
|
}
|
|
with patch.dict(os.environ, environment):
|
|
os.environ.pop("HERDR_HUB_TMUX_CONFIG", None)
|
|
hub = make_hub(argparse.Namespace())
|
|
self.assertEqual(hub.config_file, DEFAULT_TMUX_CONFIG)
|
|
self.assertEqual(hub.base_command()[-2:], ["-f", "/dev/null"])
|
|
with patch.dict(
|
|
os.environ, {**environment, "HERDR_HUB_TMUX_CONFIG": "~/hub.conf"}
|
|
):
|
|
hub = make_hub(argparse.Namespace())
|
|
self.assertEqual(hub.config_file, os.path.expanduser("~/hub.conf"))
|
|
|
|
def test_corrupt_alias_file_degrades_to_defaults_with_warning(self) -> None:
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
aliases = Path(directory) / "aliases.json"
|
|
aliases.write_text("not json", encoding="utf-8")
|
|
environment = {
|
|
"HERDR_HUB_TMUX_BIN": "/bin/true",
|
|
"HERDR_HUB_ALIASES_FILE": str(aliases),
|
|
}
|
|
warnings: list[str] = []
|
|
with patch.dict(os.environ, environment):
|
|
hub = make_hub(argparse.Namespace(), warnings=warnings)
|
|
self.assertEqual(hub.aliases, {})
|
|
self.assertEqual(len(warnings), 1)
|
|
self.assertIn("aliases.json", warnings[0])
|
|
|
|
stderr = io.StringIO()
|
|
with patch.dict(os.environ, environment), contextlib.redirect_stderr(stderr):
|
|
make_hub(argparse.Namespace())
|
|
self.assertIn("display names unavailable", stderr.getvalue())
|
|
# The unreadable file is left untouched for the user to inspect.
|
|
self.assertEqual(aliases.read_text(encoding="utf-8"), "not json")
|
|
|
|
def test_start_refuses_nested_tmux_before_discovery(self) -> None:
|
|
environment = {"TMUX": "/tmp/tmux-1/other,1,0", "HERDR_HUB_SOCKET": "herdr-hub"}
|
|
with patch.dict(os.environ, environment), patch(
|
|
"herdr_hub.cli.discover_endpoints",
|
|
side_effect=AssertionError("discovery must not run inside another tmux"),
|
|
):
|
|
with self.assertRaisesRegex(HubError, "another tmux server"):
|
|
command_start(argparse.Namespace())
|
|
|
|
def test_doctor_reports_unusable_binary_override_instead_of_crashing(self) -> None:
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
root = Path(directory)
|
|
environment = {
|
|
"HERDR_HUB_HERDR_BIN": str(root / "missing-herdr"),
|
|
"HERDR_HUB_REMOTES_FILE": str(root / "remotes.json"),
|
|
"HERDR_HUB_ALIASES_FILE": str(root / "aliases.json"),
|
|
"HERDR_HUB_PROC_ROOT": directory,
|
|
}
|
|
stdout = io.StringIO()
|
|
with patch.dict(os.environ, environment), contextlib.redirect_stdout(stdout):
|
|
status = command_doctor(argparse.Namespace())
|
|
self.assertGreaterEqual(status, 1)
|
|
self.assertIn("herdr: cannot run", stdout.getvalue())
|
|
self.assertIn("discovery failed", stdout.getvalue())
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|