Python单元测试快速指南

为什么要写单元测试

• 减少线上故障，方便重构
• TDD（测试驱动开发）
• 难测的代码就是烂代码

最佳单测库 pytest

• python 社区主流，插件多，使用 方便， pypy , requests, flask 等一众知名项目在使用
• pip install pytest
• 测试自动发现约定
  • test_xxx 开头的文件,函数，Test 开头的类
  • 单独创建 test 目录 或和 源码同级
  • 使用 pytest 即可自动运行所有测试

断言

def func(x):
    return x + 1

def test_answer():
    assert func(3) == 5

##
_______________________________ test_answer ________________________________

    def test_answer():
>       assert func(3) == 5
E       assert 4 == 5
E        +  where 4 = func(3)

test_sample.py:6: AssertionError
========================= short test summary info ==========================
FAILED test_sample.py::test_answer - assert 4 == 5
============================ 1 failed in 0.12s =============================

异常

import pytest

def f():
    raise SystemExit(1)

def test_mytest():
    with pytest.raises(SystemExit):
        f()

批量参数

import pytest

@pytest.mark.parametrize("maybe_false, expected_result", [
    ("", False),
    ([0],True ),
    (dict(), False),
    (0, False),
])
def test_is_palindrome(maybe_false, expected_result):
    assert bool(maybe_false) == expected_result

参数替换

import pytest

@pytest.fixture(scope="module")
def smtp_connection():
    return smtplib.SMTP("smtp.gmail.com", 587, timeout=5)

def test_ehlo(smtp_connection):
    response, msg = smtp_connection.ehlo()
    assert response == 250
    assert b"smtp.gmail.com" in msg
    assert 0  # for demo purposes

模拟 Mock

• pip install pytest-mock

import os

class UnixFS:

    @staticmethod
    def rm(filename):
        os.remove(filename)
# mocker
def test_unix_fs(mocker):
    mocker.patch('os.remove')
    UnixFS.rm('file')
    os.remove.assert_called_once_with('file')

class ForTest:
    field = 'origin'

    def method():
        pass

def test_for_test(mocker):
    test = ForTest()
    mock_method = mocker.patch.object(test, 'method')
    test.method()
    assert mock_method.called

    assert 'origin' == test.field
    mocker.patch.object(test, 'field', 'mocked')
    assert 'mocked' == test.field

• 更多见: https://pytest-mock.readthedocs.io/en/latest/

接口测试

• fastapi

from fastapi import FastAPI

app = FastAPI()

@app.get("/")
async def read_main():
    return {"msg": "Hello World"}

from fastapi.testclient import TestClient

from .main import app

client = TestClient(app)

def test_read_main():
    response = client.get("/")
    assert response.status_code == 200
    assert response.json() == {"msg": "Hello World"}

• pytest-django: https://pytest-django.readthedocs.io/en/latest/tutorial.html

测试覆盖率报告

• pip install pytest-cov
• 执行 pytest --cov --cov-report html

• 测试报告 选择 html, 会生成 htmlcov/index.html 可以看到测试结果 点开有详情

• 更多见https://pytest-cov.readthedocs.io/en/latest/readme.html

性能测试

• pip install pytest-benchmark
• 使用实例 benchmark

import zlib
import pytest
import zstandard as zstd

@pytest.fixture
def example_data():
    s = '{"PMCH31_02":6.7,"PMCH31_01":9.6,"PMCH31_04":3.4,"PMCH31_03":4.6,"PMCH31_06":3.4,"PMCH31_05":3.5,"PMCH31_08":3.5,"PMCH31_07":3.7,"pm100":103.80000000000001,"PMCH31_09":3.5,"PMCH31_11":4.5,"ch16_20":20.5,"lat":0,"PMCH31_10":4.3,"sat_num_pos":0,"pm4_f":84.2,"lng":0,"pm25":68,"PMCH31_24":1.7,"PMCH31_23":3.1,"PMCH31_26":0,"PMCH31_25":0,"PMCH31_28":0,"PMCH31_27":0,"gps_status":1,"PMCH31_29":0,"speed":0,"SN":"B003-02BD","sn":"B003-02BD","PMCH31_31":0,"PMCH31_30":0,"direction":0,"PMCH31_13":6,"PMCH31_12":5,"PMCH31_15":6.1,"PMCH31_14":6.3,"PMCH31_17":5.4,"PMCH31_16":4.7,"PMCH31_19":3.2,"PMCH31_18":4.5,"pm1_f":41.9,"pm10":99.00000000000001,"uploadTime":["2022-11-11 10:54:43.278"],"sensors":[{"subno":1,"DC":555},{"subno":2,"DC":674},{"subno":3,"DC":516},{"subno":4,"DC":610}],"pm03":9.6,"tm":1668135283278,"time":"2022-11-11 10:54:43.278","PMCH31_20":2.7,"PMCH31_22":1.7,"PMCH31_21":2.7,"DC":0}'
    s = s.encode()
    return s

@pytest.mark.benchmark(group="compress")
def test_compress_zlib(benchmark, example_data):
    result = benchmark(zlib.compress, example_data)
    print(
        f"zlib compress len: {len(result)} compress rate: {len(result) / len(example_data) * 100:.2f}"
    )
# benchmark
@pytest.mark.benchmark(group="compress")
def test_compress_zstd(benchmark, example_data):
    result = benchmark(zstd.compress, example_data)
    print(
        f"zstd compress len: {len(result)} compress rate: {len(result) / len(example_data) * 100:.2f}"
    )

@pytest.mark.benchmark(group="decompress")
def test_decompress_zlib(benchmark, example_data):
    compressed_data = zlib.compress(example_data)
    result = benchmark(zlib.decompress, compressed_data)
    assert result == example_data

@pytest.mark.benchmark(group="decompress")
def test_decompress_zstd(benchmark, example_data):
    compressed_data = zstd.compress(example_data)
    result = benchmark(zstd.decompress, compressed_data)
    assert result == example_data

• 运行指定测试 pytest .\test_compress.py -s

• 解读
  • Name: 测试单元名称 时间单位: 1 秒(s) = 1000 毫秒(ms) = 10^6 微秒(us) = 10^9 纳秒(ns)
  • Mean: 平均耗时， 默认按执行时间从短到长排序
  • OPS/S 每秒能执行的操作次数
• 了解更多： https://pytest-benchmark.readthedocs.io/en/latest/usage.html

其他

• 并行加速 pytest 执行: pip install pytest-xdist
  • pytest -n auto .
• pytest --durations=1 查看执行时间超过 1 秒的测试单元
• pytest-memray: 内存占用分析
• docker 启 redis， postgresql … 快速搭建测试环境，减少 mock 模拟补丁