From 07ab0df318a7ac5779be680bf9fd26d2234a3adf Mon Sep 17 00:00:00 2001 From: Nils Hamerlinck Date: Tue, 2 Jun 2026 18:38:54 +0700 Subject: [PATCH 1/3] feat(restart): add restart command with optional --watch Co-Authored-By: Claude Sonnet 4.6 --- trobz_deploy/cli.py | 2 ++ trobz_deploy/command/restart.py | 43 +++++++++++++++++++++++++++++++++ 2 files changed, 45 insertions(+) create mode 100644 trobz_deploy/command/restart.py diff --git a/trobz_deploy/cli.py b/trobz_deploy/cli.py index 21e715f..3235eec 100644 --- a/trobz_deploy/cli.py +++ b/trobz_deploy/cli.py @@ -5,6 +5,7 @@ import typer from trobz_deploy.command.configure import configure +from trobz_deploy.command.restart import restart from trobz_deploy.command.status import status from trobz_deploy.command.update import update @@ -30,4 +31,5 @@ def cli( app.command()(configure) app.command()(update) +app.command()(restart) app.command()(status) diff --git a/trobz_deploy/command/restart.py b/trobz_deploy/command/restart.py new file mode 100644 index 0000000..a86f164 --- /dev/null +++ b/trobz_deploy/command/restart.py @@ -0,0 +1,43 @@ +from __future__ import annotations + +from typing import Annotated + +import typer + +from trobz_deploy.utils.config import load_config +from trobz_deploy.utils.executor import Executor, ExecutorError + + +def restart( + ctx: typer.Context, + instance_name: Annotated[str, typer.Argument()], + ssh_host: Annotated[str | None, typer.Argument()] = None, + ssh_port: Annotated[int | None, typer.Option("-p", "--port", help="SSH port on the remote host.")] = None, + watch: Annotated[ + bool, + typer.Option("--watch", help="Stream service logs with journalctl after restarting."), + ] = False, +) -> None: + """Restart the systemd unit for a deployment instance.""" + cfg = load_config(ctx.obj["config"], instance_name) + + eff_ssh_host: str | None = ssh_host if ssh_host is not None else cfg.get("ssh_host") + eff_ssh_port: int | None = ssh_port if ssh_port is not None else cfg.get("ssh_port") + + executor = Executor(eff_ssh_host, ctx.obj["verbose"], ssh_port=eff_ssh_port) + + typer.secho(f"\nRestarting {instance_name!r}…", fg="green") + try: + executor.run(f"systemctl --user restart {instance_name}") + except ExecutorError as exc: + typer.echo(typer.style(f"Restart failed: {exc}", fg="red"), err=True) + raise typer.Exit(code=1) from exc + + typer.secho(f"\nInstance {instance_name!r} restarted.", fg="green") + + if watch: + typer.secho("\nWatching service logs (Ctrl+C to stop)…", fg="cyan") + try: + executor.stream(f"journalctl --user -u {instance_name} -f") + except KeyboardInterrupt: + typer.echo() From a03ff0cbb66e8c3ad9c53e9451d94fb3ea808fa5 Mon Sep 17 00:00:00 2001 From: Nils Hamerlinck Date: Tue, 2 Jun 2026 19:00:28 +0700 Subject: [PATCH 2/3] feat(configure): add --watch option to stream logs after setup Co-Authored-By: Claude Sonnet 4.6 --- trobz_deploy/command/configure.py | 11 +++++++++++ 1 file changed, 11 insertions(+) diff --git a/trobz_deploy/command/configure.py b/trobz_deploy/command/configure.py index 8bd658e..c1ba679 100644 --- a/trobz_deploy/command/configure.py +++ b/trobz_deploy/command/configure.py @@ -48,6 +48,10 @@ def configure( # noqa: C901 str | None, typer.Option(help="Git branch to clone and track (defaults to the repository's default branch)."), ] = None, + watch: Annotated[ + bool, + typer.Option("--watch", help="Stream service logs with journalctl after a successful configure."), + ] = False, ) -> None: """Configure a new deployment instance.""" cfg = load_config(ctx.obj["config"], instance_name) @@ -206,3 +210,10 @@ def configure( # noqa: C901 raise typer.Exit(code=1) from exc typer.secho(f"\nInstance {instance_name!r} configured successfully.", fg="green") + + if watch: + typer.secho("\nWatching service logs (Ctrl+C to stop)…", fg="cyan") + try: + executor.stream(f"journalctl --user -u {instance_name} -f") + except KeyboardInterrupt: + typer.echo() From ce53441e600d265c9dcbd80bf634f33bf9e49bdf Mon Sep 17 00:00:00 2001 From: Nils Hamerlinck Date: Tue, 2 Jun 2026 19:04:41 +0700 Subject: [PATCH 3/3] test: add watch tests for restart and configure commands Co-Authored-By: Claude Sonnet 4.6 --- tests/test_configure_watch.py | 65 +++++++++++++++++++++++++++++++++++ tests/test_restart_watch.py | 62 +++++++++++++++++++++++++++++++++ 2 files changed, 127 insertions(+) create mode 100644 tests/test_configure_watch.py create mode 100644 tests/test_restart_watch.py diff --git a/tests/test_configure_watch.py b/tests/test_configure_watch.py new file mode 100644 index 0000000..cf10266 --- /dev/null +++ b/tests/test_configure_watch.py @@ -0,0 +1,65 @@ +from __future__ import annotations + +from unittest.mock import MagicMock, patch + +import pytest +from typer.testing import CliRunner + +from trobz_deploy.cli import app + + +@pytest.fixture +def runner(): + return CliRunner() + + +def _executor_mock(): + mock = MagicMock() + mock.capture.side_effect = [ + "/home/deploy", # echo $HOME + "", # which odoo-addons-path (not reached for service type) + ] + # render_unit needs exec_start resolved; capture for server.py check returns "" + mock.capture.return_value = "" + return mock + + +def _invoke(runner, extra_args: list[str], side_effect=None): + with ( + patch("trobz_deploy.command.configure.Executor") as MockExecutor, + patch( + "trobz_deploy.command.configure.load_config", + return_value={"exec_start": "/usr/bin/myapp"}, + ), + patch("trobz_deploy.command.configure.render_unit", return_value="[Unit]\n"), + ): + mock_exec = _executor_mock() + if side_effect: + mock_exec.stream.side_effect = side_effect + MockExecutor.return_value = mock_exec + result = runner.invoke( + app, + ["configure", "service-myapp-production", "--type", "service", *extra_args], + ) + return result, mock_exec + + +def test_watch_streams_journalctl_after_configure(runner): + result, mock_exec = _invoke(runner, ["--watch"]) + + assert result.exit_code == 0 + mock_exec.stream.assert_called_once_with("journalctl --user -u service-myapp-production -f") + + +def test_no_watch_does_not_stream(runner): + result, mock_exec = _invoke(runner, []) + + assert result.exit_code == 0 + mock_exec.stream.assert_not_called() + + +def test_watch_handles_keyboard_interrupt(runner): + result, mock_exec = _invoke(runner, ["--watch"], side_effect=KeyboardInterrupt) + + assert result.exit_code == 0 + mock_exec.stream.assert_called_once() diff --git a/tests/test_restart_watch.py b/tests/test_restart_watch.py new file mode 100644 index 0000000..366ca4b --- /dev/null +++ b/tests/test_restart_watch.py @@ -0,0 +1,62 @@ +from __future__ import annotations + +from unittest.mock import MagicMock, patch + +import pytest +from typer.testing import CliRunner + +from trobz_deploy.cli import app + + +@pytest.fixture +def runner(): + return CliRunner() + + +def _executor_mock(): + mock = MagicMock() + return mock + + +def _invoke(runner, extra_args: list[str], side_effect=None): + with ( + patch("trobz_deploy.command.restart.Executor") as MockExecutor, + patch("trobz_deploy.command.restart.load_config", return_value={}), + ): + mock_exec = _executor_mock() + if side_effect: + mock_exec.stream.side_effect = side_effect + MockExecutor.return_value = mock_exec + result = runner.invoke( + app, + ["restart", "service-myapp-production", *extra_args], + ) + return result, mock_exec + + +def test_restart_runs_systemctl(runner): + result, mock_exec = _invoke(runner, []) + + assert result.exit_code == 0 + mock_exec.run.assert_called_once_with("systemctl --user restart service-myapp-production") + + +def test_watch_streams_journalctl_after_restart(runner): + result, mock_exec = _invoke(runner, ["--watch"]) + + assert result.exit_code == 0 + mock_exec.stream.assert_called_once_with("journalctl --user -u service-myapp-production -f") + + +def test_no_watch_does_not_stream(runner): + result, mock_exec = _invoke(runner, []) + + assert result.exit_code == 0 + mock_exec.stream.assert_not_called() + + +def test_watch_handles_keyboard_interrupt(runner): + result, mock_exec = _invoke(runner, ["--watch"], side_effect=KeyboardInterrupt) + + assert result.exit_code == 0 + mock_exec.stream.assert_called_once()