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 herdr_hub.cli import ( BOOTSTRAP_WINDOW_NAME, Endpoint, HubError, TmuxHub, WINDOW_SHORT_LABEL_OPTION, ) @unittest.skipUnless(shutil.which("tmux"), "tmux is required") class TmuxIntegrationTests(unittest.TestCase): 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" " exit 0\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_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("连接已断开", status_right) self.assertTrue(expanded_status_left.startswith("#[bold,fg=colour81]")) self.assertIn("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_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, "临时名称") 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_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()