研效优化实践：Python单测——从入门到起飞

腾讯技术工程官方号

发布于 2021-07-19 10:40:08

1.8K0

发布于 2021-07-19 10:40:08

文章被收录于专栏：腾讯技术工程官方号的专栏腾讯技术工程官方号的专栏腾讯技术工程官方号的专栏

作者：uniquewang，腾讯安全平台后台开发工程师

福生于微，积微成著，一行代码的精心调试，一条指令的细心验证，一个字节的研磨优化，都是影响企业研发效能工程的细节因素。而单元测试，是指针对软件中的最小可测试单元的检查验证，一个单元测试往往就是一小段代码。本文基于腾讯安全平台部的研效优化实践，介绍和总结公司第三大后端开发语言 python 的单测编写方法，面向单测 0 基础同学，欢迎共同交流探讨。

前言

本文面向单测 0 基础的同学，介绍和总结python的单测编写方法。首先会介绍主流的单测框架，重点 pytest。第二部分介绍如何使用 Mock 来辅助实现一些复杂场景测试，第三部分单测覆盖率统计。中间穿插借助 IDE 工具来提效的手段

一、python 单测框架

单测框架无外乎封装了测试相关的核心能力来辅助我们快速进行单测，例如 java 的junit，golang 的gocover，python 目前的主流单测框架有unittest，nose，pytest

unittest

unittest 是 python 官方标准库中自带的单元测试框架，又是也称为 PyUnit。类似于 JUnit 是 java 语言的标准单元测试框架一样。unittest 对于 python2.7+,python3 使用方法一致。

基本实例

# test_str.py
import unittest

class TestStringMethods(unittest.TestCase):

    def setUp(self):
        # 单测启动前的准备工作，比如初始化一个mysql连接对象
        # 为了说明函数功能，测试的时候没有CMysql模块注释掉或者换做print学习
        self.conn = CMysql()

    def tearDown(self):
        # 单测结束的收尾工作，比如数据库断开连接回收资源
        self.conn.disconnect()

    def test_upper(self):
        self.assertEqual('foo'.upper(), 'FOO')

    def test_isupper(self):
        self.assertTrue('FOO'.isupper())
        self.assertFalse('Foo'.isupper())

    def test_split(self):
        s = 'hello world'
        self.assertEqual(s.split(), ['hello', 'world'])
        # check that s.split fails when the separator is not a string
        with self.assertRaises(TypeError):
            s.split(2)

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

编写方法

