feat: add PID file utilities and tunnel status detection

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

src/autossh_mgr/process.py

@@ -0,0 +1,87 @@
+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))

tests/unit/test_process.py

@@ -0,0 +1,97 @@
+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"