116 lines
4.5 KiB
Python
116 lines
4.5 KiB
Python
import asyncio
|
|
import logging
|
|
import unittest
|
|
from pathlib import Path
|
|
from tempfile import TemporaryDirectory
|
|
from types import SimpleNamespace
|
|
from unittest.mock import AsyncMock, patch
|
|
|
|
from app.danmu_queue import BgiLogMonitor, WebServer
|
|
|
|
|
|
class BgiLogDisplayFormatTests(unittest.TestCase):
|
|
def test_formats_multi_instance_log_as_single_readable_line(self):
|
|
raw_lines = [
|
|
"[23:55:25.160] [INF] [Primary:S1:P16720:T1785426828552] "
|
|
"BetterGenshinImpact.GameTask.TaskTriggerDispatcher\n",
|
|
"游戏已退出，BetterGI 自动停止截图器\n",
|
|
"\n",
|
|
]
|
|
|
|
self.assertEqual(
|
|
BgiLogMonitor.format_display_lines(raw_lines),
|
|
["[23:55:25] 游戏已退出，BetterGI 自动停止截图器"],
|
|
)
|
|
|
|
def test_formats_legacy_log_header_the_same_way(self):
|
|
raw_lines = [
|
|
"[16:39:12.328] [INF] BetterGenshinImpact.GameTask.TaskTriggerDispatcher\n",
|
|
"游戏已退出，BetterGI 自动停止截图器\n",
|
|
]
|
|
|
|
self.assertEqual(
|
|
BgiLogMonitor.format_display_lines(raw_lines),
|
|
["[16:39:12] 游戏已退出，BetterGI 自动停止截图器"],
|
|
)
|
|
|
|
def test_preserves_multiline_message_inside_one_log_entry(self):
|
|
raw_lines = [
|
|
"[10:00:00.001] [ERR] [Primary:S1:P1:T2] BetterGI.Component\n",
|
|
"第一行\n",
|
|
"第二行\n",
|
|
"[10:00:01.999] [INF] BetterGI.OtherComponent\n",
|
|
"下一条\n",
|
|
]
|
|
|
|
self.assertEqual(
|
|
BgiLogMonitor.format_display_lines(raw_lines),
|
|
["[10:00:00] 第一行\n第二行", "[10:00:01] 下一条"],
|
|
)
|
|
|
|
def test_frontend_reader_falls_back_to_latest_log_after_midnight(self):
|
|
with TemporaryDirectory() as temp_dir:
|
|
log_dir = Path(temp_dir) / "log"
|
|
log_dir.mkdir()
|
|
(log_dir / "better-genshin-impact20260730.log").write_text(
|
|
"[23:55:25.160] [INF] [Primary:S1:P16720:T1785426828552] "
|
|
"BetterGenshinImpact.GameTask.TaskTriggerDispatcher\n"
|
|
"游戏已退出，BetterGI 自动停止截图器\n",
|
|
encoding="utf-8",
|
|
)
|
|
server = WebServer.__new__(WebServer)
|
|
server.config = SimpleNamespace(bettergi_work_dir=temp_dir)
|
|
|
|
with patch("app.danmu_queue.datetime") as mocked_datetime:
|
|
mocked_datetime.now.return_value.strftime.return_value = "20260731"
|
|
result = server._read_bgi_log(50)
|
|
|
|
self.assertEqual(
|
|
result,
|
|
["[23:55:25] 游戏已退出，BetterGI 自动停止截图器"],
|
|
)
|
|
|
|
|
|
class BgiLogCompletionTests(unittest.IsolatedAsyncioTestCase):
|
|
async def test_configuration_group_marker_still_finishes_current_group(self):
|
|
monitor = BgiLogMonitor(".", logging.getLogger("test-bgi-group-log"))
|
|
callback = AsyncMock()
|
|
monitor.set_finish_callback(callback)
|
|
with patch.object(monitor, "_get_current_log_size", return_value=0):
|
|
monitor.set_current_group("薄荷", "group-run")
|
|
|
|
monitor._process_line('配置组 "薄荷" 执行结束')
|
|
await asyncio.sleep(0)
|
|
|
|
callback.assert_awaited_once_with("薄荷", "group-run")
|
|
self.assertIsNone(monitor._current_group)
|
|
|
|
async def test_one_dragon_marker_finishes_managed_daily(self):
|
|
monitor = BgiLogMonitor(".", logging.getLogger("test-bgi-one-dragon-log"))
|
|
callback = AsyncMock()
|
|
monitor.set_finish_callback(callback)
|
|
with patch.object(monitor, "_get_current_log_size", return_value=0):
|
|
monitor.set_current_one_dragon("自动每日（委托）", "daily-run")
|
|
|
|
monitor._process_line("一条龙和配置组任务结束")
|
|
await asyncio.sleep(0)
|
|
|
|
callback.assert_awaited_once_with("自动每日（委托）", "daily-run")
|
|
self.assertIsNone(monitor._current_group)
|
|
|
|
async def test_child_group_completion_does_not_finish_one_dragon_early(self):
|
|
monitor = BgiLogMonitor(".", logging.getLogger("test-bgi-child-group-log"))
|
|
callback = AsyncMock()
|
|
monitor.set_finish_callback(callback)
|
|
with patch.object(monitor, "_get_current_log_size", return_value=0):
|
|
monitor.set_current_one_dragon("自动每日", "daily-run")
|
|
|
|
monitor._process_line('配置组 "每日委托" 执行结束')
|
|
await asyncio.sleep(0)
|
|
|
|
callback.assert_not_awaited()
|
|
self.assertEqual(monitor._current_group, "自动每日")
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|