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

irrational 发表于 2022/01/18 13:31:37 2022/01/18
【摘要】 Python有一个不那么广为人知的测试包:pytest,用它可以很好得进行接口测试,甚至还可以自动导出精美的测试报告,提高效率,组会总结展示,妥妥的!!下面,我们就来体验一下,如何用pytest测试MindSpore自定义算子的类与函数接口!

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

module1:

多玩法Python调试框架pytest

初学入门

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

img

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

img

如上图,在目录栏输入cmd(也可以直接在IDE中运行):

img

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

img

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

结果:

img

从测试结果中可以看到,该测试共执行了两个测试样例,一个失败一个成功。同样,我们也看到失败样例的详细信息,和执行过程中的中间结果。-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

img

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

报告效果:

img

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

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

img

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

pytest test_se.py::TestClassOne

img

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

pytest test_se.py::TestClassTwo::test_one

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

img

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

3.多进程运行cases

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

安装pytest-xdist:

pip install -U pytest-xdist

运行模式:

pytest test_se.py -n NUM

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

img

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

img

4.重试运行cases

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

安装pytest-rerunfailures:

pip install -U pytest-rerunfailures

运行模式:

pytest test_se.py --reruns NUM

NUM填写重试的次数。

img

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下。详细介绍请参考oneMKLoneDNN

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

算子的源文件是类的实现,主要是重载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]]

img

我们把报告输出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.]]

img

开源代码

qmckw/CPUcustom_opByPytest (gitee.com)

参考资料

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

【版权声明】本文为华为云社区用户原创内容,转载时必须标注文章的来源(华为云社区),文章链接,文章作者等基本信息,否则作者和本社区有权追究责任。如果您发现本社区中有涉嫌抄袭的内容,欢迎发送邮件至:cloudbbs@huaweicloud.com进行举报,并提供相关证据,一经查实,本社区将立刻删除涉嫌侵权内容。
  • 点赞
  • 收藏
  • 关注作者

评论(0

0/1000
抱歉,系统识别当前为高风险访问,暂不支持该操作

全部回复

上滑加载中

设置昵称

在此一键设置昵称,即可参与社区互动!

*长度不超过10个汉字或20个英文字符,设置后3个月内不可修改。

*长度不超过10个汉字或20个英文字符,设置后3个月内不可修改。