pytest框架二次开发

目录

一、背景：

二、实现思路：

2.1 报警接口

2.2 、HOOK函数：

2.2.1 pytest_runtest_makereport

2.2.2 pytest_collectstart

2.2.3 pytest_collectreport

三、项目实战：

3.1 我们先实现报警工具

3.2 跨模块的全局变量

3.3 通过HOOK函数收集报警信息&报警

---

一、背景：

我想要实现的效果，当接口自动化case运行失败时，触发企业微信机器人报警，艾特相关人员，及发送失败case的相关信息。

报警信息包括：case等级、case描述、case名称、case的开发人员。

二、实现思路：

2.1 报警接口

 实现报警接口。七、钉钉机器人报警设置_傲娇的喵酱的博客-CSDN博客_钉钉报警机器人

我们要通过企业微信实现。企业微信群聊机器人代码示例分享 - 开发者社区 - 企业微信开发者中心

 可以去查看官方文档。

case运行时，我们可以对报警做一个筛选，哪些级别的case报警，哪些级别的case不报警。或者可以再扩展一下其他逻辑。

模版：

 requests.post('https://qyapi.weixin.qq.com/cgi-bin/webhook/send?key=你的机器人的key',
 headers={'Content-Type': 'application/json'},data=json.dumps({'msgtype': 'text','text':{'mentioned_mobile_list': '你要艾特的人手机号列表','content': '{}'.format(msg).encode('utf-8').decode(
                                                           "latin1")}}, ensure_ascii=False))

注意：

1、艾特人，可以通过uid，也可以通过手机号，我们这里通过手机号 'mentioned_mobile_list'，

是一个list，如

'mentioned_mobile_list':["1871871817817",]

2、 内容为中文的话，需要注意编码。

.encode('utf-8').decode("latin1")

 

2.2 、HOOK函数：

我们需要收集一共执行了多少case、失败了多少case、每条case的用例级别、用例描述、作者、失败原因等。

实现这些数据的收集，我们需要使用pytest 提供的Hook函数。

官方文档：

Writing hook functions — pytest documentation

API Reference — pytest documentation

HOOK函数：

1、是个函数，在系统消息触发时别系统调用

2、不是用户自己触发的

3、使用时直接编写函数体

4、钩子函数的名称是确定，当系统消息触发，自动会调用

HOOK也被称作钩子。他是系统或者第三方插件暴露的一种可被回调的函数。

pytest具体提供了哪些hook函数，可以在\venv\Lib\site-packages\_pytest>hookspec.py文件中查看，里面每一个钩子函数都有相应的介绍。

插件就是用1个或者多个hook函数。如果想要编写新的插件，或者是仅仅改进现有的插件，都必须通过这个hook函数进行。所以想掌握pytest插件二次开发，必须搞定hook函数。

Hook函数都在这个路径下：site-packages/_pytest/hookspec.py

我们看一下hookspec.py源码，看一看pytest这个框架都给我们提供了哪些回调函数。

2.2.1 pytest_runtest_makereport

pytest框架，提供了pytest_runtest_makereport回调函数，来获取用例的执行结果。

官方文档：

API Reference — pytest documentation

@hookspec(firstresult=True)
def pytest_runtest_makereport(
    item: "Item", call: "CallInfo[None]"
) -> Optional["TestReport"]:
    """Called to create a :py:class:`_pytest.reports.TestReport` for each of
    the setup, call and teardown runtest phases of a test item.
 
    See :func:`pytest_runtest_protocol` for a description of the runtest protocol.
 
    :param CallInfo[None] call: The ``CallInfo`` for the phase.
 
    Stops at first non-None result, see :ref:`firstresult`.
    """

（python好像没有抽象类与接口），这里框架提供了一个pytest_runtest_makereport函数模版，我们需要去实现它，实现它的内部逻辑。

函数名称：pytest_runtest_makereport

传参：

---Item类的实例item，测试用例。

---CallInfo[None]类的实例call

返回值：Optional["TestReport"]

TestReport类的实例，或者为None（Optional更优雅的处理none）

备注：

Optional介绍：

Python:Optional和带默认值的参数_秋墓的博客-CSDN博客_optional python

每个case的执行过程，其实都是分三步的，1 setup初始化数据 2执行case 3teardown

