pytest-xdist分布式执行用例

pytest-xdist可以让自动化测试用例分布式执行，从而节省自动化测试时间， pytest-xdist是属于进程级别的并发

一. 分布式执行用例的设计原则

• 用例之间相互独立，没有依赖关系，用例可以完全独立运行
• 用例执行没有顺序，随机顺序都能正常执行
• 用例可以重复执行，运行结果不会影响其他用例
  二. 安装
  pip3 install pytest-xdist
  三.应用
  原理：xdist会产生一个或多个workers，workers都通过master来控制
  每个worker负责执行完整的测试用例集，然后按照master的要求运行测试，而master机不执行测试任务
  测试代码

# 开发时间：2023/11/6 17:45
import time

from selenium import webdriver
from selenium.webdriver.common.by import By

url = 'https://www.baidu.com'


class TestXdistDemo:
    def setup_method(self):
        self.driver = webdriver.Chrome()

    def teardown_method(self):
        self.driver.quit()

    def test_case_01(self):
        self.driver.get(url)
        time.sleep(1)
        self.driver.find_element(By.NAME, 'wd').send_keys('Python')
        self.driver.find_element(By.ID, 'su').click()
        time.sleep(3)

    def test_case_02(self):
        self.driver.get(url)
        self.driver.find_element(By.NAME, 'wd').send_keys('Java')
        self.driver.find_element(By.ID, 'su').click()
        time.sleep(3)

    def test_case_03(self):
        self.driver.get(url)
        self.driver.find_element(By.NAME, 'wd').send_keys('go')
        self.driver.find_element(By.ID, 'su').click()
        time.sleep(3)

    def test_case_04(self):
        self.driver.get(url)
        self.driver.find_element(By.NAME, 'wd').send_keys('php')
        self.driver.find_element(By.ID, 'su').click()
        time.sleep(3)

    def test_case_05(self):
        self.driver.get(url)
        self.driver.find_element(By.NAME, 'wd').send_keys('c++')
        self.driver.find_element(By.ID, 'su').click()
        time.sleep(3)

不使用分布式的执行pytest

collected 5 items                                                                                                                     

xdisttestcases/test_xdist_demo.py::TestXdistDemo::test_case_01 PASSED
xdisttestcases/test_xdist_demo.py::TestXdistDemo::test_case_02 PASSED
xdisttestcases/test_xdist_demo.py::TestXdistDemo::test_case_03 PASSED
xdisttestcases/test_xdist_demo.py::TestXdistDemo::test_case_04 PASSED
xdisttestcases/test_xdist_demo.py::TestXdistDemo::test_case_05 PASSED

========================================================= 5 passed in 33.32s 

使用分布式执行 pytest -n 3
pytest -n num (num代表使用num个CPU)
pytest -n auto(auto可以自动检测到系统的CPU核数，使用auto等于利用了所有的CPU来跑用例，CPU占用率会特别高)

3 workers [5 items]     
scheduling tests via LoadScheduling

xdisttestcases/test_xdist_demo.py::TestXdistDemo::test_case_03 
xdisttestcases/test_xdist_demo.py::TestXdistDemo::test_case_02 
xdisttestcases/test_xdist_demo.py::TestXdistDemo::test_case_01 
[gw1] PASSED xdisttestcases/test_xdist_demo.py::TestXdistDemo::test_case_02 
xdisttestcases/test_xdist_demo.py::TestXdistDemo::test_case_05 
[gw2] PASSED xdisttestcases/test_xdist_demo.py::TestXdistDemo::test_case_03 
[gw0] PASSED xdisttestcases/test_xdist_demo.py::TestXdistDemo::test_case_01 
xdisttestcases/test_xdist_demo.py::TestXdistDemo::test_case_04 
[gw1] PASSED xdisttestcases/test_xdist_demo.py::TestXdistDemo::test_case_05 
[gw0] PASSED xdisttestcases/test_xdist_demo.py::TestXdistDemo::test_case_04 
========================================================= 5 passed in 15.64s 

可见，使用三个进程执行，同样的用例执行时间从33.32s缩短到16.64s
自定义执行模式
将按照同一个作用域来分组，然后将每个测试组发给可以执行的worker，确保同一个组的测试用例在同一个进程中执行
--dist=loadscope
将按照同一个文件名来分组，然后将每个测试组发给可以执行的worker，确保同一个组的测试用例在同一个进程中执行
--dist=loadfile
将每个用例，分贝发给所有的执行器worker，相当于开了几个执行器worker
--dist=each

问题：如何让scope=session的fixture在test session中仅仅执行一次
pytest-xdist是让每个worker执行属于自己的测试用例集下的所有测试用例，这意味着在不同进程中，不同的测试用例可能会调用同一个scope范围级别较高(例如session)的fixture，该fixture就会被执行多次，这不符scope=session 的预期
解决：可以通过锁定文件进行进程通信来实现

# 开发时间：2023/11/6 18:25
import json
import os
import uuid
from os import environ

import FileLock as FileLock
import pytest


class TestXdist:
    @pytest.fixture(scope='session', autouse=True)
    def login(self, tmp_path_factory, worker_id):
        if worker_id == 'master':
            """
            自定义代码块
            比如：登陆请求，新增数据，清空数据库等等
            """
            uuid_value = uuid.uuid1()
            token = uuid_value.hex
            print('fixture:请求登陆接口，获取token', token)
            os.environ['token'] = token
            return token
        # 如果是分布式运行
        # 获取所有字节点共享的临时目录，无需修改
        root_tmp_dir = tmp_path_factory.getbasetemp().parent
        fn = root_tmp_dir / "data.json"
        with FileLock(str(fn) + ".lock"):
            if fn.is_file():
                # 缓存文件中读取数据，像登陆操作的话就是token
                token = json.loads(fn.read_text())
                print(f"读取缓存文件，token是{token}")
            else:
                """
                自定义代码块
                """
                uuid_value = uuid.uuid1()
                token = uuid_value.hex
                print('fixture:请求登陆接口，获取token', token)
                fn.write_text(json.dumps(token))
                print(f"首次执行，token是{token}")
                # 最好将后续需要保留的数据存在某个地方，比如这里是os的环境变了
            os.environ['token'] = token
        return token

示例只需要执行一次login
当第一次请求这个fixture时，则会利用FileLock仅产生一次fixture数据
当其他进程再次请求这个fixture时，则会从文件中读取数据