python 之 pytest 常用插件(八)

一些pytest常用插件介绍

[TOC]

1、失败重跑

安装:

1
pip install pytest-rerunfailures

使用方式一:命令行使用

1
2
pytest test_class.py --reruns 5 --reruns-delay 1 -vs  #(失败后重新运行5次,每次间隔1秒)
pytest --reruns 5 --reruns-delay 1 -vs test_class.py # 或者放前面也行

测试代码示例:

1
2
3
4
5
6
7
8
9
10
11
12
13
import pytest


@pytest.mark.parametrize('x,y,result', [
(1, 2, 4),
(2, 2, 4),
(100, 100, 200),
(0.1, 0.1, 0.2),
(-1, -1, -2)
], ids=['err', 'int', 'bignum', 'float', 'negative']) # 参数化
def test_add(x, y, result):
assert result == x + y

命令行运行及结果:

1
2
3
4
5
6
7
8
$ pytest --reruns 5 --reruns-delay 1 -sv testcases/test_rerun.py::test_add
# 或者 pytest testcases/test_rerun.py::test_add --reruns 5 --reruns-delay 1 -sv
============================= test session starts ==============================
# 运行结果和下面方式二的一致
=========================== short test summary info ============================
FAILED testcases/test_rerun.py::test_add[int0] - assert 4 == (1 + 2)
===================== 1 failed, 4 passed, 5 rerun in 5.06s =====================
Finished running tests!

使用方式二:装饰器装饰后运行【这个比较好用】

1
2
3
4
5
6
7
8
9
10
11
12
13
import pytest


@pytest.mark.parametrize('x,y,result', [
(1, 2, 4),
(2, 2, 4),
(100, 100, 200),
(0.1, 0.1, 0.2),
(-1, -1, -2)
], ids=['int', 'int', 'bignum', 'float', 'fushu']) # 参数化
@pytest.mark.flaky(reruns=3, reruns_delay=2) # 这里的flaky是pytest-rerunfailures插件提供的在makr函数上动态添加的属性,__getattr__方法可以获取属性
def test_add(x, y, result):
assert result == x + y

输出结果两者一样

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
28
29
30
31
32
33
34
35
36
37
38
39
40
41
============================= test session starts ==============================
platform linux -- Python 3.10.12, pytest-8.1.1, pluggy-1.4.0 -- /home/wang/code/fastApiStudy/venvadmin/bin/python
cachedir: .pytest_cache
rootdir: /home/wang/code/fastApiStudy
configfile: pytest.ini
plugins: anyio-4.2.0, rerunfailures-14.0
collecting ... collected 5 items

testcases/test_rerun.py::test_add[bignum] PASSED
testcases/test_rerun.py::test_add[fushu] PASSED
testcases/test_rerun.py::test_add[float] PASSED
testcases/test_rerun.py::test_add[int1] PASSED
testcases/test_rerun.py::test_add[int0] RERUN
testcases/test_rerun.py::test_add[int0] RERUN
testcases/test_rerun.py::test_add[int0] RERUN
testcases/test_rerun.py::test_add[int0] RERUN
testcases/test_rerun.py::test_add[int0] RERUN
testcases/test_rerun.py::test_add[int0] FAILED

=================================== FAILURES ===================================
________________________________ test_add[int0] ________________________________

x = 1, y = 2, result = 4

@pytest.mark.parametrize('x,y,result', [
(1, 2, 4),
(2, 2, 4),
(100, 100, 200),
(0.1, 0.1, 0.2),
(-1, -1, -2)
], ids=['int', 'int', 'bignum', 'float', 'fushu']) # 参数化
@pytest.mark.flaky(reruns=5, reruns_delay=1) # 这里的flaky是pytest-rerunfailures插件提供的在makr函数上动态添加的属性,__getattr__方法可以获取属性
def test_add(x, y, result):
> assert result == x + y
E assert 4 == (1 + 2)

testcases/test_rerun.py:13: AssertionError
=========================== short test summary info ============================
FAILED testcases/test_rerun.py::test_add[int0] - assert 4 == (1 + 2)
===================== 1 failed, 4 passed, 5 rerun in 5.06s =====================
Finished running tests!

2、多重校验pytest-assume

正常情况在一条测试用例内如果有多条断言,当代码走到某条断言失败了,后面的其他断言就不会继续执行下去了,而使用pytest-assume插件就可以继续执行下去

但是这个插件感觉没必要,大多数情况下,失败就是失败。不需要继续执行,除非特别情况

