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()