Files

tech e6fc5db85a Add autossh-mgr implementation plan

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

2026-05-20 21:34:04 +08:00

48 KiB

Raw Blame History

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

[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

mkdir -p src/autossh_mgr tests/unit tests/integration

src/autossh_mgr/__init__.py — empty file.

src/autossh_mgr/__main__.py:

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

uv add --dev pytest
uv sync
uv run python -c "import autossh_mgr"

Expected: no errors.

Step 4: Commit

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)

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:

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

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

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

uv run pytest tests/unit/test_config.py -v

Expected: all tests PASS.

Step 6: Commit

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:

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

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

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

uv run pytest tests/unit/test_process.py -v

Expected: all tests PASS.

Step 5: Commit

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)

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

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)

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

uv run pytest tests/unit/test_process.py -v

Expected: all tests PASS.

Step 5: Commit

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:

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

uv run pytest tests/unit/test_check.py -v

Expected: ImportError.

Step 3: Implement src/autossh_mgr/check.py

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

uv run pytest tests/unit/test_check.py -v

Expected: all tests PASS.

Step 5: Commit

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

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

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:

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

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)

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

uv run pytest tests/unit/test_cli_config.py -v

Expected: all tests PASS.

Step 5: Commit

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:

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

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:

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

uv run pytest tests/unit/ -v

Expected: all tests PASS.

Step 6: Commit

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:

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

uv run pytest tests/integration/ -v

Expected: all tests PASS.

Step 3: Run full test suite

uv run pytest -v

Expected: all tests PASS.

Step 4: Smoke-test the installed CLI

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

git add tests/integration/test_lifecycle.py
git commit -m "test: add integration tests for full tunnel lifecycle"

48 KiB Raw Blame History

autossh-mgr Implementation Plan

File Map

Task 1: Project scaffolding

Task 2: TunnelConfig dataclass + YAML I/O (config.py)

Task 3: PID file utilities and status detection (process.py part 1)

Task 4: Tunnel start and stop (process.py part 2)

Task 5: SSH connectivity check (check.py)

Task 6: Display formatters (display.py)

Task 7: CLI — config management commands

Task 8: CLI — lifecycle commands

Task 9: Integration tests

48 KiB

Raw Blame History

Task 2: TunnelConfig dataclass + YAML I/O (`config.py`)

Task 3: PID file utilities and status detection (`process.py` part 1)

Task 4: Tunnel start and stop (`process.py` part 2)

Task 5: SSH connectivity check (`check.py`)

Task 6: Display formatters (`display.py`)