AirtestIDE实战：5分钟搞定Windows桌面应用自动化测试（附网易云音乐案例）

如果你是一名测试工程师，或者是一名对自动化测试感兴趣的技术人员，那么你一定听说过Airtest。这个由网易游戏团队开源的跨平台UI自动化测试框架，凭借其“所见即所得”的图像识别能力和对多种平台的支持，在移动端测试领域已经积累了不错的口碑。但你是否想过，这套强大的工具同样可以用于Windows桌面应用的自动化测试？

在日常工作中，我们经常需要面对各种Windows桌面软件的测试需求——从安装卸载、功能验证到回归测试。传统的测试方法要么依赖人工操作，效率低下且容易出错；要么需要投入大量精力编写和维护复杂的脚本。对于中小型团队来说，专门为桌面应用搭建一套自动化测试框架往往成本过高。而AirtestIDE的出现，恰好为这个痛点提供了一个优雅的解决方案。

与常见的移动端教程不同，Windows桌面应用的自动化测试有其独特的挑战：如何获取并操作特定的窗口句柄？如何处理多窗口切换？如何在没有嵌入式UI树的情况下进行元素定位？本文将带你深入AirtestIDE的Windows测试能力，并以大家熟悉的网易云音乐Windows版为例，手把手演示如何快速搭建一套稳定可靠的桌面应用自动化测试流程。无论你是测试新手，还是希望为现有测试体系补充桌面端能力的资深工程师，相信都能从中获得实用的技巧和思路。

1. 环境准备与AirtestIDE基础配置

在开始编写第一个Windows自动化测试脚本之前，我们需要先搭建好基础环境。AirtestIDE的安装过程非常简单，但其中一些配置细节却直接影响着后续测试的顺利进行。

1.1 获取与安装AirtestIDE

AirtestIDE是一个绿色软件，无需复杂的安装过程。你可以直接从其官方GitHub仓库的Release页面下载最新版本。目前它支持Windows、macOS和Linux三大主流操作系统。对于Windows用户，建议下载.exe安装包或便携版压缩包。

注意：虽然AirtestIDE支持Linux，但官方文档明确指出，只有1.2.3及更低版本在Linux上运行稳定。因此，如果你主要在Linux环境下工作，可能需要考虑使用Windows或macOS作为主要测试环境，或者通过虚拟机方式运行。

下载完成后，如果是压缩包，直接解压到任意目录即可；如果是安装包，按照向导完成安装。首次启动时，你会看到一个简洁的界面，左侧是设备连接面板，中间是脚本编辑区，右侧是设备屏幕预览区。

1.2 理解AirtestIDE的核心组件

在深入Windows测试之前，有必要先了解AirtestIDE的几个核心组成部分：

• Airtest框架：基于图像识别的自动化测试引擎，通过截图比对实现元素定位和操作
• Poco框架：基于UI控件树的自动化测试引擎，需要应用接入SDK才能使用
• Selenium扩展：用于Web自动化测试，基于airtest-selenium封装
• 设备连接管理：支持Android真机/模拟器、iOS设备、Windows窗口和浏览器

对于Windows桌面应用测试，我们主要依赖的是Airtest的图像识别能力，因为大多数Windows应用并没有像移动端那样标准的UI控件树结构。不过，如果你的Windows应用是基于某些特定框架（如Unity3D）开发的，并且接入了Poco-SDK，那么也可以使用Poco进行更精确的控件定位。

1.3 Windows测试的特殊配置

与移动端测试不同，Windows桌面应用的测试需要一些特殊的配置。在AirtestIDE的顶部菜单中，依次选择“选项”->“设置”，这里有几个关键配置项需要注意：

设备连接设置：

• Windows嵌入模式：默认启用，适合大多数标准Windows应用
• Windows备用连接方法：当嵌入模式遇到问题时可以尝试启用
• 桌面模式：用于测试整个桌面，而非单个应用窗口

# 这是一个典型的Windows设备连接字符串示例
# 格式为：Windows:///[窗口句柄或标题]
Windows:///123456  # 通过窗口句柄连接
Windows:///?title_re=Notepad.*  # 通过窗口标题正则匹配连接
Windows:///  # 连接整个桌面（桌面模式）

图像识别设置：

• 截图质量：影响识别精度和速度，默认值通常足够
• 匹配阈值：图像相似度阈值，默认0.7，可根据需要调整
• 分辨率适配：如果测试机与录制机分辨率不同，需要启用

实际经验分享：我在测试不同DPI设置的Windows系统时发现，如果录制脚本的机器与运行脚本的机器DPI设置不同，图像识别可能会失败。这时可以通过调整Template类的resolution参数，或者统一测试环境的显示设置来解决。

2. Windows窗口连接：从基础到高级技巧

