pytest框架二次开发之机器人报警
我想要实现的效果，当接口自动化case运行失败时，触发企业微信机器人报警，艾特相关人员，及发送失败case的相关信息。报警信息包括：case等级、case描述、case名称、case的开发人员。
·
目录
2.2.1 pytest_runtest_makereport
一、背景：
我想要实现的效果，当接口自动化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 -------------------------------
用例的执行结果 <pluggy.callers._Result object at 0x7ffb48478090>
测试报告: <TestReport 'test_cs1.py::test_01' when='setup' outcome='passed'>
步骤：setup
nodeid: test_cs1.py::test_01
description: 用例描述：用例1
运行结果: passed
setup_function：每个用例开始前都会执行
-------------------------------
用例的执行结果 <pluggy.callers._Result object at 0x7ffb68786b90>
测试报告: <TestReport 'test_cs1.py::test_01' when='call' outcome='passed'>
步骤：call
nodeid: test_cs1.py::test_01
description: 用例描述：用例1
运行结果: passed
PASSED [ 50%]用例1-------
-------------------------------
用例的执行结果 <pluggy.callers._Result object at 0x7ffb48478090>
测试报告: <TestReport '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 -------------------------------
用例的执行结果 <pluggy.callers._Result object at 0x7ffb584ef710>
测试报告: <TestReport 'test_cs1.py::test_02' when='setup' outcome='passed'>
步骤：setup
nodeid: test_cs1.py::test_02
description: 用例描述：用例2
运行结果: passed
setup_function：每个用例开始前都会执行
-------------------------------
用例的执行结果 <pluggy.callers._Result object at 0x7ffb584ef710>
测试报告: <TestReport 'test_cs1.py::test_02' when='call' outcome='passed'>
步骤：call
nodeid: test_cs1.py::test_02
description: 用例描述：用例2
运行结果: passed
PASSED [100%]用例2------
-------------------------------
用例的执行结果 <pluggy.callers._Result object at 0x7ffb6876e390>
测试报告: <TestReport '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 <Session ChenShuaiTest exitstatus=<ExitCode.OK: 0> testsfailed=0 testscollected=0>
fspath /Users/zhaohui/PycharmProjects/ChenShuaiTest
_nodeid
_store <_pytest.store.Store object at 0x7fc7b8993ad0>
_dict__ {'keywords': <NodeKeywords for node <Session ChenShuaiTest exitstatus=<ExitCode.OK: 0> testsfailed=0 testscollected=0>>, 'own_markers': [], 'extra_keyword_matches': set(), 'testsfailed': 0, 'testscollected': 0, 'shouldstop': False, 'shouldfail': False, 'trace': <pluggy._tracing.TagTracerSub 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': <ExitCode.OK: 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
[<Module test_cs1.py>]
开始用例收集
集合名称：test_cs1.py
parent <Package mytest>
config <_pytest.config.Config object at 0x7fc7a8244e50>
session <Session ChenShuaiTest exitstatus=<ExitCode.OK: 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': <NodeKeywords for node <Module test_cs1.py>>, 'own_markers': [], 'extra_keyword_matches': set()}
...........................
收集的用例个数---------------------：8
[<Function test_01>, <Function test_02>, <Function test_03>, <Function test_04>, <Function test_05>, <Function test_06>, <Function test_07>, <Function test_08>]
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 "<CollectReport {!r} lenresult={} outcome={!r}>".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 [<Module test_cs1.py>]
nodeid
outcome passed
result [<Module test_cs1.py>]
sections []
收集的用例个数---------------------：8
result [<Function test_01>, <Function test_02>, <Function test_03>, <Function test_04>, <Function test_05>, <Function test_06>, <Function test_07>, <Function test_08>]
nodeid test_case/mytest/test_cs1.py
outcome passed
result [<Function test_01>, <Function test_02>, <Function test_03>, <Function test_04>, <Function test_05>, <Function test_06>, <Function test_07>, <Function test_08>]
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博客
更多推荐
1
0- 0
已为社区贡献1条内容
所有评论(0)