Appium Python分布式如何高效执行测试？-杰瑞科技汇

分布式测试的核心思想是 “主从模式”：

（图片来源网络，侵删）

主节点：负责任务调度、测试用例分发、结果汇总和报告生成，它不直接执行测试，而是像一个指挥官。
从节点：也称为 执行节点 或代理，它们是实际执行测试任务的机器，每个从节点负责运行一部分测试用例，并将执行结果返回给主节点。

这种模式的主要优势：

提升执行效率：可以同时运行多个测试任务，大大缩短测试周期。
扩大测试覆盖：可以同时在不同设备（如不同型号的手机、iOS/Android）上并行执行测试。
资源利用率高：充分利用多台机器的计算资源。
增强可扩展性：当需要更多测试能力时，只需增加新的从节点即可。

架构概览

一个典型的 Appium Python 分布式测试架构如下：

核心组件与实现步骤

我们将使用 pytest 作为测试框架，pytest-xdist 插件来实现分布式执行。pytest-xdist 会自动处理任务的分发和结果的收集。

步骤 1：环境准备

1. 所有节点（主节点和所有从节点）的环境配置

（图片来源网络，侵删）

每台机器都需要安装以下软件和 Python 库：

Python: 建议 3.7+ 版本。
Appium Server: 从 Appium 官网下载并安装。
Android/iOS SDK: 根据测试平台准备相应的环境（Android SDK, Xcode 等）。
（图片来源网络，侵删）

必要的 Python 库:

# 在所有机器上执行
pip install Appium-Python-Client pytest pytest-xdist

2. 从节点的特殊配置

每个从节点都需要启动一个 Appium Server，并监听一个 唯一的端口，这是实现并发的关键。

假设你有两台从节点机器：

从节点 A (IP: 192.168.1.10): 启动 Appium Server 在 4723 端口。
从节点 B (IP: 192.168.1.11): 启动 Appium Server 在 4724 端口。

在每台从节点的终端中执行以下命令来启动 Appium Server：

# 在从节点 A (192.168.1.10) 上执行
appium -p 4723 -bp 4723 --allow-insecure=get_session_caps
# 在从节点 B (192.168.1.11) 上执行
appium -p 4724 -bp 4724 --allow-insecure=get_session_caps

-p: Appium REST API 的监听端口。
-bp: Bootstrap 的监听端口，Appium 内部使用，每个节点也需要唯一。

3. 主节点的配置

主节点不需要启动 Appium Server，但它需要知道所有从节点的 IP 地址和端口。

步骤 2：编写测试用例

测试用例的编写方式与单机测试基本相同,关键在于 如何动态地连接到不同的 Appium 服务器。

我们可以使用 pytest 的 fixture 机制，并结合环境变量或配置文件来灵活指定目标服务器。

示例：test_calculator.py

import pytest
from appium import webdriver
from appium.webdriver.common.appiumby import AppiumBy
# 定义一个 fixture，用于获取 WebDriver 实例
# 这个 fixture 会根据传入的 'device_id' 来连接不同的 Appium 服务器
@pytest.fixture
def driver_setup(request):
    """
    Fixture to set up the Appium driver.
    It reads 'device_id' from the test node and connects to the corresponding Appium server.
    """
    # 从 pytest-xdist 的 worker 节点信息中获取设备标识符
    # worker 会是 'gw0', 'gw1', 'gw2'...
    worker_id = request.config.getoption("--dist", default="no") != "no" and request.config.worker_id
    device_id = request.config.getoption("--device")
    if not device_id:
        raise ValueError("Device ID must be provided via --device option")
    desired_caps = {
        "platformName": "Android",
        "deviceName": device_id,  # 使用传入的 device_name
        "appPackage": "com.android.calculator2",
        "appActivity": "com.android.calculator2.Calculator",
        "automationName": "UiAutomator2",
        "udid": device_id, # 对于真实设备，udid 是必需的
        "noReset": True,
        "fullReset": False
    }
    # 根据设备 ID 选择 Appium 服务器地址
    # 这里我们使用一个简单的映射，实际项目中可以用更复杂的配置（如 YAML 文件）
    appium_servers = {
        "emulator-5554": "http://localhost:4723/wd/hub", # 假设主节点也连了一个本地模拟器
        "192.168.1.10:5555": "http://192.168.1.10:4723/wd/hub", # 从节点 A 上的真实设备
        "192.168.1.11:5555": "http://192.168.1.11:4724/wd/hub"  # 从节点 B 上的真实设备
    }
    server_url = appium_servers.get(device_id)
    if not server_url:
        raise ValueError(f"No Appium server configured for device: {device_id}")
    print(f"[{worker_id}] Connecting to {server_url} for device {device_id}")
    driver = webdriver.Remote(server_url, desired_caps)
    driver.implicitly_wait(10)
    yield driver
    print(f"[{worker_id}] Teardown for device {device_id}")
    driver.quit()