连接Windows应用窗口是桌面自动化测试的第一步，也是最重要的一步。AirtestIDE提供了多种连接方式，每种方式都有其适用场景。

2.1 标准嵌入连接方式

这是最常用的连接方法，适用于大多数标准Windows应用。操作步骤如下：

1. 在AirtestIDE左侧设备面板中，点击“选择窗口”按钮
2. 将鼠标移动到目标应用窗口上，窗口边缘会显示红色边框
3. 单击左键，窗口将被嵌入到AirtestIDE的右侧设备预览区

这种方式的优点是直观易用，脚本录制和回放都在IDE内完成，便于调试。但它的局限性也很明显：某些特殊窗口（如DirectX渲染的游戏窗口、某些系统级对话框）可能无法正常嵌入。

窗口句柄的奥秘： 每个Windows窗口都有一个唯一的句柄（Handle），Airtest在嵌入连接时实际上就是获取并操作这个句柄。你可以通过以下代码查看当前连接窗口的句柄：

from airtest.core.api import *

# 连接窗口
dev = connect_device("Windows:///123456")
# 获取窗口标题
print(dev.get_title())
# 移动窗口位置
dev.move((100, 200))

2.2 非嵌入连接方式

对于无法正常嵌入的窗口，AirtestIDE提供了备用连接方案。在“选项”->“设置”->“设备”中，启用“Windows备用连接方法”后，连接方式会发生变化：

1. 点击“选择窗口”按钮
2. 鼠标移动到目标窗口上，单击左键选择
3. 窗口不会嵌入IDE，而是保持独立状态

这种方式的优点是兼容性更好，但缺点是无法在IDE内实时预览窗口内容，调试时需要在IDE和被测应用之间来回切换。

2.3 桌面模式连接

当需要测试涉及多个窗口交互的场景，或者被测应用本身不适合嵌入时，可以使用桌面模式：

1. 在设备面板点击“桌面模式”按钮
2. 整个桌面（包括任务栏）将成为测试目标
3. 脚本录制和回放都基于整个屏幕坐标

桌面模式特别适合测试安装程序、系统对话框等全屏或模态窗口。但需要注意的是，在这种模式下，图像识别可能会受到桌面背景、其他应用窗口的干扰。

2.4 多窗口管理与切换

在实际测试中，经常需要处理多个窗口的切换。Airtest提供了灵活的窗口管理机制：

# 连接第一个窗口（如网易云音乐主窗口）
dev1 = connect_device("Windows:///?title_re=网易云音乐.*")

# 执行一些操作
touch(Template(r"play_button.png"))

# 连接第二个窗口（如设置对话框）
dev2 = connect_device("Windows:///?title_re=设置.*")

# 在两个窗口间切换操作
dev1.touch(Template(r"menu_button.png"))
sleep(1)
dev2.touch(Template(r"save_button.png"))

窗口查找策略对比表：

查找方式	语法示例	优点	缺点
精确标题	Windows:///网易云音乐	精确快速	标题必须完全匹配
正则匹配	Windows:///?title_re=网易.*音乐	灵活，容错性强	性能稍差
窗口类名	Windows:///?class_name=Chrome_WidgetWin_1	最稳定可靠	需要提前知道类名
进程ID	Windows:///?process=1234	精确到进程实例	需要获取进程ID

踩坑提醒：窗口标题可能会随着应用状态变化而改变。比如网易云音乐，播放不同歌曲时标题栏会显示不同的歌曲名。这种情况下，使用正则匹配或窗口类名是更可靠的选择。

3. 网易云音乐实战：从登录到播放的完整流程

现在让我们以网易云音乐Windows版为例，构建一个完整的自动化测试用例。这个用例将覆盖登录、搜索、播放等核心功能。

3.1 测试用例设计

在开始编码之前，我们先规划测试步骤：

1. 启动网易云音乐应用
2. 登录账号（如果未登录）
3. 在搜索框输入歌曲名
4. 点击搜索结果中的播放按钮
5. 验证播放状态
6. 暂停播放并退出

3.2 脚本实现详解

首先，我们需要创建一个新的.air脚本文件。在AirtestIDE中，.air文件实际上是一个文件夹，里面包含Python脚本和相关的图像资源。

# -*- encoding=utf8 -*-
__author__ = "YourName"
from airtest.core.api import *
from airtest.core.win import *

# 自动设置脚本环境
auto_setup(__file__)

# 连接网易云音乐窗口
# 这里使用正则匹配，避免因窗口标题变化导致连接失败
netease_window = connect_device("Windows:///?title_re=网易云音乐.*")