下载安装

1
pip install pytest-assume

使用方式:不用特别操作,直接用就行

1
2
3
4
5
6
7
8
import pytest
def test_assume():
print('1、错误操作')
pytest.assume(1 == 2)
print('2、正确操作')
pytest.assume(2 == 2)
print('3、错误操作')
pytest.assume(3 == 2)

测试结果:

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
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
============================= test session starts ==============================
platform linux -- Python 3.10.12, pytest-8.1.1, pluggy-1.4.0 -- /home/wang/code/fastApiStudy/venvadmin/bin/python
cachedir: .pytest_cache
rootdir: /home/wang/code/
configfile: pytest.ini
plugins: anyio-4.2.0, assume-2.4.3, rerunfailures-14.0
collecting ... collected 1 item

testcases/test_rerun.py::test_assume 1、错误操作
2、正确操作
3、错误操作
FAILED

=================================== FAILURES ===================================
_________________________________ test_assume __________________________________

tp = <class 'pytest_assume.plugin.FailedAssumption'>, value = None, tb = None

def reraise(tp, value, tb=None):
try:
if value is None:
value = tp()
if value.__traceback__ is not tb:
> raise value.with_traceback(tb)
E pytest_assume.plugin.FailedAssumption:
E 2 Failed Assumptions:
E
E testcases/test_rerun.py:17: AssumptionFailure
E >> pytest.assume(1 == 2)
E AssertionError: assert False
E
E testcases/test_rerun.py:21: AssumptionFailure
E >> pytest.assume(3 == 2)
E AssertionError: assert False

venvadmin/lib/python3.10/site-packages/six.py:718: FailedAssumption
=============================== warnings summary ===============================
testcases/test_rerun.py::test_assume
/home/wang/code/fastApiStudy/venvadmin/lib/python3.10/site-packages/_pytest/runner.py:240: PluggyTeardownRaisedWarning: A plugin raised an exception during an old-style hookwrapper teardown.
Plugin: assume, Hook: pytest_runtest_call
FailedAssumption:
2 Failed Assumptions:

testcases/test_rerun.py:17: AssumptionFailure
>> pytest.assume(1 == 2)
AssertionError: assert False

testcases/test_rerun.py:21: AssumptionFailure
>> pytest.assume(3 == 2)
AssertionError: assert False


For more information see https://pluggy.readthedocs.io/en/stable/api_reference.html#pluggy.PluggyTeardownRaisedWarning
lambda: runtest_hook(item=item, **kwds), when=when, reraise=reraise

-- Docs: https://docs.pytest.org/en/stable/how-to/capture-warnings.html
=========================== short test summary info ============================
FAILED testcases/test_rerun.py::test_assume - pytest_assume.plugin.FailedAssu...
========================= 1 failed, 1 warning in 0.02s =========================
Finished running tests!

3、设定用例执行顺序 pytest-ordering

正常情况,用例会按照代码执行顺序,从上而下的执行,使用ordering插件可自定义执行顺序

安装:

1
pip install  pytest-ordering

使用方法:在用例上面加装饰器

1
@pytest.mark.run(order=2) # 顺序为2号

代码示例:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
@pytest.mark.run(order=1)
def test_case1():
print('case1')

@pytest.mark.run(order=4)
def test_case4():
print('case4')

@pytest.mark.run(order=3)
def test_case3():
print('case3')

@pytest.mark.run(order=2)
def test_case2():
print('case2')

运行结果

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
============================= test session starts ==============================
platform linux -- Python 3.10.12, pytest-8.1.1, pluggy-1.4.0 -- /home/wang/code/fastApiStudy/venvadmin/bin/python
cachedir: .pytest_cache
rootdir: /home/wang/code/
configfile: pytest.ini
plugins: anyio-4.2.0, ordering-0.6, rerunfailures-14.0
collecting ... collected 4 items

testcases/test_rerun.py::test_case1 case1
PASSED
testcases/test_rerun.py::test_case2 case2
PASSED
testcases/test_rerun.py::test_case3 case3
PASSED
testcases/test_rerun.py::test_case4 case4
PASSED

============================== 4 passed in 0.01s ===============================
Finished running tests!

用例依赖(pytest-dependency)

使用该插件可以标记一个testcase作为其他testcase的依赖,当依赖项执行失败时,那些依赖它的test将会被跳过。

安装 :

1
pip install pytest-dependency

使用方法:

用 @pytest.mark.dependency()对所依赖的方法进行标记,

使用@pytest.mark.dependency(depends=[“test_name”])引用依赖,test_name可以是多个。

代码示例:

1
2
3
4
5
6
7
8
9
import pytest

@pytest.mark.dependency()
def test_01():
assert False

@pytest.mark.dependency(depends=["test_01"])
def test_02():
print("执行测试2")

执行用例test_02后的运行结果:

1
2
3
4
5
6
7
8
9
10
11
12
============================= test session starts ==============================
platform linux -- Python 3.10.12, pytest-8.1.1, pluggy-1.4.0 -- /home/wang/code/fastApiStudy/venvadmin/bin/python
cachedir: .pytest_cache
rootdir: /home/wang/code/
configfile: pytest.ini
plugins: anyio-4.2.0, dependency-0.6.0, ordering-0.6, rerunfailures-14.0
collecting ... collected 1 item

testcases/test_rerun.py::test_02 SKIPPED (test_02 depends on test_01) # test_02 依赖 test_01

============================== 1 skipped in 0.01s ==============================
Finished running tests!

分布式测试(pytest-xdist)

pytest-xdist 插件使用新的测试执行模式扩展了 pytest,最常用的是将测试分布在多个 CPU 核心上以加速测试执行: 核心=进程

安装:

1
2
3
pip install pytest-xdist
# 要使用 psutil 检测可用 CPU 的数量,请额外安装 psutil,一般不需要
pip install pytest-xdist[psutil]

使用:

1
pytest -n auto test_xxx.py

这个命令行中,pytest 将生成数量等于可用 CPU 数量的工作进程,并在它们之间随机分配测试,比如说你cpu是20核的,最大分配20个进程数

由于 pytest-xdist 的实现方式, 当命令行中使用了 -s/–capture=no 选项是不起作用的。也就是说终端不会打印你的print内容,也无法禁用捕获

特征:

  • 跨多个 CPU 运行测试:可以跨多个 CPU 或主机执行测试。或使用SSH进行远程账户调用。

  • --looponfail:在子进程中重复运行测试。每次运行后 pytest 都会等待,直到项目中的文件发生更改,然后重新运行之前失败的测试。重复此操作,直到所有测试通过,然后再次执行完整运行(已弃用)。

  • 向远程 SSH 帐户发送测试覆盖范围:您可以指定不同的 Python 解释器或不同的平台,并在所有这些平台上并行运行测试。

在远程运行测试之前,pytest 会有效地将程序源代码“rsync”到远程位置。可以指定不同的 Python 版本和解释器。但是,它不会安装/同步依赖项。

注意:这种模式的存在主要是为了向后兼容,因为现代开发依赖于多平台测试的持续集成。

跨多个 CPU 运行测试

要将测试发送到多个 CPU进程中执行,使用 -n (或 --numprocesses)命令:

1
pytest -n auto
  • 使用-n auto时,pytest-xdist 将使用与本机具有 CPU 内核一样多的进程 【默认常用】

  • 使用 -n logical可使用逻辑 CPU 核心数而不是物理核心数。当前需要安装 psutil 软件包;如果没用,pytest-xdist 将回退到 -n auto 行为。【用得少】这里不多讲

  • 使用-n 8时,指定8个进程执行

  • 也可以直接接到pytest.ini文件当中去 addopts = -n auto

1
2
[pytest]
addopts = -s -v -n auto

给一段测试用例:用默认并发测试:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
import time
import pytest

def test_case1():
time.sleep(1)
print("看看会不会打印:case1") # 会禁用-s 所以不会打印

def test_case2():
time.sleep(1)
print("看看会不会打印:case2") # 会禁用-s 所以不会打印

def test_case3():
time.sleep(1)
print("看看会不会打印:case3") # 会禁用-s 所以不会打印
def test_case4():
time.sleep(1)
print("看看会不会打印:case4") # 会禁用-s 所以不会打印

测试结果:

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
28
29
30
31
32
33
34
35
36
37
38
39
40
41
============================= test session starts ==============================
platform linux -- Python 3.10.12, pytest-8.1.1, pluggy-1.4.0 -- /home/wang/code/fastApiStudy/venvadmin/bin/python
cachedir: .pytest_cache
rootdir: /home/wang/code/
configfile: pytest.ini
plugins: xdist-3.5.0, anyio-4.2.0, dependency-0.6.0, ordering-0.6, Faker-24.9.0, rerunfailures-14.0
created: 20/20 workers # 创建了20个wokers 这里的20个 是你的cpu核心数,也就是进程数
20 workers [4 items] # 20个进程4个项目