test_str.py，测试文件名约定 test_xxx
import unittest ，python 自带，无需额外 pip 安装
class 名字 Test 打头
类继承 unittest.TestCase
每个单测方法命名 test_xxx
每个测试的关键是：调用 assertEqual() 来检查预期的输出；调用 assertTrue() 或 assertFalse() 来验证一个条件；调用 assertRaises() 来验证抛出了一个特定的异常。使用这些方法而不是 assert 语句是为了让测试运行者能聚合所有的测试结果并产生结果报告。注意这些方法是 unnitest 模块的方法，需要使用 self 调用。
setUp(）方法，单测启动前的准备工作，比如初始化一个 mysql 连接对象
tearDown(）方法，单测结束的收尾工作，比如数据库断开连接回收资源。setUp 和 tearDown 非常类似于 java 里的切面编程
unittest.main() 提供了一个测试脚本的命令行接口

参数化

标准库的 unittest 自身不支持参数化测试，需要通过第三方库来支持：parameterized 和 ddt。官方文档这里完全没做介绍，暂不深入

执行结果

...
----------------------------------------------------------------------
Ran 3 tests in 0.001s

OK

nose

nose 是 Python 的一个第三方单元测试框架。这意味着，如果要使用 nose，需要先显式安装它

pip install nose

一个简单的 nose 单元测试示例如下：

import nose

def test_example ():
    pass

if __name__ == '__main__':
    nose.runmodule()

需要注意的是，nose 已经进入维护模式，最近官方已经没有提交记录，最近的 relase 是 jun 2,2015

这里由于使用经验有限，也没有深入去调研当前的一个使用情况，就不做进一步介绍，有兴趣自行 google。

pytest

先放官方 slogan

pytest: helps you write better programs The pytest framework makes it easy to write small tests, yet scales to support complex functional testing for applications and libraries.

pytest 得益于其简单的实现方案、丰富的参数化功能、易用的前后置逻辑（固件）特性，以及通用的 mock 功能，目前在是非常火爆的 python 单测框架。

安装

pytest 是第三方包，使用功能需要提前安装，支持 python2.7 和 python3.5 及以上

pip install pytest

测试发现

pytest 在所选目录中查找test_*.py或*_test.py文件。在选定的文件中，pytest 在类之外查找带前缀的测试函数，并在带前缀的测试类中查找带前缀的测试方法（无__init__()方法）。

基本实例

直接放官网的几个例子感受一下

# content of test_sample1.py
def inc(x):
    return x + 1


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

# content of test_class.py
import pytest

def f():
    raise SystemExit(1)

class TestClass:
    def test_one(self):
        x = "this"
        assert "h" in x

    def test_two(self):
        x = "hello"
        assert hasattr(x, "check")

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

运行

$ pytest

无参数，运行当前目录及子目录下所有的测试文件，发现规则见上

$ pytest test_sample1.py

运行指定测试文件

$ pytest test_class.py::TestClass

[root test]# pytest test_class.py::TestClass
============================================ test session starts =============================================
platform linux -- Python 3.6.8, pytest-6.2.4, py-1.10.0, pluggy-0.13.1
rootdir: /data/ftp/unique/mf_web_proj_v2/test
collected 3 items

test_class.py .F.                                                                                      [100%]

================================================== FAILURES ==================================================
_____________________________________________ TestClass.test_two _____________________________________________

self = <test_class.TestClass object at 0x7f55f3e0e518>

    def test_two(self):
        x = "hello"
>       assert hasattr(x, "check")
E       AssertionError: assert False
E        +  where False = hasattr('hello', 'check')

test_class.py:13: AssertionError
========================================== short test summary info ===========================================
FAILED test_class.py::TestClass::test_two - AssertionError: assert False
======================================== 1 failed, 2 passed in 0.03s =========================================

运行指定指定测试类。根据运行结果可以看出 test_two 测试方法失败

$ pytest test_class.py::TestClass::test_two

运行指定测试类下的指定测试方法

跳过指定测试

通过@pytest.mark.skip注解跳过指定的测试，跳过测试的原因比如有当前方法已经发现有 bug 还在 fix，测试的数据数据库未就绪，环境不满足等等

# content of test_skip.py
import pytest
@pytest.mark.skip
def test_min():
    values = (2, 3, 1, 4, 6)
    assert min(values) == 1

def test_max():
    values = (2, 3, 1, 4, 6)
    assert 5 in values

标记函数

类似于上面的 skip，也可指定其他的标签，使用 @pytest.mark 在函数上进行各种标记。比如 fixed，finished 等等你想要的各种标签。然后在运行的时候可以根据指定的标签跑某些标签的测试方法。

# test_with_mark.py

@pytest.mark.finished
def test_func1():
    assert 1 == 1

@pytest.mark.unfinished
def test_func2():
    assert 1 != 1

测试时使用-m选择标记的测试函数

$ pytest -m finished test_with_mark.py

参数化测试

通过参数化测试，我们可以向断言中添加多个值。这个功能使用频率非常高，我们可以模拟各种正常的、非法的入参。当然很多同学习惯直接在函数内部构造一个参数集合，通过 for 循环挨个测，通过 try/catch 的方式捕获异常使得所有参数都跑一遍，，但要分析测试结果就需要做不少额外的工作。在 pytest 中，我们有更好的解决方法，就是参数化测试，即每组参数都独立执行一次测试。使用的工具就是 @pytest.mark.parametrize(argnames, argvalues)。在函数内部的 for 循环模式，会当做一次测试用例，而采用pytest.mark.parametrize方式会产生 N 个测试用例，N=len(argnames)。

# test_parametrize_sigle.py
# 单参数
import pytest
@pytest.mark.parametrize('passwd',
                      ['123456',
                       'abcdefdfs',
                       'as52345fasdf4'])
def test_passwd_length(passwd):
    assert len(passwd) >= 8

# test_parametrize_multiple.py
# 多参数
import pytest
@pytest.mark.parametrize('user, passwd',
                         [('jack', 'abcdefgh'),
                          ('tom', 'a123456a')])
def test_passwd_md5(user, passwd):
    db = {
        'jack': 'e8dc4081b13434b45189a720b77b6818',
        'tom': '1702a132e769a623c1adb78353fc9503'
    }

    import hashlib

    assert hashlib.md5(passwd.encode()).hexdigest() == db[user]

argnames可以是用逗号分隔的字符串，也可以是列表，比如上面的'user, passwd'，或者['user','passwd']

argvalues类型是 List[Tuple]。

fixture

定义

fixture 翻译过来是固定，固定状态；固定物；【机械工程】装置器，工件夹具，直接看官方的解释更好理解。

In testing, a fixture provides a defined, reliable and consistent context for the tests. This could include environment (for example a database configured with known parameters) or content (such as a dataset). Fixtures define the steps and data that constitute the arrange phase of a test (see Anatomy of a test). In pytest, they are functions you define that serve this purpose. They can also be used to define a test’s act phase; this is a powerful technique for designing more complex tests.

谷歌翻译

在测试中，fixture 为测试提供定义的、可靠的和一致的上下文。这可能包括环境（例如配置了已知参数的数据库）或内容（例如数据集）。 Fixtures 定义了构成测试编排阶段的步骤和数据（参见 Anatomy of a test） . 在 pytest 中，它们是您定义的用于此目的的函数。它们还可用于定义测试的行为阶段；这是设计更复杂测试的强大技术。

总结下就是使用fixture可以为你的测试用例定义一些可复用的、一致的功能支持，其中最常见的可能就是数据库的初始连接和最后关闭操作，测试数据集的统一提供接口。功能类似于上面unittest框架的 setup()和 teardown()。同时也是 pytest 更加出众的地方，包括：

有独立的命名，并通过声明它们从测试函数、模块、类或整个项目中的使用来激活。
按模块化的方式实现，每个 fixture 都可以互相调用。
fixture 的范围从简单的单元测试到复杂的功能测试，可以对 fixture 配置参数，或者跨函数 function，类 class，模块 module 或整个测试 session 范围。

使用

# 官方Quick example
import pytest


class Fruit:
    def __init__(self, name):
        self.name = name
        self.cubed = False

    def cube(self):
        self.cubed = True


class FruitSalad:
    def __init__(self, *fruit_bowl):
        self.fruit = fruit_bowl
        self._cube_fruit()

    def _cube_fruit(self):
        for fruit in self.fruit:
            fruit.cube()


# Arrange
@pytest.fixture   #1
def fruit_bowl():
    return [Fruit("apple"), Fruit("banana")]


def test_fruit_salad(fruit_bowl): #2
    # Act
    fruit_salad = FruitSalad(*fruit_bowl)

    # Assert
    assert all(fruit.cubed for fruit in fruit_salad.fruit)

使用很简单：

1 通过@pytest.fixture装饰器装饰一个函数
2 直接将 fixture 作为参数传给测试用例，这样就可以做到测试用例只关心当前的测试逻辑，数据准备等交给 fixture 来搞定

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
import sqlite3
import pytest

@pytest.fixture
def connection():
    connection = sqlite3.connect(':memory:') # 1
    yield connection # 2
    connection.close() # 3


@pytest.fixture(autouse=True) # 4
def insert_one_item(connection):
    session = connection.cursor()
    session.execute('''CREATE TABLE numbers
                          (id int , existing boolean)''')
    session.execute('INSERT INTO numbers VALUES (101, 1)')
    connection.commit()

def test_exist_num(connection):
    session = connection.cursor()
    session.execute('''SELECT COUNT(1) FROM numbers WHERE id=101''')
    results = session.fetchall()
    assert len(results) > 0

这是一个通过 fixture 来管理 db 连接的例子，

1 前置准备
2 yield 关键词将 fixture 分为两部分，yield 之前的代码属于预处理，会在测试前执行；yield 之后的代码属于后处理，将在测试完成后执行。如果没有返回给yield即可
3 结束收尾
4 @pytest.fixture(autouse=True) autouse 关键字告诉框架在跑用例之前自动运行该 fixture

作用域

通过 scope 参数声明作用域，可选项有：

function: 函数级，每个测试函数都会执行一次固件；
class: 类级别，每个测试类执行一次，所有方法都可以使用；
module: 模块级，每个模块执行一次，模块内函数和方法都可使用；
session: 会话级，一次测试只执行一次，所有被找到的函数和方法都可用。

@pytest.fixture(scope='function')
def func_scope():
    pass

默认的作用域为 function

管理

可以单独在每个测试 py 文件中放置需要的 fixture
对于复杂项目，可以在不同的目录层级定义 conftest.py，其作用域为其所在的目录和子目录。例如上面的fixture/connection，就应该放在公用的 conftest.py 中，统一管理测试数据的连接。这样就很好的做到的复用。

借助 IDE 提效

已 PyCharm 为例介绍，vscode 等 ide 应该大同小异

Settings/Preferences | Tools | Python Integrated Tools选择单测框架

创建目标测试代码文件 Car.py

# content of Car.py
class Car:

    def __init__(self, speed=0):
        self.speed = speed
        self.odometer = 0
        self.time = 0

    def say_state(self):
        print("I'm going {} kph!".format(self.speed))

    def accelerate(self):
        self.speed += 5

    def brake(self):
        self.speed -= 5

    def step(self):
        self.odometer += self.speed
        self.time += 1

    def average_speed(self):
        if self.time != 0:
            return self.odometer / self.time
        else:
            pass

测试 break 方法，右键或者 Ctrl+Shift+T 选中break()，选择Go To | Test。当然也可以直接直接右键一次性为多个方法创建对应测试用例

点击Create New Test...，创建测试文件

2.png

完善测试代码逻辑

3.png

点击运行按钮，可以选择运行测试或者调试测试

4.png

运行结果，4 个测试用例，有 2 个失败。

二、Mock

上面的介绍的 pytest 框架可以辅助我们解决掉日常工作 70%的单测问题，但是对于一些不容易构造/获取的对象，需要依赖外部其他接口，特定运行环境等场景，需要借助 Mock 工具来帮我们构建全面的单测用例，而不至于卡在某一环境无法继续推进。

举一个实际工作中分布式任务下发的场景，master 节点需要通过调用资源管理服务下的 worker 节点 cpu 占用率、内存占用率等多项资源接口，来评估任务下发哪些节点。但是现在这些资源接口部署在 idc 环境，或者接口由其他同学在负责开发中，这时我们需要测试调度功能是否正常工作。

根据本人之前的经历，一个简单的办法是搭建一个测试的服务器，然后全部模拟实现一遍这些接口。之前这样做确实也挺爽，但是后边就麻烦了，调用的接口越来越来，每次都要全部实现一遍。最重要的时候测试的时候这个服务还不能挂，不然就跑步起来了。:)

其实还有一个更佳方案就是就是我们用一个 mock 对象替换掉和远端 rpc 服务的交互过程，给定任何我们期望返回的指标值。

unittest

python2.7 需要手动安装 mock 模块

pip install mock

python3.3 开始，mock 模块已经被合并到标准库中，命名为 unittest.mock，可以直接 imort 使用

from unittest import mock

mock 支持对象、方法、异常的写法，直接上代码说话，还是上面提到的分布式任务下发的场景。

ResourceManager负责资源管理，get_cpu_cost通过网络请求拉取节点利用率。

WorkerManager负责管理所有 worker 节点，get_all_works返回所有 worker 节点。

Schedule负责任务调度，核心sch方法挑选一个负载最低的节点下发任务

# content in res.py
class ResourceManager:
    # 资源管理类
    def get_cpu_cost(self,ip):
        # 已经开发完的逻辑
        import requests
        import json
        response = requests.get("http://xx.oa.com/%s"%ip)
        rs = json.loads(response.json())
        return rs["num"]

class WorkerManager:
    def get_all_works(self):
        # 已经开发完的逻辑
        import pymysql
        # 打开数据库连接
        db = pymysql.connect("localhost", "xx", "xx", "TESTDB")
        cursor = db.cursor()
        cursor.execute("SELECT ip from works where alive=1")
        ips = cursor.fetchall()
        db.close()
        return [ip[0] for ip in ips]

# content in schedule.py
from src.demo.res import WorkerManager, ResourceManager


class Schedule:
    def __init__(self):
        self.work_mgr = WorkerManager()
        self.res_mgr = ResourceManager()

    def sch(self):
        ips = self.work_mgr.get_all_works()
        cpu_cost = [self.res_mgr.get_cpu_cost(ip) for ip in ips]
        min_cost_index = cpu_cost.index(min(cpu_cost)) # 获取负载最低的ip
        return ips[min_cost_index]

现在我们要针对Schedule类中的方法进行测试。此时获取节点的方法要从 db 拉取，获取资源负责的方法要从远端网络请求拉取。两部分目前都很难搭建环境，我们希望 mock 这两部分，来测试我们调度的逻辑是否满足挑选负载最低的节点调度。

mock 对象的写法

# content in test_mock.py
import ipaddress
import unittest
from unittest import mock, TestCase

from src.demo.schedule import ResourceManager, WorkerManager #1
from src.demo.schedule import Schedule


def mock_cpu_cost(ip):
    # ip转整数
    return int(ipaddress.IPv4Address(ip))


class TestSchedule(TestCase):
    @mock.patch.object(ResourceManager, 'get_cpu_cost') #2
    @mock.patch.object(WorkerManager, 'get_all_works')
    def test_sch(self, all_workers, cpu_cost): # 3

        workers = ['1.1.1.1', '2.2.2.2']
        cpu_cost.side_effect = mock_cpu_cost #4
        all_workers.return_value = workers # 5

        sch = Schedule()
        res = sch.sch()
        self.assertEqual(res, workers[0]) #6


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

简要说下关键几步的写法：

这里import我们要 mock 的类，下面两种 import 方法都可以

 from src.demo.res import ResourceManager,WorkerManager

 from src.demo.schedule import ResourceManager, WorkerManager

使用@mock.patch.object注解写法，参数ResourceManager是要 mock 的类对象（类名）,'get_cpu_cost'是要 mock 的方法名
我们一次性 mock 两个对象，@mock.patch.object的顺序从下到上来从前到后对应参数名。all_workers和cpu_cost是两个临时名字，你可以自行定义。
重点：cpu_cost.side_effect = mock_cpu_cost用自定义方法替换目标对象方法。
重点：all_workers.return_value = workers用自定义返回值来替换目标对象返回值，直白说就是接下来调用get_all_works都会返回['1.1.1.1', '2.2.2.2']
标准 unittest 写法来判断言

mock 方法的写法

class TestSchedule(TestCase):
    @mock.patch('src.demo.res.ResourceManager.get_cpu_cost')
    @mock.patch('src.demo.res.WorkerManager.get_all_works')
    def test_sch(self, all_workers, cpu_cost): # 3
        # 其他部分完全一致

唯一不同点@mock.patch('src.demo.res.ResourceManager.get_cpu_cost')，参数写方法全路径。

当然 return_value 和 side_effect 也可一次定义

class TestSchedule(TestCase):
    # mock对象的写法
    @mock.patch.object(ResourceManager, 'get_cpu_cost')
    @mock.patch.object(WorkerManager, 'get_all_works',return_value=['1.1.1.1', '2.2.2.2'])
    # mock方法的写法
    # @mock.patch('src.demo.res.ResourceManager.get_cpu_cost')
    # @mock.patch('src.demo.res.WorkerManager.get_all_works')
    def test_sch(self, _, cpu_cost):
        cpu_cost.side_effect = mock_cpu_cost
        # all_workers.return_value = workers

        sch = Schedule()
        res = sch.sch()
        self.assertEqual(res, '1.1.1.1')

pytest

pytest 框架没有提供 mock 模块，使用需要在安装一个包 pytest-mock

pip install pytest-mock

使用方法及写法几乎与unittest.mock完全一致

class TestSchedule:

    def test_sch(self, mocker): # 1
        workers = ['1.1.1.1', '2.2.2.2']
        mocker.patch.object(ResourceManager, 'get_cpu_cost', side_effect=mock_cpu_cost) # 2
        mocker.patch.object(WorkerManager, 'get_all_works', return_value=workers)
        sch = Schedule()
        res = sch.sch()
        assert res == workers[0]

无需手动 import，test 方法参数使用mocker，pytest-mock 会自动注入。名字不能换，只能使用`mocker
写法和 unittest.mock 完全一致。
目前没有找到原生优雅写注解的办法，只能吧 mock 逻辑放到 test 方法中，后边封装后再补充

如果扫一眼源码可以看到 mock 是 pytest_mock.plugin 模块下的一个 fixture

def _mocker(pytestconfig: Any) -> Generator[MockerFixture, None, None]:
    """
    Return an object that has the same interface to the `mock` module, but
    takes care of automatically undoing all patches after each test method.
    """
    result = MockerFixture(pytestconfig)
    yield result
    result.stopall()
mocker = pytest.fixture()(_mocker)  # default scope is function
class_mocker = pytest.fixture(scope="class")(_mocker)
module_mocker = pytest.fixture(scope="module")(_mocker)
package_mocker = pytest.fixture(scope="package")(_mocker)
session_mocker = pytest.fixture(scope="session")(_mocker)

三、覆盖率

覆盖率是用来衡量单元测试对功能代码的测试情况，通过统计单元测试中对功能代码中行、分支、类等模拟场景数量，来量化说明测试的充分度

同 Java 的 JaCoCo、Golang 的 GoCover 等一样，Python 也有自己的单元测试覆盖率统计工具，Coverage 就是使用最广的一种。

安装

pip install coverage

使用

For pytest

coverage run -m pytest arg1 arg2 arg3

For unittest

coverage run -m unittest discover

上报结果

$ coverage report -m
Name                      Stmts   Miss  Cover   Missing
-------------------------------------------------------
my_program.py                20      4    80%   33-35, 39
my_other_module.py           56      6    89%   17-23
-------------------------------------------------------
TOTAL                        76     10    87%

借助 IDE 提效

右键呼出跑整个测试文件

小箭头跑单个测试用例

右侧或者左侧项目树可以看到整个覆盖情况

向上小箭头可以导出覆盖情况报告，Save会直接打开浏览器给出结果，很方便。点击具体文件还有详细说明。

接入公司覆盖率平台

如果所在公司有覆盖率检测平台，接入原理很简单。通过发布流水线集成项目代码，拉取到构建机，将上面在本地跑的 coverage 放到构建机上执行，将结果上报到远端平台。

后记

在腾讯安全平台部实际研发与测试工作中，单元测试是保证代码质量的有效手段，也是效能优化实践的重要一环。本文是笔者在学习 python 单测整个过程的总结，介绍了 python 的几种主流单测框架，Mock 的使用以及使用 coverage 来计算单测覆盖率。推荐使用 pytest 来进行日常测试框架，支持的插件足够丰富，希望可以对有需要接入 python 单测的同学有些帮助。安平研效团队仍在持续探索优化中，若大家在工作中遇到相关问题，欢迎一起交流探讨，共同把研效测试工作做好、做强。

github 仓库地址：

https://github.com/nudt681/python_unittest_guide