flomo-ai/mumu-pytest/conftest.py

"""
Mumu模拟器 pytest 测试框架
"""

import pytest
import pyautogui
import time
import os
import subprocess
import requests
from pathlib import Path

# 配置
MUMU_EXE_PATH = r"C:\Program Files\Netease\MuMu\nx_main\MuMuNxMain.exe"
MUMU_PROCESS_NAME = "MuMuNxMain.exe"
ADB_PATH = r"C:\Program Files\Netease\MuMu\shell\adb.exe"
APK_DOWNLOAD_URL = "http://your-server.com/app-release.apk"
LOCAL_APK_PATH = os.path.join(os.path.dirname(__file__), "test_app.apk")
INSTALLED_PACKAGE_NAME = "com.example.flomo_ai"
WEB_URL = "http://192.168.3.15/"

SCRIPT_DIR = os.path.dirname(__file__)


class MumuEmulator:
    """Mumu模拟器控制类"""
    
    def __init__(self):
        self.process_name = MUMU_PROCESS_NAME
        self.script_dir = SCRIPT_DIR
        pyautogui.PAUSE = 1
        pyautogui.FAILSAFE = True
    
    def is_running(self) -> bool:
        """检查模拟器是否运行"""
        try:
            result = subprocess.run(
                ["tasklist"],
                capture_output=True,
                text=True
            )
            return self.process_name in result.stdout
        except Exception:
            return False
    
    def start(self):
        """启动模拟器"""
        if not self.is_running():
            subprocess.Popen(MUMU_EXE_PATH)
            time.sleep(5)
    
    def find_and_click(self, image_name, confidence=0.8, timeout=30):
        """查找图片并点击"""
        image_path = os.path.join(self.script_dir, f"{image_name}.png")
        if not os.path.exists(image_path):
            print(f"图片不存在: {image_path}")
            return False
            
        start_time = time.time()
        while time.time() - start_time < timeout:
            try:
                location = pyautogui.locateOnScreen(image_path, confidence=confidence)
                if location:
                    center = pyautogui.center(location)
                    pyautogui.click(center)
                    print(f"点击了: {image_name}")
                    return True
            except Exception as e:
                pass
            time.sleep(1)
        print(f"未找到图片: {image_name}")
        return False
    
    def type_text(self, text):
        """输入文本"""
        pyautogui.write(text, interval=0.1)
        print(f"输入了文本: {text}")
    
    def press_enter(self):
        """按回车"""
        pyautogui.press("enter")
        print("按下回车")
    
    def wait_for_boot(self, timeout=120):
        """等待模拟器启动完成"""
        boot_completed = False
        start_time = time.time()
        while time.time() - start_time < timeout:
            if self.is_running():
                time.sleep(5)
                boot_completed = True
                break
            time.sleep(2)
        return boot_completed
    
    def check_image_exists(self, image_name, confidence=0.8) -> bool:
        """检查图片是否存在"""
        image_path = os.path.join(self.script_dir, f"{image_name}.png")
        if not os.path.exists(image_path):
            return False
        try:
            location = pyautogui.locateOnScreen(image_path, confidence=confidence)
            return location is not None
        except Exception:
            return False
    
    def run_sequence(self):
        """按顺序执行任务"""
        sequence = [
            "start",
            "running",
            "web",
            "web.address",
            "web_goon",
            "web_debug",
            "web_debug_apk",
            "web_debug_apk_download",
            "web_debug_apk_open",
            "web_debug_apk_install",
            "web_debug_apk_run"
        ]
        
        for image_name in sequence:
            if image_name == "web.address":
                if self.check_image_exists("web.address"):
                    time.sleep(1)
                    self.type_text(WEB_URL)
                    time.sleep(0.5)
                    self.press_enter()
                else:
                    print("web.address 图片不存在，跳过输入URL")
            else:
                self.find_and_click(image_name, timeout=10)
                time.sleep(2)
    
    def close(self):
        """关闭模拟器"""
        try:
            subprocess.run(["taskkill", "/F", "/IM", self.process_name])
        except Exception:
            pass


class AdbHelper:
    """ADB命令辅助类"""
    
    def __init__(self):
        self.adb_path = ADB_PATH
    
    def exec_cmd(self, *args):
        """执行ADB命令"""
        cmd = [self.adb_path] + list(args)
        result = subprocess.run(cmd, capture_output=True, text=True)
        return result.stdout, result.stderr
    
    def install_apk(self, apk_path: str) -> bool:
        """安装APK"""
        stdout, stderr = self.exec_cmd("install", "-r", apk_path)
        return "Success" in stdout
    
    def uninstall_app(self, package_name: str) -> bool:
        """卸载应用"""
        stdout, stderr = self.exec_cmd("uninstall", package_name)
        return "Success" in stdout
    
    def launch_app(self, package_name: str, activity: str):
        """启动应用"""
        self.exec_cmd("shell", "am", "start", "-n", f"{package_name}/{activity}")
    
    def is_app_installed(self, package_name: str) -> bool:
        """检查应用是否已安装"""
        stdout, _ = self.exec_cmd("shell", "pm", "list", "packages", package_name)
        return package_name in stdout
    
    def pull_file(self, remote_path: str, local_path: str):
        """从模拟器拉取文件"""
        self.exec_cmd("pull", remote_path, local_path)


@pytest.fixture(scope="session")
def emulator():
    """模拟器fixture"""
    mumu = MumuEmulator()
    yield mumu
    # 测试结束后清理
    if mumu.is_running():
        mumu.close()


@pytest.fixture(scope="session")
def adb():
    """ADB fixture"""
    return AdbHelper()


@pytest.fixture(scope="session")
def download_apk():
    """下载APK"""
    if os.path.exists(LOCAL_APK_PATH):
        return LOCAL_APK_PATH
    
    response = requests.get(APK_DOWNLOAD_URL, stream=True)
    with open(LOCAL_APK_PATH, "wb") as f:
        for chunk in response.iter_content(chunk_size=8192):
            f.write(chunk)
    
    yield LOCAL_APK_PATH
    
    # 清理
    if os.path.exists(LOCAL_APK_PATH):
        os.remove(LOCAL_APK_PATH)