scheduling tests via LoadScheduling

testcases/test_xdits_case.py::test_case4
testcases/test_xdits_case.py::test_case1
testcases/test_xdits_case.py::test_case2
testcases/test_xdits_case.py::test_case3
# 这里可以看到分配了4个进程去执行
[gw3] PASSED testcases/test_xdits_case.py::test_case4 # gw3 就是指gevent worker 3
[gw1] PASSED testcases/test_xdits_case.py::test_case2
[gw2] PASSED testcases/test_xdits_case.py::test_case3
[gw0] PASSED testcases/test_xdits_case.py::test_case1

============================== 4 passed in 1.93s ===============================
Finished running tests!
# 如过-n 是 2 的话
============================= test session starts ==============================
.........
created: 2/2 workers
2 workers [4 items] # 2个进程

scheduling tests via LoadScheduling

testcases/test_xdits_case.py::test_case3
testcases/test_xdits_case.py::test_case1
[gw1] PASSED testcases/test_xdits_case.py::test_case3
[gw0] PASSED testcases/test_xdits_case.py::test_case1
testcases/test_xdits_case.py::test_case2
testcases/test_xdits_case.py::test_case4
[gw1] PASSED testcases/test_xdits_case.py::test_case4
[gw0] PASSED testcases/test_xdits_case.py::test_case2

============================== 4 passed in 2.28s ===============================

可以使用以下选项进一步配置并发:

  • --maxprocesses=maxprocesses: 设置最大进程数–工人数 这个和-n 所指定的是一个意思
  • --max-worker-restart: 崩溃时可以重新启动的最大工作线程数(设置为零以禁用此功能)。