这里的item是测试用例，call是测试步骤，具体执行过程如下：

*先执行when="setup"返回setup的执行结果

*再执行when="call"返回call的执行结果

*最后执行when="teardown"返回teardown的执行结果。

查看一下返回结果：TestReport源码

@final
class TestReport(BaseReport):
    """Basic test report object (also used for setup and teardown calls if
    they fail)."""
 
    __test__ = False
 
    def __init__(
        self,
        nodeid: str,
        location: Tuple[str, Optional[int], str],
        keywords,
        outcome: "Literal['passed', 'failed', 'skipped']",
        longrepr: Union[
            None, ExceptionInfo[BaseException], Tuple[str, int, str], str, TerminalRepr
        ],
        when: "Literal['setup', 'call', 'teardown']",
        sections: Iterable[Tuple[str, str]] = (),
        duration: float = 0,
        user_properties: Optional[Iterable[Tuple[str, object]]] = None,
        **extra,
    ) -> None:
        #: Normalized collection nodeid.
        self.nodeid = nodeid

这里有用的信息比较多：

outcome 执行结果：passed or failed

nodeid case名称

longrepr 执行失败原因

item.function.__doc__ case描述信息

举🌰：

conftest.py

 
import pytest
 
 
 
@pytest.hookimpl(hookwrapper=True, tryfirst=True)
def pytest_runtest_makereport(item, call):
    print('-------------------------------')
    # 获取常规的钩子方法的调用结果，返回一个result对象
    out = yield
 
    print('用例的执行结果', out)
 
    # 获取调用结果的测试报告，返回一个report对象，report对象的属性
    # 包括when(setup, call, teardown三个值)、nodeid(测试用例的名字)、
    # outcome(用例执行的结果 passed, failed)
    report = out.get_result()
 
    print('测试报告: %s' % report)
    print('步骤：%s' % report.when)
    print('nodeid: %s' % report.nodeid)
 
    # 打印函数注释信息
    print('description: %s' % str(item.function.__doc__))
    print('运行结果: %s' % report.outcome)

test_cs1.py

 
 
import pytest
def setup_function():
    print(u"setup_function：每个用例开始前都会执行")
 
def teardown_function():
    print(u"teardown_function：每个用例结束后都会执行")
 
 
def test_01():
    """ 用例描述：用例1"""
    print("用例1-------")
 
 
def test_02():
    """ 用例描述：用例2"""
    print("用例2------")
 
 
 
if __name__ == '__main__':
    pytest.main(['-s'])
 

运行结果：

 
============================= test session starts ==============================
collecting ... collected 2 items
 
test_cs1.py::test_01 -------------------------------
用例的执行结果 object at 0x7ffb48478090>
测试报告: 'test_cs1.py::test_01' when='setup' outcome='passed'>
步骤：setup
nodeid: test_cs1.py::test_01
description:  用例描述：用例1
运行结果: passed
setup_function：每个用例开始前都会执行
-------------------------------
用例的执行结果 object at 0x7ffb68786b90>
测试报告: 'test_cs1.py::test_01' when='call' outcome='passed'>
步骤：call
nodeid: test_cs1.py::test_01
description:  用例描述：用例1
运行结果: passed
PASSED                                              [ 50%]用例1-------
-------------------------------
用例的执行结果 object at 0x7ffb48478090>
测试报告: 'test_cs1.py::test_01' when='teardown' outcome='passed'>
步骤：teardown
nodeid: test_cs1.py::test_01
description:  用例描述：用例1
运行结果: passed
teardown_function：每个用例结束后都会执行
 
test_cs1.py::test_02 -------------------------------
用例的执行结果 object at 0x7ffb584ef710>
测试报告: 'test_cs1.py::test_02' when='setup' outcome='passed'>
步骤：setup
nodeid: test_cs1.py::test_02
description:  用例描述：用例2
运行结果: passed
setup_function：每个用例开始前都会执行
-------------------------------
用例的执行结果 object at 0x7ffb584ef710>
测试报告: 'test_cs1.py::test_02' when='call' outcome='passed'>
步骤：call
nodeid: test_cs1.py::test_02
description:  用例描述：用例2
运行结果: passed
PASSED                                              [100%]用例2------
-------------------------------
用例的执行结果 object at 0x7ffb6876e390>
测试报告: 'test_cs1.py::test_02' when='teardown' outcome='passed'>
步骤：teardown
nodeid: test_cs1.py::test_02
description:  用例描述：用例2
运行结果: passed
teardown_function：每个用例结束后都会执行
 
 
============================== 2 passed in 0.42s ===============================
 
