diff --git a/app/models/emulator.py b/app/models/emulator.py index a5a473e31f..ce2c950b7b 100644 --- a/app/models/emulator.py +++ b/app/models/emulator.py @@ -35,6 +35,10 @@ class EmulatorRuntimePolicy: resource_path: Path +class EmulatorStartupError(RuntimeError): + """模拟器的安全启动恢复已结束，调用方不得叠加任务重试。""" + + class DeviceStatus(IntEnum): ONLINE = 0 """设备在线""" diff --git a/app/task/BAAH/AutoProxy.py b/app/task/BAAH/AutoProxy.py index d6adc1088e..70efec686e 100644 --- a/app/task/BAAH/AutoProxy.py +++ b/app/task/BAAH/AutoProxy.py @@ -42,7 +42,7 @@ ) from app.core.ws import protocol from app.models.config import BAAHConfig, BAAHPlanConfig, BAAHUserConfig -from app.models.emulator import DeviceBase +from app.models.emulator import DeviceBase, EmulatorStartupError from app.models.notification import LocalNotice from app.models.schema import HistoryWrite, WSTaskNoticeData from app.models.task import LogRecord, ScriptItem, UserItem @@ -443,6 +443,8 @@ async def _ensure_emulator_online(self) -> bool: device_info = await self.emulator_manager.open( str(self.script_config.Emulator.Index) ) + except EmulatorStartupError: + raise except Exception as e: logger.opt(exception=True).warning(f"启动模拟器失败: {e}") await self.handle_pre_script_error("启动模拟器失败", e) diff --git a/app/task/MAA/AutoProxy.py b/app/task/MAA/AutoProxy.py index 06d82657e3..f312d8a463 100644 --- a/app/task/MAA/AutoProxy.py +++ b/app/task/MAA/AutoProxy.py @@ -38,7 +38,7 @@ MaaConfig, MaaUserConfig, ) -from app.models.emulator import DeviceBase, DeviceInfo +from app.models.emulator import DeviceBase, DeviceInfo, EmulatorStartupError from app.models.notification import LocalNotice, NotificationImage from app.models.schema import ( HistoryAggregationItem, @@ -1485,6 +1485,8 @@ async def main_task(self): **mod_avd_open_kwargs(self.emulator_manager, allowed=True), ) self.emulator_info = emulator_info + except EmulatorStartupError: + raise except Exception as e: logger.opt(exception=True).warning( f"账号: {self.cur_user_uid} - 模拟器启动失败: {e}" diff --git a/app/task/MaaEnd/AutoProxy.py b/app/task/MaaEnd/AutoProxy.py index 52215d4729..fcdb2b0acd 100644 --- a/app/task/MaaEnd/AutoProxy.py +++ b/app/task/MaaEnd/AutoProxy.py @@ -32,7 +32,7 @@ from app.core.history import aggregate_data from app.core.ws import protocol from app.models.config import MaaEndConfig, MaaEndUserConfig -from app.models.emulator import DeviceBase, DeviceInfo +from app.models.emulator import DeviceBase, DeviceInfo, EmulatorStartupError from app.models.notification import LocalNotice from app.models.schema import HistoryAggregationItem, HistoryWrite, WSTaskNoticeData from app.models.task import LogRecord, ScriptItem @@ -1624,6 +1624,8 @@ async def main_task(self): self.script_config.Game.EmulatorIndex, "" if cloud_adb else "com.hypergryph.endfield", ) + except EmulatorStartupError: + raise except Exception as e: # 启动阶段失败后游戏状态不再可信：清除保留标记，避免下一轮 # 启动沿用 preserve_running_game 跳过启动器游戏设置 diff --git a/app/task/MaaFW/tools/embedded/runner_task.py b/app/task/MaaFW/tools/embedded/runner_task.py index 6452afdead..16884b242a 100644 --- a/app/task/MaaFW/tools/embedded/runner_task.py +++ b/app/task/MaaFW/tools/embedded/runner_task.py @@ -32,7 +32,7 @@ ) from app.core.notify.task_content import screenshot_entries as screenshot_entries from app.core.ws import protocol -from app.models.emulator import DeviceBase, DeviceInfo +from app.models.emulator import DeviceBase, DeviceInfo, EmulatorStartupError from app.models.notification import LocalNotice from app.models.schema import ( HistoryAttachmentInput, @@ -916,7 +916,9 @@ async def main_task(self) -> None: else f"MaaFW 运行异常: {exc}" ) self._append_log(message, warning=True) - unretryable = _is_unretryable_failure(message) + unretryable = isinstance( + exc, EmulatorStartupError + ) or _is_unretryable_failure(message) self._record_attempt( index + 1, [], diff --git a/app/task/SRC/AutoProxy.py b/app/task/SRC/AutoProxy.py index 413ac51201..8c929786f8 100644 --- a/app/task/SRC/AutoProxy.py +++ b/app/task/SRC/AutoProxy.py @@ -30,7 +30,12 @@ ) from app.core.ws import protocol from app.models.config import SrcConfig, SrcUserConfig -from app.models.emulator import DeviceBase, DeviceInfo, DeviceStatus +from app.models.emulator import ( + DeviceBase, + DeviceInfo, + DeviceStatus, + EmulatorStartupError, +) from app.models.notification import LocalNotice from app.models.schema import HistoryAggregationItem, HistoryWrite, WSTaskNoticeData from app.models.task import LogRecord, ScriptItem @@ -227,6 +232,8 @@ async def main_task(self): self.script_config.Emulator.Index, STARRAIL_PACKAGE_NAME[self.cur_user_config.Info.Server], ) + except EmulatorStartupError: + raise except Exception as e: if not await self.handle_pre_src_error("模拟器启动失败", e): return diff --git a/app/task/base.py b/app/task/base.py index a9f867722a..89603c07df 100644 --- a/app/task/base.py +++ b/app/task/base.py @@ -18,6 +18,7 @@ WhimboxConfig, ZzzOdConfig, ) +from app.models.emulator import EmulatorStartupError from app.models.task import LogRecord, ScriptItem, TaskExecuteBase, UserItem from app.services import execute_script_task from app.task.failure_capture_core import FailureCapture @@ -270,6 +271,21 @@ async def _run_main_task(self) -> None: try: async with timeout: await self.main_task() + except EmulatorStartupError as e: + # 模拟器已完成有限恢复；结束本账号本次尝试，由既有生命周期执行收尾。 + message = f"模拟器启动失败: {e}" + user = self.cur_user_item + await user.set_state(status="异常") + if user.log_record: + record = user.log_record[max(user.log_record)] + else: + record = await user.add_log_record( + datetime.now().astimezone(), LogRecord() + ) + await record.set_state(status=message) + await record.append_content(message) + await self.script_info.set_log(message) + await self.on_crash(e) except TimeoutError: # 脚本内部的 TimeoutError 仍交给原异常流程处理。 if not timeout.expired(): diff --git a/app/utils/common/emulator/mumu.py b/app/utils/common/emulator/mumu.py index 81b0c324cb..d6ffa5731c 100644 --- a/app/utils/common/emulator/mumu.py +++ b/app/utils/common/emulator/mumu.py @@ -28,6 +28,7 @@ DeviceInfo, DeviceStatus, EmulatorRuntimePolicy, + EmulatorStartupError, ) from app.utils import IS_WINDOWS, ProcessRunner, get_logger from app.utils import emulator as _utils_emulator @@ -49,6 +50,10 @@ MUMU_CLOSE_SETTLE_SECONDS = 2 MUMU_STORE_PACKAGE = "com.mumu.store" MUMU_STORE_OVERLAY_APP_OP = "SYSTEM_ALERT_WINDOW" +# 只重试已知的 mainnx 临时连接错误；所有尝试共用 MaxWaitTime，不逐轮续期。 +MUMU_LAUNCH_ATTEMPTS = 3 +MUMU_LAUNCH_RETRY_SECONDS = 2 +_INSTANCE_LOCKS: dict[tuple[asyncio.AbstractEventLoop, str, str], asyncio.Lock] = {} # MuMu 6.8.8 主进程连不上时 control launch 的返回码 MUMU_MAINNX_ERRCODE = -503 # 见到 -503 且安装目录带低完整性标签时附在异常里的修复办法。标签来自旧版安装器， @@ -120,6 +125,18 @@ def __init__( # {实例索引: 静音记录}，启动时跟随静音、关闭前还原 self._audio_mute_states: dict[str, _utils_emulator.tools.AudioMuteRecord] = {} + self._failed_launches: set[str] = set() + self._launch_errors: dict[str, str] = {} + + def _get_instance_lock(self, idx: str) -> asyncio.Lock: + # 不同配置 / 管理器对象也可能指向同一物理实例，不能按配置 UUID 加锁。 + key = ( + asyncio.get_running_loop(), + str(self.emulator_path.resolve()).casefold(), + str(int(idx)) if str(idx).isdecimal() else str(idx), + ) + return _INSTANCE_LOCKS.setdefault(key, asyncio.Lock()) + def get_adb_path(self) -> Path | None: adb_path = self.emulator_path.parent / "adb.exe" return adb_path if adb_path.exists() else None @@ -435,6 +452,138 @@ async def _block_store_overlay_ads(self, idx: str) -> None: logger.warning(f"停止 MuMu 应用商店广告进程失败: {result.stdout.strip()}") async def open(self, idx: str, package_name: str = "") -> DeviceInfo: + idx = str(int(idx)) if str(idx).isdecimal() else str(idx) + acquired = False + try: + # 包含等待实例锁、状态查询、命令、重试和就绪等待，避免多层超时相加。 + async with asyncio.timeout(self.config.Info.MaxWaitTime): + async with self._get_instance_lock(idx): + acquired = True + self._launch_errors[idx] = "" + # 新操作取得实例锁后才清除旧失败；成功时不清除等待者随后留下的 + # 失败标记，避免其收尾关闭本次刚启动的实例。 + self._failed_launches.discard(idx) + info = await self._open_locked(idx, package_name) + self._launch_errors.pop(idx, None) + return info + except asyncio.CancelledError: + self._failed_launches.add(idx) + logger.warning(f"MuMu 实例 {idx} 启动操作已取消，已释放实例锁") + raise + except Exception as e: + self._failed_launches.add(idx) + reason = ( + f"超过启动总时限 {self.config.Info.MaxWaitTime} 秒" + if isinstance(e, TimeoutError) + else str(e) + ) + detail = self._launch_errors.get(idx, "") if acquired else "等待实例锁失败" + message = ( + f"MuMu 实例 {idx} 启动失败: {reason}" + f"{('；最后命令错误: ' + detail) if detail and detail != reason else ''}" + "；安全恢复已停止，未清理 MuMu 后台进程。" + "请检查 MuMu 多开器中该实例的状态，确认其他任务已结束后手动重启 MuMu，" + "再重试任务；若手动启动仍失败，请联系 MuMu 支持" + ) + logger.error(message) + raise EmulatorStartupError(message) from e + + async def _launch_with_retries(self, idx: str, package_name: str) -> None: + commands = ( + ("setting", "-v", idx, "-k", "app_keptlive", "-val", "false"), + ( + "control", + "-v", + idx, + "launch", + *(["-pkg", package_name] if package_name else []), + ), + ) + for attempt in range(1, MUMU_LAUNCH_ATTEMPTS + 1): + status = await self.getStatus(idx) + logger.info( + f"MuMu 实例 {idx} 启动第 {attempt}/{MUMU_LAUNCH_ATTEMPTS} 轮前状态: {status.name}" + ) + if status in (DeviceStatus.ONLINE, DeviceStatus.STARTING): + logger.info(f"MuMu 实例 {idx} 已启动，跳过重复启动命令") + return + if status != DeviceStatus.OFFLINE: + raise RuntimeError(f"目标实例状态为 {status.name}，无法确认可安全启动") + + for args in commands: + if args[0] == "control": + status = await self.getStatus(idx) + if status in (DeviceStatus.ONLINE, DeviceStatus.STARTING): + logger.info(f"MuMu 实例 {idx} 在准备期间已启动，跳过 launch") + return + if status != DeviceStatus.OFFLINE: + raise RuntimeError( + f"启动前复查状态为 {status.name}，停止发送 launch" + ) + logger.info( + f"执行 MuMu 启动命令（实例 {idx}，第 {attempt} 轮）: {[str(self.emulator_path), *args]}" + ) + result = await ProcessRunner.run_process( + self.emulator_path, + *args, + timeout=self.config.Info.MaxWaitTime, + if_merge_std=True, + breakaway=True, + ) + # Windows 将负数退出码作为无符号 DWORD 返回；保留原值并显示有符号值。 + signed_code = (result.returncode + 2**31) % 2**32 - 2**31 + errcode, errmsg = None, "" + try: + payload = self._decode_polluted_json( + result.stdout, + lambda v: isinstance(v, dict) and "errcode" in v, + ) + except json.JSONDecodeError: + pass + else: + if isinstance(payload, dict): + errcode = payload.get("errcode") + errmsg = payload.get("errmsg", "") + logger.info( + f"MuMu 实例 {idx} 第 {attempt} 轮命令结果: returncode={result.returncode}, " + f"signed_returncode={signed_code}, errcode={errcode}, errmsg={errmsg!r}" + ) + if result.returncode == 0 and errcode in (None, 0): + continue + + reason = ( + f"第 {attempt}/{MUMU_LAUNCH_ATTEMPTS} 轮 {args[0]} 失败: " + f"returncode={result.returncode}, signed_returncode={signed_code}, " + f"errcode={errcode}, errmsg={errmsg!r}" + ) + if errcode is None and not errmsg: + reason += f"；原始输出: {result.failure_detail()}" + hint = _low_integrity_hint(self.emulator_path, result.stdout) + reason += hint + self._launch_errors[idx] = reason + logger.warning(reason) + status = await self.getStatus(idx) + logger.info(f"MuMu 实例 {idx} 命令失败后状态: {status.name}") + if status in (DeviceStatus.ONLINE, DeviceStatus.STARTING): + logger.warning(f"MuMu 实例 {idx} 命令报错但已启动，继续等待就绪") + return + code = errcode if errcode is not None else signed_code + if ( + bool(hint) + or code != MUMU_MAINNX_ERRCODE + or status != DeviceStatus.OFFLINE + or attempt == MUMU_LAUNCH_ATTEMPTS + ): + raise RuntimeError(reason + f"；复查状态={status.name}") + logger.warning( + f"MuMu 实例 {idx} mainnx 连接失败，{MUMU_LAUNCH_RETRY_SECONDS} 秒后重新检查并重试" + ) + await asyncio.sleep(MUMU_LAUNCH_RETRY_SECONDS) + break + else: + return + + async def _open_locked(self, idx: str, package_name: str) -> DeviceInfo: started_at = time.monotonic() logger.info(f"开始启动模拟器 {idx} - {package_name}") @@ -459,9 +608,13 @@ async def open(self, idx: str, package_name: str = "") -> DeviceInfo: if self.runtime_policy.block_ads_enabled(): await self._block_store_overlay_ads(idx) return info - elif status == DeviceStatus.OFFLINE: - logger.info(f"模拟器离线，开始执行启动流程: {idx} - {package_name}") + elif status in (DeviceStatus.OFFLINE, DeviceStatus.STARTING): + logger.info( + f"模拟器状态 {status.name}，开始执行启动流程: {idx} - {package_name}" + ) break + elif status == DeviceStatus.NOT_FOUND: + raise RuntimeError(f"未找到 MuMu 实例 {idx}") await asyncio.sleep(0.1) else: @@ -472,63 +625,15 @@ async def open(self, idx: str, package_name: str = "") -> DeviceInfo: ) raise RuntimeError(f"模拟器 {idx} 无法启动, 当前状态码: {status}") - # 启动前强力清理要放在检查多开器窗口之前：清理会把多开器一起关掉， - # 之后按「启动前没开多开器」的口径处理启动过程中弹出的窗口 if self.config.Info.ForceKillBeforeLaunch: - await self._force_clear_before_launch() + logger.warning( + "安全启动不执行启动前强力清理：无法确认后台进程归属，清理会影响其他 MuMu 实例" + ) if_close_mumu_nx = await self.find_mumu_nx_window() is None - # 启动实例前关闭 MuMu 应用保活 - logger.info(f"关闭 MuMu 应用保活: {idx}") - keepalive_started_at = time.monotonic() - result = await ProcessRunner.run_process( - self.emulator_path, - "setting", - "-v", - idx, - "-k", - "app_keptlive", - "-val", - "false", - timeout=self.config.Info.MaxWaitTime, - if_merge_std=True, - breakaway=True, - ) - if result.returncode != 0: - raise RuntimeError(f"设置 app_keptlive 失败: {result.failure_detail()}") - - logger.info( - f"MuMu 应用保活设置完成: {idx} - 用时: " - f"{time.monotonic() - keepalive_started_at:.3f}秒" - ) - - logger.info(f"执行 MuMu 启动命令: {idx} - {package_name}") launch_started_at = time.monotonic() - result = await ProcessRunner.run_process( - self.emulator_path, - "control", - "-v", - idx, - "launch", - *(["-pkg", package_name] if package_name else []), - timeout=self.config.Info.MaxWaitTime, - if_merge_std=True, - breakaway=True, - ) - # 参考命令 MuMuManager.exe control -v 2 launch - - if result.returncode != 0: - # 界面入口（emulator_manager 的 except）以前只发 WS 气泡不写日志， - # 只靠下面这条 raise，失败原因在 app.log 里是空白的。 - # 这里不拼 _describe_launch_failure：6.8.8 的 info 在实例离线时不产出 - # launch_err_* / player_state，拼出来恒为空串。 - # -503 另有环境层面的成因，能自证时把修复命令一起带到界面提示里。 - hint = _low_integrity_hint(self.emulator_path, result.stdout) - logger.error( - f"MuMu 启动命令失败: {idx} - {package_name} - {result.failure_detail()}" - ) - raise RuntimeError(f"命令执行失败: {result.failure_detail()}{hint}") + await self._launch_with_retries(idx, package_name) # 只说明命令返回 0，不代表实例会起来：真正的结论在下面的等待循环， # 措辞不要写成「已完成」，否则日志会把静默空等读成成功。 @@ -571,7 +676,12 @@ async def open(self, idx: str, package_name: str = "") -> DeviceInfo: resolve_pids=lambda: self._resolve_audio_pids(idx), silence_enabled=self.runtime_policy.silence_enabled, ) - return (await self.getInfo(idx))[idx] + info = (await self.getInfo(idx))[idx] + if info.status != DeviceStatus.ONLINE: + raise RuntimeError( + f"MuMu 实例 {idx} 就绪后状态变为 {info.status.name}" + ) + return info await asyncio.sleep(0.1) else: diagnosis = await self._describe_launch_failure(idx) @@ -621,6 +731,21 @@ def scan() -> list[int]: return await asyncio.to_thread(scan) async def close(self, idx: str) -> DeviceStatus: + idx = str(int(idx)) if str(idx).isdecimal() else str(idx) + async with self._get_instance_lock(idx): + if idx in self._failed_launches: + logger.warning( + f"MuMu 实例 {idx} 启动未完成，跳过自动关闭和强力清理，请检查实例真实状态" + ) + status = await self.getStatus(idx) + if status != DeviceStatus.OFFLINE: + raise RuntimeError( + f"MuMu 实例 {idx} 启动未完成，未自动关闭；当前状态={status.name}" + ) + return status + return await self._close_locked(idx) + + async def _close_locked(self, idx: str) -> DeviceStatus: await _utils_emulator.tools.restore_audio_before_close( self._audio_mute_states, idx ) @@ -889,6 +1014,8 @@ async def getStatus(self, idx: str, data: str | None = None) -> DeviceStatus: try: data = await self.get_device_info(idx) except Exception as e: + if idx in self._launch_errors: + self._launch_errors[idx] = f"info 查询失败: {e}" logger.error(f"获取模拟器 {idx} 信息失败: {e}") return DeviceStatus.ERROR try: @@ -905,8 +1032,25 @@ async def getStatus(self, idx: str, data: str | None = None) -> DeviceStatus: ) return DeviceStatus.UNKNOWN - entry = next( - (e for e in entries if str(e.get("index")) == str(idx)), entries[0] + entry = next((e for e in entries if str(e.get("index")) == str(idx)), None) + if entry is None: + return DeviceStatus.NOT_FOUND + logger.debug( + f"MuMu 实例 {idx} 状态: " + + str( + { + key: entry[key] + for key in ( + "is_process_started", + "is_android_started", + "launch_err_code", + "launch_err_msg", + "player_state", + "info_source", + ) + if key in entry + } + ) ) return self._get_status_from_data(entry) diff --git a/changelog.d/mumu-launch-recovery.fix.md b/changelog.d/mumu-launch-recovery.fix.md new file mode 100644 index 0000000000..c0d51ecc96 --- /dev/null +++ b/changelog.d/mumu-launch-recovery.fix.md @@ -0,0 +1,2 @@ +project: emulator +修复 MuMu 偶发启动失败时无法恢复及反复重试的问题 diff --git a/tests/task/test_mumu_startup_failure.py b/tests/task/test_mumu_startup_failure.py new file mode 100644 index 0000000000..310262d75d --- /dev/null +++ b/tests/task/test_mumu_startup_failure.py @@ -0,0 +1,132 @@ +"""模拟器恢复耗尽后的真实 MAA 入口与调度状态、持锁收尾。""" + +import asyncio +from unittest.mock import AsyncMock +from uuid import uuid4 + +import pytest + +from app.core import Publisher +from app.models.config import GeneralConfig, GeneralUserConfig, MaaConfig, MaaUserConfig +from app.models.emulator import EmulatorStartupError +from app.models.task import TaskItem, TaskMode, TaskScriptItem, TaskUserItem +from app.task.MAA.AutoProxy import AutoProxyTask +from app.task.mode_worker import AutoProxyWorker +from app.task.user_expander import UserExpander +from tests.config_support import activate_config +from tests.core.test_task_tree_dispatch import scheduler as scheduler + + +@pytest.mark.asyncio +async def test_real_maa_startup_failure_keeps_reason_and_avoids_outer_retry( + tmp_path, monkeypatch, isolated_app_paths +): + config = await activate_config( + MaaConfig.model_validate( + { + "Info": {"Path": tmp_path}, + "Run": {"RunTimesLimit": 10}, + "Emulator": {"Index": "0"}, + } + ) + ) + user_id = config.UserData.add( + MaaUserConfig, + payload={ + "Info": { + "Base": "shared", + "IfQuickConfig": False, + "Name": "测试账号", + } + }, + ) + await config.UserData.commit() + task = await activate_config(TaskItem()) + script_id = task.scripts.add( + TaskScriptItem, payload={"info": {"script_id": str(uuid4())}} + ) + await task.scripts.commit() + script = task.scripts[script_id] + script.users.add( + TaskUserItem, payload={"info": {"user_id": str(user_id), "name": "测试账号"}} + ) + await script.users.commit() + await script.set_current_index(0) + emulator = AsyncMock() + emulator.open.side_effect = EmulatorStartupError( + "实例 0 errcode=-503 mainnx connect failed" + ) + send = AsyncMock() + monkeypatch.setattr(Publisher, "send", send) + proxy = AutoProxyTask( + script, config, config.UserData, emulator, capture_adb=AsyncMock() + ) + # 保留真实 prepare、main_task 和异常记账；外部命令在模拟器边界失败，MAA 尚未运行。 + await proxy._run_main_task() + emulator.open.assert_awaited_once() + emulator.close.assert_not_awaited() + assert proxy.cur_user_item.status == "异常" + assert "mainnx connect failed" in proxy.cur_user_item.result + assert "errcode=-503" in script.current.log + assert "正在启动模拟器" not in script.current.log + assert any("-503" in call.kwargs["data"].message for call in send.await_args_list) + + +@pytest.mark.asyncio +async def test_scheduler_releases_locks_reports_error_and_runs_following_script( + scheduler, +): + manager, scripts, _, first_id = scheduler + second_id = scripts.add(GeneralConfig, payload={"Info": {"Name": "正常脚本"}}) + await scripts.commit() + scripts[second_id].UserData.add( + GeneralUserConfig, payload={"Info": {"Status": True, "RemainedDay": -1}} + ) + await scripts[second_id].UserData.commit() + finalized, following = [], [] + external_open = AsyncMock( + side_effect=EmulatorStartupError("errcode=-503 mainnx connect failed") + ) + + class Worker(AutoProxyWorker): + async def main_logic(self): + if self.script_info.script_id == str(first_id): + await external_open("0") + else: + assert not scripts[first_id].is_locked + assert not any(u.is_locked for u in scripts[first_id].UserData.values()) + following.append(self.user_info.uid) + await self.user_info.set_state(status="完成") + + async def finish(self): + assert self.user_config.is_locked + finalized.append(self.user_info.uid) + + class Expander(UserExpander): + def build_worker(self, user, user_config): + return Worker(self, user, user_config) + + manager.plugin_expander = lambda script, config: Expander(script, config) + uid = await manager.start( + TaskMode.AUTO_PROXY, + { + 0: { + "uid": "-", + "children": { + 0: {"uid": str(first_id)}, + 1: {"uid": str(second_id)}, + }, + } + }, + ) + run = manager.task_handler[uid] + await asyncio.wait_for(run.accomplish.wait(), 3) + assert run.task_info.status == "error" + assert following and len(finalized) == 3 + assert all(not config.is_locked for config in scripts.values()) + assert all( + not user.is_locked + for config in scripts.values() + for user in config.UserData.values() + ) + assert not manager.task_handler diff --git a/tests/tools/test_mumu_foreground_budget.py b/tests/tools/test_mumu_foreground_budget.py index 75a4345d4f..acc06db8c1 100644 --- a/tests/tools/test_mumu_foreground_budget.py +++ b/tests/tools/test_mumu_foreground_budget.py @@ -9,6 +9,7 @@ import pytest +from app.models.config import EmulatorConfig from app.models.emulator import DeviceInfo, DeviceStatus, EmulatorRuntimePolicy from app.utils import emulator as _utils_emulator from app.utils.common.process import runner as process_runner @@ -21,7 +22,7 @@ class Harness: """只替换时钟和命令边界；执行真实管理器流程，不访问真实设备。""" - def __init__(self, monkeypatch): + def __init__(self, monkeypatch, tmp_path): self.now = 0.0 self.calls = [] self.options = [] @@ -31,22 +32,27 @@ def __init__(self, monkeypatch): self.monkey = (0.2, 0, "") self.app_state = "running" self.app_launch = (0.2, 0, "") - self.manager = object.__new__(mumu.MumuManager) - self.manager.config = SimpleNamespace( - Info=SimpleNamespace(MaxWaitTime=300, ForceKillBeforeLaunch=False) + exe = tmp_path / "MuMuManager.exe" + exe.touch() + self.manager = mumu.MumuManager( + EmulatorConfig.model_validate( + {"Info": {"Type": "mumu", "Path": exe, "MaxWaitTime": 300}} + ), + runtime_policy=EmulatorRuntimePolicy( + lambda: False, lambda: False, tmp_path + ), ) - self.manager.emulator_path = Path("MuMuManager-test-only") - self.manager.runtime_policy = EmulatorRuntimePolicy( - silence_enabled=lambda: False, - block_ads_enabled=lambda: False, - resource_path=Path("MuMuManager-test-only"), - ) - self.manager._audio_mute_states = {} monkeypatch.setattr(mumu, "time", SimpleNamespace(monotonic=lambda: self.now)) monkeypatch.setattr( mumu, "asyncio", - SimpleNamespace(sleep=self.sleep, CancelledError=asyncio.CancelledError), + SimpleNamespace( + sleep=self.sleep, + CancelledError=asyncio.CancelledError, + timeout=asyncio.timeout, + get_running_loop=asyncio.get_running_loop, + Lock=asyncio.Lock, + ), ) monkeypatch.setattr( mumu, "ProcessRunner", SimpleNamespace(run_process=self.run) @@ -85,8 +91,8 @@ async def ensure(self, idx="0"): @pytest.fixture -def harness(monkeypatch): - return Harness(monkeypatch) +def harness(monkeypatch, tmp_path): + return Harness(monkeypatch, tmp_path) @pytest.mark.asyncio @@ -137,7 +143,7 @@ async def test_slow_cold_boot_keeps_full_startup_wait(harness, monkeypatch, fore async def status(idx): nonlocal status_calls status_calls += 1 - if status_calls == 1: + if not harness.commands("launch"): return DeviceStatus.OFFLINE # 正常但较慢：上线等待超过新的前台辅助预算。 return DeviceStatus.STARTING if harness.now < 80 else DeviceStatus.ONLINE diff --git a/tests/tools/test_mumu_launch_recovery.py b/tests/tools/test_mumu_launch_recovery.py new file mode 100644 index 0000000000..e13a6be96d --- /dev/null +++ b/tests/tools/test_mumu_launch_recovery.py @@ -0,0 +1,452 @@ +"""MuMu 启动恢复：所有命令、窗口和进程操作均为 Mock，不触碰模拟器数据。""" + +import asyncio +import json +from collections import deque +from types import SimpleNamespace +from unittest.mock import AsyncMock, Mock + +import pytest + +from app.models.config import EmulatorConfig +from app.models.emulator import DeviceStatus, EmulatorRuntimePolicy +from app.utils.common.emulator import mumu +from app.utils.common.emulator.mumu import MumuManager +from app.utils.common.process.runner import ProcessResult + + +class FakeMuMu: + def __init__(self): + self.states = {"0": DeviceStatus.OFFLINE, "1": DeviceStatus.ONLINE} + self.launch_results = deque() + self.setting_results = deque() + self.calls = [] + self.after_info = None + + async def run(self, program, *args, **kwargs): + self.calls.append(args) + idx = args[args.index("-v") + 1] + assert idx in self.states + if args[0] == "info": + state = self.states[idx] + entry = { + "index": int(idx), + "name": f"MuMu-{idx}", + "is_process_started": state + in (DeviceStatus.STARTING, DeviceStatus.ONLINE), + "is_android_started": state == DeviceStatus.ONLINE, + "adb_host_ip": "127.0.0.1", + "adb_port": 16384 + int(idx) * 32, + "player_state": "start_finished" + if state == DeviceStatus.ONLINE + else "stopped", + } + if self.after_info: + self.after_info(idx) + return ProcessResult(json.dumps(entry), "", 0) + if args[0] == "setting": + return self.setting_results.popleft() if self.setting_results else ok() + if args[0] == "control" and args[3] == "launch": + result, state = ( + self.launch_results.popleft() + if self.launch_results + else (ok(), DeviceStatus.ONLINE) + ) + self.states[idx] = state + return result + pytest.fail(f"不允许执行其他操作: {args}") + + def launches(self, idx="0"): + return [c for c in self.calls if c[:4] == ("control", "-v", idx, "launch")] + + +def ok(): + return ProcessResult('{"errcode": 0, "errmsg": ""}', "", 0) + + +def failed(code=-503, returncode=4294966793): + return ProcessResult( + json.dumps({"errcode": code, "errmsg": "mainnx connect failed"}), "", returncode + ) + + +@pytest.fixture +def harness(tmp_path, monkeypatch): + exe = tmp_path / "MuMuManager.exe" + exe.touch() + settings = { + "Path": str(exe), + "Type": "mumu", + "MaxWaitTime": 5, + "ForceKillBeforeLaunch": False, + "ForceKillOnClose": False, + } + config = EmulatorConfig.model_validate({"Info": settings}) + policy = EmulatorRuntimePolicy(lambda: False, lambda: False, tmp_path) + manager = MumuManager(config, runtime_policy=policy) + + class Settings(dict): + def __setitem__(self, key, value): + setattr(config.Info, key, value) + super().__setitem__(key, value) + + def update(self, **values): + for key, value in values.items(): + self[key] = value + + settings = Settings(settings) + fake = FakeMuMu() + monkeypatch.setattr( + mumu.ProcessRunner, "run_process", AsyncMock(side_effect=fake.run) + ) + monkeypatch.setattr(MumuManager, "find_mumu_nx_window", AsyncMock(return_value=123)) + monkeypatch.setattr(MumuManager, "_resolve_audio_pids", AsyncMock(return_value=[])) + monkeypatch.setattr( + MumuManager, + "_force_clear_before_launch", + AsyncMock(side_effect=AssertionError("不得全局清理")), + ) + monkeypatch.setattr( + MumuManager, + "_force_kill_mumu_processes", + Mock(side_effect=AssertionError("不得结束进程")), + ) + monkeypatch.setattr( + mumu._utils_emulator.tools, "apply_launch_audio_mute", AsyncMock() + ) + monkeypatch.setattr(mumu, "logger", Mock()) + real_sleep = asyncio.sleep + sleeps = [] + + async def fast_sleep(seconds): + sleeps.append(seconds) + await real_sleep(0) + + monkeypatch.setattr(mumu.asyncio, "sleep", fast_sleep) + return SimpleNamespace( + manager=manager, + fake=fake, + settings=settings, + sleeps=sleeps, + real_sleep=real_sleep, + ) + + +@pytest.mark.asyncio +async def test_normal_launch(harness): + info = await harness.manager.open("0") + assert info.status == DeviceStatus.ONLINE + assert info.adb_address == "127.0.0.1:16384" + assert len(harness.fake.launches()) == 1 + assert not harness.manager._get_instance_lock("0").locked() + + +@pytest.mark.asyncio +async def test_running_instance_is_not_launched_again(harness): + harness.fake.states["0"] = DeviceStatus.ONLINE + assert (await harness.manager.open("0")).status == DeviceStatus.ONLINE + assert all(c[0] == "info" for c in harness.fake.calls) + + +@pytest.mark.asyncio +async def test_temporary_503_recovers(harness): + harness.fake.launch_results.extend( + [(failed(), DeviceStatus.OFFLINE), (ok(), DeviceStatus.ONLINE)] + ) + assert (await harness.manager.open("0")).status == DeviceStatus.ONLINE + assert len(harness.fake.launches()) == 2 + assert harness.sleeps.count(mumu.MUMU_LAUNCH_RETRY_SECONDS) == 1 + calls = harness.fake.calls + first, second = [i for i, c in enumerate(calls) if c in harness.fake.launches()] + assert sum(c[0] == "info" for c in calls[first + 1 : second]) >= 2 + + +@pytest.mark.asyncio +async def test_persistent_503_is_bounded_and_close_does_not_kill(harness): + harness.settings.update(ForceKillBeforeLaunch=True, ForceKillOnClose=True) + harness.fake.launch_results.extend([(failed(), DeviceStatus.OFFLINE)] * 5) + with pytest.raises(RuntimeError, match="errcode=-503") as exc: + await harness.manager.open("0") + assert "4294966793" in str(exc.value) and "mainnx connect failed" in str(exc.value) + assert len(harness.fake.launches()) == mumu.MUMU_LAUNCH_ATTEMPTS == 3 + assert harness.sleeps.count(mumu.MUMU_LAUNCH_RETRY_SECONDS) == 2 + assert not harness.manager._get_instance_lock("0").locked() + assert await harness.manager.close("0") == DeviceStatus.OFFLINE + assert harness.fake.states["1"] == DeviceStatus.ONLINE + assert all(c[2] == "0" for c in harness.fake.calls) + + +@pytest.mark.asyncio +@pytest.mark.parametrize("state", [DeviceStatus.ONLINE, DeviceStatus.STARTING]) +async def test_command_errors_but_instance_started(harness, state): + harness.fake.launch_results.append((failed(), state)) + if state == DeviceStatus.STARTING: + + def after_info(idx): + if harness.fake.states[idx] == DeviceStatus.STARTING: + harness.fake.states[idx] = DeviceStatus.ONLINE + + harness.fake.after_info = after_info + assert (await harness.manager.open("0")).status == DeviceStatus.ONLINE + assert len(harness.fake.launches()) == 1 + + +@pytest.mark.asyncio +async def test_instance_started_during_retry_delay(harness): + harness.fake.launch_results.append((failed(), DeviceStatus.OFFLINE)) + original_sleep = mumu.asyncio.sleep + + async def sleep(seconds): + if seconds == mumu.MUMU_LAUNCH_RETRY_SECONDS: + harness.fake.states["0"] = DeviceStatus.ONLINE + await original_sleep(seconds) + + from unittest.mock import patch + + with patch.object(mumu.asyncio, "sleep", sleep): + assert (await harness.manager.open("0")).status == DeviceStatus.ONLINE + assert len(harness.fake.launches()) == 1 + + +@pytest.mark.asyncio +@pytest.mark.parametrize( + "result", + [failed(-503, 0), ProcessResult("", "", 4294966793), ProcessResult("", "", -503)], +) +async def test_json_and_signed_exit_codes_are_recognized(harness, result): + harness.fake.launch_results.append((result, DeviceStatus.OFFLINE)) + assert (await harness.manager.open("0")).status == DeviceStatus.ONLINE + assert len(harness.fake.launches()) == 2 + + +@pytest.mark.asyncio +async def test_non_retryable_error_is_reported_once(harness): + harness.fake.launch_results.append((failed(-999, 123), DeviceStatus.OFFLINE)) + with pytest.raises(RuntimeError, match="errcode=-999"): + await harness.manager.open("0") + assert len(harness.fake.launches()) == 1 + + +@pytest.mark.asyncio +async def test_setting_503_also_has_bounded_recovery(harness): + harness.fake.setting_results.extend([failed(), ok()]) + assert (await harness.manager.open("0")).status == DeviceStatus.ONLINE + assert len(harness.fake.launches()) == 1 + + +@pytest.mark.asyncio +async def test_initial_starting_instance_is_only_observed(harness): + harness.fake.states["0"] = DeviceStatus.STARTING + reads = 0 + + def after_info(idx): + nonlocal reads + reads += 1 + if reads >= 3: + harness.fake.states[idx] = DeviceStatus.ONLINE + + harness.fake.after_info = after_info + assert (await harness.manager.open("0")).status == DeviceStatus.ONLINE + assert not harness.fake.launches() + assert all(c[0] == "info" for c in harness.fake.calls) + + +@pytest.mark.asyncio +async def test_same_physical_instance_is_serialized_across_managers(harness): + second = MumuManager( + harness.manager.config, runtime_policy=harness.manager.runtime_policy + ) + first_info, second_info = await asyncio.gather( + harness.manager.open("0"), second.open("00") + ) + assert first_info == second_info + assert len(harness.fake.launches()) == 1 + assert harness.manager._get_instance_lock("0") is second._get_instance_lock("00") + + +@pytest.mark.asyncio +async def test_failure_on_one_instance_does_not_block_another(harness): + harness.fake.launch_results.extend([(failed(), DeviceStatus.OFFLINE)] * 3) + results = await asyncio.gather( + harness.manager.open("0"), harness.manager.open("1"), return_exceptions=True + ) + assert isinstance(results[0], RuntimeError) + assert results[1].status == DeviceStatus.ONLINE + assert harness.fake.states["1"] == DeviceStatus.ONLINE + assert not harness.fake.launches("1") + + +@pytest.mark.asyncio +async def test_total_timeout_bounds_polling_and_releases_lock(harness): + harness.settings["MaxWaitTime"] = 1 + harness.fake.states["0"] = DeviceStatus.STARTING + with pytest.raises(RuntimeError, match="总时限|启动超时"): + await asyncio.wait_for(harness.manager.open("0"), 2) + assert not harness.manager._get_instance_lock("0").locked() + assert not harness.fake.launches() + + +@pytest.mark.asyncio +async def test_cancel_releases_lock_and_next_open_can_recover(harness): + entered = asyncio.Event() + + async def hang(program, *args, **kwargs): + entered.set() + await asyncio.Event().wait() + + mumu.ProcessRunner.run_process.side_effect = hang + task = asyncio.create_task(harness.manager.open("0")) + await entered.wait() + task.cancel() + with pytest.raises(asyncio.CancelledError): + await task + assert not harness.manager._get_instance_lock("0").locked() + mumu.ProcessRunner.run_process.side_effect = harness.fake.run + assert (await harness.manager.open("0")).status == DeviceStatus.ONLINE + assert "0" not in harness.manager._failed_launches + + +@pytest.mark.asyncio +async def test_missing_target_is_not_confused_with_another_instance(harness): + mumu.ProcessRunner.run_process.return_value = ProcessResult( + json.dumps({"index": 1, "name": "other", "is_android_started": True}), "", 0 + ) + mumu.ProcessRunner.run_process.side_effect = None + assert await harness.manager.getStatus("0") == DeviceStatus.NOT_FOUND + with pytest.raises(RuntimeError, match="未找到"): + await harness.manager.open("0") + + +@pytest.mark.asyncio +async def test_unreadable_state_never_triggers_launch(harness): + harness.settings["MaxWaitTime"] = 1 + mumu.ProcessRunner.run_process.side_effect = None + mumu.ProcessRunner.run_process.return_value = failed() + with pytest.raises(RuntimeError): + await harness.manager.open("0") + assert all( + call.args[1] == "info" for call in mumu.ProcessRunner.run_process.call_args_list + ) + + +@pytest.mark.asyncio +async def test_failed_start_does_not_claim_starting_instance_was_closed(harness): + harness.manager._failed_launches.add("0") + harness.settings["ForceKillOnClose"] = True + harness.fake.states["0"] = DeviceStatus.STARTING + with pytest.raises(RuntimeError, match="未自动关闭"): + await harness.manager.close("0") + assert harness.fake.states["1"] == DeviceStatus.ONLINE + assert all(c[0] == "info" for c in harness.fake.calls) + + +@pytest.mark.asyncio +async def test_retry_delay_is_part_of_total_budget(harness): + harness.settings["MaxWaitTime"] = 1 + harness.fake.launch_results.append((failed(), DeviceStatus.OFFLINE)) + from unittest.mock import patch + + with patch.object(mumu.asyncio, "sleep", harness.real_sleep): + with pytest.raises(RuntimeError, match="总时限.*errcode=-503"): + await asyncio.wait_for(harness.manager.open("0"), 2) + assert len(harness.fake.launches()) == 1 + assert not harness.manager._get_instance_lock("0").locked() + + +@pytest.mark.asyncio +async def test_unknown_state_after_error_stops_recovery(harness): + harness.fake.launch_results.append((failed(), DeviceStatus.OFFLINE)) + original = harness.manager.getStatus + + async def status(idx): + if harness.fake.launches(): + return DeviceStatus.UNKNOWN + return await original(idx) + + from unittest.mock import patch + + with patch.object(harness.manager, "getStatus", status): + with pytest.raises(RuntimeError, match="复查状态=UNKNOWN"): + await harness.manager.open("0") + assert len(harness.fake.launches()) == 1 + + +@pytest.mark.asyncio +async def test_close_waits_for_inflight_launch(harness): + entered, release = asyncio.Event(), asyncio.Event() + original = harness.fake.run + + async def run(program, *args, **kwargs): + if args[:4] == ("control", "-v", "0", "launch"): + entered.set() + await release.wait() + return await original(program, *args, **kwargs) + + mumu.ProcessRunner.run_process.side_effect = run + from unittest.mock import patch + + with patch.object( + harness.manager, "_close_locked", AsyncMock(return_value=DeviceStatus.OFFLINE) + ) as close: + opening = asyncio.create_task(harness.manager.open("0")) + await entered.wait() + closing = asyncio.create_task(harness.manager.close("0")) + await harness.real_sleep(0) + close.assert_not_awaited() + release.set() + await opening + assert await closing == DeviceStatus.OFFLINE + close.assert_awaited_once_with("0") + + +@pytest.mark.asyncio +async def test_queued_start_timeout_cannot_close_another_successful_start(harness): + entered, release = asyncio.Event(), asyncio.Event() + original = harness.fake.run + + async def run(program, *args, **kwargs): + if args[:4] == ("control", "-v", "0", "launch"): + entered.set() + await release.wait() + return await original(program, *args, **kwargs) + + mumu.ProcessRunner.run_process.side_effect = run + opening = asyncio.create_task(harness.manager.open("0")) + await entered.wait() + harness.settings["MaxWaitTime"] = 1 + try: + with pytest.raises(RuntimeError, match="等待实例锁失败"): + await harness.manager.open("0") + closing = asyncio.create_task(harness.manager.close("0")) + release.set() + assert (await opening).status == DeviceStatus.ONLINE + with pytest.raises(RuntimeError, match="未自动关闭"): + await closing + assert harness.fake.states["0"] == DeviceStatus.ONLINE + assert not any( + call[:4] == ("control", "-v", "0", "shutdown") + for call in harness.fake.calls + ) + finally: + release.set() + await opening + + +@pytest.mark.asyncio +async def test_confirmed_low_integrity_keeps_manual_hint_and_stops_retries( + harness, monkeypatch +): + monkeypatch.setattr(mumu, "IS_WINDOWS", True) + monkeypatch.setattr( + mumu, + "platform_integrity", + SimpleNamespace( + INTEGRITY_LOW=0x1000, + mandatory_label_rid=Mock(return_value=0x1000), + ), + ) + harness.fake.launch_results.append((failed(), DeviceStatus.OFFLINE)) + with pytest.raises(RuntimeError, match="icacls"): + await harness.manager.open("0") + assert len(harness.fake.launches()) == 1 + assert all(call[0] in ("info", "setting", "control") for call in harness.fake.calls)