# 如果连接失败，尝试启动应用
if not netease_window:
    print("未找到已启动的网易云音乐，尝试启动...")
    start_app("C:\\Program Files\\Netease\\CloudMusic\\cloudmusic.exe")
    sleep(5)  # 等待应用启动
    netease_window = connect_device("Windows:///?title_re=网易云音乐.*")

# 检查登录状态
if exists(Template(r"tpl_login_button.png", threshold=0.8)):
    print("检测到未登录状态，执行登录操作")
    touch(Template(r"tpl_login_button.png"))
    sleep(2)
    
    # 切换到二维码登录标签（如果默认不是）
    if exists(Template(r"tpl_qrcode_tab.png")):
        touch(Template(r"tpl_qrcode_tab.png"))
        sleep(1)
    
    print("请使用手机网易云音乐扫描二维码登录...")
    # 等待用户扫码登录，这里设置较长的等待时间
    wait(Template(r"tpl_login_success.png"), timeout=60)
    print("登录成功")
else:
    print("已处于登录状态")

# 定位搜索框并搜索歌曲
search_box = exists(Template(r"tpl_search_box.png"))
if search_box:
    touch(search_box)
    sleep(1)
    
    # 清空搜索框（如果有内容）
    keyevent("{CTRL}a")  # 全选
    keyevent("{DEL}")    # 删除
    
    # 输入搜索关键词
    text("周杰伦 晴天")
    sleep(1)
    keyevent("{ENTER}")
    
    # 等待搜索结果加载
    sleep(3)
    
    # 点击第一个搜索结果中的播放按钮
    # 注意：这里需要根据实际UI调整图像模板
    if exists(Template(r"tpl_first_result_play.png", threshold=0.85)):
        touch(Template(r"tpl_first_result_play.png"))
        print("开始播放歌曲")
    else:
        print("未找到播放按钮，尝试其他定位方式")
        # 备用方案：通过相对坐标点击
        # 先找到歌曲条目，然后计算播放按钮的相对位置
        song_item = exists(Template(r"tpl_song_item.png"))
        if song_item:
            # 假设播放按钮在歌曲条目右侧固定位置
            play_pos = (song_item[0] + 100, song_item[1])
            touch(play_pos)
else:
    print("未找到搜索框，测试终止")
    raise AssertionError("搜索框元素缺失")

# 验证播放状态
sleep(2)  # 等待播放开始
if exists(Template(r"tpl_playing_indicator.png", threshold=0.9)):
    print("播放状态验证成功")
    
    # 等待播放10秒
    print("播放中...")
    sleep(10)
    
    # 暂停播放
    if exists(Template(r"tpl_pause_button.png")):
        touch(Template(r"tpl_pause_button.png"))
        print("已暂停播放")
    else:
        # 尝试通过快捷键暂停
        keyevent(" ")
        print("通过空格键暂停播放")
else:
    print("播放状态验证失败")
    # 截图保存错误状态
    snapshot(filename="playback_failed.png")
    raise AssertionError("播放功能异常")

# 退出应用（可选）
# keyevent("{ALT}{F4}")

print("网易云音乐播放测试执行完成")

3.3 图像模板的采集与管理

在Airtest中，图像模板的质量直接决定了测试的稳定性。以下是一些实用的图像采集技巧：

采集最佳实践：

1. 选择特征明显的区域：避免选择纯色、重复图案的区域
2. 适当包含周围上下文：包含一些独特的背景元素可以提高识别率
3. 使用合适的阈值：默认0.7适合大多数情况，对于模糊或变形的图像可适当降低
4. 多分辨率适配：如果需要在不同分辨率的机器上运行，需要采集多套模板

模板管理策略：

test_project.air/
├── main.py              # 主脚本
├── images/              # 图像模板目录
│   ├── login/          # 登录相关图像
│   │   ├── login_button.png
│   │   ├── qrcode_tab.png
│   │   └── login_success.png
│   ├── search/         # 搜索相关图像
│   │   ├── search_box.png
│   │   └── clear_button.png
│   ├── player/         # 播放器相关图像
│   │   ├── play_button.png
│   │   ├── pause_button.png
│   │   └── playing_indicator.png
│   └── common/         # 通用图像
│       └── close_button.png
└── data/               # 测试数据
    └── test_cases.json

代码中的模板引用：

# 使用相对路径引用模板
login_btn = Template(r"images/login/login_button.png", 
                     record_pos=(-0.2, 0.3), 
                     resolution=(1920, 1080),
                     threshold=0.8)

3.4 异常处理与重试机制

在实际测试中，网络延迟、应用响应慢等因素都可能导致操作失败。良好的异常处理机制是测试脚本健壮性的关键。