Process finished with exit code 0

我们在conftest.py中实现pytest_runtest_makereport这个方法，系统在调用这个函数时，会自动注入传参。

2.2.2 pytest_collectstart

pytest框架，提供了pytest_collectstart 开始收集，收集每个模块的case信息

源码：

 
def pytest_collectstart(collector: "Collector") -> None:
    """Collector starts collecting."""

我们看一下传参collector: "Collector"源码

 
class Collector(Node):
    """Collector instances create children through collect() and thus
    iteratively build a tree."""
 
    class CollectError(Exception):
        """An error during collection, contains a custom message."""
 
    def collect(self) -> Iterable[Union["Item", "Collector"]]:
        """Return a list of children (items and collectors) for this
        collection node."""
        raise NotImplementedError("abstract")
 
    # TODO: This omits the style= parameter which breaks Liskov Substitution.
    def repr_failure(  # type: ignore[override]
        self, excinfo: ExceptionInfo[BaseException]
    ) -> Union[str, TerminalRepr]:
        """Return a representation of a collection failure.
        :param excinfo: Exception information for the failure.
        """
        if isinstance(excinfo.value, self.CollectError) and not self.config.getoption(
            "fulltrace", False
        ):
            exc = excinfo.value
            return str(exc.args[0])
 
        # Respect explicit tbstyle option, but default to "short"
        # (_repr_failure_py uses "long" with "fulltrace" option always).
        tbstyle = self.config.getoption("tbstyle", "auto")
        if tbstyle == "auto":
            tbstyle = "short"
 
        return self._repr_failure_py(excinfo, style=tbstyle)
 
    def _prunetraceback(self, excinfo: ExceptionInfo[BaseException]) -> None:
        if hasattr(self, "fspath"):
            traceback = excinfo.traceback
            ntraceback = traceback.cut(path=self.fspath)
            if ntraceback == traceback:
                ntraceback = ntraceback.cut(excludepath=tracebackcutdir)
            excinfo.traceback = ntraceback.filter()
 
 
def _check_initialpaths_for_relpath(session, fspath):
    for initial_path in session._initialpaths:
        if fspath.common(initial_path) == initial_path:
            return fspath.relto(initial_path)

类Collector继承了Node类。我们查看一下Node源码。

 
class Node(metaclass=NodeMeta):
    """Base class for Collector and Item, the components of the test
    collection tree.
    Collector subclasses have children; Items are leaf nodes.
    """
 
    # Use __slots__ to make attribute access faster.
    # Note that __dict__ is still available.
    __slots__ = (
        "name",
        "parent",
        "config",
        "session",
        "fspath",
        "_nodeid",
        "_store",
        "__dict__",
    )
 
    def __init__(
        self,
        name: str,
        parent: "Optional[Node]" = None,
        config: Optional[Config] = None,
        session: "Optional[Session]" = None,
        fspath: Optional[py.path.local] = None,
        nodeid: Optional[str] = None,
    ) -> None:
        #: A unique name within the scope of the parent node.
        self.name = name
 
        #: The parent collector node.
        self.parent = parent

属性包括

        "name",
        "parent",
        "config",
        "session",
        "fspath",
        "_nodeid",
        "_store",
        "__dict__",

我们在 conftest.py文件下，打印一下这个几个属性。

def pytest_collectstart(collector:Collector):
    print("开始用例收集")
    print("集合名称：%s"%collector.name)
    print("parent",collector.parent)
    print("config",collector.config)
    print("session",collector.session)
    print("fspath",collector.fspath)
    print("_nodeid",collector._nodeid)
    print("_store", collector._store)
    print("_dict__", collector.__dict__)
    print("...........................")

输出结果：

