From 35507c12383857a225c7dc298359f4af9f7ab19a Mon Sep 17 00:00:00 2001 From: tech Date: Thu, 28 May 2026 17:49:56 +0800 Subject: [PATCH] feat: add daemon module with cross-platform process lifecycle Co-Authored-By: Claude Opus 4.7 --- src/jianda_proxy/daemon.py | 129 +++++++++++++++++++++++++++++++++++++ tests/test_daemon.py | 96 +++++++++++++++++++++++++++ 2 files changed, 225 insertions(+) create mode 100644 src/jianda_proxy/daemon.py create mode 100644 tests/test_daemon.py diff --git a/src/jianda_proxy/daemon.py b/src/jianda_proxy/daemon.py new file mode 100644 index 0000000..3b3819f --- /dev/null +++ b/src/jianda_proxy/daemon.py @@ -0,0 +1,129 @@ +import os +import sys +import time +import signal +from jianda_proxy.config import get_pid_path + + +def read_pid(): + path = get_pid_path() + if not os.path.exists(path): + return None + try: + with open(path) as f: + return int(f.read().strip()) + except (ValueError, OSError): + return None + + +def write_pid(pid): + path = get_pid_path() + os.makedirs(os.path.dirname(path), exist_ok=True) + with open(path, "w") as f: + f.write(str(pid)) + + +def remove_pid(): + path = get_pid_path() + if os.path.exists(path): + os.remove(path) + + +def is_process_alive(pid): + if os.name == "nt": + import ctypes + import ctypes.wintypes + + kernel32 = ctypes.windll.kernel32 + PROCESS_QUERY_LIMITED_INFORMATION = 0x1000 + STILL_ACTIVE = 259 + handle = kernel32.OpenProcess(PROCESS_QUERY_LIMITED_INFORMATION, False, pid) + if not handle: + return False + try: + exit_code = ctypes.wintypes.DWORD() + kernel32.GetExitCodeProcess(handle, ctypes.byref(exit_code)) + return exit_code.value == STILL_ACTIVE + finally: + kernel32.CloseHandle(handle) + else: + try: + os.kill(pid, 0) + return True + except ProcessLookupError: + return False + except PermissionError: + return True + + +def is_running(): + pid = read_pid() + if pid is None: + return False + return is_process_alive(pid) + + +def stop_daemon(): + pid = read_pid() + if pid is None: + return "not_running" + + if not is_process_alive(pid): + remove_pid() + return "not_running" + + if os.name == "nt": + os.system(f"taskkill /PID {pid} /T") + else: + try: + os.kill(pid, signal.SIGTERM) + except ProcessLookupError: + remove_pid() + return "stopped" + + for _ in range(6): + time.sleep(0.5) + if not is_process_alive(pid): + remove_pid() + return "stopped" + + if os.name == "nt": + os.system(f"taskkill /F /PID {pid} /T") + else: + try: + os.kill(pid, signal.SIGKILL) + except ProcessLookupError: + pass + + remove_pid() + return "stopped" + + +def get_status(): + pid = read_pid() + if pid is None or not is_process_alive(pid): + return {"running": False, "pid": None} + return {"running": True, "pid": pid} + + +def daemonize(): + if os.name == "nt": + import subprocess + + creation_flags = subprocess.CREATE_NEW_PROCESS_GROUP | subprocess.DETACHED_PROCESS + subprocess.Popen( + [sys.executable] + sys.argv + ["--_daemon_child"], + creationflags=creation_flags, + close_fds=True, + stdin=subprocess.DEVNULL, + stdout=subprocess.DEVNULL, + stderr=subprocess.DEVNULL, + ) + else: + pid = os.fork() + if pid > 0: + os._exit(0) + os.setsid() + pid = os.fork() + if pid > 0: + os._exit(0) diff --git a/tests/test_daemon.py b/tests/test_daemon.py new file mode 100644 index 0000000..eaf492c --- /dev/null +++ b/tests/test_daemon.py @@ -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