深夜：在？我用本地环境pytest带你玩自定义算子

module1：

多玩法Python调试框架pytest

初学入门

大家好python通用测试框架的是unittest+HTMLTestRunner，这段时间看到了pytest文档，发现这个框架和丰富的plugins很好用，所以来学习下pytest.

pytest是一个非常成熟的全功能的Python测试框架，主要有以下几个特点：

• 简单灵活，容易上手
• 支持参数化
• 能够支持简单的单元测试和复杂的功能测试，还可以用来做selenium/appnium等自动化测试、接口自动化测试（pytest+requests）
• pytest具有很多第三方插件，并且可以自定义扩展，比较好用的如pytest-selenium（集成selenium）、pytest-html（完美html测试报告生成）、pytest-rerunfailures（失败case重复执行）、pytest-xdist（多CPU分发）等
• 测试用例的skip和xfail处理
• 可以很好的和jenkins集成
• report框架----allure 也支持了pytest

安装pytest：

pip install -U pytest

验证安装的版本：

pytest --version

pytest documentation中的例子:

例1：

首先我们找一个目录，新建文件夹Pytest，然后新建test_sample.py，输入以下内容。

import pytest

# content of test_sample.py
def func(x):
    return x + 1
def test_answer():
    assert func(3) == 5

如上图，在目录栏输入cmd（也可以直接在IDE中运行）：

回车，打开命令行，输入“pytest”。

pytest返回一个错误报告，因为func(3)不返回5。

例子2：
当需要编写多个测试样例的时候，我们可以将其放到一个测试类当中，新建一个目录test，在目录下新建test_class.py,输入一下内容：

class TestClass:  
    def test_one(self):  
        x = "this"  
        assert 'h' in x  
  
    def test_two(self):  
        x = "hello"  
        assert hasattr(x, 'check') 

pytest -q test_class.py

结果：

从测试结果中可以看到，该测试共执行了两个测试样例，一个失败一个成功。同样，我们也看到失败样例的详细信息，和执行过程中的中间结果。-q即-quiet，作用是减少冗长，具体就是不再展示pytest的版本信息。

如何编写pytest测试样例

通过上面2个实例，我们发现编写pytest测试样例非常简单，只需要按照下面的规则：

• 测试文件以test_开头（以_test结尾也可以）
• 测试类以Test开头，并且不能带有 init 方法
• 测试函数以test_开头
• 断言使用基本的assert即可

运行模式

Pytest的多种运行模式，让测试和调试变得更加得心应手，下面介绍5种常用的模式。需要注意的是，运行测试的文件是有命名的要求的，否则需要额外的命令行指示。pytest命令会找当前目录及其子目录中的所有test_*.py 或 *_test.py格式的文件以及以test开头的方法或者class，不然就会提示找不到可以运行的case了。

1.运行后生成测试报告（htmlReport）

安装pytest-html：

pip install -U pytest-html

运行模式：

pytest --html=report.html

我们可以看到，目录下生成了html文件，点开看看：

报告效果：

我们可以看到排版非常整洁的错误报告，在以上报告中可以清晰的看到测试结果和错误原因，定位问题很容易。

2.运行指定的case

当我们写了较多的cases时，如果每次都要全部运行一遍，无疑是很浪费时间的，通过指定case来运行就很方便了。

测试用例，新建test2目录，目录下新建test_se.py文件：

class TestClassOne(object):
    def test_one(self):
        x = "this"
        assert 't'in x

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


class TestClassTwo(object):
    def test_one(self):
        x = "iphone"
        assert 'p'in x

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

运行模式：

模式1：直接运行test_se.py文件中的所有cases:

pytest test_se.py

模式2：运行test_se.py文件中的TestClassOne这个class下的两个cases:

pytest test_se.py::TestClassOne

模式3：运行test_se.py文件中的TestClassTwo这个class下的test_one:

pytest test_se.py::TestClassTwo::test_one

注意：定义class时，需要以T开头，不然pytest是不会去运行该class的。

我们可以看到，三者的用时是递减的。

3.多进程运行cases

当cases量很多时，运行时间也会变的很长，如果想缩短脚本运行的时长，就可以用多进程来运行。

安装pytest-xdist：

pip install -U pytest-xdist

运行模式：

pytest test_se.py -n NUM

其中NUM填写并发的进程数。

对比最开始，我们这次用两个进程花了0.93s，一个进程花了0.83s，一百个进程花了十多秒。在我们测试代码量不大的情况下，多进程没有什么优势。

