fix: remove incorrect port-in-use check for reverse tunnels

The `_is_port_in_use` bind check was wrong for `-R` tunnels: local_port is
the forward target (e.g. a Docker container), so it should be listening,
not free. Drop the check — autossh doesn't require the target to be
reachable at startup.

Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>

src/autossh_mgr/cli.py

@@ -0,0 +1,247 @@
+import click
+from dataclasses import replace
+from autossh_mgr.config import (
+    TunnelConfig, get_config_dir, ensure_dirs,
+    load_tunnels, save_tunnels, get_tunnel,
+)
+from autossh_mgr.check import check_connectivity
+from autossh_mgr.process import (
+    get_status, start_tunnel, stop_tunnel, delete_pid_file,
+)
+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 user is None:
+        user = click.prompt("SSH user")
+    if local_port is None:
+        local_port = click.prompt("Local port", type=int)
+    if remote_port is None:
+        remote_port = click.prompt("Remote port", type=int)
+    if port is None:
+        port = 22
+    if identity_file is None:
+        identity_file = "~/.ssh/id_ed25519"
+    if local_host is None:
+        local_host = "127.0.0.1"
+    if remote_host is None:
+        remote_host = "0.0.0.0"
+    if ssh_options is None:
+        ssh_options = ""
+
+    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"]
+    tunnels = load_tunnels(config_dir)
+    if not any(t.name == name for t in tunnels):
+        raise click.ClickException(f"No tunnel named '{name}'")
+    status = get_status(config_dir, name)
+    if status.state == "running":
+        raise click.ClickException(f"Stop '{name}' before removing it")
+    if not click.confirm(f"Remove tunnel '{name}'?"):
+        return
+    delete_pid_file(config_dir, name)
+    save_tunnels(config_dir, [t for t in tunnels if t.name != name])
+    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)
+    if not any(t.name == name for t in tunnels):
+        raise click.ClickException(f"No tunnel named '{name}'")
+    tunnels = [replace(t, **{key: typed_value}) if t.name == name else t for t in tunnels]
+    save_tunnels(config_dir, tunnels)
+    click.echo(f"Updated {name}.{key} = {typed_value}")
+
+
+@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)
+        if status.state == "stale":
+            click.echo(f"{tunnel.name}: cleaned up stale pid")
+        else:
+            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}")

src/autossh_mgr/display.py

@@ -0,0 +1,57 @@
+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}")

src/autossh_mgr/process.py

@@ -1,7 +1,7 @@
 import os
+import shlex
 import shutil
 import signal
-import socket
 import subprocess
 import time
 from dataclasses import dataclass
@@ -109,25 +109,14 @@ def build_autossh_cmd(tunnel: TunnelConfig) -> list[str]:
         f"{tunnel.user}@{tunnel.host}",
     ]
     if tunnel.ssh_options:
-        for opt in tunnel.ssh_options.split():
+        for opt in shlex.split(tunnel.ssh_options):
             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"
     log_path.parent.mkdir(parents=True, exist_ok=True)

tests/integration/test_lifecycle.py

@@ -0,0 +1,135 @@
+import os
+import signal
+import time
+import pytest
+from datetime import datetime, timezone
+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, write_pid_file
+
+
+@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(autouse=True)
+def cleanup_processes(config_dir):
+    """Kill any processes recorded in PID files after each test."""
+    yield
+    pids_dir = config_dir / "pids"
+    if not pids_dir.exists():
+        return
+    for pid_file in pids_dir.glob("*.pid"):
+        try:
+            lines = pid_file.read_text().splitlines()
+            if lines:
+                os.kill(int(lines[0]), signal.SIGKILL)
+        except (ValueError, OSError, ProcessLookupError):
+            pass
+
+
+@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, _ = pid_data
+    assert is_process_alive(pid)
+
+
+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
+
+
+def test_stale_pid_cleaned_on_start(runner, config_dir, tunnel, fake_autossh):
+    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
+
+
+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)
+    assert not is_process_alive(first_pid)
+
+
+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
+
+
+def test_idempotent_stop(runner, tunnel):
+    result = runner.invoke(cli, ["stop", "test-tunnel"])
+    assert result.exit_code == 0
+    assert "not running" in result.output

tests/unit/test_check.py

@@ -1,5 +1,4 @@
 import pytest
-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

tests/unit/test_cli_config.py

@@ -0,0 +1,140 @@
+import os
+import pytest
+from datetime import datetime, timezone
+from click.testing import CliRunner
+from autossh_mgr.cli import cli
+from autossh_mgr.config import load_tunnels, save_tunnels
+from autossh_mgr.process import write_pid_file
+
+
+@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
+
+
+def test_remove_running_tunnel_fails(runner, config_dir, sample_tunnel):
+    save_tunnels(config_dir, [sample_tunnel])
+    write_pid_file(config_dir, "web-service", os.getpid(), datetime.now(timezone.utc))
+    result = runner.invoke(cli, ["remove", "web-service"], input="y\n")
+    assert result.exit_code != 0
+    assert "Stop" in result.output
+    assert len(load_tunnels(config_dir)) == 1

tests/unit/test_cli_lifecycle.py

@@ -0,0 +1,161 @@
+import os
+import signal
+import pytest
+from click.testing import CliRunner
+from unittest.mock import patch, MagicMock
+from autossh_mgr.cli import cli
+from autossh_mgr.config import TunnelConfig, save_tunnels
+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
+
+
+def test_restart_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"])
+    first_pid = read_pid_file(config_dir, "web-service")[0]
+
+    result = runner.invoke(cli, ["restart", "web-service"])
+    assert result.exit_code == 0, result.output
+
+    second_pid_data = read_pid_file(config_dir, "web-service")
+    assert second_pid_data is not None
+    assert second_pid_data[0] != first_pid
+
+    os.kill(second_pid_data[0], signal.SIGKILL)
+
+
+def test_start_all(runner, config_dir, 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']}")
+
+    t1 = TunnelConfig(name="alpha", host="h.com", user="u", local_port=8001, remote_port=18001)
+    t2 = TunnelConfig(name="beta", host="h.com", user="u", local_port=8002, remote_port=18002)
+    save_tunnels(config_dir, [t1, t2])
+
+    result = runner.invoke(cli, ["start"])
+    assert result.exit_code == 0, result.output
+    assert "alpha" in result.output
+    assert "beta" in result.output
+
+    for name in ["alpha", "beta"]:
+        pid_data = read_pid_file(config_dir, name)
+        if pid_data:
+            os.kill(pid_data[0], signal.SIGKILL)
+
+
+def test_stop_all(runner, config_dir, 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']}")
+
+    t1 = TunnelConfig(name="alpha", host="h.com", user="u", local_port=8001, remote_port=18001)
+    t2 = TunnelConfig(name="beta", host="h.com", user="u", local_port=8002, remote_port=18002)
+    save_tunnels(config_dir, [t1, t2])
+
+    runner.invoke(cli, ["start"])
+    result = runner.invoke(cli, ["stop"])
+    assert result.exit_code == 0, result.output
+    assert "alpha" in result.output
+    assert "beta" in result.output
+    assert read_pid_file(config_dir, "alpha") is None
+    assert read_pid_file(config_dir, "beta") is None