def safe_touch(template, max_retries=3, interval=1):
    """带重试机制的点击操作"""
    for attempt in range(max_retries):
        try:
            pos = exists(template)
            if pos:
                touch(pos)
                return True
            else:
                print(f"第{attempt+1}次尝试：未找到元素 {template.filename}")
                sleep(interval)
        except Exception as e:
            print(f"第{attempt+1}次尝试发生异常：{str(e)}")
            sleep(interval)
    
    print(f"经过{max_retries}次尝试仍未找到元素 {template.filename}")
    return False

def wait_and_assert(template, timeout=10, assert_msg="元素未出现"):
    """等待元素出现并断言"""
    result = wait(template, timeout=timeout)
    if result:
        return result
    else:
        snapshot(filename=f"assert_failed_{int(time.time())}.png")
        raise AssertionError(assert_msg)

# 在测试中使用
if not safe_touch(Template(r"tpl_play_button.png"), max_retries=3):
    # 备用方案：尝试键盘快捷键
    keyevent("P")  # 假设P是播放快捷键
    sleep(2)
    
    # 验证播放是否成功
    wait_and_assert(Template(r"tpl_playing_indicator.png"), 
                   timeout=5,
                   assert_msg="播放功能异常")

4. 高级技巧与最佳实践

掌握了基础操作后，让我们来看看一些提升测试效率和稳定性的高级技巧。

4.1 使用相对坐标与动态定位

完全依赖图像识别有时会过于脆弱，特别是当UI有微小变化时。结合相对坐标可以大大提高脚本的适应性。

def click_relative_to_template(base_template, offset_x=0, offset_y=0):
    """基于基准模板的相对坐标点击"""
    base_pos = exists(base_template)
    if base_pos:
        target_x = base_pos[0] + offset_x
        target_y = base_pos[1] + offset_y
        touch((target_x, target_y))
        return True
    return False

# 示例：点击搜索框右侧的"清除"按钮
# 先找到搜索框，然后向右偏移100像素点击
search_box = wait(Template(r"tpl_search_box.png"), timeout=5)
if search_box:
    click_relative_to_template(Template(r"tpl_search_box.png"), 
                               offset_x=150, 
                               offset_y=0)

4.2 处理动态内容与等待策略

Windows应用中的动态内容（如加载动画、进度条）需要特殊的等待策略。

def wait_for_stable(template, stable_time=2, check_interval=0.5, timeout=30):
    """等待元素稳定出现（不再闪烁或变化）"""
    start_time = time.time()
    last_seen_time = 0
    
    while time.time() - start_time < timeout:
        if exists(template):
            current_time = time.time()
            if current_time - last_seen_time >= stable_time:
                return True
            last_seen_time = current_time
        else:
            last_seen_time = 0
        
        sleep(check_interval)
    
    return False

def wait_for_disappear(template, timeout=10, check_interval=0.5):
    """等待元素消失"""
    start_time = time.time()
    while time.time() - start_time < timeout:
        if not exists(template):
            return True
        sleep(check_interval)
    return False

# 使用示例：等待加载动画消失后再继续操作
print("等待加载完成...")
if wait_for_disappear(Template(r"tpl_loading_spinner.png"), timeout=30):
    print("加载完成，继续操作")
else:
    print("加载超时，尝试继续或记录错误")

4.3 数据驱动测试

将测试数据与脚本逻辑分离，可以提高测试用例的复用性和可维护性。

import json
import os

class TestData:
    def __init__(self, data_file="test_data.json"):
        self.data_file = data_file
        self.load_data()
    
    def load_data(self):
        if os.path.exists(self.data_file):
            with open(self.data_file, 'r', encoding='utf-8') as f:
                self.data = json.load(f)
        else:
            self.data = {
                "users": [
                    {"username": "test1@example.com", "password": "pass123"},
                    {"username": "test2@example.com", "password": "pass456"}
                ],
                "songs": [
                    {"name": "晴天", "artist": "周杰伦"},
                    {"name": "七里香", "artist": "周杰伦"},
                    {"name": "青花瓷", "artist": "周杰伦"}
                ],
                "playlists": [
                    {"name": "我的最爱", "count": 50},
                    {"name": "工作专注", "count": 30}
                ]
            }
            self.save_data()
    
    def save_data(self):
        with open(self.data_file, 'w', encoding='utf-8') as f:
            json.dump(self.data, f, ensure_ascii=False, indent=2)
    
    def get_test_user(self, index=0):
        return self.data["users"][index % len(self.data["users"])]
    
    def get_test_song(self, index=0):
        return self.data["songs"][index % len(self.data["songs"])]

# 在测试脚本中使用
test_data = TestData()

