From 6416f8a8e2a00303efbd7370915d109ee86a4e66 Mon Sep 17 00:00:00 2001 From: tech Date: Thu, 21 May 2026 12:10:37 +0800 Subject: [PATCH] feat: add PID file utilities and tunnel status detection Co-Authored-By: Claude Sonnet 4.6 --- src/autossh_mgr/process.py | 87 ++++++++++++++++++++++++++++++++++ tests/unit/test_process.py | 97 ++++++++++++++++++++++++++++++++++++++ 2 files changed, 184 insertions(+) create mode 100644 src/autossh_mgr/process.py create mode 100644 tests/unit/test_process.py diff --git a/src/autossh_mgr/process.py b/src/autossh_mgr/process.py new file mode 100644 index 0000000..b480a17 --- /dev/null +++ b/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)) diff --git a/tests/unit/test_process.py b/tests/unit/test_process.py new file mode 100644 index 0000000..531fe08 --- /dev/null +++ b/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"