62 lines
3.1 KiB
Python
62 lines
3.1 KiB
Python
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import tempfile
|
|
import unittest
|
|
from pathlib import Path
|
|
from unittest.mock import AsyncMock, patch
|
|
|
|
from src.hub_routes import configure_hub, fallback_live_event_lines, hub_dashboard
|
|
|
|
|
|
class HubRoutesTests(unittest.TestCase):
|
|
def test_fallback_live_event_lines_reads_local_process_logs(self) -> None:
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
out_path = Path(tmp) / "youtube-out.log"
|
|
err_path = Path(tmp) / "youtube-error.log"
|
|
out_path.write_text("booting\nstream ready\n", encoding="utf-8")
|
|
err_path.write_text("warning line\ntraceback line\n", encoding="utf-8")
|
|
rows = [
|
|
{
|
|
"label": "YouTube",
|
|
"pm2": "youtube",
|
|
"out_path": str(out_path),
|
|
"err_path": str(err_path),
|
|
}
|
|
]
|
|
lines = fallback_live_event_lines(rows, max_lines=4)
|
|
self.assertEqual(len(lines), 4)
|
|
self.assertTrue(lines[0].startswith("[YouTube stderr]"))
|
|
self.assertTrue(lines[-1].endswith("stream ready"))
|
|
|
|
def test_hub_dashboard_marks_local_declared_process_as_configured(self) -> None:
|
|
backend = type("StubBackend", (), {})()
|
|
backend.mode = "local"
|
|
backend.ensure_mode = AsyncMock()
|
|
backend.process_infos = AsyncMock(
|
|
return_value={"youtube": {"process_status": "not_found", "out_path": None, "err_path": None, "pid": 0}}
|
|
)
|
|
backend.parse_pm2_describe = AsyncMock(return_value={"process_status": "not_found", "out_path": None, "err_path": None})
|
|
backend.live_ffmpeg_pids = AsyncMock(return_value=[])
|
|
configure_hub(backend, [{"pm2": "youtube", "label": "YouTube", "script": "youtube.py"}])
|
|
|
|
with patch("src.hub_routes.system_snapshot", return_value={}):
|
|
with patch("src.hub_routes.query_services", new=AsyncMock(return_value=[])):
|
|
with patch("src.hub_routes.adb_available", return_value=False):
|
|
with patch("src.hub_routes.stack_summary", return_value={}):
|
|
with patch("src.hub_routes.stack_stream_probes", new=AsyncMock(return_value={})):
|
|
with patch("src.hub_routes.recent_pm2_log_lines", return_value=[]):
|
|
with patch("src.hub_routes.network_iface_stats", return_value={}):
|
|
with patch("src.hub_routes.load_hardware_profile", new=AsyncMock(return_value={})):
|
|
with patch("src.hub_routes.redroid_runtime_summary", new=AsyncMock(return_value={})):
|
|
with patch("src.hub_routes.hdmi_capture_summary", new=AsyncMock(return_value={})):
|
|
payload = asyncio.run(hub_dashboard())
|
|
|
|
row = payload["live"]["processes"][0]
|
|
self.assertEqual(row["process_status"], "configured")
|
|
self.assertEqual(row["business_status"], "ready")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|