# 数据驱动的搜索测试
for i in range(len(test_data.data["songs"])):
    song = test_data.get_test_song(i)
    print(f"测试搜索歌曲: {song['name']} - {song['artist']}")
    
    # 执行搜索操作
    touch(Template(r"tpl_search_box.png"))
    text(f"{song['name']} {song['artist']}")
    keyevent("{ENTER}")
    
    # 验证搜索结果
    sleep(2)
    if exists(Template(r"tpl_search_results.png")):
        print(f"歌曲 '{song['name']}' 搜索成功")
    else:
        print(f"歌曲 '{song['name']}' 搜索失败")
        snapshot(filename=f"search_failed_{song['name']}.png")

4.4 测试报告生成与分析

AirtestIDE内置了测试报告生成功能，但我们可以进一步定制化以满足团队需求。

import datetime
import json
from airtest.report.report import LogToHtml

class EnhancedReporter:
    def __init__(self, script_root, log_root, export_dir):
        self.script_root = script_root
        self.log_root = log_root
        self.export_dir = export_dir
        self.test_start_time = datetime.datetime.now()
        self.test_cases = []
    
    def start_test_case(self, case_name, description=""):
        """开始记录一个测试用例"""
        test_case = {
            "name": case_name,
            "description": description,
            "start_time": datetime.datetime.now().isoformat(),
            "steps": [],
            "status": "running"
        }
        self.test_cases.append(test_case)
        return len(self.test_cases) - 1  # 返回用例索引
    
    def add_step(self, case_index, step_name, action, expected, actual, status="passed", screenshot=None):
        """添加测试步骤"""
        step = {
            "name": step_name,
            "action": action,
            "expected": expected,
            "actual": actual,
            "status": status,
            "timestamp": datetime.datetime.now().isoformat(),
            "screenshot": screenshot
        }
        self.test_cases[case_index]["steps"].append(step)
    
    def end_test_case(self, case_index, status="passed", notes=""):
        """结束测试用例"""
        self.test_cases[case_index]["end_time"] = datetime.datetime.now().isoformat()
        self.test_cases[case_index]["status"] = status
        self.test_cases[case_index]["notes"] = notes
    
    def generate_report(self):
        """生成增强版测试报告"""
        # 生成标准Airtest报告
        rpt = LogToHtml(self.script_root, self.log_root, self.export_dir)
        rpt.report()
        
        # 生成自定义JSON报告
        report_data = {
            "project": "网易云音乐自动化测试",
            "test_date": self.test_start_time.date().isoformat(),
            "start_time": self.test_start_time.isoformat(),
            "end_time": datetime.datetime.now().isoformat(),
            "duration": (datetime.datetime.now() - self.test_start_time).total_seconds(),
            "test_cases": self.test_cases,
            "summary": self._generate_summary()
        }
        
        report_file = os.path.join(self.export_dir, "custom_report.json")
        with open(report_file, 'w', encoding='utf-8') as f:
            json.dump(report_data, f, ensure_ascii=False, indent=2)
        
        # 生成简化的HTML报告
        self._generate_html_report(report_data, self.export_dir)
        
        return report_data
    
    def _generate_summary(self):
        """生成测试摘要"""
        total = len(self.test_cases)
        passed = sum(1 for tc in self.test_cases if tc["status"] == "passed")
        failed = total - passed
        
        return {
            "total_cases": total,
            "passed": passed,
            "failed": failed,
            "pass_rate": passed / total * 100 if total > 0 else 0
        }
    
    def _generate_html_report(self, data, output_dir):
        """生成简单的HTML报告"""
        html_template = """
        <!DOCTYPE html>
        <html>
        <head>
            <title>测试报告 - {project}</title>
            <style>
                body {{ font-family: Arial, sans-serif; margin: 20px; }}
                .summary {{ background: #f5f5f5; padding: 15px; border-radius: 5px; margin-bottom: 20px; }}
                .case {{ border: 1px solid #ddd; margin: 10px 0; padding: 10px; border-radius: 5px; }}
                .passed {{ background: #d4edda; }}
                .failed {{ background: #f8d7da; }}
                .step {{ margin: 5px 0; padding: 5px; border-left: 3px solid #ccc; }}
            </style>
        </head>
        <body>
            <h1>{project} - 测试报告</h1>
            <div class="summary">
                <h2>测试摘要</h2>
                <p>测试日期: {test_date}</p>
                <p>总用例数: {total_cases}</p>
                <p>通过: <span style="color: green">{passed}</span></p>
                <p>失败: <span style="color: red">{failed}</span></p>
                <p>通过率: {pass_rate:.1f}%</p>
            </div>
            <h2>详细结果</h2>
            {cases_html}
        </body>
        </html>
        """
        
        # 生成用例HTML
        cases_html = ""
        for case in data["test_cases"]:
            case_class = "case passed" if case["status"] == "passed" else "case failed"
            steps_html = "".join([
                f'<div class="step">{step["name"]} - {step["status"]}</div>'
                for step in case["steps"]
            ])
            
            cases_html += f"""
            <div class="{case_class}">
                <h3>{case["name"]}</h3>
                <p>{case["description"]}</p>
                <p>状态: <strong>{case["status"]}</strong></p>
                <div>{steps_html}</div>
            </div>
            """
        
        html_content = html_template.format(
            project=data["project"],
            test_date=data["test_date"],
            total_cases=data["summary"]["total_cases"],
            passed=data["summary"]["passed"],
            failed=data["summary"]["failed"],
            pass_rate=data["summary"]["pass_rate"],
            cases_html=cases_html
        )
        
        html_file = os.path.join(output_dir, "custom_report.html")
        with open(html_file, 'w', encoding='utf-8') as f:
            f.write(html_content)

