autossh-mgr/tests/unit/test_process.py

import os
import pytest
from datetime import datetime, timezone, timedelta
from autossh_mgr.process import (
    write_pid_file, read_pid_file, delete_pid_file,
    is_process_alive, check_stale, get_status, format_uptime,
)


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"


def test_read_pid_file_corrupt(config_dir):
    import click
    pid_path = config_dir / "pids" / "web-service.pid"
    pid_path.write_text("only-one-line\n")
    with pytest.raises(click.ClickException, match="Corrupt PID file"):
        read_pid_file(config_dir, "web-service")


# ---------------------------------------------------------------------------
# Task 4: build_autossh_cmd, start_tunnel, stop_tunnel
# ---------------------------------------------------------------------------

import shutil
import signal
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 "-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