接口测试多线程执行用例-Pytest-xdist

前言

在目前实际项目下的接口自动化用例会非常多，如果采用单进程串行执行的话会非常耗费时间，在实际项目中如服务端开发对底层的公用模块进行改动，此时我们测试QA就需要回归所有服务的接口，这时执行所有服务下接口自动化用例可能少则需要半小时、多则需要好几个小时，这是很低效率的。为了节省项目回归测试的时间，需要多个测试用例同时并行执行，这就是一种分布式场景来缩短测试用例的执行时间，提高效率。

但是在分布式执行用例要遵循以下的原则：
1.用例之间是相互独立的，没有依赖关系，可以独立运行；
2.用例执行之间没有顺序要求，随机顺序都能正常执行；
3.每个用例都能重复执行，运行结果不会影响到其他用例

Pytest-xdist介绍

目前自己用的框架论对多进程执行用例的优化使用的是pytest-xdist。下面根据网上的资料收集，简单介绍下pytest-xdist。
pytest-xdist是属于进程级别的并发，让自动化测试用例可以分布式执行，从而节省自动化回归测试的时间。pytest-xdist可以通过独特的测试执行模式扩展pytest：

1. 测试运行并行化：如果有多个CPU或主机，则可以将它们用于组合的测试运行。这样可以加快开发速度或使用远程计算机的特殊资源。
2. –looponfail：在子进程中重复运行测试。每次运行之后，pytest 都会等到项目中的文件更改后再运行之前失败的测试。重复此过程，直到所有测试通过，然后再次执行完整运行。
3. 跨平台覆盖：可以指定不同的 Python 解释器或不同的平台，并在所有这些平台上并行运行测试。

pytest-xdist的分布式测试原理

pytest-xdist比较类似于一主多从的结构，master机负责下发命令，控制slave机，而slave机根据master机的命令执行特定测试任务。xdist会产生一个或多个的workers，它都通过master来控制；每个worker负责执行完成的测试用例集，然后按照master的要求运行测试，而master机不执行测试任务。

pytest-xdist实现

安装：在命令行种运行命令进行安装

pip install pytest-xdist。
1

使用：使用pytest命令行参数中指定-n auto/x(自动识别CPU/运行的进程数量)实现多进程执行测试用例。

-n auto:自动获取系统的CPU核数，缺点是启用该参数CPU占用率会非常高，每个进程执行速度会非常慢，所以，worker越多，并不会按照理论时间来进行测试脚本运行效率的提升
-n x:手动指定CPU数量
1
2

下面用项目的部分用例来对比单线程与多线程执行用例的情况。
为了更加突出多线程的执行效率，在http_request函数请求前会加上sleep(1)的等待时间。

单线程执行命令：

pytest -s -v  ./test_case/test_demo1/ ./test_case/test_demo2/
1

总共收集到25条用例，单线程的执行耗时是35.61s

pytest-xdist多线程执行命令：

pytest -s -v -n auto ./test_case/test_demo1/ ./test_case/test_demo2/
1

从执行命令后打印的日志可以看到，本地一共启用了8个worker,gw0-gw7，读取的系统默认CPU核数为8，可以看到已经开启多线程执行用例。

总共收集到25条用例，多线程的执行耗时是9.1s

踩坑点，收货知识点

在使用多线程执行用例的过程中遇到一个问题，在用例执行前，会执行fixture函数，设置的范围是session，就是fixture函数get_header会被执行多次，导致多线程执行用例时不能用同一个用户token去测试，会导致token过期。

如何让scope=session的fixture在测试用例的session中仅执行一次？

通过排查问题发现原因：
pytest-xdist是让每个worker进程执行属于自己的测试用例集下的所有测试用例。
这意味着在不同进程中，不同的测试用例可能会调用同一个scope范围级别较高（例如session）的fixture，该fixture则会被执行多次，这不符合scope=session的预期。

在网上查询资料得到成功的解决办法：
虽然pytest-xdist没有内置的支持来确保会话范围的fixture仅执行一次，但是可以通过使用锁定文件进行进程间通信来实现。

官方的解决办法

以下用修改get_header函数来介绍解决逻辑

只要根据官方的代码套自己的操作就可了

@pytest.fixture(autouse=True,scope="session")
def get_header(tmp_path_factory,worker_id):
    if worker_id == "master":
        # not executing in with multiple workers, just produce the data and let
        # pytest's fixture caching do its job
        logger.write_msg(INFO, "执行confest.get_header方法,更新header_data")
        #调用账号登录接口方法更新header
        header_data = normal_login().normal_login_request()
        # 登录后会更新env_config的header,ypcookie
        ReadConfig().set_config(file_name=Get_project_path(Get_ENV().get_env_directly()), section='header',option='header_data', value=str(header_data))
        logger.write_msg(INFO, f"master首次执行，数据是{header_data}")
        return header_data
    # get the temp directory shared by all workers
    root_tmp_dir = tmp_path_factory.getbasetemp().parent
    fn = root_tmp_dir / "data.json"
    with FileLock(str(fn) + ".lock"):
        if fn.is_file():
            data = json.loads(fn.read_text())
            logger.write_msg(INFO, f"worker读取缓存文件，数据是{data}")
        else:
            logger.write_msg(INFO, "执行confest.get_header方法,更新header_data")
            # 调用账号登录接口方法更新header
            header_data = normal_login().normal_login_request()
            # 登录后会更新env_config的header,ypcookie
            ReadConfig().set_config(file_name=Get_project_path(Get_ENV().get_env_directly()), section='header',option='header_data', value=str(header_data))
            fn.write_text(json.dumps(header_data))
            logger.write_msg(INFO,f"worker首次执行，数据是{header_data}")
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27

1. 示例是对单线程与多线程做了判断条件，兼容两种方式执行用例
2. 如果worker_id等于master，代表是单线程，则只会执行一次fixture函数
3. 如果worker_id不等于master，代表是多线程，会获取并创建所有子节点共享的临时目录文件，当session首次执行fixture函数会将header_data信息写进缓存文件，第二次执行fixture函数就会直接读取缓存文件里的header_data，不再更新token信息，解决了问题。