# --- 测试用例 ---
def test_addition(driver_setup):
    driver = driver_setup
    # 执行加法 1 + 2 = 3
    driver.find_element(AppiumBy.ID, "digit_1").click()
    driver.find_element(AppiumBy.ID, "plus").click()
    driver.find_element(AppiumBy.ID, "digit_2").click()
    driver.find_element(AppiumBy.ID, "equals").click()
    result_element = driver.find_element(AppiumBy.ID, "result")
    result_text = result_element.text
    assert result_text == "3"
def test_subtraction(driver_setup):
    driver = driver_setup
    # 执行减法 6 - 3 = 3
    driver.find_element(AppiumBy.ID, "digit_6").click()
    driver.find_element(AppiumBy.ID, "minus").click()
    driver.find_element(AppiumBy.ID, "digit_3").click()
    driver.find_element(AppiumBy.ID, "equals").click()
    result_element = driver.find_element(AppiumBy.ID, "result")
    result_text = result_element.text
    assert result_text == "3"

步骤 3：在主节点上执行分布式测试

我们在 主节点 的终端中，使用 pytest 的命令来启动分布式测试。

关键命令：

# 基础命令
pytest -n auto
# 更推荐的命令，明确指定使用 exec 模式和节点数
pytest -d --dist=worksteal --numprocesses=2

-n auto: pytest-xdist 会自动检测到可用的 CPU 核心数，并启动相应数量的 worker 进程。
-d 或 --dist=worksteal: -d 是 --dist=worksteal 的简写，它表示当一个 worker 完成任务后，会从任务队列中“偷取”一个新任务，能更好地平衡负载。
--numprocesses=2: 明确指定启动 2 个 worker 进程，这个数字应该等于你的 从节点数量（或者如果你想在主节点上也运行一个 worker，可以是 从节点数 + 1）。

如何将设备分配给测试用例？

pytest-xdist 会自动将测试用例分发到不同的 worker，我们的 driver_setup fixture 通过 --device 参数来接收设备信息。

执行方式：

我们需要将设备列表和测试用例关联起来,最简单的方式是使用 pytest 的 parametrize 和 --dist=loadscope 策略，确保同一个设备的测试用例被分配到同一个 worker。

假设我们有 3 台设备：emulator-5554, 168.1.10:5555, 168.1.11:5555。

修改 conftest.py (项目根目录下的配置文件)

# conftest.py
import pytest
# 定义所有可用的设备
DEVICES = [
    "emulator-5554",
    "192.168.1.10:5555",
    "192.168.1.11:5555"
]
# 使用 fixture 来提供设备列表
@pytest.fixture(params=DEVICES, scope="session")
def device(request):
    return request.param
# 使用 pytest 的钩子，将设备参数添加到命令行选项中
def pytest_addoption(parser):
    parser.addoption(
        "--device", action="store", default=None, help="UDID or device name to run tests on"
    )
# 使用钩子，在收集测试用例时，将设备参数注入到测试函数中
def pytest_configure(config):
    if config.getoption("--dist") in ("loadscope", "worksteal", "all"):
        # 获取通过 --device 指定的设备
        device_param = config.getoption("--device")
        if device_param:
            # 如果指定了设备，则只为该设备收集测试用例
            config._inicache["device_to_run"] = device_param
        else:
            # 如果没有指定设备，则默认使用 DEVICES 列表中的第一个
            # 更复杂的逻辑可以在这里实现，比如循环使用设备列表
            config._inicache["device_to_run"] = DEVICES[0]
def pytest_collection_modifyitems(config, items):
    device_to_run = getattr(config, "_inicache", {}).get("device_to_run")
    if device_to_run:
        # 过滤出只针对特定设备的测试用例
        # 这需要你的测试用例有特定的标记，@pytest.mark.device("emulator-5554")
        # 这里为了简化，我们假设所有测试用例都使用同一个 driver_setup
        # 更好的做法是给测试用例打上标记，然后根据标记和设备进行过滤
        pass

更简单的执行方式 (推荐)

上面的 conftest.py 比较复杂，一个更直接的方法是，每次只为一个设备执行测试，然后通过脚本循环为所有设备执行。

