's'

d2ypp2/tests/test_process_runtime_status.py

@@ -0,0 +1,84 @@
+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()