15. Pytest常用插件：pytest-xdist分布式执行用例

一、前言

当我们自动化测试用例非常多的时候，一条条按顺序执行会非常慢，pytest-xdist的出现就是为了让自动化测试用例可以分布式执行，从而节省自动化测试时间，pytest-xdist是属于进程级别的并发。

二、学习目标

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

2.pytest-xdist插件安装

3.pytest-xdist插件应用

三、知识点

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

用例之间是独立的，用例之间没有依赖关系，用例可以完全独立运行【独立运行】
用例执行没有顺序，随机顺序都能正常执行【随机执行】
每个用例都能重复运行，运行结果不会影响其他用例【不影响其他用例】

2.pytest-xdist插件安装

插件安装：

pip install pytest-xdist

3.pytest-xdist插件应用

原理：xdist会产生一个或多个workers，workers都通过master来控制；每个worker负责执行完整的测试用例集，然后按照master的要求运行测试，而master机不执行测试任务

测试用例示例代码：

#test_xdist_demo.py
import time
import pytest
from selenium import webdriver

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

class TestCase():
    '''测试类'''

    def setup(self):
        '''方法级前置操作-每条用例开始前初始化driver'''
        self.driver = webdriver.Chrome(executable_path=r"C:\Users\19344\Desktop\xdist_demo\chromedriver.exe")

    def teardown(self):
        '''方法级后置操作-每条用例结束后关闭浏览器'''
        self.driver.quit()

    def test_case_01(self):
        '''测试用例一'''
        self.driver .get(url)
        self.driver .find_element_by_name('wd').send_keys('Pthon')
        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)

不使用分布式的执行：

============================= test session starts =============================
collecting ... collected 5 items

test_demo.py::TestCase::test_case_01 
test_demo.py::TestCase::test_case_02 
test_demo.py::TestCase::test_case_03 
test_demo.py::TestCase::test_case_04 
test_demo.py::TestCase::test_case_05 

============================= 5 passed in 34.31s ==============================

从结果可以看到，五条测试用例执行完用时34秒，这是不用分布式执行的结果，这次执行是单线程的。

使用分布式的执行：

pytest -n num (代表使用num个CPU)
pytest -n auto
- n auto：可以自动检测到系统的CPU核数；从测试结果来看，检测到的是逻辑处理器的数量
- 使用auto等于利用了所有CPU来跑用例，此时CPU占用率会特别高

说明：建议最多使用1/2的CPU个数来进行执行，消耗资源太多，导致电脑太卡。

#使用命令运行：pytest -n 3 test_demo.py
================================== test session starts =======================================
platform win32 -- Python 3.6.8, pytest-6.2.5, py-1.10.0, pluggy-1.0.0
rootdir: C:\Users\19344\Desktop\xdist_demo, configfile: pytest.ini
plugins: forked-1.3.0, html-3.1.1, metadata-1.11.0, xdist-2.4.0
gw0 [5] / gw1 [5] / gw2 [5]

DevTools listening on ws://127.0.0.1:58590/devtools/browser/0de25300-b3bf-406e-99d0-fec38953e1db

DevTools listening on ws://127.0.0.1:58591/devtools/browser/314de6af-89a9-443d-9477-99202a1b56c0

DevTools listening on ws://127.0.0.1:58592/devtools/browser/ebeb1620-ee4c-413f-9de2-cb84478ccbde
[13268:15344:1022/104131.731:ERROR:chrome_browser_main_extra_parts_metrics.cc(230)] crbug.com/1216328: Checking Bluetooth availability started. Please report if there is no report that this ends.
[13268:3332:1022/104131.732:ERROR:device_event_log_impl.cc(214)] [10:41:31.731] USB: usb_device_handle_win.cc:1048 Failed to read descriptor from node connection: 连到系统上的设备没有发挥作用。 (0x1F)
[13268:15344:1022/104131.733:ERROR:chrome_browser_main_extra_parts_metrics.cc(233)] crbug.com/1216328: Checking Bluetooth availability ended.
[13268:15344:1022/104131.733:ERROR:chrome_browser_main_extra_parts_metrics.cc(236)] crbug.com/1216328: Checking default browser status started. Please report if there is no report that this ends.
...
DevTools listening on ws://127.0.0.1:58767/devtools/browser/45411608-ed69-4109-a5cc-6d76b7764df9

DevTools listening on ws://127.0.0.1:58793/devtools/browser/52da91b3-ba03-426c-8cfe-c1f1a266fb59
..                                                                                                                                                                                                                            [100%]
--------------------- generated html file: file://C:\Users\19344\Desktop\xdist_demo\auto_reports.html --------------
=================================== 5 passed in 15.10s =========================================

可以看到，使用三个进程进行执行，同样的用例用时缩短为15秒，这次的执行是多进程执行的。

自定义执行模式

将按照同一个作用域方法来分组，然后将每个测试组发给可以执行的worker，确保同一个组的测试用例在同一个进程中执行:
```
--dist=loadscope #每个worker按类执行

示例：pytest -v -n 3 --dist=loadscope test_demo.py
```
按照同一个文件名来分组，然后将每个测试组发给可以执行的worker，确保同一个组的测试用例在同一个进程中执行：
```
--dist=loadfile #每个worker按文件执行

示例：pytest -v -n 3 --dist=loadfile test_xdist.py test_xdist_02.py test_xdist_03.py
```
是将每个用例，分别发给所有的执行器worker，相当于开了几个执行器worker，同一个用例就执行几遍：
```
--dist=each

示例：pytest -v -n 3 --dist=each test_xdist.py
```
将待运行的用例随机发给可用的执行器worker，用例执行顺序随机的，目前默认采用这种方式：
```
--dist=load 和 --dist==no

示例：pytest -v -n 3 --dist=load test_xdist.py
```

如何让scope=session的fixture在test session中仅仅执行一次

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

虽然pytest-xdist没有内置的支持来确保会话范围的夹具仅执行一次，但是可以通过使用锁定文件进行进程间通信来实现。

import pytest
from filelock import FileLock


@pytest.fixture(scope="session",autouse=True)
def login(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
        # 如果测试用例有需要，可以返回对应的数据，比如 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:
            """
            【自定义代码块】
            跟上面 if 的代码块一样就行
            """
            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