#!/bin/bash
# run_all_tests.sh
DEVICES=("emulator-5554" "192.168.1.10:5555" "192.168.1.11:5555")
for device in "${DEVICES[@]}"; do
    echo "========================================"
    echo "Running tests on device: $device"
    echo "========================================"
    # 为每个设备启动一个独立的 pytest 进程
    # -n 1 表示这个设备只使用一个 worker (即一个从节点)
    # --dist=worksteal 确保任务在单个 worker 内部平衡
    pytest -n 1 --dist=worksteal --device=$device
    # 如果想在一个命令里并行跑所有设备，可以使用 xdist 的 exec 模式
    # pytest -n 3 --dist=exec --device=$device
    # 但这需要更复杂的 fixture 来处理 worker 和设备的映射
done

最简单且最常用的方法：

直接使用 pytest-xdist 的 exec 模式，它会将测试用例分发到所有可用的 worker，然后我们在 fixture 中通过 worker_id 来映射设备。

最终执行命令（在主节点上）：

# 假设你有3个设备，对应3个从节点
# --dist=exec 会将每个测试用例的执行交给一个 worker
# pytest 会自动将测试用例分组，确保每个 worker 上的测试可以独立运行
pytest -n 3 --dist=exec --devices="emulator-5554,192.168.1.10:5555,192.168.1.11:5555"

然后修改 driver_setup fixture，让它能解析 --devices 列表并根据 worker_id 来选择设备。

# 在 conftest.py 或 fixture 中
@pytest.fixture
def driver_setup(request):
    devices_str = request.config.getoption("--devices")
    devices = devices_str.split(',')
    # worker_id 通常是 'gw0', 'gw1', 'gw2' ...
    worker_id = request.config.worker_id
    # 获取 worker 的索引
    worker_index = int(worker_id.replace('gw', ''))
    if worker_index >= len(devices):
        raise RuntimeError(f"Not enough devices for all workers. Worker {worker_id} has no device.")
    device_id = devices[worker_index]
    # ... 后续的连接代码与之前相同 ...

步骤 4：生成报告

pytest-xdist 会自动收集所有从节点的测试结果，你可以使用 pytest-html 或 Allure 等插件来生成统一的测试报告。

安装报告插件：

pip install pytest-html

执行测试并生成报告：

pytest -n 2 --dist=worksteal --html=report.html --self-contained-html

执行完成后,在主节点的当前目录下会生成一个包含所有测试结果的 report.html 文件。

总结与最佳实践

环境隔离：确保每个从节点上的 Appium Server 端口不冲突，并且能被主节点访问。
动态连接：测试代码（特别是 fixture）必须具备动态连接到不同 Appium 服务器的能力，这是分布式测试的核心。
任务分发策略：
- -n auto / --numprocesses=N: 控制并发度。
- --dist=worksteal: 负载均衡，推荐使用。
- --dist=loadscope: 保证同一个 scope（如一个类或一个模块）的测试用例在同一个 worker 上执行，适用于共享状态的测试。
设备管理：通过命令行参数（如 --device）或配置文件来管理设备列表，并将其与测试用例或 worker 关联起来。
报告统一：利用 pytest 的报告插件，在主节点生成一份完整的测试报告，方便查看和分析。
日志管理：为了方便调试，可以在每个从节点的测试输出中加入 worker ID 或设备信息，print(f"[gw0] ...")。

通过以上步骤,你就可以成功搭建一个基于 Appium 和 Python 的移动端分布式自动化测试框架，极大地提升你的测试效率和覆盖率。

Appium Python分布式如何高效执行测试？

架构概览

核心组件与实现步骤

步骤 1：环境准备

步骤 2：编写测试用例

步骤 3：在主节点上执行分布式测试

步骤 4：生成报告

总结与最佳实践

99ANYc3cd6

Java JDK 1.6.0 已过时，该如何升级？

三星9502如何刷机？教程步骤有哪些？

python opencv2 文字

SQLite Java驱动如何选择与使用？

C视频教程哪里能打包下载？

Java Socket编程实例具体怎么实现？

Win7如何配置Java环境变量？

Java JSON如何转XML？

Linux Python MySQL如何协同工作？

Java中super与extends究竟有何区别？

Java应用与Web开发有何关联？

python mkhtmltopdf

python easyexcel行数

Outlook 2007教程，新手如何快速上手？

3d max 2012教程怎么学？新手入门技巧有哪些？

Java字符串如何转字符串数组？

Appium Python分布式如何高效执行测试？

架构概览

核心组件与实现步骤

步骤 1：环境准备

步骤 2：编写测试用例

步骤 3：在主节点上执行分布式测试

步骤 4：生成报告

总结与最佳实践

相关推荐

Java Socket编程实例具体怎么实现？