--dist设定用于控制分布式测试中的测试用例加载和分发方式。它提供了不同的选项来控制测试用例的加载和分发行为,以满足不同的需求。(distribution)

  • --dist load (默认): 将挂起的测试发送给任何可用的worker工作线程,但是顺序是不固定的。可以使用 –maxschedchunk 选项进行微调来调整,请参阅 pytest –help 的输出。

  • ——dist loadscope: 在主进程中加载和运行测试用例,然后将测试用例按照其作用域(scope)分发到工作进程。这种方式可以更好地控制测试用例的加载行为,特别是在使用_fixture_等 pytest 功能时。按类分组优先于按模块分组。看代码:

  • import time
    import pytest
    
    
    
    class TestScope1: # 一个类算一个级别
        
        def test_case1(self):
            time.sleep(1)
            print("看看会不会打印:case1")
            
        def test_case2(self):
            time.sleep(1)
            print("看看会不会打印:case2")
            
            
    class TestScope2:    
        def test_case3(self):
            time.sleep(1)
            print("看看会不会打印:case3")
        def test_case4(self):
            time.sleep(1)
            print("看看会不会打印:case4")
    def test_case5(): # 一个方法算一个级别
            time.sleep(1)
            print("看看会不会打印:case4")
    # 分析下测试结果
    ============================= test session starts ==============================
    ........
    created: 20/20 workers
    20 workers [5 items] 
    
    scheduling tests via LoadScopeScheduling
    
    testcases/test_xdits_case.py::TestScope1::test_case1 
    testcases/test_xdits_case.py::test_case5 
    testcases/test_xdits_case.py::TestScope2::test_case3 
    [gw2] PASSED testcases/test_xdits_case.py::test_case5  # gw2 来测test_case5
    [gw1] PASSED testcases/test_xdits_case.py::TestScope2::test_case3  # gw1 来测TestScope2
    [gw0] PASSED testcases/test_xdits_case.py::TestScope1::test_case1  # gw0 来测TestScope1
    testcases/test_xdits_case.py::TestScope1::test_case2 
    testcases/test_xdits_case.py::TestScope2::test_case4 
    [gw1] PASSED testcases/test_xdits_case.py::TestScope2::test_case4 # gw1 来测TestScope2
    [gw0] PASSED testcases/test_xdits_case.py::TestScope1::test_case2 # gw0 来测TestScope1
    # 总共分了3个进程去做测试,一个类一个进程,一个方法一个进程,
    
    ============================== 5 passed in 2.94s ===============================
    
    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
    28
    29
    30
    31
    32
    33
    34
    35
    36
    37
    38
    39
    40
    41
    42
    43
    44
    45
    46
    47
    48
    49
    50
    51

    - `——dist loadgroup`:对测试用例按`xdist_group标记`分组。以组为单位分配给可用的worker工人。这保证了具有相同`xdist_group标记`的所有测试都在同一工作线程中运行。如下:

    ```python

    class TestScope1:
    @pytest.mark.xdist_group("group1")
    def test_case1(self):
    time.sleep(1)
    print("看看会不会打印:case1")

    @pytest.mark.xdist_group("group2")
    def test_case2(self):
    time.sleep(1)
    print("看看会不会打印:case2")


    class TestScope2:
    @pytest.mark.xdist_group("group2")
    def test_case3(self):
    time.sleep(1)
    print("看看会不会打印:case3")
    @pytest.mark.xdist_group("group1")
    def test_case4(self):
    time.sleep(1)
    print("看看会不会打印:case4")
    @pytest.mark.xdist_group("group2")
    def test_case5():
    time.sleep(1)
    print("看看会不会打印:case4")
    # 测试结果分析一下
    ============================ test session starts ==============================
    ............
    created: 20/20 workers
    20 workers [5 items]

    scheduling tests via LoadGroupScheduling
    # 这里就很明显了,可以看到对应的测试被分到对应的组去运行了
    testcases/test_xdits_case.py::TestScope1::test_case1@group1
    testcases/test_xdits_case.py::TestScope1::test_case2@group2
    [gw0] PASSED testcases/test_xdits_case.py::TestScope1::test_case2@group2 # 组2
    [gw1] PASSED testcases/test_xdits_case.py::TestScope1::test_case1@group1
    testcases/test_xdits_case.py::TestScope2::test_case3@group2
    testcases/test_xdits_case.py::TestScope2::test_case4@group1
    [gw0] PASSED testcases/test_xdits_case.py::TestScope2::test_case3@group2
    testcases/test_xdits_case.py::test_case5@group2
    [gw1] PASSED testcases/test_xdits_case.py::TestScope2::test_case4@group1
    [gw0] PASSED testcases/test_xdits_case.py::test_case5@group2

    ============================== 5 passed in 3.96s ===============================
    # 这将确保 test1 和 TestA::test2 将在同一个工作线程中运行。没有 xdist_group 标记的测试按照 --dist=load 模式正常分发。
  • ——dist worksteal: 最开始的时候,测试用例会被均匀的分配给可用的woker中,当工作线程完成了大部分分配的测试用例并且没有足够的测试用例来继续(当前每个工作线程在其队列中至少需要两个测试)时,将会尝试从其他工作线程的队列中重新分配(“窃取”)一部分测试用例来跑。运行结果与--dist load方法相似,但是worksteal可以更好地处理持续时间明显不同的测试用例,同时,它可以提供类似或更好的fixtures重用。——这个代码不好演示,大致的意思就是进程1和2都有100条用例,但是1先跑完了,那1就会去窃取2的用例过来跑,

  • --dist no: 正常的pytest执行模式,一次运行一个测试(根本没有分发)。禁用分发

当我们测试用例非常多的时候,比如有1千条用例,假设每个用例执行需要1分钟,如果单个会话去测试,就需要1000分钟才能跑完

使用该插件可以并发去执行,

要求:

  • 各个用例之间互相独立,没有依赖,可完全独立运行
  • 没有顺序要求
  • 可重复运行,且运行结果互不影响

远程 SSH 分布式测试

远程 SSH 分布式测试

官方文档说即将被弃用

假设您有一个包mypkg,其中包含一些可以在本地成功运行的测试。你有一个ssh可达的机器myhost。然后,您可以通过键入:

使用方式:

1
pytest -d  --rsyncdir mypkg --tx ssh=myhostpopen mypkg/tests/unit/test_something.py

可以使用——rsyncdir指定要发送到远程端的多个目录。

为了使pytest正确地收集和发送测试,您不仅需要确保所有代码和测试目录都是同步的,而且任何测试(子)目录也都有一个__init__.py文件,因为pytest内部将tests引用为完全限定的python模块路径。否则,您将在远程端设置过程中得到奇怪的错误。

将测试发送到远程套接字服务器

没有尝试过,就不写了,感觉也不常用


python 之 pytest 常用插件(八)
http://example.com/2024/01/20/668python-之-pytest-常用插件(八)/
作者
Wangxiaowang
发布于
2024年1月20日
许可协议