代码库

单元测试

Xunit风格

import pytest


def setup_module(module):
    print("每个模块之前执行一次")


def teardown_module(module):
    print("每个模块之后执行一次")


class TestCase:
    @classmethod
    def setup_class(cls):
        print("父类的setup_class")


class TestMaShang(TestCase):
    workage = 8

    def setup_class(self):
        """也支持方法的形式实现setupClass"""
        super().setup_class()
        print("每个类之前执行一次")

    def teardown_class(self):
        print("每个类之后执行一次")

    def setup_method(self):
        print("每个用例之前执行一次")

    def teardown_method(self):
        print("每个用例之后执行一次")

    @pytest.mark.smoke  # 标记
    def test_baili(self):
        print("测试百里老师")

    def test_xingyao(self):
        print("测试星瑶老师")

    @pytest.mark.smoke
    def test_yiran(self):
        print("测试依然老师")

    @pytest.mark.skipif(reason="无理由跳过")
    def test_skip_if(self):
        """跳过测试

        使用场景: 反例测试用例
        """

    @pytest.mark.skipif(workage < 10, reason="有条件跳过")
    def test_skip_if_condition(self):
        pass


if __name__ == '__main__':
    pytest.main()

单脚本场景下的单元测试

场景: 一个main脚本放在服务器用定时任务去跑

def main():
    pass

if __name__ == '__main__':
    main()

创建一个main_test.py编写测试用例

接口测试

"""https://www.django-rest-framework.org/api-guide/testing/#api-test-cases"""
from django.urls import reverse
from rest_framework import status
from rest_framework.test import APITestCase
from django.db import models


class Account(models.Model):
    pass


class AccountTest(APITestCase):
    def test_create_account(self):
        """确保我们可以创建一个新的账号"""
        url = reverse('account-list')
        data = {"name": "DabApps"}
        response = self.client.post(url, data, format='json')
        self.assertEqual(response.status_code, status.HTTP_201_CREATED)
        self.assertEqual(Account.objects.count(), 1)
        self.assertEqual(AccountTest.objects.get().name, 'DabApps')

DDT风格

import pytest


# https://docs.pytest.org/en/7.4.x/how-to/parametrize.html#how-to-parametrize-fixtures-and-test-functions
@pytest.mark.parametrize("test_input,expected", [("3+5", 8), ("2+4", 6), ("6*9", 42)], ids=["加法", "加法2", "乘法"])
def test_eval(test_input, expected):
    assert eval(test_input) == expected


# https://docs.pytest.org/en/7.4.x/how-to/fixtures.html#using-markers-to-pass-data-to-fixtures
# 将参数传递到固件
@pytest.fixture
def fixt(request):
    marker = request.node.get_closest_marker("fixt_data")
    if marker is None:
        data = None
    else:
        data = marker.args[0]
    # 处理data
    return data


@pytest.mark.fixt_data(42)
def test_fixt(fixt):
    assert fixt == 42

基于容器的外部依赖

https://gitee.com/luzhenxiong/bilibili-tests/tree/master/testcontainers_

import os

import pytest
import sqlalchemy


# mysql
# =======================================
@pytest.fixture()
def c():
    from testcontainers.mysql import MySqlContainer

    # 3.7.1的testcontainers需要设置环境变量
    os.environ.setdefault('TC_HOST', 'localhost')

    with MySqlContainer('mysql:8.0.35') as mysql:
        engine = sqlalchemy.create_engine(mysql.get_connection_url())
        print(mysql.get_connection_url())
        with engine.connect() as conn:
            yield conn


def test_case(c):
    result = c.execute(sqlalchemy.text("select version()"))
    version, = result.fetchone()
    print(version)


# redis
# =======================================
from testcontainers.redis import RedisContainer


@pytest.fixture(scope='session')
def redis_container():
    # 3.7.1的testcontainers需要设置环境变量
    # os.environ.setdefault('TC_HOST', 'localhost')
    with RedisContainer('redis:7.2.3') as redis_container:
        yield redis_container


@pytest.fixture
def redis_client(redis_container):
    yield redis_container.get_client()


def test_redis(redis_container):
    print(redis_container.get_container_host_ip())
    # 端口号
    # 3.7.1
    # print(redis_container.get_exposed_port(redis_container.port_to_expose))

mock技术

import datetime
import os
from unittest.mock import MagicMock

from pytest_mock import MockFixture, MockType


class UnixFS:
    @staticmethod
    def rm(filename):
        os.remove(filename)


def test_unix_fs(mocker):
    mocker.patch('os.remove')
    UnixFS.rm('file')
    os.remove.assert_called_once_with('file')


# Stub: https://pytest-mock.readthedocs.io/en/latest/usage.html#stub
# 等价于调用MagicMock
def test_stub(mocker: MockFixture):
    def foo(on_something):
        on_something('foo', 'bar')

    # 【pk】MockType: 关联issue[#414](https://github.com/pytest-dev/pytest-mock/issues/414)
    stub: MockType = mocker.stub(name='on_something_stub')
    foo(on_something=stub)
    stub.assert_called_once_with('foo', 'bar')

    # 关联issue: [mocker.stub() is a coroutine function](https://github.com/pytest-dev/pytest-mock/issues/375)
    import inspect
    assert inspect.iscoroutinefunction(stub)
    assert not inspect.iscoroutinefunction(MagicMock(name='on_something_stub'))


# mock类的方法的同时让原方法调用生效
def test_real_call(mocker: MockFixture):
    class Client:
        def create(self):
            print('create')

    c = Client()

    def call(client: Client):
        client.create()

    stub = mocker.stub(name='create')
    stub.side_effect = c.create

    # new参数是关键
    mocker.patch.object(Client, 'create', new=stub)
    call(c)  # create方法仍然生效

    stub.assert_called_once()


# mock时间类型
# https://github.com/spulec/freezegun/blob/master/README.rst
@freeze_time("2012-01-14")
def test():
    assert datetime.datetime.now() == datetime.datetime(2012, 1, 14)

命令行执行测试

pytest -v

测试报告

https://github.com/gotestyourself/gotestsum

go run gotest.tools/gotestsum@latest

测试异步函数

pytest-asyncio

@pytest.mark.asyncio
async def test_some_asyncio_code():
    res = await library.do_something()
    assert b'expected result' == res

测试覆盖

# 运行
$ coverage run --source=app -m pytest
# 命令行输出报告
$ coverage report --show-missing
# 测试结果保存到html
$ coverage html --title "${@-coverage}"

参见

fastapi模板项目的 test.sh