import asyncio import json import logging import tempfile import unittest from pathlib import Path from types import SimpleNamespace from unittest.mock import AsyncMock, MagicMock, call, patch from app.bettergi_current_party import ( CurrentPartyReadError, CurrentPartyReadResult, PreparedCurrentPartyRead, ) from app.bettergi_daily import ( DailyAutomationError, DailyRequest, JsonUpdate, PreparedDailyRun, ) from app.danmu_queue import ( COMMAND_ALLOWED_ROLE_DEFAULTS, BetterGIRunner, CommandHandler, Config, build_public_command_hints, ) class DailyCommandIntegrationTests(unittest.IsolatedAsyncioTestCase): def setUp(self): self.temp_dir = tempfile.TemporaryDirectory() self.root = Path(self.temp_dir.name) self.config = SimpleNamespace( data={ "commands": { "run": {"enabled": True, "aliases": ["执行", "跑"]}, "daily": { "enabled": True, "aliases": ["自动每日", "每日", "日常", "做每日", "做日常"], }, "switch_party": { "enabled": True, "aliases": ["切换队伍", "更换队伍", "换队伍", "换队"], }, "edit_party": { "enabled": True, "aliases": ["修改队员", "更换队员", "换队员", "换角色"], }, } }, admin_uids=set(), bettergi_work_dir=str(self.root), daily_cfg={ "one_dragon_template": "默认配置", "managed_one_dragon_name": "直播系统自动每日", "ley_line_craft_resin_before": True, "commission_use_current_party": True, "current_party_read_timeout_sec": 45, }, broadcast_cfg={"tts_categories": {"execution": True}}, music_monitor_cfg={"request_player": {}}, ) self.queue_mgr = MagicMock() self.queue_mgr.state = { "current_admin_uid": 123, "login_status": "logged_in", "current_group": None, "default_running": False, } self.queue_mgr.is_admin.return_value = True self.queue_mgr.start_group.return_value = { "success": True, "msg": "ok", "run_id": "daily-run", } self.runner = MagicMock() self.runner.kill_bgi = AsyncMock() self.runner.start_groups = AsyncMock(return_value=True) self.runner.start_one_dragon = AsyncMock(return_value=True) self.runner.sync_js_script.return_value = True self.log_monitor = MagicMock() self.broadcaster = SimpleNamespace(broadcast=AsyncMock()) self.handler = CommandHandler( self.config, MagicMock(), self.queue_mgr, self.runner, self.log_monitor, MagicMock(), logging.getLogger("test-daily-command"), broadcaster=self.broadcaster, ) def tearDown(self): self.temp_dir.cleanup() def _prepared_daily(self, *, requires_current_party: bool = False) -> PreparedDailyRun: return PreparedDailyRun( request=DailyRequest(mode="commission" if requires_current_party else "none"), config_name="直播系统自动每日", updates=( JsonUpdate( self.root / "User" / "OneDragon" / "直播系统自动每日.json", {"Name": "直播系统自动每日"}, "test", ), ), requires_current_party=requires_current_party, ) async def test_daily_replaces_old_task_writes_config_and_starts_one_dragon(self): self.queue_mgr.state["current_group"] = "旧任务" prepared = self._prepared_daily() with ( patch("app.danmu_queue.prepare_daily_run", return_value=prepared) as prepare, patch("app.danmu_queue.apply_json_updates") as apply_updates, patch("app.danmu_queue.uuid.uuid4", return_value=SimpleNamespace(hex="daily-run")), ): await self.handler._cmd_daily(123, "测试用户", "") prepare.assert_called_once() self.queue_mgr.interrupt_group.assert_called_once_with( "daily_replaced", status="cancelled", ) self.runner.kill_bgi.assert_awaited_once_with(reason="自动每日配置更新") apply_updates.assert_called_once_with(prepared.updates) self.queue_mgr.start_group.assert_called_once_with( 123, "自动每日", run_id="daily-run", ) self.log_monitor.set_current_one_dragon.assert_called_once_with( "自动每日", "daily-run", ) self.runner.start_one_dragon.assert_awaited_once_with("直播系统自动每日") async def test_daily_validation_failure_does_not_stop_current_task(self): self.queue_mgr.state["current_group"] = "旧任务" with patch( "app.danmu_queue.prepare_daily_run", side_effect=DailyAutomationError("未找到一条龙模板"), ): await self.handler._cmd_daily(123, "测试用户", "秘境 风本") self.queue_mgr.interrupt_group.assert_not_called() self.runner.kill_bgi.assert_not_awaited() self.runner.start_one_dragon.assert_not_awaited() async def test_daily_commission_reads_current_party_before_starting(self): prepared = self._prepared_daily(requires_current_party=True) reader_update = JsonUpdate( self.root / "User" / "ScriptGroup" / "直播系统读取当前队伍.json", {"projects": []}, "reader", ) reader = PreparedCurrentPartyRead( request_id="daily-run", group_name="直播系统读取当前队伍", status_path=self.root / "status.json", updates=(reader_update,), ) party_update = JsonUpdate( self.root / "User" / "JsScript" / "AutoCommissionNova" / "Data" / "user-config.json", { "party": { "global": { "battleTeamName": "好感队", "elementTeamName": "好感队", } } }, "party", ) capture = AsyncMock(return_value="好感队") with ( patch("app.danmu_queue.prepare_daily_run", return_value=prepared) as prepare, patch("app.danmu_queue.prepare_current_party_read", return_value=reader), patch( "app.danmu_queue.prepare_commission_current_party_update", return_value=party_update, ) as prepare_party, patch("app.danmu_queue.apply_json_updates") as apply_updates, patch.object(self.handler, "_capture_current_party_name", capture), patch("app.danmu_queue.uuid.uuid4", return_value=SimpleNamespace(hex="daily-run")), ): await self.handler._cmd_daily(123, "测试用户", "委托") self.assertTrue(prepare.call_args.kwargs["commission_use_current_party"]) capture.assert_awaited_once_with(reader, 45) prepare_party.assert_called_once_with(str(self.root), "好感队") apply_updates.assert_called_once_with((party_update, *prepared.updates)) self.runner.start_one_dragon.assert_awaited_once_with("直播系统自动每日") async def test_daily_commission_reader_failure_does_not_start_billed_task(self): prepared = self._prepared_daily(requires_current_party=True) reader = PreparedCurrentPartyRead( request_id="daily-run", group_name="直播系统读取当前队伍", status_path=self.root / "status.json", updates=(), ) with ( patch("app.danmu_queue.prepare_daily_run", return_value=prepared), patch("app.danmu_queue.prepare_current_party_read", return_value=reader), patch.object( self.handler, "_capture_current_party_name", new=AsyncMock(side_effect=CurrentPartyReadError("未识别到队伍名称")), ), patch("app.danmu_queue.apply_json_updates") as apply_updates, patch("app.danmu_queue.uuid.uuid4", return_value=SimpleNamespace(hex="daily-run")), ): await self.handler._cmd_daily(123, "测试用户", "委托") apply_updates.assert_not_called() self.queue_mgr.start_group.assert_not_called() self.runner.start_one_dragon.assert_not_awaited() async def test_current_party_capture_syncs_group_and_waits_for_matching_status(self): reader_update = JsonUpdate( self.root / "User" / "ScriptGroup" / "直播系统读取当前队伍.json", {"projects": []}, "reader", ) reader = PreparedCurrentPartyRead( request_id="request-123", group_name="直播系统读取当前队伍", status_path=self.root / "status.json", updates=(reader_update,), ) with ( patch("app.danmu_queue.apply_json_updates") as apply_updates, patch( "app.danmu_queue.read_current_party_status", return_value=CurrentPartyReadResult("好感队", ("好感队",)), ), patch("app.danmu_queue.asyncio.sleep", new=AsyncMock()), ): party_name = await self.handler._capture_current_party_name(reader, 45) self.assertEqual(party_name, "好感队") self.runner.sync_js_script.assert_called_once_with("LiveCurrentParty") apply_updates.assert_called_once_with(reader.updates) self.runner.start_groups.assert_awaited_once_with(["直播系统读取当前队伍"]) async def test_daily_requires_confirmed_login_before_preparing_files(self): self.queue_mgr.state["login_status"] = "confirming" with patch("app.danmu_queue.prepare_daily_run") as prepare: await self.handler._cmd_daily(123, "测试用户", "") prepare.assert_not_called() self.runner.kill_bgi.assert_not_awaited() async def test_daily_start_failure_clears_monitor_and_enters_reset_flow(self): prepared = self._prepared_daily() self.runner.start_one_dragon.return_value = False reset = AsyncMock() self.handler.system = SimpleNamespace(_reset_wait_and_retry_login=reset) with ( patch("app.danmu_queue.prepare_daily_run", return_value=prepared), patch("app.danmu_queue.apply_json_updates"), patch("app.danmu_queue.uuid.uuid4", return_value=SimpleNamespace(hex="daily-run")), ): await self.handler._cmd_daily(123, "测试用户", "") self.log_monitor.set_current_one_dragon.assert_has_calls( [call("自动每日", "daily-run"), call(None)] ) reset.assert_awaited_once_with( 123, "测试用户", "自动每日启动失败,正在关闭原神并启动扫码上号", ) async def test_switch_party_updates_group_then_reuses_group_execution(self): update = JsonUpdate( self.root / "User" / "ScriptGroup" / "切换队伍.json", {"projects": []}, "test", ) with ( patch("app.danmu_queue.prepare_switch_party_update", return_value=update), patch("app.danmu_queue.apply_json_updates") as apply_updates, patch("app.danmu_queue.uuid.uuid4", return_value=SimpleNamespace(hex="party-run")), patch.object( self.handler, "_resolve_group_name", return_value={ "success": True, "name": "切换队伍", "matched": False, "suggestions": [], }, ), patch.object(self.handler, "_group_uses_nahida_collect", return_value=False), ): await self.handler._cmd_switch_party(123, "测试用户", "永冻队") apply_updates.assert_called_once_with((update,)) self.queue_mgr.start_group.assert_called_once_with( 123, "切换队伍", run_id="party-run", ) self.runner.start_groups.assert_awaited_once_with(["切换队伍"]) def test_execution_command_aliases_support_no_space_arguments(self): self.assertEqual(self.handler._split_command_text("执行薄荷"), ("执行", "薄荷")) self.assertEqual( self.handler._split_command_text("自动每日秘境 风本"), ("自动每日", "秘境 风本"), ) self.assertEqual( self.handler._split_command_text("更换队伍永冻队"), ("更换队伍", "永冻队"), ) self.assertEqual( self.handler._split_command_text("修改队员芙宁娜纳西妲钟离雷电将军"), ("修改队员", "芙宁娜纳西妲钟离雷电将军"), ) self.assertEqual( self.handler._split_command_text("日常秘境 风本"), ("日常", "秘境 风本"), ) self.assertEqual( self.handler._split_command_text("换队永冻队"), ("换队", "永冻队"), ) self.assertEqual( self.handler._split_command_text("换角色芙宁娜纳西妲钟离雷电将军"), ("换角色", "芙宁娜纳西妲钟离雷电将军"), ) def test_longest_new_alias_wins_for_contiguous_arguments(self): self.assertEqual( self.handler._split_command_text("换队员芙宁娜纳西妲钟离雷电将军"), ("换队员", "芙宁娜纳西妲钟离雷电将军"), ) async def test_confirmed_login_announces_daily_and_party_usage(self): self.queue_mgr.state["login_status"] = "confirming" await self.handler._cmd_confirm_login(123, "测试用户", "是") self.assertEqual(self.broadcaster.broadcast.await_count, 2) first, second = self.broadcaster.broadcast.await_args_list self.assertIn("发送“每日”完成基础每日", first.args[0]) self.assertTrue(first.kwargs["tts"]) self.assertIn("每日 秘境 风本", second.args[0]) self.assertIn("换队 永冻队", second.args[0]) self.assertIn("换角色 四名角色", second.args[0]) self.assertIn("执行 组名", second.args[0]) self.assertFalse(second.kwargs["tts"]) self.assertTrue(second.kwargs["danmu"]) def test_public_command_hints_are_sanitized_and_use_shortest_alias(self): config_data = { "commands": self.config.data["commands"], "bilibili": {"cookie": "secret"}, } hints = build_public_command_hints(config_data) self.assertEqual(set(hints), {"daily", "switch_party", "edit_party"}) self.assertEqual(hints["daily"]["preferred_alias"], "每日") self.assertEqual(hints["switch_party"]["preferred_alias"], "换队") self.assertEqual(hints["edit_party"]["preferred_alias"], "换角色") self.assertNotIn("bilibili", hints) def test_new_command_alias_defaults_are_complete(self): config_path = self.root / "config.json" config_path.write_text(json.dumps({}), encoding="utf-8") config = Config(str(config_path)) self.assertEqual( config.data["commands"]["daily"]["aliases"], ["自动每日", "每日", "日常", "做每日", "做日常"], ) self.assertEqual( config.data["commands"]["switch_party"]["aliases"], ["切换队伍", "更换队伍", "换队伍", "换队"], ) self.assertEqual( config.data["commands"]["edit_party"]["aliases"], ["修改队员", "更换队员", "换队员", "换角色"], ) def test_disabled_command_hint_has_no_public_alias(self): config_data = { "commands": { "daily": {"enabled": False, "aliases": ["每日"]}, } } hints = build_public_command_hints(config_data) self.assertFalse(hints["daily"]["enabled"]) self.assertEqual(hints["daily"]["aliases"], []) self.assertEqual(hints["daily"]["preferred_alias"], "") def test_new_command_permissions_match_run(self): for key in ("daily", "switch_party", "edit_party"): self.assertEqual( COMMAND_ALLOWED_ROLE_DEFAULTS[key], COMMAND_ALLOWED_ROLE_DEFAULTS["run"], ) class BetterGIRunnerOneDragonTests(unittest.IsolatedAsyncioTestCase): async def test_start_one_dragon_uses_supported_cli_shape(self): runner = BetterGIRunner( r"C:\BetterGI\BetterGI.exe", r"C:\BetterGI", logging.getLogger("test-one-dragon-runner"), ) create_process = AsyncMock(return_value=SimpleNamespace()) with ( patch("app.danmu_queue.asyncio.create_subprocess_exec", create_process), patch("app.danmu_queue.asyncio.sleep", new=AsyncMock()), ): result = await runner.start_one_dragon("直播系统自动每日") self.assertTrue(result) create_process.assert_awaited_once_with( r"C:\BetterGI\BetterGI.exe", "startOneDragon", "直播系统自动每日", cwd=r"C:\BetterGI", stdout=asyncio.subprocess.DEVNULL, stderr=asyncio.subprocess.DEVNULL, ) if __name__ == "__main__": unittest.main()