diff --git a/docs/superpowers/plans/2026-05-20-autossh-mgr.md b/docs/superpowers/plans/2026-05-20-autossh-mgr.md new file mode 100644 index 0000000..35acd55 --- /dev/null +++ b/docs/superpowers/plans/2026-05-20-autossh-mgr.md @@ -0,0 +1,1689 @@ +# 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" +```