# 在测试脚本中使用增强报告
reporter = EnhancedReporter(
    script_root=os.path.dirname(__file__),
    log_root=os.path.join(os.path.dirname(__file__), "log"),
    export_dir=os.path.join(os.path.dirname(__file__), "report")
)

# 记录测试用例
case_id = reporter.start_test_case(
    "网易云音乐播放功能测试",
    "测试歌曲搜索、播放、暂停等核心功能"
)

try:
    # 测试步骤1：搜索歌曲
    reporter.add_step(case_id, "搜索歌曲", 
                     "在搜索框输入歌曲名并搜索", 
                     "显示搜索结果", 
                     "成功显示搜索结果",
                     screenshot="search_success.png")
    
    # 测试步骤2：播放歌曲
    reporter.add_step(case_id, "播放歌曲",
                     "点击搜索结果中的播放按钮",
                     "开始播放歌曲",
                     "歌曲开始播放",
                     screenshot="play_success.png")
    
    # ... 更多测试步骤
    
    reporter.end_test_case(case_id, "passed", "所有功能正常")
    
except Exception as e:
    reporter.add_step(case_id, "异常处理", 
                     "执行测试用例", 
                     "无异常发生", 
                     f"发生异常: {str(e)}",
                     status="failed",
                     screenshot="error.png")
    reporter.end_test_case(case_id, "failed", f"测试执行异常: {str(e)}")
    raise

finally:
    # 生成报告
    report = reporter.generate_report()
    print(f"测试报告已生成: {os.path.join(reporter.export_dir, 'custom_report.html')}")

4.5 集成到CI/CD流水线

将Airtest测试集成到持续集成流程中，可以实现自动化测试的常态化运行。

# ci_runner.py - CI/CD集成运行脚本
import os
import sys
import subprocess
import argparse
from datetime import datetime

def run_airtest_test(script_path, device_str, report_dir=None):
    """运行Airtest测试脚本"""
    if report_dir is None:
        report_dir = os.path.join(os.path.dirname(script_path), 
                                 f"report_{datetime.now().strftime('%Y%m%d_%H%M%S')}")
    
    # 确保报告目录存在
    os.makedirs(report_dir, exist_ok=True)
    
    # 构建命令行
    cmd = [
        "airtest",
        "run",
        script_path,
        "--device",
        device_str,
        "--log",
        os.path.join(report_dir, "log"),
        "--export",
        report_dir
    ]
    
    print(f"执行命令: {' '.join(cmd)}")
    
    # 执行测试
    result = subprocess.run(cmd, capture_output=True, text=True)
    
    # 输出结果
    print("标准输出:", result.stdout)
    if result.stderr:
        print("错误输出:", result.stderr)
    
    # 检查测试结果
    if result.returncode == 0:
        print("测试执行成功")
        
        # 解析测试结果
        success = parse_test_result(report_dir)
        return success
    else:
        print(f"测试执行失败，返回码: {result.returncode}")
        return False

def parse_test_result(report_dir):
    """解析测试结果"""
    log_file = os.path.join(report_dir, "log", "log.txt")
    
    if not os.path.exists(log_file):
        print(f"日志文件不存在: {log_file}")
        return False
    
    with open(log_file, 'r', encoding='utf-8') as f:
        content = f.read()
    
    # 简单的成功判断逻辑
    # 实际项目中可能需要更复杂的解析
    if "FAIL" in content or "AssertionError" in content:
        print("测试用例执行失败")
        return False
    elif "PASS" in content or "所有测试通过" in content:
        print("所有测试用例通过")
        return True
    else:
        # 如果没有明确的成功失败标记，检查是否有错误
        if "ERROR" in content or "Traceback" in content:
            print("测试执行过程中出现错误")
            return False
        else:
            print("测试执行完成，状态未知")
            return True  # 保守起见，返回True

