path: root/tests/tmux-util/test_tmux_util.py

"""Tests for dotfiles/common/.local/bin/tmux-util.

The script is a bash wrapper around `tmux` with subcommand dispatch. Tests
invoke the real script with a faked `tmux`, `kill`, and `sleep` on PATH. The
fakes read canned state from a temp dir and append a one-line record to a
call log on each invocation. Assertions compare the call log to the expected
sequence for the scenario — we test behavior (what the script causes tmux /
kill to do), not implementation.

Run from repo root:
    python3 -m unittest tests.tmux-util.test_tmux_util
"""

import os
import shutil
import stat
import subprocess
import tempfile
import unittest


REPO_ROOT = os.path.abspath(os.path.join(os.path.dirname(__file__), "..", ".."))
SCRIPT = os.path.join(REPO_ROOT, "dotfiles/common/.local/bin/tmux-util")
FAKES_DIR = os.path.dirname(__file__)


class TmuxUtilHarness(unittest.TestCase):
    """Shared harness: run tmux-util with faked tmux/kill/sleep on PATH."""

    def setUp(self):
        self.tmp = tempfile.mkdtemp(prefix="tmux-util-test-")

        # bin dir with the fakes symlinked under their canonical names
        self.bin_dir = os.path.join(self.tmp, "bin")
        os.makedirs(self.bin_dir)
        for name in ("tmux", "kill", "sleep"):
            fake = os.path.join(FAKES_DIR, f"fake-{name}")
            os.chmod(fake, os.stat(fake).st_mode | stat.S_IEXEC)
            os.symlink(fake, os.path.join(self.bin_dir, name))

        # Pre-create empty state + log files so the fakes never have to
        self.state_file = os.path.join(self.tmp, "sessions.txt")
        self.calls_log = os.path.join(self.tmp, "calls.log")
        self.kill_log = os.path.join(self.tmp, "kill.log")
        open(self.state_file, "w").close()
        open(self.calls_log, "w").close()
        open(self.kill_log, "w").close()

    def tearDown(self):
        shutil.rmtree(self.tmp, ignore_errors=True)

    def set_sessions(self, sessions):
        """sessions: list of (name, attached, [pane_pids])."""
        with open(self.state_file, "w") as f:
            for name, attached, pids in sessions:
                pids_csv = ",".join(str(p) for p in pids) if pids else "-"
                f.write(f"{name} {attached} {pids_csv}\n")

    def run_script(self, *args, env_extra=None):
        env = os.environ.copy()
        # Prepend the bin dir so the fakes win
        env["PATH"] = self.bin_dir + os.pathsep + env.get("PATH", "")
        env["FAKE_TMUX_DIR"] = self.tmp
        if env_extra:
            env.update(env_extra)
        return subprocess.run(
            [SCRIPT] + list(args),
            env=env,
            capture_output=True,
            text=True,
            timeout=10,
        )

    def tmux_calls(self):
        with open(self.calls_log) as f:
            return [line.rstrip("\n") for line in f if line.strip()]

    def kill_calls(self):
        with open(self.kill_log) as f:
            return [line.rstrip("\n") for line in f if line.strip()]

    def remaining_sessions(self):
        with open(self.state_file) as f:
            return [line.split()[0] for line in f if line.strip()]


# -----------------------------------------------------------------------------
# Dispatch + usage
# -----------------------------------------------------------------------------

class TestDispatch(TmuxUtilHarness):

    def test_no_args_prints_usage_to_stdout_and_exits_zero(self):
        result = self.run_script()
        self.assertEqual(result.returncode, 0)
        self.assertIn("Usage: tmux-util", result.stdout)
        self.assertIn("reap", result.stdout)

    def test_dash_h_prints_usage(self):
        result = self.run_script("-h")
        self.assertEqual(result.returncode, 0)
        self.assertIn("Usage: tmux-util", result.stdout)

    def test_double_dash_help_prints_usage(self):
        result = self.run_script("--help")
        self.assertEqual(result.returncode, 0)
        self.assertIn("Usage: tmux-util", result.stdout)

    def test_help_subcommand_prints_usage(self):
        result = self.run_script("help")
        self.assertEqual(result.returncode, 0)
        self.assertIn("Usage: tmux-util", result.stdout)

    def test_unknown_subcommand_exits_nonzero_and_prints_usage_to_stderr(self):
        result = self.run_script("frobnicate")
        self.assertNotEqual(result.returncode, 0)
        self.assertIn("unknown subcommand", result.stderr)
        self.assertIn("Usage: tmux-util", result.stderr)

    def test_unimplemented_subcommand_exits_nonzero(self):
        # ls/go/pick/find/rename will land later; for now they stub out.
        result = self.run_script("ls")
        self.assertNotEqual(result.returncode, 0)
        self.assertIn("not implemented yet", result.stderr)