4.重试运行cases

在做接口测试时，有事会遇到503或短时的网络波动，导致case运行失败，而这并非是我们期望的结果，此时可以就可以通过重试运行cases的方式来解决。

安装pytest-rerunfailures：

pip install -U pytest-rerunfailures

运行模式：

pytest test_se.py --reruns NUM

NUM填写重试的次数。

5.显示print内容

在运行测试脚本时，为了调试或打印一些内容，我们会在代码中加一些print内容，但是在运行pytest时，这些内容不会显示出来。如果带上-s，就可以显示了。

运行模式：

pytest test_se.py -s

另外，pytest的多种运行模式是可以叠加执行的，比如说，你想同时运行4个进程，又想打印出print的内容。可以用：

pytest test_se.py -s -n 4

module2：

pytest带你玩转mindspore自定义算子

有关算子的前置知识，请阅读官网的参考文档：MindSpore算子

当开发网络遇到内置算子不足以满足需求时，你可以利用MindSpore的Python API和C++ API方便快捷地扩展CPU端的自定义算子。

自定义算子分为三步

• 算子原语：定义了算子在网络中的前端接口原型，也是组成网络模型的基础单元，主要包括算子的名称、属性（可选）、输入输出名称、输出shape推理方法、输出dtype推理方法等信息。
• 算子实现：利用框架提供的C++ API，结合算子具体特性实现算子内部计算逻辑。
• 算子信息：描述CPU算子的基本信息，如算子名称、支持的输入输出类型等。它是后端做算子选择和映射时的依据。

下面我们以自定义Transpose算子为例，介绍自定义算子和利用pytest进行测试。

注册算子原语

算子的原语是一个继承于PrimitiveWithInfer的子类，其类型名称即是算子名称。

CPU算子原语定义在mindspore/ops/operations路径下，根据算子类型选择适合的文件，接口定义如下：

• 属性由构造函数__init__的入参定义。本用例的算子没有init属性，因此__init__没有额外的入参。
• 输入输出的名称通过init_prim_io_names函数定义。
• 输出Tensor的shape和dtype(数据类型对象，可以参考本人这篇文章)检验在__infer__函数中实现。

以Transpose算子原语为例，给出如下示例代码，这里用到了装饰器语法糖。

Transpose算子原语中参数“perm”作为输入传入，但是在解析时元组类型的“perm”实际被认为是算子的属性。

from mindspore.ops import PrimitiveWithInfer

class Transpose(PrimitiveWithInfer):
    """
    The definition of the Transpose primitive.
    """
    @prim_attr_register
    def __init__(self):
        """Initialize Transpose"""
        self.init_prim_io_names(inputs=['x', 'perm'], outputs=['output'])
    def __infer__(self, x, perm):
        x_shape = x['shape']
        p_value = perm['value']
        if len(x_shape) != len(p_value):
            raise ValueError('The dimension of x and perm must be equal.')
        out_shapes = []
        for i in p_value:
            out_shapes.append(x_shape[i])
        out = {'shape': tuple(out_shapes),
               'dtype': x['dtype'],
               'value': None}
        return out

实现CPU算子和注册算子信息（此处仅供简单了解，详见官网）

实现CPU算子

通常一个CPU算子的实现，需要编写一个头文件和一个源文件，文件路径为mindspore/ccsrc/backend/kernel_compiler/cpu，如果算子的逻辑实现是通过调用第三方库MKL-DNN，则放在子目录mkldnn下。详细介绍请参考oneMKL和oneDNN 。

算子的头文件中包括算子的注册信息和类的声明。算子类继承于CPUKernel父类，重载InitKernel和Launch两个成员函数。

算子的源文件是类的实现，主要是重载InitKernel和Launch两个函数，InitKernel中AnfRuntimeAlgorithm类中的函数实现了各种对算子节点的操作，shape_表示算子第1个输入的shape，axis_表示算子的属性perm。

AnfRuntimeAlgorithm类的详细内容可参考MindSpore源码中mindspore/ccsrc/backend/session/anf_runtime_algorithm.h下的声明。

源文件中Launch函数首先依次获取每个输入输出的地址，然后根据axis_变换维度，把值赋给输出地址指向的空间。

这里对实现过程就不详细阐述了。

注册算子信息

