feat: add daemon module with cross-platform process lifecycle

Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
This commit is contained in:
2026-05-28 17:49:56 +08:00
parent db4a64319c
commit 35507c1238
2 changed files with 225 additions and 0 deletions

96
tests/test_daemon.py Normal file
View File

@@ -0,0 +1,96 @@
import os
import tempfile
import pytest
from unittest.mock import patch, MagicMock
from jianda_proxy.daemon import (
read_pid,
write_pid,
remove_pid,
is_process_alive,
is_running,
stop_daemon,
get_status,
)
class TestPidFile:
def test_write_and_read_pid(self, tmp_path, monkeypatch):
pid_file = str(tmp_path / "test.pid")
monkeypatch.setattr("jianda_proxy.daemon.get_pid_path", lambda: pid_file)
write_pid(12345)
assert read_pid() == 12345
def test_read_missing_pid_returns_none(self, tmp_path, monkeypatch):
pid_file = str(tmp_path / "nonexistent.pid")
monkeypatch.setattr("jianda_proxy.daemon.get_pid_path", lambda: pid_file)
assert read_pid() is None
def test_remove_pid(self, tmp_path, monkeypatch):
pid_file = str(tmp_path / "test.pid")
monkeypatch.setattr("jianda_proxy.daemon.get_pid_path", lambda: pid_file)
write_pid(12345)
remove_pid()
assert read_pid() is None
class TestProcessCheck:
@patch("os.kill")
def test_alive_on_unix(self, mock_kill):
mock_kill.return_value = None
assert is_process_alive(12345) is True
@patch("os.kill", side_effect=ProcessLookupError)
def test_dead_on_unix(self, mock_kill):
assert is_process_alive(12345) is False
@patch("os.kill", side_effect=PermissionError)
def test_permission_means_alive(self, mock_kill):
assert is_process_alive(12345) is True
class TestIsRunning:
@patch("jianda_proxy.daemon.is_process_alive", return_value=True)
@patch("jianda_proxy.daemon.read_pid", return_value=12345)
def test_running_when_pid_and_alive(self, mock_read, mock_alive):
assert is_running() is True
@patch("jianda_proxy.daemon.is_process_alive", return_value=False)
@patch("jianda_proxy.daemon.read_pid", return_value=12345)
def test_not_running_when_pid_dead(self, mock_read, mock_alive):
assert is_running() is False
@patch("jianda_proxy.daemon.read_pid", return_value=None)
def test_not_running_when_no_pid(self, mock_read):
assert is_running() is False
class TestStopDaemon:
@patch("jianda_proxy.daemon.is_process_alive", return_value=True)
@patch("jianda_proxy.daemon.read_pid", return_value=12345)
@patch("time.sleep")
@patch("os.kill")
def test_stop_sends_sigterm(self, mock_kill, mock_sleep, mock_read, mock_alive):
mock_alive.side_effect = [True, False]
stop_daemon()
mock_kill.assert_called_once_with(12345, 15) # SIGTERM=15
@patch("jianda_proxy.daemon.remove_pid")
@patch("jianda_proxy.daemon.is_process_alive", return_value=False)
@patch("jianda_proxy.daemon.read_pid", return_value=None)
def test_stop_noop_when_not_running(self, mock_read, mock_alive, mock_remove):
result = stop_daemon()
assert result == "not_running"
class TestGetStatus:
@patch("jianda_proxy.daemon.is_process_alive", return_value=True)
@patch("jianda_proxy.daemon.read_pid", return_value=12345)
def test_status_running(self, mock_read, mock_alive):
status = get_status()
assert status["running"] is True
assert status["pid"] == 12345
@patch("jianda_proxy.daemon.read_pid", return_value=None)
def test_status_not_running(self, mock_read):
status = get_status()
assert status["running"] is False