# -----------------------------------------------------------------------------
# Reap — Normal cases
# -----------------------------------------------------------------------------

class TestReapNormal(TmuxUtilHarness):

    def test_reap_unattached_sends_sighup_then_kill_session(self):
        # bar is unattached, foo is attached. Only bar should be reaped.
        self.set_sessions([
            ("foo", 1, [101, 102]),
            ("bar", 0, [201]),
        ])
        result = self.run_script("reap")
        self.assertEqual(result.returncode, 0, msg=result.stderr)
        self.assertIn("Reaping: bar", result.stdout)
        self.assertNotIn("Reaping: foo", result.stdout)
        # SIGHUP went to bar's pid only, never to foo's
        self.assertIn("kill -HUP 201", self.kill_calls())
        self.assertNotIn("kill -HUP 101", self.kill_calls())
        self.assertNotIn("kill -HUP 102", self.kill_calls())
        # Fallback kill-session ran (our fake-kill is a no-op, so the session
        # stays "alive" and the force-kill branch fires)
        tmux = self.tmux_calls()
        self.assertTrue(
            any("kill-session -t bar" in c for c in tmux),
            f"expected kill-session -t bar in {tmux!r}",
        )

    def test_reap_skips_attached_sessions(self):
        self.set_sessions([
            ("foo", 1, [101]),
            ("bar", 1, [201]),
        ])
        result = self.run_script("reap")
        self.assertEqual(result.returncode, 0)
        self.assertIn("No unattached sessions", result.stdout)
        self.assertEqual(self.kill_calls(), [])

    def test_reap_skips_aiv_prefix_by_default(self):
        self.set_sessions([
            ("aiv-claude", 0, [301]),
            ("aiv-gemini", 0, [302]),
            ("worker", 0, [401]),
        ])
        result = self.run_script("reap")
        self.assertEqual(result.returncode, 0)
        self.assertIn("Reaping: worker", result.stdout)
        self.assertNotIn("Reaping: aiv-claude", result.stdout)
        self.assertNotIn("Reaping: aiv-gemini", result.stdout)
        self.assertIn("kill -HUP 401", self.kill_calls())
        self.assertNotIn("kill -HUP 301", self.kill_calls())
        self.assertNotIn("kill -HUP 302", self.kill_calls())

    def test_reap_skip_pattern_overridable_via_env(self):
        # Override to match nothing → aiv-claude should now be reaped.
        self.set_sessions([
            ("aiv-claude", 0, [301]),
        ])
        result = self.run_script("reap", env_extra={"TMUX_UTIL_REAP_SKIP": "^never-matches-anything$"})
        self.assertEqual(result.returncode, 0)
        self.assertIn("Reaping: aiv-claude", result.stdout)
        self.assertIn("kill -HUP 301", self.kill_calls())

    def test_reap_session_with_multiple_panes_sighups_each(self):
        self.set_sessions([
            ("multi", 0, [501, 502, 503]),
        ])
        result = self.run_script("reap")
        self.assertEqual(result.returncode, 0)
        # xargs may collapse into one call; verify each PID is mentioned
        kc = "\n".join(self.kill_calls())
        for pid in ("501", "502", "503"):
            self.assertIn(pid, kc, f"expected pid {pid} in kill calls: {kc!r}")


# -----------------------------------------------------------------------------
# Reap — Boundary cases
# -----------------------------------------------------------------------------

class TestReapBoundary(TmuxUtilHarness):

    def test_reap_no_sessions_at_all(self):
        self.set_sessions([])
        result = self.run_script("reap")
        self.assertEqual(result.returncode, 0)
        self.assertIn("No unattached sessions", result.stdout)
        self.assertEqual(self.kill_calls(), [])

    def test_reap_session_with_no_panes_does_not_invoke_kill(self):
        # A session listed but with no PIDs — should still get kill-session.
        self.set_sessions([
            ("ghost", 0, []),
        ])
        result = self.run_script("reap")
        self.assertEqual(result.returncode, 0)
        self.assertIn("Reaping: ghost", result.stdout)
        # No kill -HUP because no PIDs
        self.assertEqual(self.kill_calls(), [])
        # But kill-session still ran (force kill, since fake-tmux didn't
        # auto-remove the session on its own)
        tmux = self.tmux_calls()
        self.assertTrue(
            any("kill-session -t ghost" in c for c in tmux),
            f"expected kill-session -t ghost in {tmux!r}",
        )

    def test_reap_only_attached_and_skipped_sessions(self):
        # Mix that leaves no candidates after filtering.
        self.set_sessions([
            ("foo", 1, [101]),         # attached
            ("aiv-bar", 0, [201]),     # skipped by pattern
        ])
        result = self.run_script("reap")
        self.assertEqual(result.returncode, 0)
        self.assertIn("No unattached sessions", result.stdout)
        self.assertEqual(self.kill_calls(), [])


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