collecting ... 开始用例收集
集合名称：ChenShuaiTest
parent None
config <_pytest.config.Config object at 0x7fc7a8244e50>
session 0> testsfailed=0 testscollected=0>
fspath /Users/zhaohui/PycharmProjects/ChenShuaiTest
_nodeid 
_store <_pytest.store.Store object at 0x7fc7b8993ad0>
_dict__ {'keywords': for node 0> testsfailed=0 testscollected=0>>, 'own_markers': [], 'extra_keyword_matches': set(), 'testsfailed': 0, 'testscollected': 0, 'shouldstop': False, 'shouldfail': False, 'trace': object at 0x7fc7c83bc210>, 'startdir': local('/Users/zhaohui/PycharmProjects/ChenShuaiTest/test_case/mytest'), '_initialpaths': frozenset({local('/Users/zhaohui/PycharmProjects/ChenShuaiTest/test_case/mytest/test_cs1.py')}), '_bestrelpathcache': _bestrelpath_cache(path=PosixPath('/Users/zhaohui/PycharmProjects/ChenShuaiTest')), 'exitstatus': 0>, '_fixturemanager': <_pytest.fixtures.FixtureManager object at 0x7fc7b89936d0>, '_setupstate': <_pytest.runner.SetupState object at 0x7fc7c8502bd0>, '_notfound': [], '_initial_parts': [(local('/Users/zhaohui/PycharmProjects/ChenShuaiTest/test_case/mytest/test_cs1.py'), [])], 'items': []}
...........................
收集的用例个数---------------------：1
[]
开始用例收集
集合名称：test_cs1.py
parent 
config <_pytest.config.Config object at 0x7fc7a8244e50>
session 0> testsfailed=0 testscollected=0>
fspath /Users/zhaohui/PycharmProjects/ChenShuaiTest/test_case/mytest/test_cs1.py
_nodeid test_case/mytest/test_cs1.py
_store <_pytest.store.Store object at 0x7fc7a83088d0>
_dict__ {'keywords': for node >, 'own_markers': [], 'extra_keyword_matches': set()}
...........................
收集的用例个数---------------------：8
[, , , , , , , ]
collected 8 items

这里主要展示的就是收集时每个模块的case信息。感觉用处不是很大。

2.2.3 pytest_collectreport

pytest框架，提供了pytest_collectreport回调函数，收集完成后case的报告

源码：

 
def pytest_collectreport(report: "CollectReport") -> None:
    """Collector finished collecting."""

传参CollectReport 

 
@final
class CollectReport(BaseReport):
    """Collection report object."""
 
    when = "collect"
 
    def __init__(
        self,
        nodeid: str,
        outcome: "Literal['passed', 'skipped', 'failed']",
        longrepr,
        result: Optional[List[Union[Item, Collector]]],
        sections: Iterable[Tuple[str, str]] = (),
        **extra,
    ) -> None:
        #: Normalized collection nodeid.
        self.nodeid = nodeid
 
        #: Test outcome, always one of "passed", "failed", "skipped".
        self.outcome = outcome
 
        #: None or a failure representation.
        self.longrepr = longrepr
 
        #: The collected items and collection nodes.
        self.result = result or []
 
        #: List of pairs ``(str, str)`` of extra information which needs to
        #: marshallable.
        # Used by pytest to add captured text : from ``stdout`` and ``stderr``,
        # but may be used by other plugins : to add arbitrary information to
        # reports.
        self.sections = list(sections)
 
        self.__dict__.update(extra)
 
    @property
    def location(self):
        return (self.fspath, None, self.fspath)
 
    def __repr__(self) -> str:
        return "".format(
            self.nodeid, len(self.result), self.outcome
        )
 

类CollectReport继承了

属性：

nodeid: str,
outcome: "Literal['passed', 'skipped', 'failed']",
longrepr,
result: Optional[List[Union[Item, Collector]]],
sections: Iterable[Tuple[str, str]] = (),

我们把这些属性打印一下：

def pytest_collectreport(report: CollectReport):
    print("收集的用例个数---------------------：%s"%len(report.result))
    print("result",report.result)
    print("nodeid", report.nodeid)
    print("outcome", report.outcome)
    print("result", report.result)
    print("sections", report.sections)

collecting ... 收集的用例个数---------------------：1
result []
nodeid 
outcome passed
result []
sections []
收集的用例个数---------------------：8
result [, , , , , , , ]
nodeid test_case/mytest/test_cs1.py
outcome passed
result [, , , , , , , ]
sections []
collected 8 items

感觉也没啥太有用的东西.... 

三、项目实战：

3.1 我们先实现报警工具

