python

关注公众号 jb51net

关闭
首页 > 脚本专栏 > python > pytest多进程或多线程测试

pytest多进程或多线程执行测试实例

作者：网名余先生

这篇文章介绍了pytest多进程或多线程执行测试的实例，文中通过示例代码介绍的非常详细。对大家的学习或工作具有一定的参考借鉴价值，需要的朋友可以参考下

前言：

分布式执行用例的原则：

项目结构

测试脚本

# test1/test_1.py
import time

def test1_test1():
	time.sleep(1)
	assert 1 == 1, "1==1"


def test1_test2():
	time.sleep(1)
	assert 1 == 1, "1==1"
	
	
class TestDemo1:
	def test_inner_1(self):
		time.sleep(1)
		assert 1 == 1, "1==1"


class TestDemo2:
	def test_inner_2(self):
		time.sleep(1)
		assert 1 == 1, "1==1"
# test1/inner/test_3.py
import time

def test3_test1():
	time.sleep(1)
	assert 1 == 1, "1==1"


def test3_test2():
	time.sleep(1)
	assert 1 == 1, "1==1"
	
# test2/test_2.py
import time

def test2_test1():
	time.sleep(1)
	assert 1 == 1, "1==1"


def test2_test2():
	time.sleep(1)
	assert 1 == 1, "1==1"
	
# test2/inner/test_3.py
import time

def test4_test1():
	time.sleep(1)
	assert 1 == 1, "1==1"


def test4_test2():
	time.sleep(1)
	assert 1 == 1, "1==1"

正常执行：需要8.10s

多进程执行用例之pytest-xdist

多cpu并行执行用例，直接加个-n参数即可，后面num参数就是并行数量，比如num设置为3

pytest -v -n num

参数：

多进程并行执行：耗时2.66s大大的缩短了测试用例的执行时间。

pytest-xdist分布式测试的原理：

pytest-xdist分布式测试的流程：

第一步：master创建worker

第二步：workers收集测试项用例

注意：分布式测试（pytest-xdist）方式执行测试时不会输出测试用例中的print内容，因为master并不执行测试用例。

第三步：master检测workers收集到的测试用例集

第四步：master分发测试用例

有以下四种分发策略：命令行参数 --dist=mode选项(默认load)

each：master将完整的测试索引列表分发到每个worker，即每个worker都会执行一遍所有的用例。

load：master将大约$\frac{1}{n}$的测试用例以轮询的方式分发到各个worker，剩余的测试用例则会等待worker执行完测试用例以后再分发；每个用例只会被其中一个worker执行一次。

loadfile：master分发用例的策略为按ids中的文件名(test_xx.py/xx_test.py)进行分发，即同一个测试文件中的测试用例只会分发给其中一个worker；具有一定的隔离性。

loadscope：master分发用例对策略为按作用域进行分发，同一个模块下的测试函数或某个测试类中的测试函数会分发给同一个worker来执行；即py文件中无测试类的话(只有测试function)将该模块分发给同一个worker执行，如果有测试类则会将该文件中的测试类只会分发给同一个worker执行，多个类可能分发给多个worker；目前无法自定义分组，按类 class 分组优先于按模块 module 分组。

注意：可以使用pytest_xdist_make_scheduler这个hook来实现自定义测试分发逻辑。
如：想按目录级别来分发测试用例：

from xdist.scheduler import LoadScopeScheduling


class CustomizeScheduler(LoadScopeScheduling):
	def _split_scope(self, nodeid):
		return nodeid.split("/", 1)[0]


def pytest_xdist_make_scheduler(config, log):
	return CustomizeScheduler(config, log)
pytest -v -n 4 --dist=loadfile

第五步：worker执行测试用例

第六步：测试结束

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

pytest-xdist 没有内置的支持来确保会话范围的 fixture 仅执行一次，但是可以通过使用锁定文件进行进程间通信来实现；让scope=session 的 fixture 在 test session 中仅执行一次。

示例：需要安装 filelock 包，安装命令pip install filelock

import pytest
from filelock import FileLock

 
@pytest.fixture(scope="session")
def login(tmp_path_factory, worker_id):
    # 代表是单机运行
    if worker_id == "master":
        token = str(random())
        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():  # 代表已经有进程执行过该fixture
            token = json.loads(fn.read_text())
        else:  # 代表该fixture第一次被执行
            token = str(random())
            fn.write_text(json.dumps(token))
        # 最好将后续需要保留的数据存在某个地方，比如这里是os的环境变量
        os.environ['token'] = token
	return token

多线程执行用例之pytest-parallel

用于并行和并发测试的 pytest 插件

pip install pytest-parallel

常用参数配置

如果两个参数都配置了，就是进程并行；每个进程最多n个线程，总线程数：进程数*线程数

【注意】

示例：

import pytest


def test_01():
    print('测试用例1操作')

def test_02():
    print('测试用例2操作')

def test_03():
    print('测试用例3操作')

def test_04():
    print('测试用例4操作')
    
def test_05():
    print('测试用例5操作')

def test_06():
    print('测试用例6操作')
    
def test_07():
    print('测试用例7操作')

def test_08():
    print('测试用例8操作')


if __name__ == "__main__":
    pytest.main(["-s", "test_b.py", '--workers=2', '--tests-per-worker=4'])

pytest-parallel与pytest-xdist对比说明:

简而言之，pytest-xdist并行性pytest-parallel是并行性和并发性。

到此这篇关于pytest多进程或多线程执行测试的文章就介绍到这了。希望对大家的学习有所帮助，也希望大家多多支持脚本之家。

您可能感兴趣的文章:
阅读全文