def main():
    parser = argparse.ArgumentParser(description='Airtest CI/CD运行器')
    parser.add_argument('--script', required=True, help='Airtest脚本路径')
    parser.add_argument('--device', default='Windows:///', help='设备连接字符串')
    parser.add_argument('--report-dir', help='报告输出目录')
    parser.add_argument('--slack-webhook', help='Slack Webhook URL用于通知')
    
    args = parser.parse_args()
    
    # 运行测试
    success = run_airtest_test(args.script, args.device, args.report_dir)
    
    # 发送通知（可选）
    if args.slack_webhook:
        send_slack_notification(args.slack_webhook, success, args.script)
    
    # 根据测试结果返回适当的退出码
    sys.exit(0 if success else 1)

def send_slack_notification(webhook_url, success, script_name):
    """发送Slack通知"""
    import requests
    import json
    
    status = "成功" if success else "失败"
    color = "good" if success else "danger"
    
    message = {
        "attachments": [{
            "color": color,
            "title": f"Airtest测试执行{status}",
            "fields": [
                {"title": "测试脚本", "value": script_name, "short": True},
                {"title": "执行时间", "value": datetime.now().strftime("%Y-%m-%d %H:%M:%S"), "short": True},
                {"title": "状态", "value": status, "short": True}
            ]
        }]
    }
    
    try:
        response = requests.post(webhook_url, 
                                data=json.dumps(message),
                                headers={'Content-Type': 'application/json'})
        if response.status_code == 200:
            print("Slack通知发送成功")
        else:
            print(f"Slack通知发送失败: {response.status_code}")
    except Exception as e:
        print(f"发送Slack通知时出错: {str(e)}")

if __name__ == "__main__":
    main()

Jenkins Pipeline配置示例：

pipeline {
    agent any
    
    stages {
        stage('Checkout') {
            steps {
                git branch: 'main', url: 'https://github.com/your-org/your-test-repo.git'
            }
        }
        
        stage('Setup Environment') {
            steps {
                bat 'python -m pip install --upgrade pip'
                bat 'pip install airtest'
                bat 'pip install pocoui'
            }
        }
        
        stage('Run Windows Tests') {
            steps {
                script {
                    // 确保网易云音乐已启动
                    bat 'start "" "C:\\Program Files\\Netease\\CloudMusic\\cloudmusic.exe"'
                    sleep(time: 10, unit: 'SECONDS')
                    
                    // 运行Airtest测试
                    bat 'python ci_runner.py --script tests\\netease_player.air --device "Windows:///?title_re=网易云音乐.*" --report-dir test-reports'
                }
            }
            post {
                always {
                    // 清理：关闭应用
                    bat 'taskkill /F /IM cloudmusic.exe'
                    
                    // 归档测试报告
                    archiveArtifacts artifacts: 'test-reports/**/*', fingerprint: true
                    
                    // 发布HTML报告
                    publishHTML(target: [
                        allowMissing: false,
                        alwaysLinkToLastBuild: false,
                        keepAll: true,
                        reportDir: 'test-reports',
                        reportFiles: 'custom_report.html',
                        reportName: 'Airtest测试报告'
                    ])
                }
            }
        }
    }
}

4.6 性能监控与资源检查

除了功能测试，我们还可以在自动化测试中加入性能监控和资源检查。

import psutil
import time
from datetime import datetime

class PerformanceMonitor:
    def __init__(self, process_name):
        self.process_name = process_name
        self.metrics = {
            "cpu_percent": [],
            "memory_mb": [],
            "io_read_mb": [],
            "io_write_mb": [],
            "timestamps": []
        }
    
    def start_monitoring(self, interval=1.0):
        """开始监控"""
        self.monitoring = True
        while self.monitoring:
            self._record_metrics()
            time.sleep(interval)
    
    def stop_monitoring(self):
        """停止监控"""
        self.monitoring = False
    
    def _record_metrics(self):
        """记录性能指标"""
        for proc in psutil.process_iter(['pid', 'name']):
            if proc.info['name'] and self.process_name.lower() in proc.info['name'].lower():
                try:
                    p = psutil.Process(proc.info['pid'])
                    
                    # CPU使用率
                    cpu_percent = p.cpu_percent(interval=0.1)
                    
                    # 内存使用（MB）
                    memory_mb = p.memory_info().rss / 1024 / 1024
                    
                    # IO统计
                    io_counters = p.io_counters()
                    io_read_mb = io_counters.read_bytes / 1024 / 1024
                    io_write_mb = io_counters.write_bytes / 1024 / 1024
                    
                    # 记录数据
                    self.metrics["cpu_percent"].append(cpu_percent)
                    self.metrics["memory_mb"].append(memory_mb)
                    self.metrics["io_read_mb"].append(io_read_mb)
                    self.metrics["io_write_mb"].append(io_write_mb)
                    self.metrics["timestamps"].append(datetime.now().isoformat())
                    
                    print(f"[性能监控] {self.process_name}: "
                          f"CPU={cpu_percent:.1f}%, "
                          f"内存={memory_mb:.1f}MB, "
                          f"读IO={io_read_mb:.1f}MB, "
                          f"写IO={io_write_mb:.1f}MB")
                    
                except (psutil.NoSuchProcess, psutil.AccessDenied):
                    continue
    
    def generate_report(self, output_file="performance_report.json"):
        """生成性能报告"""
        import json
        
        report = {
            "process_name": self.process_name,
            "monitoring_start": self.metrics["timestamps"][0] if self.metrics["timestamps"] else None,
            "monitoring_end": self.metrics["timestamps"][-1] if self.metrics["timestamps"] else None,
            "samples": len(self.metrics["timestamps"]),
            "summary": self._calculate_summary(),
            "detailed_metrics": self.metrics
        }
        
        with open(output_file, 'w', encoding='utf-8') as f:
            json.dump(report, f, ensure_ascii=False, indent=2)
        
        return report
    
    def _calculate_summary(self):
        """计算性能指标摘要"""
        if not self.metrics["cpu_percent"]:
            return {}
        
        return {
            "cpu_avg": sum(self.metrics["cpu_percent"]) / len(self.metrics["cpu_percent"]),
            "cpu_max": max(self.metrics["cpu_percent"]),
            "memory_avg_mb": sum(self.metrics["memory_mb"]) / len(self.metrics["memory_mb"]),
            "memory_max_mb": max(self.metrics["memory_mb"]),
            "total_io_read_mb": sum(self.metrics["io_read_mb"]),
            "total_io_write_mb": sum(self.metrics["io_write_mb"])
        }

