85 lines
2.9 KiB
Python
85 lines
2.9 KiB
Python
from __future__ import annotations
|
|
|
|
import unittest
|
|
|
|
from src.process_runtime_status import (
|
|
coerce_declared_process_status,
|
|
normalize_log_text,
|
|
summarize_process_activity,
|
|
tail_log_text,
|
|
)
|
|
|
|
|
|
class ProcessRuntimeStatusTests(unittest.TestCase):
|
|
def test_normalize_and_tail_log_text_handles_carriage_returns(self) -> None:
|
|
raw = "line1\rline2\r\nline3\nline4"
|
|
self.assertEqual(normalize_log_text(raw), "line1\nline2\nline3\nline4")
|
|
self.assertEqual(tail_log_text(raw, 2), "line3\nline4\n")
|
|
|
|
def test_summarize_process_activity_marks_streaming_when_ffmpeg_is_active(self) -> None:
|
|
state = summarize_process_activity(
|
|
"youtube",
|
|
"online",
|
|
"监测 1 | 格式 YOUTUBE",
|
|
"",
|
|
log_age_seconds=3.0,
|
|
ffmpeg_pids=[1234],
|
|
)
|
|
self.assertEqual(state["business_status"], "streaming")
|
|
self.assertTrue(state["is_pushing"])
|
|
|
|
def test_summarize_process_activity_marks_waiting_source(self) -> None:
|
|
state = summarize_process_activity(
|
|
"youtube",
|
|
"online",
|
|
"等待直播...\n检测直播间中...",
|
|
"",
|
|
log_age_seconds=4.0,
|
|
ffmpeg_pids=[],
|
|
)
|
|
self.assertEqual(state["business_status"], "waiting_source")
|
|
self.assertFalse(state["is_pushing"])
|
|
|
|
def test_summarize_process_activity_marks_misconfigured(self) -> None:
|
|
state = summarize_process_activity(
|
|
"youtube",
|
|
"online",
|
|
"",
|
|
"请在 config/youtube.ini 中配置 key=你的stream_key",
|
|
log_age_seconds=2.0,
|
|
ffmpeg_pids=[],
|
|
)
|
|
self.assertEqual(state["business_status"], "misconfigured")
|
|
|
|
def test_summarize_process_activity_marks_stale_when_heartbeat_is_old(self) -> None:
|
|
state = summarize_process_activity(
|
|
"youtube",
|
|
"online",
|
|
"监测 1 | 格式 YOUTUBE",
|
|
"",
|
|
log_age_seconds=360.0,
|
|
ffmpeg_pids=[],
|
|
)
|
|
self.assertEqual(state["business_status"], "stale")
|
|
|
|
def test_summarize_process_activity_marks_configured_process_as_ready(self) -> None:
|
|
state = summarize_process_activity(
|
|
"youtube",
|
|
"configured",
|
|
"",
|
|
"",
|
|
log_age_seconds=None,
|
|
ffmpeg_pids=[],
|
|
)
|
|
self.assertEqual(state["business_status"], "ready")
|
|
self.assertEqual(state["business_note"], "Process is configured but not running")
|
|
|
|
def test_coerce_declared_process_status_marks_local_missing_process_as_configured(self) -> None:
|
|
self.assertEqual(coerce_declared_process_status("local", "not_found", True), "configured")
|
|
self.assertEqual(coerce_declared_process_status("pm2", "not_found", True), "not_found")
|
|
self.assertEqual(coerce_declared_process_status("local", "online", True), "online")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|