herdr-session-switch/tests/test_cli_integration.py

from __future__ import annotations

import os
from pathlib import Path
import shutil
import subprocess
import tempfile
import unittest
import uuid

from herdr_hub.aliases import AliasRegistry
from herdr_hub.cli import TmuxHub


@unittest.skipUnless(shutil.which("tmux"), "tmux is required")
class CliIntegrationTests(unittest.TestCase):
    def test_remote_registry_cli_round_trip(self) -> None:
        repository_root = Path(__file__).resolve().parents[1]
        launcher = repository_root / "bin" / "herdr-hub"
        with tempfile.TemporaryDirectory() as directory:
            environment = {
                **os.environ,
                "HERDR_HUB_REMOTES_FILE": str(Path(directory) / "remotes.json"),
            }
            add = subprocess.run(
                [
                    str(launcher),
                    "remote",
                    "add",
                    "orangepi",
                    "--remote-keybindings",
                    "server",
                    "--transport",
                    "ssh",
                ],
                cwd=repository_root,
                env=environment,
                capture_output=True,
                text=True,
                check=False,
            )
            self.assertEqual(add.returncode, 0, add.stderr)
            listing = subprocess.run(
                [str(launcher), "remote", "list"],
                cwd=repository_root,
                env=environment,
                capture_output=True,
                text=True,
                check=False,
            )
            self.assertEqual(listing.stdout.strip(), "orangepi\tserver\tssh")
            remove = subprocess.run(
                [str(launcher), "remote", "remove", "orangepi"],
                cwd=repository_root,
                env=environment,
                capture_output=True,
                text=True,
                check=False,
            )
            self.assertEqual(remove.returncode, 0, remove.stderr)
            self.assertIn("removed orangepi", remove.stdout)

    def test_repository_launcher_refreshes_and_stops_isolated_hub(self) -> None:
        repository_root = Path(__file__).resolve().parents[1]
        launcher = repository_root / "bin" / "herdr-hub"
        socket_name = f"herdr-hub-cli-test-{os.getpid()}-{uuid.uuid4().hex[:8]}"
        socket_path = Path(
            os.environ.get("TMUX_TMPDIR", f"/tmp/tmux-{os.getuid()}")
        ) / socket_name
        with tempfile.TemporaryDirectory() as directory:
            root = Path(directory)
            proc_root = root / "proc"
            proc_root.mkdir()
            fake_herdr = root / "herdr"
            fake_herdr.write_text(
                "#!/bin/sh\n"
                "if [ \"${1:-}\" = \"session\" ]; then\n"
                "  printf '%s\\n' '{\"sessions\":[{\"name\":\"default\",\"running\":true}]}'\n"
                "  exit 0\n"
                "fi\n"
                "exec sleep 30\n",
                encoding="utf-8",
            )
            fake_herdr.chmod(0o755)
            environment = {
                **os.environ,
                "HERDR_HUB_HERDR_BIN": str(fake_herdr),
                "HERDR_HUB_TMUX_BIN": shutil.which("tmux") or "tmux",
                "HERDR_HUB_SOCKET": socket_name,
                "HERDR_HUB_SESSION": "hub",
                "HERDR_HUB_TMUX_CONFIG": "/dev/null",
                "HERDR_HUB_PROC_ROOT": str(proc_root),
                "HERDR_HUB_ALIASES_FILE": str(root / "aliases.json"),
            }
            AliasRegistry(root / "aliases.json").set("local:default", "主工作")

            refresh = subprocess.run(
                [str(launcher), "refresh"],
                cwd=repository_root,
                env=environment,
                capture_output=True,
                text=True,
                check=False,
            )
            self.assertEqual(refresh.returncode, 0, refresh.stderr)
            self.assertIn("added 1 session", refresh.stdout)

            hub = TmuxHub(
                tmux_bin=shutil.which("tmux") or "tmux",
                socket_name=socket_name,
                session_name="hub",
                config_file="/dev/null",
                script_path=str(launcher),
            )

            def cleanup_hub() -> None:
                if hub.exists():
                    hub.stop()
                socket_path.unlink(missing_ok=True)

            self.addCleanup(cleanup_hub)
            self.assertEqual(
                [window.key for window in hub.managed_windows()],
                ["local:default"],
            )
            self.assertEqual(hub.managed_windows()[0].name, "主工作")

            # Operational recovery must remain possible even if optional
            # display-name configuration is later corrupted.
            (root / "aliases.json").write_text("not json", encoding="utf-8")
            degraded = subprocess.run(
                [str(launcher), "refresh"],
                cwd=repository_root,
                env=environment,
                capture_output=True,
                text=True,
                check=False,
            )
            self.assertEqual(degraded.returncode, 0, degraded.stderr)
            self.assertIn("display names unavailable", degraded.stderr)
            self.assertEqual(hub.managed_windows()[0].name, "L default")
            stop = subprocess.run(
                [str(launcher), "stop"],
                cwd=repository_root,
                env=environment,
                capture_output=True,
                text=True,
                check=False,
            )
            self.assertEqual(stop.returncode, 0, stop.stderr)
            self.assertIn("Herdr servers were left running", stop.stdout)
            self.assertFalse(hub.exists())

    def test_empty_refresh_does_not_leave_a_bootstrap_hub(self) -> None:
        repository_root = Path(__file__).resolve().parents[1]
        launcher = repository_root / "bin" / "herdr-hub"
        socket_name = f"herdr-hub-empty-test-{os.getpid()}-{uuid.uuid4().hex[:8]}"
        with tempfile.TemporaryDirectory() as directory:
            root = Path(directory)
            proc_root = root / "proc"
            proc_root.mkdir()
            environment = {
                **os.environ,
                "HERDR_HUB_HERDR_BIN": "/bin/false",
                "HERDR_HUB_TMUX_BIN": shutil.which("tmux") or "tmux",
                "HERDR_HUB_SOCKET": socket_name,
                "HERDR_HUB_SESSION": "hub",
                "HERDR_HUB_TMUX_CONFIG": "/dev/null",
                "HERDR_HUB_PROC_ROOT": str(proc_root),
                "HERDR_HUB_ALIASES_FILE": str(root / "aliases.json"),
                "HERDR_HUB_REMOTES_FILE": str(root / "remotes.json"),
            }
            refresh = subprocess.run(
                [str(launcher), "refresh"],
                cwd=repository_root,
                env=environment,
                capture_output=True,
                text=True,
                check=False,
            )
            self.assertEqual(refresh.returncode, 0, refresh.stderr)
            hub = TmuxHub(
                tmux_bin=shutil.which("tmux") or "tmux",
                socket_name=socket_name,
                session_name="hub",
                config_file="/dev/null",
            )
            self.addCleanup(hub.stop)
            self.assertFalse(hub.exists())


if __name__ == "__main__":
    unittest.main()