diff --git a/AGENTS.md b/AGENTS.md index a11b5825c8..5065b6c1c0 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -201,6 +201,7 @@ Pass structured data on every log call where useful for filtering, searching, or | `tmux_pane` | `str` | pane identifier | | `tmux_config_path` | `str` | workspace config file path | | `tmux_layout` | `str` | window layout string | +| `tmux_hook_cmd` | `str` | lifecycle hook shell command line | **Heavy/optional keys** (DEBUG only, potentially large): diff --git a/conftest.py b/conftest.py index 1f0583439e..587ee0cf9c 100644 --- a/conftest.py +++ b/conftest.py @@ -114,6 +114,8 @@ def socket_name(request: pytest.FixtureRequest) -> str: # Modules that actually need tmux fixtures in their doctests DOCTEST_NEEDS_TMUX = { + "tmuxp.cli.stop", + "tmuxp.util", "tmuxp.workspace.builder", } diff --git a/docs/cli/index.md b/docs/cli/index.md index fd38b681ea..47d1e6e633 100644 --- a/docs/cli/index.md +++ b/docs/cli/index.md @@ -25,6 +25,12 @@ Interactive Python shell with tmux context. Export running sessions to config files. ::: +:::{grid-item-card} tmuxp stop +:link: stop +:link-type: doc +Stop running sessions with cleanup hooks. +::: + :::{grid-item-card} tmuxp convert :link: convert :link-type: doc @@ -53,6 +59,7 @@ load shell ls search +stop ``` ```{toctree} diff --git a/docs/cli/stop.md b/docs/cli/stop.md new file mode 100644 index 0000000000..cbd6bae747 --- /dev/null +++ b/docs/cli/stop.md @@ -0,0 +1,38 @@ +(cli-stop)= + +(cli-stop-reference)= + +# tmuxp stop + +Stop (kill) a running tmux session. If the session was created from a workspace +with `on_project_stop`, that hook runs before the session is killed. + +## Command + +```{eval-rst} +.. argparse:: + :module: tmuxp.cli + :func: create_parser + :prog: tmuxp + :path: stop +``` + +## Basic Usage + +Stop a session by name: + +```console +$ tmuxp stop mysession +``` + +Stop the currently attached session: + +```console +$ tmuxp stop +``` + +Use a custom socket: + +```console +$ tmuxp stop -L mysocket mysession +``` diff --git a/docs/configuration/examples.md b/docs/configuration/examples.md index 084cf2015c..5bb47645fd 100644 --- a/docs/configuration/examples.md +++ b/docs/configuration/examples.md @@ -797,6 +797,19 @@ The `synchronize` window key provides a shorthand for enabling ``` ```` +## Lifecycle Hooks + +Run shell commands at different stages of the session lifecycle: + +````{tab} YAML +```{literalinclude} ../../examples/lifecycle-hooks.yaml +:language: yaml + +``` +```` + +See {ref}`top-level` for full hook documentation. + ## Pane Titles Pane title keys turn on tmux pane border titles and label individual panes: diff --git a/docs/configuration/top-level.md b/docs/configuration/top-level.md index 0f4d914c85..59491da586 100644 --- a/docs/configuration/top-level.md +++ b/docs/configuration/top-level.md @@ -41,6 +41,42 @@ Notes: Above: Use `tmux` directly to attach _banana_. +## Lifecycle Hooks + +Workspace configs support four lifecycle hooks: + +```yaml +session_name: myproject +on_project_start: notify-send "Starting myproject" +on_project_restart: notify-send "Reattaching to myproject" +on_project_exit: notify-send "Detached from myproject" +on_project_stop: notify-send "Stopping myproject" +windows: + - window_name: main + panes: + - +``` + +| Hook | When it runs | +|------|-------------| +| `on_project_start` | Before a new session is built. | +| `on_project_restart` | Before reattaching to an existing session. | +| `on_project_exit` | When the last client detaches. | +| `on_project_stop` | Before `tmuxp stop` kills the session. | + +Each hook accepts a string command or a list of command strings: + +```yaml +on_project_start: + - notify-send "Starting" + - ./setup.sh +``` + +`on_project_start`, `on_project_restart`, and `on_project_stop` run through the +shell and block tmuxp until they finish. `on_project_exit` is different: it runs +via tmux's `client-detached` hook after tmuxp has already returned, so it never +blocks the command. Hook failures are logged and do not stop the tmuxp command. + ## Pane Titles Enable pane border titles to display labels on each pane: diff --git a/docs/internals/api/cli/index.md b/docs/internals/api/cli/index.md index 1381fbc90f..7416cd7948 100644 --- a/docs/internals/api/cli/index.md +++ b/docs/internals/api/cli/index.md @@ -19,6 +19,7 @@ ls progress search shell +stop utils ``` diff --git a/docs/internals/api/cli/stop.md b/docs/internals/api/cli/stop.md new file mode 100644 index 0000000000..7f01b8a4d3 --- /dev/null +++ b/docs/internals/api/cli/stop.md @@ -0,0 +1,8 @@ +# tmuxp stop - `tmuxp.cli.stop` + +```{eval-rst} +.. automodule:: tmuxp.cli.stop + :members: + :show-inheritance: + :undoc-members: +``` diff --git a/examples/lifecycle-hooks.yaml b/examples/lifecycle-hooks.yaml new file mode 100644 index 0000000000..8b8a3668ec --- /dev/null +++ b/examples/lifecycle-hooks.yaml @@ -0,0 +1,9 @@ +session_name: lifecycle hooks +on_project_start: echo "project starting" +on_project_restart: echo "project restarting" +on_project_exit: echo "project exiting" +on_project_stop: echo "project stopping" +windows: + - window_name: main + panes: + - diff --git a/src/tmuxp/cli/__init__.py b/src/tmuxp/cli/__init__.py index 860a9200cb..999ce3c816 100644 --- a/src/tmuxp/cli/__init__.py +++ b/src/tmuxp/cli/__init__.py @@ -57,6 +57,12 @@ command_shell, create_shell_subparser, ) +from .stop import ( + STOP_DESCRIPTION, + CLIStopNamespace, + command_stop, + create_stop_subparser, +) from .utils import tmuxp_echo logger = logging.getLogger(__name__) @@ -130,6 +136,13 @@ "tmuxp edit myproject", ], ), + ( + "stop", + [ + "tmuxp stop mysession", + "tmuxp stop -L mysocket mysession", + ], + ), ( "debug-info", [ @@ -155,6 +168,7 @@ "import", "search", "shell", + "stop", "debug-info", ] CLIImportSubparserName: TypeAlias = t.Literal["teamocil", "tmuxinator"] @@ -262,6 +276,14 @@ def create_parser() -> argparse.ArgumentParser: ) create_freeze_subparser(freeze_parser) + stop_parser = subparsers.add_parser( + "stop", + help="stop (kill) a tmux session", + description=STOP_DESCRIPTION, + formatter_class=formatter_class, + ) + create_stop_subparser(stop_parser) + return parser @@ -353,6 +375,11 @@ def cli(_args: list[str] | None = None) -> None: args=CLIFreezeNamespace(**vars(args)), parser=parser, ) + elif args.subparser_name == "stop": + command_stop( + args=CLIStopNamespace(**vars(args)), + parser=parser, + ) elif args.subparser_name == "ls": command_ls( args=CLILsNamespace(**vars(args)), diff --git a/src/tmuxp/cli/load.py b/src/tmuxp/cli/load.py index 375cdb1b22..e7f478c179 100644 --- a/src/tmuxp/cli/load.py +++ b/src/tmuxp/cli/load.py @@ -234,6 +234,7 @@ def _reattach(builder: WorkspaceBuilder, colors: Colors | None = None) -> None: def _load_attached( builder: WorkspaceBuilder, detached: bool, + pre_build_hook: t.Callable[[], None] | None = None, pre_attach_hook: t.Callable[[], None] | None = None, ) -> None: """ @@ -243,10 +244,15 @@ def _load_attached( ---------- builder: :class:`workspace.builder.WorkspaceBuilder` detached : bool + pre_build_hook : callable, optional + called immediately before ``builder.build()`` for new-session load paths. pre_attach_hook : callable, optional called after build, before attach/switch_client; use to stop the spinner so its cleanup sequences don't appear inside the tmux pane. """ + if pre_build_hook is not None: + pre_build_hook() + builder.build() assert builder.session is not None @@ -267,6 +273,7 @@ def _load_attached( def _load_detached( builder: WorkspaceBuilder, colors: Colors | None = None, + pre_build_hook: t.Callable[[], None] | None = None, pre_output_hook: t.Callable[[], None] | None = None, ) -> None: """ @@ -277,9 +284,14 @@ def _load_detached( builder: :class:`workspace.builder.WorkspaceBuilder` colors : Colors | None Optional Colors instance for styled output. + pre_build_hook : Callable | None + Called immediately before ``builder.build()`` for new-session load paths. pre_output_hook : Callable | None Called after build but before printing, e.g. to stop a spinner. """ + if pre_build_hook is not None: + pre_build_hook() + builder.build() assert builder.session is not None @@ -325,6 +337,7 @@ def _dispatch_build( append: bool, answer_yes: bool, cli_colors: Colors, + pre_build_hook: t.Callable[[], None] | None = None, pre_attach_hook: t.Callable[[], None] | None = None, on_error_hook: t.Callable[[], None] | None = None, pre_prompt_hook: t.Callable[[], None] | None = None, @@ -347,6 +360,8 @@ def _dispatch_build( Skip interactive prompts. cli_colors : Colors Colors instance for styled output. + pre_build_hook : callable, optional + Called before the build only for code paths that create a new session. pre_attach_hook : callable, optional Called before attach/switch_client (e.g. stop spinner). on_error_hook : callable, optional @@ -368,20 +383,35 @@ def _dispatch_build( """ try: if detached: - _load_detached(builder, cli_colors, pre_output_hook=pre_attach_hook) + _load_detached( + builder, + cli_colors, + pre_build_hook=pre_build_hook, + pre_output_hook=pre_attach_hook, + ) return _setup_plugins(builder) if append: if "TMUX" in os.environ: # tmuxp ran from inside tmux _load_append_windows_to_current_session(builder) else: - _load_attached(builder, detached, pre_attach_hook=pre_attach_hook) + _load_attached( + builder, + detached, + pre_build_hook=pre_build_hook, + pre_attach_hook=pre_attach_hook, + ) return _setup_plugins(builder) # append and answer_yes have no meaning if specified together if answer_yes: - _load_attached(builder, detached, pre_attach_hook=pre_attach_hook) + _load_attached( + builder, + detached, + pre_build_hook=pre_build_hook, + pre_attach_hook=pre_attach_hook, + ) return _setup_plugins(builder) if "TMUX" in os.environ: # tmuxp ran from inside tmux @@ -395,13 +425,27 @@ def _dispatch_build( choice = prompt_choices(msg, choices=options, color_mode=cli_colors.mode) if choice == "y": - _load_attached(builder, detached, pre_attach_hook=pre_attach_hook) + _load_attached( + builder, + detached, + pre_build_hook=pre_build_hook, + pre_attach_hook=pre_attach_hook, + ) elif choice == "a": _load_append_windows_to_current_session(builder) else: - _load_detached(builder, cli_colors) + _load_detached( + builder, + cli_colors, + pre_build_hook=pre_build_hook, + ) else: - _load_attached(builder, detached, pre_attach_hook=pre_attach_hook) + _load_attached( + builder, + detached, + pre_build_hook=pre_build_hook, + pre_attach_hook=pre_attach_hook, + ) except exc.TmuxpException as e: if on_error_hook is not None: @@ -599,17 +643,33 @@ def load_workspace( # Session-exists check — outside spinner so prompt_yes_no is safe if builder.session_exists(session_name) and not append: - if not detached and ( + should_attach = not detached and ( answer_yes or prompt_yes_no( f"{cli_colors.highlight(session_name)} is already running. Attach?", default=True, color_mode=cli_colors.mode, ) - ): + ) + if should_attach: + if "on_project_restart" in expanded_workspace: + hook_cwd = expanded_workspace.get("start_directory") + util.run_hook_commands( + expanded_workspace["on_project_restart"], + cwd=hook_cwd, + ) _reattach(builder, cli_colors) return None + def _run_on_project_start() -> None: + if "on_project_start" not in expanded_workspace: + return + hook_cwd = expanded_workspace.get("start_directory") + util.run_hook_commands( + expanded_workspace["on_project_start"], + cwd=hook_cwd, + ) + if _progress_disabled: _private_path = str(PrivatePath(workspace_file)) result = _dispatch_build( @@ -618,6 +678,7 @@ def load_workspace( append, answer_yes, cli_colors, + pre_build_hook=_run_on_project_start, ) if result is not None: summary = "" @@ -693,6 +754,7 @@ def _emit_success() -> None: append, answer_yes, cli_colors, + pre_build_hook=_run_on_project_start, pre_attach_hook=_emit_success, on_error_hook=spinner.stop, pre_prompt_hook=spinner.stop, diff --git a/src/tmuxp/cli/stop.py b/src/tmuxp/cli/stop.py new file mode 100644 index 0000000000..130ba34b37 --- /dev/null +++ b/src/tmuxp/cli/stop.py @@ -0,0 +1,140 @@ +"""CLI for ``tmuxp stop`` subcommand.""" + +from __future__ import annotations + +import argparse +import logging +import os +import sys +import typing as t + +from libtmux.server import Server + +from tmuxp import exc, util +from tmuxp.exc import TmuxpException + +from ._colors import Colors, build_description, get_color_mode +from .utils import tmuxp_echo + +logger = logging.getLogger(__name__) + +STOP_DESCRIPTION = build_description( + """ + Stop (kill) a tmux session. + """, + ( + ( + None, + [ + "tmuxp stop mysession", + "tmuxp stop -L mysocket mysession", + ], + ), + ), +) + +if t.TYPE_CHECKING: + CLIColorModeLiteral: t.TypeAlias = t.Literal["auto", "always", "never"] + + +class CLIStopNamespace(argparse.Namespace): + """Typed :class:`argparse.Namespace` for tmuxp stop command.""" + + color: CLIColorModeLiteral + session_name: str | None + socket_name: str | None + socket_path: str | None + + +def create_stop_subparser( + parser: argparse.ArgumentParser, +) -> argparse.ArgumentParser: + """Augment :class:`argparse.ArgumentParser` with ``stop`` subcommand. + + Examples + -------- + >>> import argparse + >>> parser = create_stop_subparser(argparse.ArgumentParser()) + >>> args = parser.parse_args(["mysession"]) + >>> args.session_name + 'mysession' + """ + parser.add_argument( + dest="session_name", + metavar="session-name", + nargs="?", + action="store", + ) + parser.add_argument( + "-S", + dest="socket_path", + metavar="socket-path", + help="pass-through for tmux -S", + ) + parser.add_argument( + "-L", + dest="socket_name", + metavar="socket-name", + help="pass-through for tmux -L", + ) + parser.set_defaults(print_help=parser.print_help) + return parser + + +def command_stop( + args: CLIStopNamespace, + parser: argparse.ArgumentParser | None = None, +) -> None: + """Entrypoint for ``tmuxp stop``, kill a tmux session. + + Examples + -------- + >>> test_session = server.new_session(session_name="doctest_stop") + >>> args = CLIStopNamespace() + >>> args.session_name = "doctest_stop" + >>> args.color = "never" + >>> args.socket_name = server.socket_name + >>> args.socket_path = None + >>> command_stop(args) # doctest: +ELLIPSIS + Stopped doctest_stop + >>> server.sessions.get(session_name="doctest_stop", default=None) is None + True + """ + color_mode = get_color_mode(args.color) + colors = Colors(color_mode) + + server = Server(socket_name=args.socket_name, socket_path=args.socket_path) + + try: + if args.session_name: + session = server.sessions.get( + session_name=args.session_name, + default=None, + ) + elif os.environ.get("TMUX"): + session = util.get_session(server, require_pane_resolution=True) + else: + tmuxp_echo( + colors.error("No session name given and not inside tmux."), + ) + sys.exit(1) + + if not session: + raise exc.SessionNotFound(args.session_name) + except TmuxpException as e: + tmuxp_echo(colors.error(str(e))) + sys.exit(1) + + session_name = session.name + + on_stop_cmd = session.getenv("TMUXP_ON_PROJECT_STOP") + if on_stop_cmd and isinstance(on_stop_cmd, str): + start_dir = session.getenv("TMUXP_START_DIRECTORY") + stop_cwd = str(start_dir) if isinstance(start_dir, str) else None + util.run_hook_commands(on_stop_cmd, cwd=stop_cwd) + + session.kill() + logger.info("session stopped", extra={"tmux_session": session_name or ""}) + tmuxp_echo( + colors.success("Stopped ") + colors.highlight(session_name or ""), + ) diff --git a/src/tmuxp/util.py b/src/tmuxp/util.py index 152b1f6c06..f52bd5f1f7 100644 --- a/src/tmuxp/util.py +++ b/src/tmuxp/util.py @@ -23,6 +23,7 @@ logger = logging.getLogger(__name__) PY2 = sys.version_info[0] == 2 +_REDACTED_HOOK_COMMAND = "" def run_before_script( @@ -105,6 +106,66 @@ def run_before_script( return return_code +def run_hook_commands( + commands: str | list[str], + cwd: pathlib.Path | str | None = None, +) -> None: + """Run lifecycle hook shell commands. + + Hooks use ``shell=True`` so project configs can use normal shell syntax + such as pipes, redirects, and command chaining. Hook failures are logged + and do not raise. + + Examples + -------- + >>> run_hook_commands("") + + >>> run_hook_commands("printf hook >/dev/null") + """ + if isinstance(commands, str): + commands = [commands] + joined = "; ".join(commands) + if not joined.strip(): + return + + logger.info( + "hook commands started", extra={"tmux_hook_cmd": _REDACTED_HOOK_COMMAND} + ) + try: + result = subprocess.run( + joined, + shell=True, + cwd=cwd, + check=False, + capture_output=True, + text=True, + ) + except OSError: + logger.warning( + "hook command failed", + extra={"tmux_hook_cmd": _REDACTED_HOOK_COMMAND}, + ) + return + + if result.returncode != 0: + logger.warning( + "hook command failed", + extra={ + "tmux_hook_cmd": _REDACTED_HOOK_COMMAND, + "tmux_exit_code": result.returncode, + }, + ) + # Hook output may carry the same expanded credentials as the + # redacted command text; persist lengths only. + logger.debug( + "hook output suppressed", + extra={ + "tmux_stdout_len": len(result.stdout), + "tmux_stderr_len": len(result.stderr), + }, + ) + + def oh_my_zsh_auto_title() -> None: """Give warning and offer to fix ``DISABLE_AUTO_TITLE``. @@ -145,27 +206,41 @@ def get_session( server: Server, session_name: str | None = None, current_pane: Pane | None = None, + require_pane_resolution: bool = False, ) -> Session: - """Get tmux session for server by session name, respects current pane, if passed.""" + """Get tmux session for server by name or current pane. + + Examples + -------- + >>> from tmuxp.util import get_session + >>> get_session(server, session_name=session.name) == session + True + """ + session_result: Session | None = None try: if session_name: - session = server.sessions.get(session_name=session_name) + session_result = server.sessions.get(session_name=session_name) elif current_pane is not None: - session = server.sessions.get(session_id=current_pane.session_id) + session_result = server.sessions.get(session_id=current_pane.session_id) else: current_pane = get_current_pane(server) if current_pane: - session = server.sessions.get(session_id=current_pane.session_id) - else: - session = server.sessions[0] + session_result = server.sessions.get(session_id=current_pane.session_id) + elif require_pane_resolution: + pass + elif server.sessions: + session_result = server.sessions[0] except Exception as e: if session_name: raise exc.SessionNotFound(session_name) from e raise exc.SessionNotFound from e - assert session is not None - return session + if session_result is None: + if session_name: + raise exc.SessionNotFound(session_name) + raise exc.SessionNotFound + return session_result def get_window( diff --git a/src/tmuxp/workspace/builder.py b/src/tmuxp/workspace/builder.py index 6ad3fd9445..0105762abf 100644 --- a/src/tmuxp/workspace/builder.py +++ b/src/tmuxp/workspace/builder.py @@ -4,6 +4,7 @@ import logging import os +import shlex import shutil import time import typing as t @@ -422,6 +423,8 @@ def build(self, session: Session | None = None, append: bool = False) -> None: append : bool append windows in current active session """ + session_created = session is None + if not session: if not self.server: msg = ( @@ -538,6 +541,35 @@ def build(self, session: Session | None = None, append: bool = False) -> None: for option, value in self.session_config["environment"].items(): self.session.set_environment(option, value) + if session_created and "on_project_exit" in self.session_config: + exit_cmds = self.session_config["on_project_exit"] + if isinstance(exit_cmds, str): + exit_cmds = [exit_cmds] + joined = "; ".join(exit_cmds) + start_dir = self.session_config.get("start_directory") + if start_dir: + joined = f"cd {shlex.quote(start_dir)} && {joined}" + guarded = f"if [ #{{session_attached}} -eq 0 ]; then {joined}; fi" + self.session.set_hook( + "client-detached", + f"run-shell {shlex.quote(guarded)}", + ) + + if session_created and "on_project_stop" in self.session_config: + stop_cmds = self.session_config["on_project_stop"] + if isinstance(stop_cmds, str): + stop_cmds = [stop_cmds] + self.session.set_environment( + "TMUXP_ON_PROJECT_STOP", + "; ".join(stop_cmds), + ) + + if session_created and "start_directory" in self.session_config: + self.session.set_environment( + "TMUXP_START_DIRECTORY", + self.session_config["start_directory"], + ) + for window, window_config in self.iter_create_windows(session, append): assert isinstance(window, Window) diff --git a/src/tmuxp/workspace/loader.py b/src/tmuxp/workspace/loader.py index f6e32629fd..8f69edf0bc 100644 --- a/src/tmuxp/workspace/loader.py +++ b/src/tmuxp/workspace/loader.py @@ -171,6 +171,19 @@ def expand( if any(workspace_dict["before_script"].startswith(a) for a in [".", "./"]): workspace_dict["before_script"] = str(cwd / workspace_dict["before_script"]) + for hook_key in ( + "on_project_start", + "on_project_restart", + "on_project_exit", + "on_project_stop", + ): + if hook_key in workspace_dict: + hook_value = workspace_dict[hook_key] + if isinstance(hook_value, str): + workspace_dict[hook_key] = expandshell(hook_value) + elif isinstance(hook_value, list): + workspace_dict[hook_key] = [expandshell(v) for v in hook_value] + if "shell_command" in workspace_dict and isinstance( workspace_dict["shell_command"], str, diff --git a/tests/cli/test_help_examples.py b/tests/cli/test_help_examples.py index 9cbe365db2..adff949017 100644 --- a/tests/cli/test_help_examples.py +++ b/tests/cli/test_help_examples.py @@ -114,6 +114,7 @@ def test_main_help_examples_are_valid_subcommands() -> None: "edit", "freeze", "search", + "stop", } for example in examples: @@ -137,6 +138,7 @@ def test_main_help_examples_are_valid_subcommands() -> None: "edit", "freeze", "search", + "stop", ], ) def test_subcommand_help_has_examples(subcommand: str) -> None: @@ -236,6 +238,15 @@ def test_search_subcommand_examples_are_valid() -> None: assert example.startswith("tmuxp search"), f"Bad example format: {example}" +def test_stop_subcommand_examples_are_valid() -> None: + """Stop subcommand examples should have valid flags.""" + help_text = _get_help_text("stop") + examples = extract_examples_from_help(help_text) + + for example in examples: + assert example.startswith("tmuxp stop"), f"Bad example format: {example}" + + def test_search_no_args_shows_help() -> None: """Running 'tmuxp search' with no args shows help. diff --git a/tests/cli/test_load.py b/tests/cli/test_load.py index ec045dcf3c..a924e92bf3 100644 --- a/tests/cli/test_load.py +++ b/tests/cli/test_load.py @@ -12,6 +12,7 @@ from libtmux.server import Server from libtmux.session import Session +import tmuxp.cli.load as load_module from tests.constants import FIXTURE_PATH from tests.fixtures import utils as test_utils from tmuxp import cli @@ -887,6 +888,119 @@ def test_load_workspace_env_progress_disabled( assert session.name == "sample workspace" +def test_load_on_project_start_runs_hook( + tmp_path: pathlib.Path, + server: Server, + monkeypatch: pytest.MonkeyPatch, +) -> None: + """load_workspace() runs on_project_start before creating a new session.""" + monkeypatch.delenv("TMUX", raising=False) + + marker = tmp_path / "start-hook-ran" + workspace_file = tmp_path / "hook-start.yaml" + workspace_file.write_text( + f"""\ +session_name: hook-start-test +on_project_start: "touch {marker}" +windows: + - window_name: main + panes: + - echo hello +""", + encoding="utf-8", + ) + + session = load_workspace( + workspace_file, + socket_name=server.socket_name, + detached=True, + ) + + assert marker.exists() + assert session is not None + session.kill() + + +class ExistingSessionRestartFixture(t.NamedTuple): + """Fixture for existing-session restart hook behavior.""" + + test_id: str + detached: bool + expect_restart_hook: bool + expect_reattach: bool + + +EXISTING_SESSION_RESTART_FIXTURES: list[ExistingSessionRestartFixture] = [ + ExistingSessionRestartFixture( + test_id="confirmed-attach-runs-hook", + detached=False, + expect_restart_hook=True, + expect_reattach=True, + ), + ExistingSessionRestartFixture( + test_id="detached-existing-session-skips-hook", + detached=True, + expect_restart_hook=False, + expect_reattach=False, + ), +] + + +@pytest.mark.parametrize( + list(ExistingSessionRestartFixture._fields), + EXISTING_SESSION_RESTART_FIXTURES, + ids=[fixture.test_id for fixture in EXISTING_SESSION_RESTART_FIXTURES], +) +def test_load_existing_session_restart_hook( + tmp_path: pathlib.Path, + server: Server, + monkeypatch: pytest.MonkeyPatch, + test_id: str, + detached: bool, + expect_restart_hook: bool, + expect_reattach: bool, +) -> None: + """on_project_restart runs only when attaching to an existing session.""" + monkeypatch.delenv("TMUX", raising=False) + server.new_session(session_name="hook-restart-test") + + marker = tmp_path / f"{test_id}.marker" + workspace_file = tmp_path / "hook-restart.yaml" + workspace_file.write_text( + f"""\ +session_name: hook-restart-test +on_project_restart: "touch {marker}" +windows: + - window_name: main + panes: + - +""", + encoding="utf-8", + ) + + reattach_called = False + + def fake_reattach( + _builder: WorkspaceBuilder, + _colors: object | None = None, + ) -> None: + nonlocal reattach_called + reattach_called = True + + monkeypatch.setattr(load_module, "_reattach", fake_reattach) + + result = load_module.load_workspace( + workspace_file, + socket_name=server.socket_name, + answer_yes=True, + detached=detached, + ) + + assert result is None + assert marker.exists() is expect_restart_hook + assert reattach_called is expect_reattach + + def test_load_masks_home_in_spinner_message(monkeypatch: pytest.MonkeyPatch) -> None: """Spinner message should mask home directory via PrivatePath.""" monkeypatch.setattr(pathlib.Path, "home", lambda: pathlib.Path("/home/testuser")) diff --git a/tests/cli/test_stop.py b/tests/cli/test_stop.py new file mode 100644 index 0000000000..21fc4ecd53 --- /dev/null +++ b/tests/cli/test_stop.py @@ -0,0 +1,93 @@ +"""CLI tests for tmuxp stop.""" + +from __future__ import annotations + +import argparse +import pathlib + +import pytest +from libtmux.server import Server + +from tmuxp.cli.stop import CLIStopNamespace, command_stop, create_stop_subparser + + +def _stop_args( + server: Server, + session_name: str | None, +) -> CLIStopNamespace: + args = CLIStopNamespace() + args.color = "never" + args.session_name = session_name + args.socket_name = server.socket_name + args.socket_path = None + return args + + +def test_create_stop_subparser() -> None: + """create_stop_subparser() parses an optional session name.""" + parser = create_stop_subparser(argparse.ArgumentParser()) + + args = parser.parse_args(["mysession"]) + + assert args.session_name == "mysession" + + +def test_command_stop_kills_named_session( + server: Server, + capsys: pytest.CaptureFixture[str], +) -> None: + """command_stop() kills a named tmux session.""" + session = server.new_session(session_name="stop-named") + + command_stop(_stop_args(server, session.name)) + + assert server.sessions.get(session_name="stop-named", default=None) is None + assert "Stopped stop-named" in capsys.readouterr().out + + +def test_command_stop_runs_on_project_stop( + tmp_path: pathlib.Path, + server: Server, +) -> None: + """command_stop() runs on_project_stop before killing the session.""" + marker = tmp_path / "stop-hook-ran" + session = server.new_session(session_name="stop-hook") + session.set_environment("TMUXP_ON_PROJECT_STOP", f"touch {marker}") + session.set_environment("TMUXP_START_DIRECTORY", str(tmp_path)) + + command_stop(_stop_args(server, session.name)) + + assert marker.exists() + assert server.sessions.get(session_name="stop-hook", default=None) is None + + +def test_command_stop_current_session_inside_tmux( + server: Server, + monkeypatch: pytest.MonkeyPatch, +) -> None: + """command_stop() stops the current tmux session when no name is given.""" + session = server.new_session(session_name="stop-current") + pane = session.active_window.active_pane + assert pane is not None + assert pane.pane_id is not None + + monkeypatch.setenv("TMUX", "/tmp/tmux-test/default,123,0") + monkeypatch.setenv("TMUX_PANE", pane.pane_id) + + command_stop(_stop_args(server, None)) + + assert server.sessions.get(session_name="stop-current", default=None) is None + + +def test_command_stop_requires_name_outside_tmux( + server: Server, + monkeypatch: pytest.MonkeyPatch, +) -> None: + """command_stop() exits if no session name is given outside tmux.""" + monkeypatch.delenv("TMUX", raising=False) + monkeypatch.delenv("TMUX_PANE", raising=False) + + with pytest.raises(SystemExit) as excinfo: + command_stop(_stop_args(server, None)) + + assert excinfo.value.code == 1 diff --git a/tests/test_util.py b/tests/test_util.py index 098c8c212b..8d82a47d6f 100644 --- a/tests/test_util.py +++ b/tests/test_util.py @@ -12,7 +12,13 @@ from tmuxp import exc from tmuxp.exc import BeforeLoadScriptError, BeforeLoadScriptNotExists -from tmuxp.util import get_pane, get_session, oh_my_zsh_auto_title, run_before_script +from tmuxp.util import ( + get_pane, + get_session, + oh_my_zsh_auto_title, + run_before_script, + run_hook_commands, +) from .constants import FIXTURE_PATH @@ -140,6 +146,99 @@ def test_beforeload_returns_stderr_messages() -> None: assert excinfo.match(r"failed with returncode") +class RunHookCommandsFixture(t.NamedTuple): + """Fixture for lifecycle hook command execution.""" + + test_id: str + command_templates: tuple[str, ...] + expected_text: str + + +RUN_HOOK_COMMANDS_FIXTURES: list[RunHookCommandsFixture] = [ + RunHookCommandsFixture( + test_id="single-command", + command_templates=("printf single > {marker}",), + expected_text="single", + ), + RunHookCommandsFixture( + test_id="command-list", + command_templates=( + "printf first > {marker}", + "printf second >> {marker}", + ), + expected_text="firstsecond", + ), +] + + +@pytest.mark.parametrize( + list(RunHookCommandsFixture._fields), + RUN_HOOK_COMMANDS_FIXTURES, + ids=[fixture.test_id for fixture in RUN_HOOK_COMMANDS_FIXTURES], +) +def test_run_hook_commands( + tmp_path: pathlib.Path, + test_id: str, + command_templates: tuple[str, ...], + expected_text: str, +) -> None: + """run_hook_commands() executes string and list hooks.""" + marker = tmp_path / f"{test_id}.txt" + commands = [template.format(marker=marker) for template in command_templates] + hook_commands: str | list[str] = commands[0] if len(commands) == 1 else commands + + run_hook_commands(hook_commands) + + assert marker.read_text(encoding="utf-8") == expected_text + + +def test_run_hook_commands_empty_is_noop() -> None: + """run_hook_commands() ignores empty command strings.""" + run_hook_commands("") + + +def test_run_hook_commands_logs_failure( + caplog: pytest.LogCaptureFixture, +) -> None: + """run_hook_commands() logs non-zero hook exits without raising.""" + with caplog.at_level(logging.WARNING, logger="tmuxp.util"): + run_hook_commands("exit 7") + + records = [ + record + for record in caplog.records + if record.levelno == logging.WARNING and hasattr(record, "tmux_exit_code") + ] + assert len(records) == 1 + assert records[0].tmux_exit_code == 7 + + +def test_run_hook_commands_redacts_logged_command( + caplog: pytest.LogCaptureFixture, +) -> None: + """run_hook_commands() keeps expanded hook commands out of logs.""" + secret = "secret-token-for-hook-log" + + with caplog.at_level(logging.INFO, logger="tmuxp.util"): + run_hook_commands(f"false # {secret}") + + for record in caplog.records: + assert secret not in record.getMessage() + assert secret not in str(record.__dict__) + + hook_records = [ + record for record in caplog.records if hasattr(record, "tmux_hook_cmd") + ] + assert hook_records + assert all(record.tmux_hook_cmd == "" for record in hook_records) + + failure_records = [ + record for record in caplog.records if hasattr(record, "tmux_exit_code") + ] + assert len(failure_records) == 1 + assert failure_records[0].tmux_exit_code == 1 + + def test_get_session_should_default_to_local_attached_session( server: Server, monkeypatch: pytest.MonkeyPatch, @@ -171,6 +270,19 @@ def test_get_session_should_return_first_session_if_no_active_session( assert get_session(server) == first_session +def test_get_session_require_pane_resolution_raises_without_current_pane( + server: Server, + monkeypatch: pytest.MonkeyPatch, +) -> None: + """get_session() can require TMUX_PANE resolution for destructive commands.""" + monkeypatch.delenv("TMUX_PANE", raising=False) + monkeypatch.delenv("TMUX", raising=False) + server.new_session(session_name="myfirstsession") + + with pytest.raises(exc.SessionNotFound): + get_session(server, require_pane_resolution=True) + + def test_get_pane_logs_debug_on_failure( server: Server, monkeypatch: pytest.MonkeyPatch, @@ -234,3 +346,59 @@ def patched_exists(path: str) -> bool: warning_records = [r for r in caplog.records if r.levelno == logging.WARNING] assert len(warning_records) >= 1 assert "DISABLE_AUTO_TITLE" in warning_records[0].message + + +class HookOutputLeakFixture(t.NamedTuple): + """Fixture for hook output suppression in failure logs.""" + + test_id: str + command: str + expect_stdout: bool + + +HOOK_OUTPUT_LEAK_FIXTURES: list[HookOutputLeakFixture] = [ + HookOutputLeakFixture( + test_id="stdout_leak", + command="echo {secret}; exit 1", + expect_stdout=True, + ), + HookOutputLeakFixture( + test_id="stderr_leak", + command="echo {secret} >&2; exit 1", + expect_stdout=False, + ), +] + + +@pytest.mark.parametrize( + list(HookOutputLeakFixture._fields), + HOOK_OUTPUT_LEAK_FIXTURES, + ids=[f.test_id for f in HOOK_OUTPUT_LEAK_FIXTURES], +) +def test_run_hook_commands_suppresses_failure_output( + caplog: pytest.LogCaptureFixture, + test_id: str, + command: str, + expect_stdout: bool, +) -> None: + """run_hook_commands() keeps hook output out of failure logs. + + Hook commands may expand credentials; redacting the command text + while logging its raw output would leak the same secrets. + """ + secret = "hook-output-secret-marker" + + with caplog.at_level(logging.DEBUG, logger="tmuxp.util"): + run_hook_commands(command.format(secret=secret)) + + for record in caplog.records: + assert secret not in record.getMessage() + assert secret not in str(record.__dict__) + + suppressed = [r for r in caplog.records if hasattr(r, "tmux_stdout_len")] + assert len(suppressed) == 1 + assert hasattr(suppressed[0], "tmux_stderr_len") + if expect_stdout: + assert suppressed[0].tmux_stdout_len > 0 + else: + assert suppressed[0].tmux_stderr_len > 0 diff --git a/tests/workspace/test_builder.py b/tests/workspace/test_builder.py index 74a9e480ad..f55f954124 100644 --- a/tests/workspace/test_builder.py +++ b/tests/workspace/test_builder.py @@ -2228,3 +2228,110 @@ def spy_sleep(seconds: float) -> None: option_sleeps = [s for s in slept if s in (0.21, 0.31)] assert sorted(option_sleeps) == sorted(expected_sleeps) + + +class ProjectExitHookFixture(t.NamedTuple): + """Fixture for on_project_exit hook metadata.""" + + test_id: str + hook_value: str | list[str] + expected_fragment: str + + +PROJECT_EXIT_HOOK_FIXTURES: list[ProjectExitHookFixture] = [ + ProjectExitHookFixture( + test_id="string-command", + hook_value="echo goodbye", + expected_fragment="echo goodbye", + ), + ProjectExitHookFixture( + test_id="list-command", + hook_value=["echo first", "echo second"], + expected_fragment="echo first; echo second", + ), +] + + +@pytest.mark.parametrize( + list(ProjectExitHookFixture._fields), + PROJECT_EXIT_HOOK_FIXTURES, + ids=[fixture.test_id for fixture in PROJECT_EXIT_HOOK_FIXTURES], +) +def test_on_project_exit_sets_guarded_hook( + server: Server, + tmp_path: pathlib.Path, + test_id: str, + hook_value: str | list[str], + expected_fragment: str, +) -> None: + """on_project_exit stores a guarded client-detached hook on new sessions.""" + workspace: dict[str, t.Any] = { + "session_name": f"hook-exit-{test_id}", + "start_directory": str(tmp_path), + "on_project_exit": hook_value, + "windows": [{"window_name": "main", "panes": [{"shell_command": []}]}], + } + workspace = loader.trickle(loader.expand(workspace)) + + builder = WorkspaceBuilder(session_config=workspace, server=server) + builder.build() + assert builder.session is not None + + hook_values = [str(value) for value in builder.session.show_hooks().values()] + assert any("#{session_attached}" in value for value in hook_values) + assert any(str(tmp_path) in value for value in hook_values) + assert any(expected_fragment in value for value in hook_values) + + builder.session.kill() + + +def test_on_project_stop_sets_environment( + server: Server, + tmp_path: pathlib.Path, +) -> None: + """on_project_stop stores commands and cwd metadata on new sessions.""" + workspace: dict[str, t.Any] = { + "session_name": "hook-stop-env-test", + "start_directory": str(tmp_path), + "on_project_stop": ["echo stop", "echo done"], + "windows": [{"window_name": "main", "panes": [{"shell_command": []}]}], + } + workspace = loader.trickle(loader.expand(workspace)) + + builder = WorkspaceBuilder(session_config=workspace, server=server) + builder.build() + assert builder.session is not None + + assert builder.session.getenv("TMUXP_ON_PROJECT_STOP") == "echo stop; echo done" + assert builder.session.getenv("TMUXP_START_DIRECTORY") == str(tmp_path) + + builder.session.kill() + + +def test_lifecycle_metadata_not_overwritten_on_reused_session( + session: Session, + tmp_path: pathlib.Path, +) -> None: + """Lifecycle hooks do not overwrite metadata on appended/reused sessions.""" + original_hook = "run-shell 'printf %s original-exit >/dev/null'" + session.set_hook("client-detached", original_hook) + session.set_environment("TMUXP_ON_PROJECT_STOP", "original stop") + session.set_environment("TMUXP_START_DIRECTORY", "/original/start") + + workspace: dict[str, t.Any] = { + "session_name": session.name, + "start_directory": str(tmp_path), + "on_project_exit": "printf '%s' new-exit >/dev/null", + "on_project_stop": "printf '%s' new-stop >/dev/null", + "windows": [{"window_name": "reused-window", "panes": [{"shell_command": []}]}], + } + workspace = loader.trickle(loader.expand(workspace)) + + builder = WorkspaceBuilder(session_config=workspace, server=session.server) + builder.build(session=session, append=True) + + hook_values = [str(value) for value in session.show_hooks().values()] + assert any("original-exit" in value for value in hook_values) + assert all("new-exit" not in value for value in hook_values) + assert session.getenv("TMUXP_ON_PROJECT_STOP") == "original stop" + assert session.getenv("TMUXP_START_DIRECTORY") == "/original/start" diff --git a/tests/workspace/test_config.py b/tests/workspace/test_config.py index 44a1445056..883c3f177e 100644 --- a/tests/workspace/test_config.py +++ b/tests/workspace/test_config.py @@ -590,3 +590,66 @@ def test_validate_schema_logs_debug( records = [r for r in caplog.records if r.msg == "validating workspace schema"] assert len(records) >= 1 assert getattr(records[0], "tmux_session", None) == "test_validate" + + +class LifecycleHookExpandFixture(t.NamedTuple): + """Fixture for lifecycle hook expansion.""" + + test_id: str + hook_key: str + hook_value: str | list[str] + env: dict[str, str] + expected: str | list[str] + + +LIFECYCLE_HOOK_EXPAND_FIXTURES: list[LifecycleHookExpandFixture] = [ + LifecycleHookExpandFixture( + test_id="start-string-env", + hook_key="on_project_start", + hook_value="$TMUXP_HOOK_CMD", + env={"TMUXP_HOOK_CMD": "docker compose up"}, + expected="docker compose up", + ), + LifecycleHookExpandFixture( + test_id="stop-string-with-suffix", + hook_key="on_project_stop", + hook_value="$TMUXP_HOOK_CMD down", + env={"TMUXP_HOOK_CMD": "docker compose"}, + expected="docker compose down", + ), + LifecycleHookExpandFixture( + test_id="restart-list-env", + hook_key="on_project_restart", + hook_value=["$TMUXP_HOOK_CMD", "echo world"], + env={"TMUXP_HOOK_CMD": "echo hello"}, + expected=["echo hello", "echo world"], + ), +] + + +@pytest.mark.parametrize( + list(LifecycleHookExpandFixture._fields), + LIFECYCLE_HOOK_EXPAND_FIXTURES, + ids=[fixture.test_id for fixture in LIFECYCLE_HOOK_EXPAND_FIXTURES], +) +def test_expand_lifecycle_hooks( + monkeypatch: pytest.MonkeyPatch, + test_id: str, + hook_key: str, + hook_value: str | list[str], + env: dict[str, str], + expected: str | list[str], +) -> None: + """expand() expands environment variables in lifecycle hook values.""" + for key, value in env.items(): + monkeypatch.setenv(key, value) + + workspace: dict[str, t.Any] = { + "session_name": f"hook-{test_id}", + hook_key: hook_value, + "windows": [{"window_name": "main", "panes": [{"shell_command": []}]}], + } + + result = loader.expand(workspace) + + assert result[hook_key] == expected