import asyncio import ctypes import logging import tempfile import unittest from datetime import datetime, timedelta from pathlib import Path from types import SimpleNamespace from unittest.mock import AsyncMock, MagicMock, patch from app.danmu_queue import CommandHandler, QueueSystem, SystemScheduler class _FakeConfig: def __init__(self, system_cfg): self.system_cfg = system_cfg class RebootAfterStopTests(unittest.IsolatedAsyncioTestCase): def setUp(self): self.logger = logging.getLogger("test-system-scheduler") self.scheduler = SystemScheduler( _FakeConfig({ "reboot_after_stop_enabled": True, "reboot_after_stop_delay_sec": 60, }), self.logger, ) @patch("app.danmu_queue.os.name", "nt") @patch("app.danmu_queue.subprocess.Popen") def test_schedules_one_minute_reboot_only_once(self, popen): stopped_at = datetime(2026, 7, 23, 23, 0) self.assertTrue(self.scheduler._schedule_reboot_after_stop(stopped_at)) self.assertFalse(self.scheduler._schedule_reboot_after_stop(stopped_at)) popen.assert_called_once() command = popen.call_args.args[0] self.assertEqual(command[:5], ["shutdown", "/r", "/t", "60", "/c"]) async def test_failed_stop_still_schedules_reboot(self): self.scheduler._live_occurrences = MagicMock( return_value=[("stop", datetime.now().replace(second=0, microsecond=0))] ) self.scheduler.stop_bilibili_live = AsyncMock(return_value=False) self.scheduler._schedule_reboot_after_stop = MagicMock() self.scheduler._end_live_statistics = MagicMock() async def stop_after_first_sleep(_seconds): self.scheduler._stop = True with patch("app.danmu_queue.asyncio.sleep", side_effect=stop_after_first_sleep): await self.scheduler.run() self.scheduler._schedule_reboot_after_stop.assert_called_once() async def test_stop_exception_still_schedules_reboot(self): self.scheduler._live_occurrences = MagicMock( return_value=[("stop", datetime.now().replace(second=0, microsecond=0))] ) self.scheduler.stop_bilibili_live = AsyncMock(side_effect=RuntimeError("click failed")) self.scheduler._schedule_reboot_after_stop = MagicMock() self.scheduler._end_live_statistics = MagicMock() async def stop_after_first_sleep(_seconds): self.scheduler._stop = True with patch("app.danmu_queue.asyncio.sleep", side_effect=stop_after_first_sleep): await self.scheduler.run() self.scheduler._schedule_reboot_after_stop.assert_called_once() class StartupLiveCompensationTests(unittest.IsolatedAsyncioTestCase): def setUp(self): self.logger = logging.getLogger("test-startup-live-compensation") self.scheduler = SystemScheduler( _FakeConfig({ "live_start_time": "09:00", "live_end_time": "23:00", "launch_bilibili_live_enabled": True, "launch_genshin_enabled": True, "bilibili_push_enabled": True, }), self.logger, ) self.start_at = datetime.now() - timedelta(hours=1) self.end_at = datetime.now() + timedelta(hours=1) self.scheduler._startup_window_active = MagicMock(return_value=True) self.scheduler._is_exe_running = MagicMock(return_value=False) self.scheduler._launch_exe = AsyncMock(side_effect=[True, True]) self.scheduler.push_bilibili_live = AsyncMock(return_value=True) async def test_launches_apps_after_one_minute_then_pushes_after_another(self): with patch("app.danmu_queue.asyncio.sleep", new_callable=AsyncMock) as sleep: await self.scheduler._run_startup_live_compensation(self.start_at, self.end_at) self.assertEqual( [call.args[0] for call in sleep.await_args_list], [60, 60], ) self.assertEqual( [call.args for call in self.scheduler._launch_exe.await_args_list], [("B站直播姬", "bilibili_live_exe"), ("原神", "genshin_exe")], ) self.scheduler.push_bilibili_live.assert_awaited_once() async def test_skips_push_when_livehime_was_already_running(self): self.scheduler._is_exe_running = MagicMock(return_value=True) self.scheduler._launch_exe = AsyncMock(side_effect=[False, False]) with patch("app.danmu_queue.asyncio.sleep", new_callable=AsyncMock): await self.scheduler._run_startup_live_compensation(self.start_at, self.end_at) self.scheduler.push_bilibili_live.assert_not_awaited() async def test_cancels_before_push_when_live_window_ends(self): self.scheduler._startup_window_active = MagicMock(side_effect=[True, False]) with patch("app.danmu_queue.asyncio.sleep", new_callable=AsyncMock): await self.scheduler._run_startup_live_compensation(self.start_at, self.end_at) self.scheduler.push_bilibili_live.assert_not_awaited() async def test_scheduling_marks_prepare_and_start_to_avoid_duplicate_clicks(self): with patch.object(self.scheduler, "_run_startup_live_compensation", new_callable=AsyncMock): scheduled = self.scheduler._schedule_startup_live_compensation(self.start_at, self.end_at) await asyncio.sleep(0) if self.scheduler._startup_compensation_task: await self.scheduler._startup_compensation_task self.assertTrue(scheduled) self.assertFalse(self.scheduler._schedule_startup_live_compensation(self.start_at, self.end_at)) self.assertIn( self.scheduler._event_key("prepare_tts", self.start_at - timedelta(minutes=15)), self.scheduler._triggered_events, ) self.assertIn( self.scheduler._event_key("prepare", self.start_at - timedelta(minutes=10)), self.scheduler._triggered_events, ) self.assertIn( self.scheduler._event_key("start", self.start_at), self.scheduler._triggered_events, ) def test_current_live_window_supports_cross_midnight_schedule(self): self.scheduler.config.system_cfg["live_start_time"] = "23:00" self.scheduler.config.system_cfg["live_end_time"] = "02:00" current = datetime(2026, 7, 31, 1, 0) window = self.scheduler._current_live_window(current) self.assertEqual( window, (datetime(2026, 7, 30, 23, 0), datetime(2026, 7, 31, 2, 0)), ) class LivehimePushVerificationTests(unittest.IsolatedAsyncioTestCase): def setUp(self): self.logger = logging.getLogger("test-livehime-push-verification") self.scheduler = SystemScheduler( _FakeConfig({ "bilibili_push_window_keyword": "直播姬", "bilibili_push_click_x_ratio": 0.787, "bilibili_push_click_y_ratio": 0.927, "bilibili_stop_push_click_x_ratio": 0.787, "bilibili_stop_push_click_y_ratio": 0.927, }), self.logger, ) async def test_push_skips_click_when_bilibili_is_already_live(self): self.scheduler._get_bilibili_live_status = AsyncMock(return_value=1) self.scheduler._click_bilibili_live = AsyncMock() self.assertTrue(await self.scheduler.push_bilibili_live()) self.scheduler._click_bilibili_live.assert_not_awaited() async def test_push_clicks_then_waits_for_confirmed_live_status(self): self.scheduler._get_bilibili_live_status = AsyncMock(side_effect=[0, 0, 1]) self.scheduler._click_bilibili_live = AsyncMock(return_value=True) with patch("app.danmu_queue.asyncio.sleep", new_callable=AsyncMock): result = await self.scheduler.push_bilibili_live() self.assertTrue(result) self.scheduler._click_bilibili_live.assert_awaited_once() async def test_stop_skips_click_when_bilibili_is_already_offline(self): self.scheduler._get_bilibili_live_status = AsyncMock(return_value=0) self.scheduler._click_bilibili_live = AsyncMock() self.assertTrue(await self.scheduler.stop_bilibili_live()) self.scheduler._click_bilibili_live.assert_not_awaited() async def test_live_status_timeout_returns_specific_error(self): self.scheduler._get_bilibili_live_status = AsyncMock(return_value=0) with patch("app.danmu_queue.asyncio.sleep", new_callable=AsyncMock): result = await self.scheduler._wait_for_bilibili_live_status( 1, timeout_seconds=0.01, ) self.assertFalse(result) self.assertIn("未确认开播", self.scheduler.last_live_action_error) @patch("app.danmu_queue.os.name", "nt") async def test_click_restores_and_focuses_before_reading_physical_window_rect(self): hwnd = 12345 user32 = MagicMock() call_order = [] user32.IsIconic.return_value = True user32.GetForegroundWindow.return_value = hwnd user32.GetSystemMetrics.side_effect = lambda index: { 76: 0, 77: 0, 78: 2560, 79: 1440, }[index] user32.SetCursorPos.return_value = True user32.GetCursorPos.return_value = False user32.SetThreadDpiAwarenessContext.return_value = None def show_window(_hwnd, _command): call_order.append("restore") return True def get_window_rect(_hwnd, pointer): call_order.append("rect") rect = ctypes.cast(pointer, ctypes.POINTER(ctypes.wintypes.RECT)).contents rect.left = -11 rect.top = -11 rect.right = 2571 rect.bottom = 1379 return True user32.ShowWindow.side_effect = show_window user32.GetWindowRect.side_effect = get_window_rect self.scheduler._find_window_by_keyword = MagicMock(return_value=hwnd) self.scheduler._force_foreground_window = MagicMock(return_value=True) with ( patch("app.danmu_queue.ctypes.windll.user32", user32), patch("app.danmu_queue.asyncio.sleep", new_callable=AsyncMock), ): result = await self.scheduler._click_bilibili_live( x_ratio_key="bilibili_push_click_x_ratio", y_ratio_key="bilibili_push_click_y_ratio", action_name="开启推流按钮", ) self.assertTrue(result) self.assertLess(call_order.index("restore"), call_order.index("rect")) user32.SetCursorPos.assert_called_once_with(2021, 1277) self.assertEqual(user32.mouse_event.call_count, 2) class ManualLiveOverrideTests(unittest.TestCase): def setUp(self): self.system = QueueSystem.__new__(QueueSystem) self.system.logger = logging.getLogger("test-manual-live-override") self.system._manual_live_override_until_monotonic = 0.0 self.system._manual_live_override_until_wall = 0.0 self.system._bilibili_live_status = None self.system._bilibili_live_status_updated_monotonic = 0.0 self.system._bilibili_live_status_updated_wall = 0.0 def test_manual_live_override_allows_commands_outside_schedule(self): handler = CommandHandler.__new__(CommandHandler) handler.system = self.system handler.config = SimpleNamespace( system_cfg={"live_start_time": "08:30", "live_end_time": "23:30"} ) with patch("app.danmu_queue.time.monotonic", return_value=100.0): self.system.enable_manual_live_override(300) self.assertTrue(handler._is_live_time()) snapshot = self.system.manual_live_override_snapshot() self.assertTrue(snapshot["active"]) self.assertGreater(snapshot["until"], 0) def test_clear_manual_live_override_restores_schedule_gate(self): handler = CommandHandler.__new__(CommandHandler) handler.system = self.system handler.config = SimpleNamespace( system_cfg={"live_start_time": "08:30", "live_end_time": "23:30"} ) with ( patch("app.danmu_queue.time.monotonic", return_value=100.0), patch("app.danmu_queue.datetime") as mocked_datetime, ): mocked_datetime.now.return_value = datetime(2026, 8, 16, 3, 30) mocked_datetime.strptime = datetime.strptime self.system.enable_manual_live_override(300) self.system.clear_manual_live_override() self.assertFalse(handler._is_live_time()) def test_confirmed_bilibili_live_status_allows_commands_outside_schedule(self): handler = CommandHandler.__new__(CommandHandler) handler.system = self.system handler.config = SimpleNamespace( system_cfg={"live_start_time": "08:30", "live_end_time": "23:30"} ) with ( patch("app.danmu_queue.time.monotonic", return_value=100.0), patch("app.danmu_queue.datetime") as mocked_datetime, ): mocked_datetime.now.return_value = datetime(2026, 8, 16, 3, 30) mocked_datetime.strptime = datetime.strptime self.system.set_bilibili_live_status(1) self.assertTrue(handler._is_live_time()) class StartupTaskSelfHealTests(unittest.TestCase): def setUp(self): self.logger = logging.getLogger("test-startup-self-heal") self.scheduler = SystemScheduler( _FakeConfig({ "enable_startup_shortcut": True, "startup_bat": "run.bat", }), self.logger, ) @patch("app.danmu_queue.os.name", "nt") @patch("app.danmu_queue.project_path") def test_rebuilds_when_task_missing(self, project_path_mock): with tempfile.TemporaryDirectory() as root: root_path = Path(root) (root_path / "run.bat").touch() (root_path / ".venv-tts" / "Scripts").mkdir(parents=True) (root_path / ".venv-tts" / "Scripts" / "pythonw.exe").touch() (root_path / "app").mkdir() (root_path / "app" / "main.py").touch() project_path_mock.side_effect = lambda relative: root_path / relative self.scheduler._startup_task_exists = MagicMock(return_value=False) self.scheduler._create_startup_task = MagicMock(return_value=True) self.scheduler.ensure_startup_task_self_heal() self.scheduler._create_startup_task.assert_called_once() args = self.scheduler._create_startup_task.call_args.args self.assertEqual(args[0], "BetterGI弹幕排队系统") # 无窗口模式：pythonw 直接运行 main.py，不再弹 cmd 窗口 self.assertIn(str(root_path / "app" / "main.py"), args[1]) self.assertIn("--role all", args[1]) self.assertIn(f"--port {self.scheduler.port}", args[1]) @patch("app.danmu_queue.os.name", "nt") @patch("app.danmu_queue.project_path") def test_skips_when_task_present(self, project_path_mock): project_path_mock.return_value = MagicMock( exists=MagicMock(return_value=True), __str__=lambda _s: r"C:\proj\run.bat", ) self.scheduler._startup_task_exists = MagicMock(return_value=True) self.scheduler._create_startup_task = MagicMock() self.scheduler.ensure_startup_task_self_heal() self.scheduler._create_startup_task.assert_not_called() @patch("app.danmu_queue.os.name", "nt") def test_skips_when_disabled(self): self.scheduler.config.system_cfg["enable_startup_shortcut"] = False self.scheduler._startup_task_exists = MagicMock() self.scheduler._create_startup_task = MagicMock() self.scheduler.ensure_startup_task_self_heal() self.scheduler._startup_task_exists.assert_not_called() self.scheduler._create_startup_task.assert_not_called() class ScriptGroupSyncTests(unittest.TestCase): def setUp(self): self.logger = logging.getLogger("test-scriptgroup-sync") self.scheduler = SystemScheduler( _FakeConfig({}), self.logger, ) def test_copies_and_overwrites_files(self): with tempfile.TemporaryDirectory() as src_dir, tempfile.TemporaryDirectory() as bgi_dir: src = Path(src_dir) (src / "薄荷.json").write_text('{"v": 2}', encoding="utf-8") (src / "晶蝶.json").write_text("{}", encoding="utf-8") dst = Path(bgi_dir) / "User" / "ScriptGroup" dst.mkdir(parents=True) (dst / "薄荷.json").write_text('{"v": 1, "corrupt": true}', encoding="utf-8") (dst / "本地多余.json").write_text("{}", encoding="utf-8") self.scheduler.config.bettergi_work_dir = bgi_dir self.scheduler.config.system_cfg["scriptgroup_backup_dir"] = src_dir self.scheduler._sync_script_groups() self.assertEqual( (dst / "薄荷.json").read_text(encoding="utf-8"), '{"v": 2}' ) self.assertTrue((dst / "晶蝶.json").exists()) # 目标目录里多出的文件不删除 self.assertTrue((dst / "本地多余.json").exists()) def test_missing_source_dir_is_skipped(self): with tempfile.TemporaryDirectory() as bgi_dir: self.scheduler.config.bettergi_work_dir = bgi_dir self.scheduler.config.system_cfg["scriptgroup_backup_dir"] = ( str(Path(bgi_dir) / "不存在") ) # 不抛异常即通过 self.scheduler._sync_script_groups() if __name__ == "__main__": unittest.main()