32 lines
1.2 KiB
Python
32 lines
1.2 KiB
Python
import os
|
|
from pathlib import Path
|
|
import subprocess
|
|
import sys
|
|
|
|
|
|
def test_watchdog_exits_a_process_with_stalled_event_loop(tmp_path):
|
|
# A blocked event loop cannot run coroutine-based timeout handlers.
|
|
code = """import time
|
|
from src.core.health import start_watchdog
|
|
start_watchdog(grace_seconds=0.1, interval=0.02)
|
|
time.sleep(10)
|
|
"""
|
|
result = subprocess.run([sys.executable, "-c", code], cwd=Path(__file__).resolve().parents[1],
|
|
env=dict(os.environ, HEARTBEAT_FILE=str(tmp_path / "absent")),
|
|
capture_output=True, timeout=8)
|
|
assert result.returncode == 1
|
|
assert b"Heartbeat expired" in result.stderr
|
|
|
|
|
|
def test_stopped_watchdog_allows_normal_shutdown(tmp_path):
|
|
code = """import time
|
|
from src.core.health import start_watchdog
|
|
stopped = start_watchdog(grace_seconds=0.1, interval=0.02)
|
|
stopped.set()
|
|
time.sleep(0.2)
|
|
"""
|
|
result = subprocess.run([sys.executable, "-c", code], cwd=Path(__file__).resolve().parents[1],
|
|
env=dict(os.environ, HEARTBEAT_FILE=str(tmp_path / "absent")),
|
|
capture_output=True, timeout=8)
|
|
assert result.returncode == 0
|