# 在测试中使用性能监控
def test_with_performance_monitoring():
    """带性能监控的测试"""
    monitor = PerformanceMonitor("cloudmusic.exe")
    
    # 在后台启动监控
    import threading
    monitor_thread = threading.Thread(target=monitor.start_monitoring, args=(2.0,))
    monitor_thread.daemon = True
    monitor_thread.start()
    
    try:
        # 执行测试用例
        print("开始执行功能测试...")
        
        # 测试步骤1：启动应用
        start_time = time.time()
        start_app("C:\\Program Files\\Netease\\CloudMusic\\cloudmusic.exe")
        startup_time = time.time() - start_time
        print(f"应用启动时间: {startup_time:.2f}秒")
        
        # 等待应用完全加载
        sleep(5)
        
        # 测试步骤2：执行一系列操作
        # ... 这里放置你的测试代码 ...
        
        # 测试步骤3：检查内存泄漏
        # 在执行操作前后对比内存使用
        initial_memory = psutil.Process().memory_info().rss / 1024 / 1024
        print(f"初始内存使用: {initial_memory:.1f}MB")
        
        # 执行一些可能引起内存增长的操作
        for i in range(10):
            # 模拟多次操作
            touch(Template(r"tpl_next_song.png"))
            sleep(1)
        
        final_memory = psutil.Process().memory_info().rss / 1024 / 1024
        memory_increase = final_memory - initial_memory
        print(f"最终内存使用: {final_memory:.1f}MB")
        print(f"内存增长: {memory_increase:.1f}MB")
        
        # 断言内存增长在可接受范围内
        assert memory_increase < 50, f"内存泄漏嫌疑：增长{memory_increase:.1f}MB超过阈值"
        
    finally:
        # 停止监控并生成报告
        monitor.stop_monitoring()
        monitor.generate_report("netease_performance.json")
        
        # 分析性能数据
        report = monitor.generate_report()
        summary = report["summary"]
        
        print("\n=== 性能测试摘要 ===")
        print(f"平均CPU使用率: {summary.get('cpu_avg', 0):.1f}%")
        print(f"峰值CPU使用率: {summary.get('cpu_max', 0):.1f}%")
        print(f"平均内存使用: {summary.get('memory_avg_mb', 0):.1f}MB")
        print(f"峰值内存使用: {summary.get('memory_max_mb', 0):.1f}MB")
        print(f"总读取IO: {summary.get('total_io_read_mb', 0):.1f}MB")
        print(f"总写入IO: {summary.get('total_io_write_mb', 0):.1f}MB")
        
        # 性能断言
        assert summary.get('cpu_avg', 100) < 80, "平均CPU使用率过高"
        assert summary.get('memory_max_mb', 1000) < 500, "内存使用超过阈值"

# 运行带性能监控的测试
if __name__ == "__main__":
    test_with_performance_monitoring()

通过这些高级技巧和最佳实践，你可以构建出更加健壮、可维护的Windows桌面应用自动化测试脚本。在实际项目中，我建议先从简单的功能测试开始，逐步添加异常处理、数据驱动、性能监控等高级特性，最终形成一个完整的自动化测试体系。