移动应用迭代迅速，手工回归测试耗时且易出错。现有自动化方案（如Appium脚本）多分散在开发者本地，缺乏统一调度、设备共享和结果可视化能力。平台集成Appium、ADB和Celery异步任务队列，实现用例在线管理、多设备并发执行、实时日志推送和报告自动生成，帮助中小团队低成本构建私有化测试服务。

一，选型

Flask：轻量级Python Web框架，配合Jinja2模板、Flask-SocketIO扩展，能快速构建RESTful API和实时通信。

Appium + uiautomator2：作为Android UI自动化引擎，支持多设备并发会话，通过appium-python-client驱动。

ADB (Android Debug Bridge)：管理设备连接、状态监控、截图回传。

Celery加Redis：异步执行测试任务，解耦Web请求与长时间运行的脚本，确保前端响应速度。

数据库：MySQL存储用户、项目、用例、设备、任务等结构化数据；Redis用作消息队列和缓存。

前端：Bootstrap+jQuery+Vue.js（部分动态表格/日志组件），与Flask后端通过AJAX/WebSocket交互。

二，系统需求分析

功能性需求：

用户与权限：注册登录，角色分离（管理员/测试员/观察者）。

设备管理：通过ADB发现并注册设备，实时显示在线状态、型号、系统版本、占用情况。

用例管理：支持上传Python测试脚本、在线编辑、分组（项目/模块），并关联所需的Appium Capabilities。

测试任务：选择目标设备和用例集创建任务，可立即执行或定时；执行中实时展示控制台日志和当前屏幕截图。

报告中心：解析测试框架（如pytest）的输出，生成包含通过率、失败详情、步骤截图的HTML报告，并留存历史记录。

非功能性需求：

设备并发执行能力不少于4台；

WebSocket日志延迟<1s；

任务异常时自动回收设备并标记状态。

三，系统架构设计

采用分层B/S架构：

表示层：浏览器端，负责交互、日志展示、报告渲染。

服务层：Flask应用处理请求，调用Celery任务接口；Flask-SocketIO维护WebSocket连接，向客户端推送任务进度。

任务执行层：Celery Worker进程监听Redis队列，从数据库获取任务详情，启动Appium会话并执行用例脚本，结束后采集结果和截图。

设备代理层：每台测试机上运行Appium Server和ADB Daemon，Worker通过adb -s <device_id>和Appium URL分配设备。

数据层：MySQL持久化存储，Redis缓存设备状态和任务队列。

模块划分：

设备管理模块：定期扫描adb devices，入库新设备，更新在线状态；通过设备自旋锁防止并发占用。

用例管理模块：存储脚本文件到服务器，支持基于Page Object模式的模板生成。

任务调度模块：提供任务创建API，Celery异步执行，Worker内使用subprocess或直接调用pytest.main()，并hook日志流。

实时通信模块：Celery任务通过Redis Pub/Sub将日志发布到Flask-SocketIO，再推送到前端。

报告模块：整合Allure框架，在任务完成后调用allure generate，并将报告目录通过Nginx代理访问。

四，实现细节

设备监控占用

python

# 设备状态轮询（Flask中通过APScheduler定时任务）

def update_device_status():

    result = subprocess.run(['adb', 'devices'], capture_output=True, text=True)

    online_devices = parse_adb_result(result.stdout)

    for dev in online_devices:

        Device.query.filter_by(serial=dev.serial).update({'online': True})

    # 占用状态通过Celery任务在开始/结束时标记occupied字段

异步任务执行

python

# celery任务示例

@celery.task(bind=True)

def run_test_task(self, task_id):

    task = TestTask.query.get(task_id)

    device = Device.query.get(task.device_id)

    device.occupied = True

    db.session.commit()

    try:

        # 动态修改Appium配置，指定设备

        cmd = f"pytest {task.case.script_path} --device={device.serial} --alluredir=reports/{task_id}"

        # 捕获输出并通过redis发布

        process = subprocess.Popen(cmd, shell=True, stdout=subprocess.PIPE, stderr=subprocess.STDOUT)

        for line in iter(process.stdout.readline, b''):

            redis_client.publish(f'task_log_{task_id}', line.decode())

        process.wait()

        # 生成报告

        subprocess.run(f"allure generate reports/{task_id} -o reports/{task_id}/html")

    finally:

        device.occupied = False

        db.session.commit()

实时日志推送

Flask-SocketIO监听Redis频道，将消息转发给对应SocketIO房间：

python

@socketio.on('join_task_room')

def handle_join(data):

    join_room(data['task_id'])

    # 启动后台线程监听redis

    socketio.start_background_task(listen_redis_and_emit, data['task_id'])

用例脚本模板

平台内置Appium启动封装，用户只需关注业务逻辑：

python

# 用户用例示例

from test_base import BaseTest

class LoginTest(BaseTest):

    def test_valid_login(self):

        self.driver.find_element_by_id("username").send_keys("admin")

        self.driver.find_element_by_id("password").send_keys("123456")

        self.driver.find_element_by_id("login_btn").click()

        assert self.driver.find_element_by_id("welcome").text == "欢迎"

BaseTest完成driver初始化、连接指定设备、测试后清理，降低使用门槛。

五，系统测试评估

搭建2台Android真机（Android 10/12）和2台模拟器进行功能验证：

设备上线成功率100%，异常断连自动标记离线；

单任务执行与手动运行pytest结果一致，并发4任务无设备争夺错误；

WebSocket日志平均延迟0.3s，截图实时回传正常；

Allure报告正确生成，点击历史可回溯。

性能方面，Celery Worker增加至4时，CPU占用低于30%，内存无泄漏。

平台利用Flask简洁高效的特性，成功整合了Android UI自动化测试的全流程管理，降低了测试人员的使用和维护成本。

集成iOS设备支持、引入AI图像断言、提供分布式执行节点、支持更多测试框架（如Robot Framework），并实现CI/CD流水线对接。为中小团队自动化测试平台的建设提供了可参考的实现路径。