alarm_utils.py 下，AlarmUtils工具类

# -*- coding:utf-8 -*-
# @Author: 喵酱
# @time: 2022 - 09 -15
# @File: alarm_utils.py
import requests
import json
 
 
class AlarmUtils:
    @staticmethod
    def default_alram(alarm_content:str,phone_list:list):
        requests.post('https://qyapi.weixin.qq.com/cgi-bin/webhook/send?key=你的机器人的key',
                      headers={'Content-Type': 'application/json'},
                      data=json.dumps(
                          {'msgtype': 'text', 'text': {'mentioned_mobile_list':phone_list,
                                                       'content': '{}'.format(alarm_content).encode('utf-8').decode(
                                                           "latin1")}},
                          ensure_ascii=False))

这里艾特人的方式，通过手机号，是一个list。这个方法需要传报警内容及艾特的手机号。

3.2 跨模块的全局变量

Python设置跨文件的全局变量_he2693836的博客-CSDN博客_python跨文件全局变量

# -*- coding:utf-8 -*-
# @Author: 喵酱
# @time: 2022 - 09 -14
# @File: gol.py
# -*- coding: utf-8 -*-
 
def _init():  # 初始化
    global _global_dict
    _global_dict = {}
 
 
def set_value(key, value):
    # 定义一个全局变量
    _global_dict[key] = value
 
 
def get_value(key):
    # 获得一个全局变量，不存在则提示读取对应变量失败
    try:
        return _global_dict[key]
    except:
        print('读取' + key + '失败\r\n')

3.3 通过HOOK函数收集报警信息&报警

conftest.py

 
 
@pytest.hookimpl(hookwrapper=True, tryfirst=True)
def pytest_runtest_makereport(item, call)-> Optional[TestReport]:
    print('-------------------------------')
    # 获取常规的钩子方法的调用结果，返回一个result对象
    out = yield
    # '用例的执行结果', out
 
 
 
    # 获取调用结果的测试报告，返回一个report对象，report对象的属性
    # 包括when(setup, call, teardown三个值)、nodeid(测试用例的名字)、
    # outcome(用例执行的结果 passed, failed)
    report = out.get_result()
 
    # 只关注用例本身结果
    if report.when == "call":
 
        num = gol.get_value("total")
        gol.set_value("total",num+1)
        if report.outcome != "passed":
            failnum = gol.get_value("fail_num")
            gol.set_value("fail_num", failnum + 1)
            single_fail_content = "{}.报错case名称：{},{}，失败原因：{}".format(gol.get_value("fail_num"), report.nodeid,
                                                                    str(item.function.__doc__),report.longrepr)
            list_content:list = gol.get_value("fail_content")
            list_content.append(single_fail_content)
     
 
 
 
 
 
@pytest.fixture(scope="session", autouse=True)
def fix_a():
    gol._init()
    gol.set_value("total",0)
    gol.set_value("fail_num", 0)
    # 失败内容
    gol.set_value("fail_content", [])
 
    yield
 
    # 执行case总数
    all_num = str(gol.get_value("total"))
    # 失败case总数:
    fail_num = str(gol.get_value("fail_num"))
    if int(gol.get_value("fail_num")):
        list_content: list = gol.get_value("fail_content")
        final_alarm_content = "项目名称：{},\n执行case总数：{},失败case总数：{},\n详情：{}".format("陈帅的测试项目", all_num, fail_num,
                                                                                 str(list_content))
        print(final_alarm_content )
        AlarmUtils.default_alram(final_alarm_content, ['1871xxxxxx'])
 
 

大概是这个样子了，后续一些细节慢慢优化吧

pytest进行二次开发，主要依赖于Hook函数。在conftest.py文件里调用Hook函数。

conftest.py是pytest特有的本地测试配置文件，既可以用来设置项目级的fixture，也可以用来导入外部插件，还可以指定钩子函数。

conftest.py文件名称是固定的，pytest会自动设别该文件，只作用在它的目录以及子目录。

通过装饰器@pytest.fixture来告诉pytest某个特定的函数是一个fixture，然后用例可以直接把fixture当参数来调用

这里的前置后置@pytest.fixture(scope="session", autouse=True)，是项目级的。只运行一次。

下一章：

pytest框架二次开发之自定义注解_傲娇的喵酱的博客-CSDN博客