# autossh-mgr Implementation Plan > **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking. **Goal:** Build a Python + Click CLI tool that manages autossh reverse SSH tunnel lifecycle (config CRUD, start/stop/restart, status monitoring, connectivity check). **Architecture:** Domain-services approach — `config.py` owns YAML I/O, `process.py` owns PID-file-based process management, `check.py` owns SSH connectivity, `display.py` owns output formatting. `cli.py` is a thin Click wrapper over all of them. Config stored at `~/.config/autossh-mgr/` (overridable via `AUTOSSH_MGR_CONFIG_DIR` env var). **Tech Stack:** Python 3.11+, Click 8.x, PyYAML 6.x, pytest, uv --- ## File Map | File | Responsibility | |---|---| | `pyproject.toml` | Packaging, deps, script entry point | | `src/autossh_mgr/__init__.py` | Package marker | | `src/autossh_mgr/__main__.py` | `python -m autossh_mgr` support | | `src/autossh_mgr/config.py` | `TunnelConfig` dataclass + YAML read/write + dir setup | | `src/autossh_mgr/process.py` | Start/stop/status via PID files and `os.kill` | | `src/autossh_mgr/check.py` | SSH connectivity verification | | `src/autossh_mgr/display.py` | `click.echo` table and detail formatters | | `src/autossh_mgr/cli.py` | Click group + all 11 commands | | `tests/conftest.py` | Shared pytest fixtures | | `tests/unit/test_config.py` | Config I/O unit tests | | `tests/unit/test_process.py` | Process management unit tests | | `tests/unit/test_check.py` | Connectivity check unit tests | | `tests/integration/test_lifecycle.py` | Full start/stop/status lifecycle | --- ### Task 1: Project scaffolding **Files:** - Create: `pyproject.toml` - Create: `src/autossh_mgr/__init__.py` - Create: `src/autossh_mgr/__main__.py` - Create: `tests/__init__.py` - Create: `tests/unit/__init__.py` - Create: `tests/integration/__init__.py` - Create: `tests/conftest.py` - [ ] **Step 1: Create `pyproject.toml`** ```toml [project] name = "autossh-mgr" version = "0.1.0" description = "CLI tool to manage autossh reverse SSH tunnels" requires-python = ">=3.11" dependencies = [ "click>=8.1", "pyyaml>=6.0", ] [project.scripts] autossh-mgr = "autossh_mgr.cli:cli" [build-system] requires = ["hatchling"] build-backend = "hatchling.build" [tool.hatch.build.targets.wheel] packages = ["src/autossh_mgr"] [tool.pytest.ini_options] testpaths = ["tests"] ``` - [ ] **Step 2: Create package skeleton** ```bash mkdir -p src/autossh_mgr tests/unit tests/integration ``` `src/autossh_mgr/__init__.py` — empty file. `src/autossh_mgr/__main__.py`: ```python from autossh_mgr.cli import cli if __name__ == "__main__": cli() ``` `tests/__init__.py`, `tests/unit/__init__.py`, `tests/integration/__init__.py` — all empty. - [ ] **Step 3: Install dev dependencies and verify structure** ```bash uv add --dev pytest uv sync uv run python -c "import autossh_mgr" ``` Expected: no errors. - [ ] **Step 4: Commit** ```bash git add pyproject.toml src/ tests/ git commit -m "feat: scaffold autossh-mgr project structure" ``` --- ### Task 2: TunnelConfig dataclass + YAML I/O (`config.py`) **Files:** - Create: `src/autossh_mgr/config.py` - Create: `tests/unit/test_config.py` - [ ] **Step 1: Create `tests/conftest.py`** (requires `config.py` to import — do this before writing tests) ```python import pytest from autossh_mgr.config import TunnelConfig, ensure_dirs @pytest.fixture def config_dir(tmp_path): ensure_dirs(tmp_path) return tmp_path @pytest.fixture def sample_tunnel(): return TunnelConfig( name="web-service", host="relay.example.com", user="deploy", local_port=8080, remote_port=18080, ) ``` - [ ] **Step 2: Write failing tests** `tests/unit/test_config.py`: ```python import pytest import click from autossh_mgr.config import ( TunnelConfig, ensure_dirs, load_tunnels, save_tunnels, get_tunnel ) def test_ensure_dirs_creates_structure(tmp_path): ensure_dirs(tmp_path) assert (tmp_path / "tunnels.yaml").exists() assert (tmp_path / "pids").is_dir() assert (tmp_path / "logs").is_dir() def test_ensure_dirs_is_idempotent(tmp_path): ensure_dirs(tmp_path) ensure_dirs(tmp_path) # must not raise def test_load_tunnels_empty(config_dir): assert load_tunnels(config_dir) == [] def test_save_and_load_roundtrip(config_dir, sample_tunnel): save_tunnels(config_dir, [sample_tunnel]) loaded = load_tunnels(config_dir) assert len(loaded) == 1 t = loaded[0] assert t.name == "web-service" assert t.host == "relay.example.com" assert t.user == "deploy" assert t.local_port == 8080 assert t.remote_port == 18080 assert t.port == 22 assert t.identity_file == "~/.ssh/id_ed25519" assert t.local_host == "127.0.0.1" assert t.remote_host == "0.0.0.0" assert t.ssh_options == "" def test_save_atomic_write(config_dir, sample_tunnel): save_tunnels(config_dir, [sample_tunnel]) # No .tmp file left behind assert not (config_dir / "tunnels.yaml.tmp").exists() def test_get_tunnel_found(config_dir, sample_tunnel): save_tunnels(config_dir, [sample_tunnel]) t = get_tunnel(config_dir, "web-service") assert t.name == "web-service" def test_get_tunnel_not_found(config_dir): with pytest.raises(click.ClickException, match="No tunnel named 'missing'"): get_tunnel(config_dir, "missing") def test_get_config_dir_env_override(monkeypatch, tmp_path): monkeypatch.setenv("AUTOSSH_MGR_CONFIG_DIR", str(tmp_path)) from autossh_mgr.config import get_config_dir assert get_config_dir() == tmp_path ``` - [ ] **Step 3: Run tests to confirm they fail** ```bash uv run pytest tests/unit/test_config.py -v ``` Expected: `ModuleNotFoundError` or `ImportError` — `config.py` does not exist yet. - [ ] **Step 4: Implement `src/autossh_mgr/config.py`** ```python import os from dataclasses import dataclass, asdict, field from pathlib import Path from typing import Optional import click import yaml def get_config_dir() -> Path: env = os.environ.get("AUTOSSH_MGR_CONFIG_DIR") if env: return Path(env) return Path.home() / ".config" / "autossh-mgr" @dataclass class TunnelConfig: name: str host: str user: str local_port: int remote_port: int port: int = 22 identity_file: str = "~/.ssh/id_ed25519" local_host: str = "127.0.0.1" remote_host: str = "0.0.0.0" ssh_options: str = "" _DEFAULTS = TunnelConfig( name="", host="", user="", local_port=0, remote_port=0 ) def ensure_dirs(config_dir: Path) -> None: config_dir.mkdir(parents=True, exist_ok=True) (config_dir / "pids").mkdir(exist_ok=True) (config_dir / "logs").mkdir(exist_ok=True) tunnels_file = config_dir / "tunnels.yaml" if not tunnels_file.exists(): tunnels_file.write_text("tunnels: []\n") def load_tunnels(config_dir: Path) -> list[TunnelConfig]: tunnels_file = config_dir / "tunnels.yaml" if not tunnels_file.exists(): return [] data = yaml.safe_load(tunnels_file.read_text()) or {} return [TunnelConfig(**entry) for entry in data.get("tunnels", [])] def save_tunnels(config_dir: Path, tunnels: list[TunnelConfig]) -> None: rows = [] for t in tunnels: d = asdict(t) # Omit optional fields that are at their default values for key in ("port", "identity_file", "local_host", "remote_host", "ssh_options"): if d[key] == getattr(_DEFAULTS, key): del d[key] rows.append(d) tmp = config_dir / "tunnels.yaml.tmp" tmp.write_text(yaml.dump({"tunnels": rows}, default_flow_style=False)) tmp.replace(config_dir / "tunnels.yaml") def get_tunnel(config_dir: Path, name: str) -> TunnelConfig: for t in load_tunnels(config_dir): if t.name == name: return t raise click.ClickException(f"No tunnel named '{name}'") ``` - [ ] **Step 5: Run tests to confirm they pass** ```bash uv run pytest tests/unit/test_config.py -v ``` Expected: all tests PASS. - [ ] **Step 6: Commit** ```bash git add src/autossh_mgr/config.py tests/conftest.py tests/unit/test_config.py git commit -m "feat: add TunnelConfig dataclass and YAML I/O" ``` --- ### Task 3: PID file utilities and status detection (`process.py` part 1) **Files:** - Create: `src/autossh_mgr/process.py` - Create: `tests/unit/test_process.py` - [ ] **Step 1: Write failing tests** `tests/unit/test_process.py`: ```python import os import pytest from datetime import datetime, timezone, timedelta from unittest.mock import patch from autossh_mgr.process import ( write_pid_file, read_pid_file, delete_pid_file, is_process_alive, check_stale, get_status, format_uptime, TunnelStatus, ) def test_write_and_read_pid_file(config_dir): started = datetime(2026, 5, 20, 10, 0, 0, tzinfo=timezone.utc) write_pid_file(config_dir, "web-service", 12345, started) pid, ts = read_pid_file(config_dir, "web-service") assert pid == 12345 assert ts == started def test_read_pid_file_missing(config_dir): assert read_pid_file(config_dir, "missing") is None def test_delete_pid_file(config_dir): started = datetime.now(timezone.utc) write_pid_file(config_dir, "web-service", 12345, started) delete_pid_file(config_dir, "web-service") assert read_pid_file(config_dir, "web-service") is None def test_delete_pid_file_missing_is_noop(config_dir): delete_pid_file(config_dir, "missing") # must not raise def test_is_process_alive_running(): # os.getpid() is always alive assert is_process_alive(os.getpid()) is True def test_is_process_alive_dead(): assert is_process_alive(9999999) is False def test_check_stale_no_pid_file(config_dir): assert check_stale(config_dir, "web-service") is False def test_check_stale_running_process(config_dir): write_pid_file(config_dir, "web-service", os.getpid(), datetime.now(timezone.utc)) assert check_stale(config_dir, "web-service") is False def test_check_stale_dead_process(config_dir): write_pid_file(config_dir, "web-service", 9999999, datetime.now(timezone.utc)) assert check_stale(config_dir, "web-service") is True def test_get_status_stopped(config_dir): status = get_status(config_dir, "web-service") assert status.state == "stopped" assert status.pid is None assert status.uptime is None def test_get_status_running(config_dir): write_pid_file(config_dir, "web-service", os.getpid(), datetime.now(timezone.utc)) status = get_status(config_dir, "web-service") assert status.state == "running" assert status.pid == os.getpid() assert status.uptime is not None def test_get_status_stale(config_dir): write_pid_file(config_dir, "web-service", 9999999, datetime.now(timezone.utc)) status = get_status(config_dir, "web-service") assert status.state == "stale" assert status.pid == 9999999 def test_format_uptime_seconds(): started = datetime.now(timezone.utc) - timedelta(seconds=45) assert format_uptime(started) == "45s" def test_format_uptime_minutes(): started = datetime.now(timezone.utc) - timedelta(minutes=3, seconds=15) assert format_uptime(started) == "3m" def test_format_uptime_hours(): started = datetime.now(timezone.utc) - timedelta(hours=2, minutes=30) assert format_uptime(started) == "2h 30m" def test_format_uptime_days(): started = datetime.now(timezone.utc) - timedelta(days=1, hours=5) assert format_uptime(started) == "1d 5h" ``` - [ ] **Step 2: Run tests to confirm they fail** ```bash uv run pytest tests/unit/test_process.py -v ``` Expected: `ImportError` — `process.py` does not exist yet. - [ ] **Step 3: Implement PID file utilities and status in `src/autossh_mgr/process.py`** ```python import os import signal from dataclasses import dataclass from datetime import datetime, timezone from pathlib import Path from typing import Optional import click @dataclass class TunnelStatus: state: str # "running", "stopped", "stale" pid: Optional[int] = None uptime: Optional[str] = None def _pid_path(config_dir: Path, name: str) -> Path: return config_dir / "pids" / f"{name}.pid" def write_pid_file(config_dir: Path, name: str, pid: int, started_at: datetime) -> None: path = _pid_path(config_dir, name) path.write_text(f"{pid}\n{started_at.isoformat()}\n") with open(path, "a") as f: os.fsync(f.fileno()) def read_pid_file(config_dir: Path, name: str) -> Optional[tuple[int, datetime]]: path = _pid_path(config_dir, name) if not path.exists(): return None lines = path.read_text().splitlines() if len(lines) < 2: raise click.ClickException( f"Corrupt PID file for '{name}', try: autossh-mgr stop {name}" ) return int(lines[0]), datetime.fromisoformat(lines[1]) def delete_pid_file(config_dir: Path, name: str) -> None: path = _pid_path(config_dir, name) path.unlink(missing_ok=True) def is_process_alive(pid: int) -> bool: try: os.kill(pid, 0) return True except (ProcessLookupError, PermissionError): return False def check_stale(config_dir: Path, name: str) -> bool: result = read_pid_file(config_dir, name) if result is None: return False pid, _ = result return not is_process_alive(pid) def format_uptime(started_at: datetime) -> str: delta = datetime.now(timezone.utc) - started_at total = int(delta.total_seconds()) days, rem = divmod(total, 86400) hours, rem = divmod(rem, 3600) minutes, seconds = divmod(rem, 60) parts = [] if days: parts.append(f"{days}d") if hours: parts.append(f"{hours}h") if minutes: parts.append(f"{minutes}m") if not parts: parts.append(f"{seconds}s") return " ".join(parts) def get_status(config_dir: Path, name: str) -> TunnelStatus: result = read_pid_file(config_dir, name) if result is None: return TunnelStatus(state="stopped") pid, started_at = result if not is_process_alive(pid): return TunnelStatus(state="stale", pid=pid) return TunnelStatus(state="running", pid=pid, uptime=format_uptime(started_at)) ``` - [ ] **Step 4: Run tests to confirm they pass** ```bash uv run pytest tests/unit/test_process.py -v ``` Expected: all tests PASS. - [ ] **Step 5: Commit** ```bash git add src/autossh_mgr/process.py tests/unit/test_process.py git commit -m "feat: add PID file utilities and tunnel status detection" ``` --- ### Task 4: Tunnel start and stop (`process.py` part 2) **Files:** - Modify: `src/autossh_mgr/process.py` - Modify: `tests/unit/test_process.py` - [ ] **Step 1: Write failing tests** (append to `tests/unit/test_process.py`) ```python import shutil import subprocess import time from autossh_mgr.config import TunnelConfig from autossh_mgr.process import start_tunnel, stop_tunnel, build_autossh_cmd def test_build_autossh_cmd_basic(sample_tunnel): cmd = build_autossh_cmd(sample_tunnel) assert cmd[0] == "autossh" assert "-M" in cmd and "0" in cmd assert "-N" in cmd assert "-i" in cmd assert os.path.expanduser(sample_tunnel.identity_file) in cmd assert "-p" in cmd and str(sample_tunnel.port) in cmd assert f"-R" in cmd r_idx = cmd.index("-R") assert cmd[r_idx + 1] == ( f"{sample_tunnel.remote_host}:{sample_tunnel.remote_port}" f":{sample_tunnel.local_host}:{sample_tunnel.local_port}" ) assert f"{sample_tunnel.user}@{sample_tunnel.host}" in cmd def test_build_autossh_cmd_with_ssh_options(): t = TunnelConfig( name="test", host="h.com", user="u", local_port=80, remote_port=8080, ssh_options="-o StrictHostKeyChecking=no", ) cmd = build_autossh_cmd(t) assert "-o" in cmd assert "StrictHostKeyChecking=no" in cmd def test_start_tunnel_no_autossh(config_dir, sample_tunnel, monkeypatch): monkeypatch.setattr(shutil, "which", lambda _: None) import click with pytest.raises(click.ClickException, match="autossh not found"): start_tunnel(config_dir, sample_tunnel) def test_start_creates_pid_file(config_dir, sample_tunnel, monkeypatch, tmp_path): # Replace autossh with a real long-lived process for testing fake = tmp_path / "autossh" fake.write_text("#!/bin/sh\nexec sleep 60\n") fake.chmod(0o755) monkeypatch.setenv("PATH", f"{tmp_path}:{os.environ['PATH']}") pid = start_tunnel(config_dir, sample_tunnel) try: assert pid > 0 result = read_pid_file(config_dir, "web-service") assert result is not None assert result[0] == pid assert is_process_alive(pid) finally: os.kill(pid, signal.SIGKILL) def test_stop_tunnel(config_dir, sample_tunnel, monkeypatch, tmp_path): fake = tmp_path / "autossh" fake.write_text("#!/bin/sh\nexec sleep 60\n") fake.chmod(0o755) monkeypatch.setenv("PATH", f"{tmp_path}:{os.environ['PATH']}") pid = start_tunnel(config_dir, sample_tunnel) assert is_process_alive(pid) stop_tunnel(config_dir, "web-service") time.sleep(0.2) assert not is_process_alive(pid) assert read_pid_file(config_dir, "web-service") is None def test_stop_tunnel_stale_pid(config_dir): write_pid_file(config_dir, "web-service", 9999999, datetime.now(timezone.utc)) stop_tunnel(config_dir, "web-service") # must not raise assert read_pid_file(config_dir, "web-service") is None ``` - [ ] **Step 2: Run tests to confirm they fail** ```bash uv run pytest tests/unit/test_process.py::test_build_autossh_cmd_basic tests/unit/test_process.py::test_start_tunnel_no_autossh -v ``` Expected: `ImportError` for `build_autossh_cmd`, `start_tunnel`, `stop_tunnel`. - [ ] **Step 3: Implement start/stop in `src/autossh_mgr/process.py`** (append to existing file) ```python import shutil import signal import socket import subprocess import time import click from .config import TunnelConfig def build_autossh_cmd(tunnel: TunnelConfig) -> list[str]: cmd = [ "autossh", "-M", "0", "-N", "-o", "ServerAliveInterval=30", "-o", "ServerAliveCountMax=3", "-i", os.path.expanduser(tunnel.identity_file), "-p", str(tunnel.port), "-R", ( f"{tunnel.remote_host}:{tunnel.remote_port}" f":{tunnel.local_host}:{tunnel.local_port}" ), f"{tunnel.user}@{tunnel.host}", ] if tunnel.ssh_options: for opt in tunnel.ssh_options.split(): cmd.append(opt) return cmd def _is_port_in_use(port: int) -> bool: with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s: try: s.bind(("127.0.0.1", port)) return False except OSError: return True def start_tunnel(config_dir: Path, tunnel: TunnelConfig) -> int: if not shutil.which("autossh"): raise click.ClickException("autossh not found. Install it first.") if _is_port_in_use(tunnel.local_port): raise click.ClickException(f"Port {tunnel.local_port} is already in use") log_path = config_dir / "logs" / f"{tunnel.name}.log" env = os.environ.copy() env["AUTOSSH_GATETIME"] = "0" with open(log_path, "a") as log: proc = subprocess.Popen( build_autossh_cmd(tunnel), stdin=subprocess.DEVNULL, stdout=log, stderr=log, env=env, ) started_at = datetime.now(timezone.utc) write_pid_file(config_dir, tunnel.name, proc.pid, started_at) time.sleep(1) if not is_process_alive(proc.pid): delete_pid_file(config_dir, tunnel.name) raise click.ClickException( f"autossh process for '{tunnel.name}' exited immediately. " f"Check logs: {log_path}" ) return proc.pid def stop_tunnel(config_dir: Path, name: str) -> None: result = read_pid_file(config_dir, name) if result is None: return pid, _ = result if not is_process_alive(pid): delete_pid_file(config_dir, name) return try: os.kill(pid, signal.SIGTERM) except ProcessLookupError: delete_pid_file(config_dir, name) return for _ in range(50): time.sleep(0.1) if not is_process_alive(pid): break else: try: os.kill(pid, signal.SIGKILL) except ProcessLookupError: pass delete_pid_file(config_dir, name) ``` - [ ] **Step 4: Run all process tests** ```bash uv run pytest tests/unit/test_process.py -v ``` Expected: all tests PASS. - [ ] **Step 5: Commit** ```bash git add src/autossh_mgr/process.py tests/unit/test_process.py git commit -m "feat: add tunnel start/stop with PID file tracking" ``` --- ### Task 5: SSH connectivity check (`check.py`) **Files:** - Create: `src/autossh_mgr/check.py` - Create: `tests/unit/test_check.py` - [ ] **Step 1: Write failing tests** `tests/unit/test_check.py`: ```python import subprocess from unittest.mock import patch, MagicMock from autossh_mgr.check import check_connectivity, build_ssh_check_cmd from autossh_mgr.config import TunnelConfig @pytest.fixture def tunnel(): return TunnelConfig( name="web", host="relay.example.com", user="deploy", local_port=8080, remote_port=18080, ) def test_build_ssh_check_cmd(tunnel): cmd = build_ssh_check_cmd(tunnel) assert cmd[0] == "ssh" assert "-o" in cmd and "BatchMode=yes" in cmd assert "ConnectTimeout=5" in " ".join(cmd) assert f"{tunnel.user}@{tunnel.host}" in cmd assert "true" in cmd assert "-p" in cmd and str(tunnel.port) in cmd def test_check_connectivity_success(tunnel): mock_result = MagicMock() mock_result.returncode = 0 mock_result.stderr = "" with patch("subprocess.run", return_value=mock_result) as mock_run: success, msg = check_connectivity(tunnel) assert success is True assert msg == "" called_cmd = mock_run.call_args[0][0] assert called_cmd == build_ssh_check_cmd(tunnel) def test_check_connectivity_failure(tunnel): mock_result = MagicMock() mock_result.returncode = 255 mock_result.stderr = "Connection refused" with patch("subprocess.run", return_value=mock_result): success, msg = check_connectivity(tunnel) assert success is False assert "Connection refused" in msg ``` - [ ] **Step 2: Run tests to confirm they fail** ```bash uv run pytest tests/unit/test_check.py -v ``` Expected: `ImportError`. - [ ] **Step 3: Implement `src/autossh_mgr/check.py`** ```python import os import subprocess from autossh_mgr.config import TunnelConfig def build_ssh_check_cmd(tunnel: TunnelConfig) -> list[str]: return [ "ssh", "-o", "BatchMode=yes", "-o", "ConnectTimeout=5", "-i", os.path.expanduser(tunnel.identity_file), "-p", str(tunnel.port), f"{tunnel.user}@{tunnel.host}", "true", ] def check_connectivity(tunnel: TunnelConfig) -> tuple[bool, str]: result = subprocess.run( build_ssh_check_cmd(tunnel), capture_output=True, text=True, ) return result.returncode == 0, result.stderr.strip() ``` - [ ] **Step 4: Run tests to confirm they pass** ```bash uv run pytest tests/unit/test_check.py -v ``` Expected: all tests PASS. - [ ] **Step 5: Commit** ```bash git add src/autossh_mgr/check.py tests/unit/test_check.py git commit -m "feat: add SSH connectivity check" ``` --- ### Task 6: Display formatters (`display.py`) **Files:** - Create: `src/autossh_mgr/display.py` No separate unit tests — display output is verified through CLI tests in Tasks 7 and 8. - [ ] **Step 1: Implement `src/autossh_mgr/display.py`** ```python import click from autossh_mgr.config import TunnelConfig from autossh_mgr.process import TunnelStatus def _server_str(t: TunnelConfig) -> str: return f"{t.user}@{t.host}:{t.port}" def _ports_str(t: TunnelConfig) -> str: return f"{t.local_port} -> {t.remote_port}" def print_tunnel_table(tunnels: list[TunnelConfig], statuses: dict[str, TunnelStatus]) -> None: if not tunnels: click.echo("No tunnels configured.") return fmt = f"{'NAME':<20} {'SERVER':<32} {'PORTS':<15} STATUS" click.echo(fmt) click.echo("-" * 72) for t in tunnels: s = statuses.get(t.name) state = s.state if s else "unknown" click.echo(f"{t.name:<20} {_server_str(t):<32} {_ports_str(t):<15} {state}") def print_status_table(tunnels: list[TunnelConfig], statuses: dict[str, TunnelStatus]) -> None: if not tunnels: click.echo("No tunnels configured.") return fmt = f"{'NAME':<20} {'STATE':<10} {'PID':<8} UPTIME" click.echo(fmt) click.echo("-" * 52) for t in tunnels: s = statuses.get(t.name) if s: pid_str = str(s.pid) if s.pid else "-" uptime_str = s.uptime or "-" click.echo(f"{t.name:<20} {s.state:<10} {pid_str:<8} {uptime_str}") def print_tunnel_detail(tunnel: TunnelConfig, status: TunnelStatus | None) -> None: click.echo(f"Name: {tunnel.name}") click.echo(f"Host: {tunnel.host}") click.echo(f"SSH port: {tunnel.port}") click.echo(f"User: {tunnel.user}") click.echo(f"Identity file: {tunnel.identity_file}") click.echo(f"Local: {tunnel.local_host}:{tunnel.local_port}") click.echo(f"Remote: {tunnel.remote_host}:{tunnel.remote_port}") if tunnel.ssh_options: click.echo(f"SSH options: {tunnel.ssh_options}") if status: click.echo(f"State: {status.state}") if status.pid: click.echo(f"PID: {status.pid}") if status.uptime: click.echo(f"Uptime: {status.uptime}") ``` - [ ] **Step 2: Commit** ```bash git add src/autossh_mgr/display.py git commit -m "feat: add display formatting helpers" ``` --- ### Task 7: CLI — config management commands **Files:** - Create: `src/autossh_mgr/cli.py` - Create: `tests/unit/test_cli_config.py` Commands covered: `init`, `add`, `remove`, `list`, `show`, `config` - [ ] **Step 1: Write failing tests** `tests/unit/test_cli_config.py`: ```python import pytest from click.testing import CliRunner from autossh_mgr.cli import cli from autossh_mgr.config import load_tunnels, save_tunnels, TunnelConfig, ensure_dirs @pytest.fixture(autouse=True) def set_config_dir(config_dir, monkeypatch): monkeypatch.setenv("AUTOSSH_MGR_CONFIG_DIR", str(config_dir)) @pytest.fixture def runner(): return CliRunner() def test_init_creates_structure(runner, tmp_path, monkeypatch): new_dir = tmp_path / "new-config" monkeypatch.setenv("AUTOSSH_MGR_CONFIG_DIR", str(new_dir)) result = runner.invoke(cli, ["init"]) assert result.exit_code == 0 assert (new_dir / "tunnels.yaml").exists() assert (new_dir / "pids").is_dir() assert (new_dir / "logs").is_dir() def test_init_idempotent(runner): result = runner.invoke(cli, ["init"]) assert result.exit_code == 0 result = runner.invoke(cli, ["init"]) assert result.exit_code == 0 def test_add_non_interactive(runner, config_dir): result = runner.invoke(cli, [ "add", "web-service", "--host", "relay.example.com", "--user", "deploy", "--local-port", "8080", "--remote-port", "18080", ]) assert result.exit_code == 0, result.output tunnels = load_tunnels(config_dir) assert len(tunnels) == 1 assert tunnels[0].name == "web-service" assert tunnels[0].host == "relay.example.com" def test_add_duplicate_name_fails(runner, config_dir, sample_tunnel): save_tunnels(config_dir, [sample_tunnel]) result = runner.invoke(cli, [ "add", "web-service", "--host", "other.com", "--user", "u", "--local-port", "9090", "--remote-port", "19090", ]) assert result.exit_code != 0 assert "already exists" in result.output def test_remove_non_running(runner, config_dir, sample_tunnel): save_tunnels(config_dir, [sample_tunnel]) result = runner.invoke(cli, ["remove", "web-service"], input="y\n") assert result.exit_code == 0 assert load_tunnels(config_dir) == [] def test_remove_unknown_name(runner): result = runner.invoke(cli, ["remove", "missing"], input="y\n") assert result.exit_code != 0 assert "No tunnel named" in result.output def test_remove_aborted(runner, config_dir, sample_tunnel): save_tunnels(config_dir, [sample_tunnel]) result = runner.invoke(cli, ["remove", "web-service"], input="n\n") assert result.exit_code == 0 assert len(load_tunnels(config_dir)) == 1 def test_list_empty(runner): result = runner.invoke(cli, ["list"]) assert result.exit_code == 0 assert "No tunnels" in result.output def test_list_shows_tunnels(runner, config_dir, sample_tunnel): save_tunnels(config_dir, [sample_tunnel]) result = runner.invoke(cli, ["list"]) assert result.exit_code == 0 assert "web-service" in result.output assert "relay.example.com" in result.output def test_show_tunnel(runner, config_dir, sample_tunnel): save_tunnels(config_dir, [sample_tunnel]) result = runner.invoke(cli, ["show", "web-service"]) assert result.exit_code == 0 assert "relay.example.com" in result.output assert "8080" in result.output assert "18080" in result.output def test_config_update_field(runner, config_dir, sample_tunnel): save_tunnels(config_dir, [sample_tunnel]) result = runner.invoke(cli, ["config", "web-service", "remote_port", "19000"]) assert result.exit_code == 0 t = load_tunnels(config_dir)[0] assert t.remote_port == 19000 def test_config_unknown_key(runner, config_dir, sample_tunnel): save_tunnels(config_dir, [sample_tunnel]) result = runner.invoke(cli, ["config", "web-service", "bad_key", "val"]) assert result.exit_code != 0 assert "Unknown field" in result.output def test_config_rename_disallowed(runner, config_dir, sample_tunnel): save_tunnels(config_dir, [sample_tunnel]) result = runner.invoke(cli, ["config", "web-service", "name", "new-name"]) assert result.exit_code != 0 assert "Cannot rename" in result.output def test_config_invalid_type(runner, config_dir, sample_tunnel): save_tunnels(config_dir, [sample_tunnel]) result = runner.invoke(cli, ["config", "web-service", "port", "notanumber"]) assert result.exit_code != 0 ``` - [ ] **Step 2: Run tests to confirm they fail** ```bash uv run pytest tests/unit/test_cli_config.py -v ``` Expected: `ImportError` — `cli.py` does not exist. - [ ] **Step 3: Implement `src/autossh_mgr/cli.py`** (config management portion) ```python import click from pathlib import Path from autossh_mgr.config import ( TunnelConfig, get_config_dir, ensure_dirs, load_tunnels, save_tunnels, get_tunnel, ) from autossh_mgr.process import get_status from autossh_mgr.display import ( print_tunnel_table, print_tunnel_detail, print_status_table, ) _FIELD_TYPES: dict[str, type] = { "host": str, "port": int, "user": str, "identity_file": str, "local_host": str, "local_port": int, "remote_host": str, "remote_port": int, "ssh_options": str, } @click.group() @click.pass_context def cli(ctx: click.Context) -> None: ctx.ensure_object(dict) config_dir = get_config_dir() ensure_dirs(config_dir) ctx.obj["config_dir"] = config_dir @cli.command() @click.pass_context def init(ctx: click.Context) -> None: """Initialize config directory.""" config_dir = ctx.obj["config_dir"] ensure_dirs(config_dir) click.echo(f"Initialized {config_dir}") @cli.command() @click.argument("name") @click.option("--host", default=None) @click.option("--port", type=int, default=None) @click.option("--user", default=None) @click.option("--identity-file", default=None) @click.option("--local-host", default=None) @click.option("--local-port", type=int, default=None) @click.option("--remote-host", default=None) @click.option("--remote-port", type=int, default=None) @click.option("--ssh-options", default=None) @click.pass_context def add( ctx, name, host, port, user, identity_file, local_host, local_port, remote_host, remote_port, ssh_options, ): """Add a new tunnel configuration.""" config_dir = ctx.obj["config_dir"] tunnels = load_tunnels(config_dir) if any(t.name == name for t in tunnels): raise click.ClickException(f"Tunnel '{name}' already exists") if host is None: host = click.prompt("Public server host") if port is None: port = click.prompt("SSH port", default=22, type=int) if user is None: user = click.prompt("SSH user") if identity_file is None: identity_file = click.prompt("Identity file", default="~/.ssh/id_ed25519") if local_host is None: local_host = click.prompt("Local host", default="127.0.0.1") if local_port is None: local_port = click.prompt("Local port", type=int) if remote_host is None: remote_host = click.prompt("Remote host", default="0.0.0.0") if remote_port is None: remote_port = click.prompt("Remote port", type=int) if ssh_options is None: ssh_options = click.prompt("Extra SSH options (optional)", default="") tunnel = TunnelConfig( name=name, host=host, port=port, user=user, identity_file=identity_file, local_host=local_host, local_port=local_port, remote_host=remote_host, remote_port=remote_port, ssh_options=ssh_options, ) tunnels.append(tunnel) save_tunnels(config_dir, tunnels) click.echo(f"Added tunnel '{name}'") @cli.command() @click.argument("name") @click.pass_context def remove(ctx, name): """Remove a tunnel configuration.""" config_dir = ctx.obj["config_dir"] tunnel = get_tunnel(config_dir, name) status = get_status(config_dir, name) if status.state == "running": raise click.ClickException(f"Stop '{name}' before removing it") click.confirm(f"Remove tunnel '{name}'?", abort=True) tunnels = [t for t in load_tunnels(config_dir) if t.name != name] save_tunnels(config_dir, tunnels) click.echo(f"Removed tunnel '{name}'") @cli.command(name="list") @click.pass_context def list_cmd(ctx): """List all tunnel configurations.""" config_dir = ctx.obj["config_dir"] tunnels = load_tunnels(config_dir) statuses = {t.name: get_status(config_dir, t.name) for t in tunnels} print_tunnel_table(tunnels, statuses) @cli.command() @click.argument("name") @click.pass_context def show(ctx, name): """Show full details for a tunnel.""" config_dir = ctx.obj["config_dir"] tunnel = get_tunnel(config_dir, name) status = get_status(config_dir, name) print_tunnel_detail(tunnel, status) @cli.command(name="config") @click.argument("name") @click.argument("key") @click.argument("value") @click.pass_context def config_cmd(ctx, name, key, value): """Update a single tunnel configuration field.""" config_dir = ctx.obj["config_dir"] if key == "name": raise click.ClickException("Cannot rename a tunnel via config. Remove and re-add.") if key not in _FIELD_TYPES: raise click.ClickException(f"Unknown field '{key}'") try: typed_value = _FIELD_TYPES[key](value) except ValueError: raise click.ClickException( f"Invalid value for '{key}': expected {_FIELD_TYPES[key].__name__}" ) tunnels = load_tunnels(config_dir) found = False for t in tunnels: if t.name == name: setattr(t, key, typed_value) found = True break if not found: raise click.ClickException(f"No tunnel named '{name}'") save_tunnels(config_dir, tunnels) click.echo(f"Updated {name}.{key} = {typed_value}") ``` - [ ] **Step 4: Run tests to confirm they pass** ```bash uv run pytest tests/unit/test_cli_config.py -v ``` Expected: all tests PASS. - [ ] **Step 5: Commit** ```bash git add src/autossh_mgr/cli.py tests/unit/test_cli_config.py git commit -m "feat: add CLI config management commands (init, add, remove, list, show, config)" ``` --- ### Task 8: CLI — lifecycle commands **Files:** - Modify: `src/autossh_mgr/cli.py` - Create: `tests/unit/test_cli_lifecycle.py` Commands: `start`, `stop`, `restart`, `status`, `check` - [ ] **Step 1: Write failing tests** `tests/unit/test_cli_lifecycle.py`: ```python import os import signal import time import pytest from click.testing import CliRunner from unittest.mock import patch, MagicMock from autossh_mgr.cli import cli from autossh_mgr.config import save_tunnels, TunnelConfig, ensure_dirs from autossh_mgr.process import write_pid_file, read_pid_file from datetime import datetime, timezone @pytest.fixture(autouse=True) def set_config_dir(config_dir, monkeypatch): monkeypatch.setenv("AUTOSSH_MGR_CONFIG_DIR", str(config_dir)) @pytest.fixture def runner(): return CliRunner() @pytest.fixture def with_tunnel(config_dir, sample_tunnel): save_tunnels(config_dir, [sample_tunnel]) return sample_tunnel def test_start_already_running(runner, config_dir, with_tunnel): write_pid_file(config_dir, "web-service", os.getpid(), datetime.now(timezone.utc)) result = runner.invoke(cli, ["start", "web-service"]) assert result.exit_code == 0 assert "already running" in result.output def test_start_stale_pid_cleaned(runner, config_dir, with_tunnel, tmp_path, monkeypatch): fake = tmp_path / "autossh" fake.write_text("#!/bin/sh\nexec sleep 60\n") fake.chmod(0o755) monkeypatch.setenv("PATH", f"{tmp_path}:{os.environ['PATH']}") write_pid_file(config_dir, "web-service", 9999999, datetime.now(timezone.utc)) result = runner.invoke(cli, ["start", "web-service"]) assert result.exit_code == 0, result.output assert "stale" in result.output.lower() # Clean up launched process pid_data = read_pid_file(config_dir, "web-service") if pid_data: os.kill(pid_data[0], signal.SIGKILL) def test_start_unknown_tunnel(runner): result = runner.invoke(cli, ["start", "missing"]) assert result.exit_code != 0 assert "No tunnel named" in result.output def test_stop_running_tunnel(runner, config_dir, with_tunnel, tmp_path, monkeypatch): fake = tmp_path / "autossh" fake.write_text("#!/bin/sh\nexec sleep 60\n") fake.chmod(0o755) monkeypatch.setenv("PATH", f"{tmp_path}:{os.environ['PATH']}") runner.invoke(cli, ["start", "web-service"]) result = runner.invoke(cli, ["stop", "web-service"]) assert result.exit_code == 0 assert read_pid_file(config_dir, "web-service") is None def test_stop_already_stopped(runner, with_tunnel): result = runner.invoke(cli, ["stop", "web-service"]) assert result.exit_code == 0 assert "not running" in result.output def test_status_all_stopped(runner, config_dir, with_tunnel): result = runner.invoke(cli, ["status"]) assert result.exit_code == 0 assert "web-service" in result.output assert "stopped" in result.output def test_status_single(runner, config_dir, with_tunnel): result = runner.invoke(cli, ["status", "web-service"]) assert result.exit_code == 0 assert "stopped" in result.output assert "relay.example.com" in result.output def test_check_success(runner, with_tunnel): with patch("autossh_mgr.check.subprocess.run") as mock_run: mock_run.return_value = MagicMock(returncode=0, stderr="") result = runner.invoke(cli, ["check", "web-service"]) assert result.exit_code == 0 assert "OK" in result.output or "success" in result.output.lower() def test_check_failure(runner, with_tunnel): with patch("autossh_mgr.check.subprocess.run") as mock_run: mock_run.return_value = MagicMock(returncode=255, stderr="Connection refused") result = runner.invoke(cli, ["check", "web-service"]) assert result.exit_code != 0 assert "Connection refused" in result.output ``` - [ ] **Step 2: Run tests to confirm they fail** ```bash uv run pytest tests/unit/test_cli_lifecycle.py -v ``` Expected: failures due to missing `start`, `stop`, `restart`, `status`, `check` commands. - [ ] **Step 3: Implement lifecycle commands** — add these imports to the **top** of `src/autossh_mgr/cli.py` (alongside the existing imports), then append the commands below to the end of the file. Additional imports to add at top of `cli.py`: ```python from autossh_mgr.check import check_connectivity from autossh_mgr.process import ( start_tunnel, stop_tunnel, check_stale, delete_pid_file, ) ``` Commands to append to `cli.py`: @cli.command() @click.argument("name", required=False) @click.pass_context def start(ctx, name): """Start a tunnel (or all tunnels if no name given).""" config_dir = ctx.obj["config_dir"] tunnels = load_tunnels(config_dir) targets = ( [get_tunnel(config_dir, name)] if name else sorted(tunnels, key=lambda t: t.name) ) for tunnel in targets: status = get_status(config_dir, tunnel.name) if status.state == "running": click.echo(f"{tunnel.name} already running (pid: {status.pid})") continue if status.state == "stale": click.echo(f"{tunnel.name}: stale pid detected, cleaning up") delete_pid_file(config_dir, tunnel.name) pid = start_tunnel(config_dir, tunnel) click.echo(f"Started {tunnel.name} (pid: {pid})") @cli.command() @click.argument("name", required=False) @click.pass_context def stop(ctx, name): """Stop a tunnel (or all running tunnels if no name given).""" config_dir = ctx.obj["config_dir"] tunnels = load_tunnels(config_dir) targets = ( [get_tunnel(config_dir, name)] if name else sorted(tunnels, key=lambda t: t.name) ) for tunnel in targets: status = get_status(config_dir, tunnel.name) if status.state == "stopped": click.echo(f"{tunnel.name} is not running") continue stop_tunnel(config_dir, tunnel.name) click.echo(f"Stopped {tunnel.name}") @cli.command() @click.argument("name", required=False) @click.pass_context def restart(ctx, name): """Restart a tunnel (or all tunnels if no name given).""" ctx.invoke(stop, name=name) ctx.invoke(start, name=name) @cli.command() @click.argument("name", required=False) @click.pass_context def status(ctx, name): """Show tunnel status (all tunnels if no name given).""" config_dir = ctx.obj["config_dir"] if name: tunnel = get_tunnel(config_dir, name) s = get_status(config_dir, name) print_tunnel_detail(tunnel, s) else: tunnels = load_tunnels(config_dir) statuses = {t.name: get_status(config_dir, t.name) for t in tunnels} print_status_table(tunnels, statuses) @cli.command(name="check") @click.argument("name") @click.pass_context def check_cmd(ctx, name): """Check SSH connectivity to a tunnel's server.""" config_dir = ctx.obj["config_dir"] tunnel = get_tunnel(config_dir, name) success, msg = check_connectivity(tunnel) if success: click.echo(f"OK: {tunnel.user}@{tunnel.host}:{tunnel.port} is reachable") else: raise click.ClickException(f"Connection failed: {msg}") ``` - [ ] **Step 4: Run all lifecycle tests** ```bash uv run pytest tests/unit/test_cli_lifecycle.py -v ``` Expected: all tests PASS. - [ ] **Step 5: Run full unit test suite** ```bash uv run pytest tests/unit/ -v ``` Expected: all tests PASS. - [ ] **Step 6: Commit** ```bash git add src/autossh_mgr/cli.py tests/unit/test_cli_lifecycle.py git commit -m "feat: add CLI lifecycle commands (start, stop, restart, status, check)" ``` --- ### Task 9: Integration tests **Files:** - Create: `tests/integration/test_lifecycle.py` These tests use a real subprocess substituting `autossh` with a dummy `sleep 60` script. They exercise the full PID file lifecycle end-to-end. - [ ] **Step 1: Write integration tests** `tests/integration/test_lifecycle.py`: ```python import os import signal import time import pytest from click.testing import CliRunner from autossh_mgr.cli import cli from autossh_mgr.config import save_tunnels, TunnelConfig, ensure_dirs from autossh_mgr.process import read_pid_file, is_process_alive @pytest.fixture def fake_autossh(tmp_path, monkeypatch): """Replace autossh with a long-running dummy process.""" fake = tmp_path / "autossh" fake.write_text("#!/bin/sh\nexec sleep 60\n") fake.chmod(0o755) monkeypatch.setenv("PATH", f"{tmp_path}:{os.environ['PATH']}") return fake @pytest.fixture def config_dir(tmp_path): ensure_dirs(tmp_path) return tmp_path @pytest.fixture(autouse=True) def set_config_dir(config_dir, monkeypatch): monkeypatch.setenv("AUTOSSH_MGR_CONFIG_DIR", str(config_dir)) @pytest.fixture def tunnel(config_dir): t = TunnelConfig( name="test-tunnel", host="relay.example.com", user="deploy", local_port=19999, remote_port=29999, ) save_tunnels(config_dir, [t]) return t @pytest.fixture def runner(): return CliRunner() def test_start_writes_pid_file(runner, config_dir, tunnel, fake_autossh): result = runner.invoke(cli, ["start", "test-tunnel"]) assert result.exit_code == 0, result.output pid_data = read_pid_file(config_dir, "test-tunnel") assert pid_data is not None pid, started_at = pid_data assert is_process_alive(pid) # cleanup os.kill(pid, signal.SIGKILL) def test_stop_kills_process(runner, config_dir, tunnel, fake_autossh): runner.invoke(cli, ["start", "test-tunnel"]) pid_data = read_pid_file(config_dir, "test-tunnel") assert pid_data is not None pid = pid_data[0] assert is_process_alive(pid) result = runner.invoke(cli, ["stop", "test-tunnel"]) assert result.exit_code == 0 time.sleep(0.2) assert not is_process_alive(pid) assert read_pid_file(config_dir, "test-tunnel") is None def test_status_shows_running(runner, config_dir, tunnel, fake_autossh): runner.invoke(cli, ["start", "test-tunnel"]) result = runner.invoke(cli, ["status", "test-tunnel"]) assert result.exit_code == 0 assert "running" in result.output pid_data = read_pid_file(config_dir, "test-tunnel") if pid_data: os.kill(pid_data[0], signal.SIGKILL) def test_stale_pid_cleaned_on_start(runner, config_dir, tunnel, fake_autossh): from autossh_mgr.process import write_pid_file from datetime import datetime, timezone write_pid_file(config_dir, "test-tunnel", 9999999, datetime.now(timezone.utc)) result = runner.invoke(cli, ["start", "test-tunnel"]) assert result.exit_code == 0, result.output assert "stale" in result.output.lower() pid_data = read_pid_file(config_dir, "test-tunnel") assert pid_data is not None assert pid_data[0] != 9999999 os.kill(pid_data[0], signal.SIGKILL) def test_restart(runner, config_dir, tunnel, fake_autossh): runner.invoke(cli, ["start", "test-tunnel"]) first_pid = read_pid_file(config_dir, "test-tunnel")[0] result = runner.invoke(cli, ["restart", "test-tunnel"]) assert result.exit_code == 0, result.output second_pid = read_pid_file(config_dir, "test-tunnel")[0] assert second_pid != first_pid assert is_process_alive(second_pid) os.kill(second_pid, signal.SIGKILL) def test_idempotent_start(runner, config_dir, tunnel, fake_autossh): runner.invoke(cli, ["start", "test-tunnel"]) first_pid = read_pid_file(config_dir, "test-tunnel")[0] result = runner.invoke(cli, ["start", "test-tunnel"]) assert result.exit_code == 0 assert "already running" in result.output assert read_pid_file(config_dir, "test-tunnel")[0] == first_pid os.kill(first_pid, signal.SIGKILL) def test_idempotent_stop(runner, tunnel): result = runner.invoke(cli, ["stop", "test-tunnel"]) assert result.exit_code == 0 assert "not running" in result.output ``` - [ ] **Step 2: Run integration tests** ```bash uv run pytest tests/integration/ -v ``` Expected: all tests PASS. - [ ] **Step 3: Run full test suite** ```bash uv run pytest -v ``` Expected: all tests PASS. - [ ] **Step 4: Smoke-test the installed CLI** ```bash uv run autossh-mgr --help uv run autossh-mgr init uv run autossh-mgr add test --host example.com --user deploy --local-port 8080 --remote-port 18080 uv run autossh-mgr list uv run autossh-mgr show test uv run autossh-mgr config test port 2222 uv run autossh-mgr show test uv run autossh-mgr remove test ``` Expected: each command runs without errors and output is sensible. - [ ] **Step 5: Commit** ```bash git add tests/integration/test_lifecycle.py git commit -m "test: add integration tests for full tunnel lifecycle" ```