算子信息是指导后端选择算子实现的关键信息，MS_REG_CPU_KERNEL中第一个参数是注册算子的名称，和原语中算子名称一致，第二个参数依次指明每个输入输出的类型，最后一个参数是算子实现的类名。Transpose算子注册代码如下：

MS_REG_CPU_KERNEL(Transpose, KernelAttr().AddInputAttr(kNumberTypeFloat32).AddOutputAttr(kNumberTypeFloat32),
                  TransposeCPUFwdKernel);

算子信息中定义输入输出信息的个数和顺序、算子实现中的输入输出信息的个数和顺序、算子原语中输入输出名称列表的个数和顺序，三者要完全一致。

编译MindSpore

写好自定义CPU算子后，需要重新编译安装MindSpore，具体请参考安装文档。

使用自定义CPU算子并用pytest测试

编译并安装完成后，自定义CPU算子可以通过导入原语直接使用。下面以Transpose的单算子网络测试为例进行说明。

在test_transpose.py文件中定义网络。

import numpy as np
import mindspore.nn as nn
import mindspore.context as context
from mindspore import Tensor
import mindspore.ops as ops

context.set_context(mode=context.GRAPH_MODE, device_target="CPU")

class Net(nn.Cell):
    def __init__(self):
        super(Net, self).__init__()
        self.transpose = ops.Transpose()

    def construct(self, data):
        return self.transpose(data, (1, 0))

def test_net():
    x = np.arange(2 * 3).reshape(2, 3).astype(np.float32)
    transpose = Net()
    output = transpose(Tensor(x))
    print("output: ", output)

执行用例（这里我们可以看到，利用pytest是测试其中一个类中的一个函数，并且均以test_开头，可以回到上文的pytest用法复习一下）:

pytest -s test_transpose.py::test_net

执行结果:

output: [[0, 3]
        [1, 4]
        [2, 5]]

我们把报告输出html：

pytest --html=report.html -s test_transpose.py::test_net

可以查看html报告。

定义算子反向传播函数

如果算子要支持自动微分，需要在其原语中定义其反向传播函数（bprop）。你需要在bprop中描述利用正向输入、正向输出和输出梯度得到输入梯度的反向计算逻辑。反向计算逻辑可以使用内置算子或自定义反向算子构成。

定义算子反向传播函数时需注意以下几点：

• bprop函数的入参顺序约定为正向的输入、正向的输出、输出梯度。若算子为多输出算子，正向输出和输出梯度将以元组的形式提供。
• bprop函数的返回值形式约定为输入梯度组成的元组，元组中元素的顺序与正向输入参数顺序一致。即使只有一个输入梯度，返回值也要求是元组的形式。

例如，Transpose的反向原语为：

import mindspore as ms
import mindspore.ops as ops
from mindspore.ops._grad.grad_base import bprop_getters
fill = ops.Fill()
invert_permutation = ops.InvertPermutation()
transpose = ops.Transpose()
@bprop_getters.register(ops.Transpose)
def get_bprop_transpose(self):
    """Generate bprop for Transpose"""

    def bprop(x, perm, out, dout):
        return transpose(dout, invert_permutation(perm)), fill(ms.int32, (len(perm), ), 0)

    return bprop

• Transpose反向算子中用到了InvertPermutation算子，该算子和Transpose算子开发一样，需要有算子的原语，注册，实现等完整的流程。

在test_transpose.py文件中加入一下内容，定义反向用例。

import mindspore.ops as ops
class Grad(nn.Cell):
    def __init__(self, network):
        super(Grad, self).__init__()
        self.grad = ops.GradOperation(sens_param=True)
        self.network = network

    def construct(self, input_data, sens):
        gout = self.grad(self.network)(input_data, sens)
        return gout

def test_grad_net():
    x = np.arange(2 * 3).reshape(2, 3).astype(np.float32)
    sens = np.arange(2 * 3).reshape(3, 2).astype(np.float32)
    grad = Grad(Net())
    dx = grad(Tensor(x), Tensor(sens))
    print("dx: ", dx.asnumpy())

执行用例:

pytest -s test_transpose.py::test_grad_net

执行结果:

dx:  [[0. 2. 4.]
     [1. 3. 5.]]

开源代码

qmckw/CPUcustom_opByPytest (gitee.com)

参考资料

pytest documentation
好用的Pytest单元测试框架（《51测试天地》四十九（下）- 44）
Pytest学习笔记
pytest单元测试框架
MindSpore参考文档