feat: add restart command and --watch to configure

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()

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()

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)

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()

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()