pytest框架插件源码_关于钩子方法调用部分的简单理解（pytest_runtest_makereport）
作者：互联网
前言：
因为想不明白写的pytest_runtest_makereport里的yield是怎么把结果传出来的？pytest是怎么调用的我们自己写的pytest_runtest_makereport方法？一不小心给自己开了新坑……熬了两个晚上，终于对整个流程稍微有点思路……
P.S. 参考1中的教程非常详细的解释了pluggy源码，对pytest插件执行流程的理解非常有帮助，建议深读
因为是边单步执行源码，边查资料理解，边写完这篇博客，所有前面部分会有点乱，尽可能把我理解的东西写出来。
首先，贴源码
我在conftest.py里写的pytest_runtest_makereport方法代码如下
@pytest.hookimpl(hookwrapper=True, tryfirst=True)
def pytest_runtest_makereport(item, call):
print("ininin")
out = yield
res = out.get_result()
print(res)
if res.when == "call":
logging.info(f"item：{item}")
logging.info(f"异常：{call.excinfo}")
logging.info(f"故障表示：{res.longrepr}")
logging.info(f"测试结果：{res.outcome}")
logging.info(f"用例耗时：{res.duration}")
logging.info("**************************************")
经过打断点，知道pytest_runtest_makereport是由这方法调用的
# site-packages\pluggy\callers.py
def _multicall(hook_impls, caller_kwargs, firstresult=False):
"""Execute a call into multiple python functions/methods and return the
result(s).
``caller_kwargs`` comes from _HookCaller.__call__().
"""
__tracebackhide__ = True
results = []
excinfo = None
try: # run impl and wrapper setup functions in a loop
teardowns = []
try:
for hook_impl in reversed(hook_impls):
try:
args = [caller_kwargs[argname] for argname in hook_impl.argnames]
except KeyError:
for argname in hook_impl.argnames:
if argname not in caller_kwargs:
raise HookCallError(
"hook call must provide argument %r" % (argname,)
)
if hook_impl.hookwrapper:
try:
gen = hook_impl.function(*args)
next(gen) # first yield
teardowns.append(gen)
except StopIteration:
_raise_wrapfail(gen, "did not yield")
else:
res = hook_impl.function(*args)
if res is not None:
results.append(res)
if firstresult: # halt further impl calls
break
except BaseException:
excinfo = sys.exc_info()
finally:
if firstresult: # first result hooks return a single value
outcome = _Result(results[0] if results else None, excinfo)
else:
outcome = _Result(results, excinfo)
# run all wrapper post-yield blocks
for gen in reversed(teardowns):
try:
gen.send(outcome)
_raise_wrapfail(gen, "has second yield")
except StopIteration:
pass
return outcome.get_result()
其中根据大佬的解析可知：
- 插件会先注册使得存在这个接口类
- 调用这个接口会跳到实现函数，也就是我们写的pytest_runtest_makereport
具体来一步步看
一、 实现函数使用装饰器
@pytest.hookimpl(hookwrapper=True, tryfirst=True)
def pytest_runtest_makereport(item, call):
- 根据pycharm跳转hookimpl的来源，可知
hookimpl = HookimplMarker("pytest")
hookspec = HookspecMarker("pytest")
hookimpl 是HookimplMarker()的实例化
- HookimplMarker()类
# site-packages\pluggy\hooks.py
class HookimplMarker(object):
""" Decorator helper class for marking functions as hook implementations.
You can instantiate with a ``project_name`` to get a decorator.
Calling :py:meth:`.PluginManager.register` later will discover all marked functions
if the :py:class:`.PluginManager` uses the same project_name.
"""
def __init__(self, project_name):
self.project_name = project_name
def __call__(
self,
function=None,
hookwrapper=False,
optionalhook=False,
tryfirst=False,
trylast=False,
):
def setattr_hookimpl_opts(func):
setattr(
func,
self.project_name + "_impl",
dict(
hookwrapper=hookwrapper,
optionalhook=optionalhook,
tryfirst=tryfirst,
trylast=trylast,
),
)
return func
if function is None:
return setattr_hookimpl_opts
else:
return setattr_hookimpl_opts(function)
# 其中还有
可知，HookimplMarker类存在__call__魔法方法，也就是类在实例化之后，可以想普通函数一样进行调用。
-
hookimpl = HookimplMarker("pytest")这一步实例化，走__init__魔法方法，即hookimpl 拥有了变量project_name，值为"pytest" -
回到@pytest.hookimpl(hookwrapper=True, tryfirst=True)
也就是说hookimpl这里就进到了__call__里面
传了两个参数hookwrapper、tryfirst，其他为默认值- setattr(object, name, value)
给object设置属性name的属性值value（不存在name属性就新增）
- setattr(object, name, value)
这段代码简单来说就是给被装饰的函数添加属性值return setattr_hookimpl_opts(function)
属性名为self.project_name + "_impl"，也就是"pytest_impl"
属性值为一个字典，包括hookwrapper、optionalhook、tryfirst、trylast这几个key
最后返回被装饰的函数return func
这个时候pytest_runtest_makereport函数就有了pytest_impl属性值
二、 接下来就是使用PluginManager类创建接口类，并加到钩子定义中，注册实现函数，这部分先略过
简单来说，经过这步这个函数就可以作为钩子调用了
-
接口方法拥有project_name+"_spec"（即"pytest_spec")属性，属性值为一个字典，包括firstresult，historic，warn_on_impl这3个key
-
hookwrapper=Ture则把实现函数放到了_wrappers列表中 -
实例化HookImpl对象，存放实现函数的信息
-
给self.hook 添加了名为实现方法的函数名的属性，属性值为
_HookCaller(name, self._hookexec) -
_HookCaller(name, self._hookexec)这里依然是调了_HookCaller类的__call__方法，返回了
self._hookexec(self, self.get_hookimpls(), kwargs) -
self.get_hookimpls() 返回的是
self._nonwrappers + self._wrappers,也就是实现函数列表
三、跳转到实现函数
应该是触发钩子接口后，跳转到_multicall方法，接下来就是进入实现函数的控制执行了
- 首先是循环该接口的实现函数
也就是所有注册的pytest_runtest_makereport方法
def _multicall(hook_impls, caller_kwargs, firstresult=False):
"""Execute a call into multiple python functions/methods and return the
result(s).
``caller_kwargs`` comes from _HookCaller.__call__().
"""
__tracebackhide__ = True
results = []
excinfo = None
try: # run impl and wrapper setup functions in a loop
teardowns = []
try:
for hook_impl in reversed(hook_impls):
……
由代码可知，for hook_impl in reversed(hook_impls)，hook_impls里存放的是所有的实现函数，reversed倒序返回列表（先注册的实现函数会存在hook_impls[0]，也就是说这里会先执行后注册的实现函数）
pytest_runtest_makereport共有4个插件，也就是有4个实现函数
2. 把caller_kwargs[argname]存到args
也就是(iten,call)，为了传参给实现函数
args = [caller_kwargs[argname] for argname in hook_impl.argnames]
3. 跳转到实现函数
if hook_impl.hookwrapper: # 取实现函数的hookwrapper属性进行判断，如果hookwrapper为Ture，则说明实现函数为生成器
try:
gen = hook_impl.function(*args) # gen为pytest_runtest_makereport生成器
next(gen) # first yield # 走到这步的时候跳转到实现函数
teardowns.append(gen) # 执行到实现函数的yeild回到这里，把生成器放入teardowns
except StopIteration:
_raise_wrapfail(gen, "did not yield")
执行完这一步，又继续循环reversed(hook_impls)
跳转到pytest_runtest_makereport的实现函数（这部分应该是pytest原有的实现函数）
代码如下
# _pytest.skipping.py
@hookimpl(hookwrapper=True)
def pytest_runtest_makereport(item: Item, call: CallInfo[None]):
outcome = yield
rep = outcome.get_result()
xfailed = item._store.get(xfailed_key, None)
# unittest special case, see setting of unexpectedsuccess_key
if unexpectedsuccess_key in item._store and rep.when == "call":
reason = item._store[unexpectedsuccess_key]
if reason:
rep.longrepr = f"Unexpected success: {reason}"
else:
rep.longrepr = "Unexpected success"
rep.outcome = "failed"
elif item.config.option.runxfail:
pass # don't interfere
elif call.excinfo and isinstance(call.excinfo.value, xfail.Exception):
assert call.excinfo.value.msg is not None
rep.wasxfail = "reason: " + call.excinfo.value.msg
rep.outcome = "skipped"
elif not rep.skipped and xfailed:
if call.excinfo:
raises = xfailed.raises
if raises is not None and not isinstance(call.excinfo.value, raises):
rep.outcome = "failed"
else:
rep.outcome = "skipped"
rep.wasxfail = xfailed.reason
elif call.when == "call":
if xfailed.strict:
rep.outcome = "failed"
rep.longrepr = "[XPASS(strict)] " + xfailed.reason
else:
rep.outcome = "passed"
rep.wasxfail = xfailed.reason
if (
item._store.get(skipped_by_mark_key, True)
and rep.skipped
and type(rep.longrepr) is tuple
):
# Skipped by mark.skipif; change the location of the failure
# to point to the item definition, otherwise it will display
# the location of where the skip exception was raised within pytest.
_, _, reason = rep.longrepr
filename, line = item.reportinfo()[:2]
assert line is not None
rep.longrepr = str(filename), line + 1, reason
之后循环实现函数_pytest.unittest.py、runner.py的实现函数，就不重复贴代码了
进入实现函数都会执行一次各个实现函数的代码
- 接下来会跑pytest_runtest_logreport、pytest_report_teststatus、pytest_runtest_protocol、pytest_runtest_logstart、pytest_runtest_setup、pytest_fixture_setup等接口的实现函数（可能需要调用这些函数返回什么信息吧）
这块的流程不太清楚，感觉可能在_multicall的上一层应该还有一个控制函数，触发了哪些接口，再调_multicall跑这些接口的实现函数？也有可能debug调试的时候，我点太快跑飞了……
- 跑完实现函数后，进入finally部分,赋值outcome
finally:
if firstresult: # first result hooks return a single value
outcome = _Result(results[0] if results else None, excinfo)
else:
outcome = _Result(results, excinfo)
- 跑完实现函数之后，最后会把之前存在teardown里的生成器（为生成器的实现函数）跑完，把outcome的值传给生成器
# run all wrapper post-yield blocks
for gen in reversed(teardowns):
try:
gen.send(outcome)
_raise_wrapfail(gen, "has second yield")
except StopIteration:
pass
`gen.send(outcome)` 把outcome的值传给生成器，生成器会从上一次yeild的地方往下跑
也就是回到的conftest.py的pytest_runtest_makereport的实现函数里的
outcome = yield这行
def pytest_runtest_makereport(item: Item, call: CallInfo[None]):
outcome = yield # 这里
rep = outcome.get_result()
新建变量outcome接收了传过来的outcome
这里涉及到生成器的知识
- 调用生成器执行到yield，返回到调用函数，生成器的变量状态保留
- 使用send()方法，可把调用函数的值传给生成器
- 这里还有一个小知识点，生成器第一次调用的时候不可以使用send()方法传值，会报错TypeError: can't send non-None value to a just-started generator
简单写个生成器调用，流程和pytest里执行实现函数是一样的，单步执行跑一下代码就理解了
def fun2():
print("fun2")
out = yield
print("fun22")
print(f"out:{out}")
def fun3():
print("fun3")
f = fun2()
next(f)
f.send("00")
print("fun33")
if __name__ == '__main__':
fun3()
四、 之后执行pytest_runtest_makereport方法的代码就没什么可说的，自己写的逻辑很简单
最后跳出来到了_pytest/runner.py的call_and_report方法
report: TestReport = hook.pytest_runtest_makereport(item=item, call=call)
return report
再跳到runtestprotocol方法
总结：
一、所谓的钩子函数hook
有一个方法A，还有另外一个方法B，执行到方法A的时候跳转到方法B，这就是实现了hook的作用。
如何能把方法A和方法B关联起来，就用到一个起到注册功能的方法，通过这个方法实现两个方法的关联。
def fun1():
print("fun1")
return "out"
class TestHook:
def __init__(self):
self.hook_fun = None
def register_fun2_hook(self,fun):
self.hook_fun = fun
def fun2(self):
print("这里是fun2")
if self.hook_fun:
self.hook_fun()
else:
print("no hook")
if __name__ == '__main__':
xxx = TestHook()
xxx.register_fun2_hook(fun1)
xxx.hook_fun()
print('*********')
xxx.fun2()
# -----输出-----
# fun1
# *********
# 这里是fun2
# fun1
-
实例化TestHook这个类，hook_fun为None
-
调用register_fun2_hook方法，注册self.hook_fun，使得self.hook_fun与传入的参数fun进行关联，这个fun就是我们另外自定义的方法B，self.hook_fun就是钩子函数
-
执行xxx.fun2()，就会去执行fun1
-
说回pytest，self.hook_fun 就是 runner.py 定义的接口函数 pytest_runtest_makereport ，fun1 就是我们在 conftest.py 写的实现函数pytest_runtest_makereport
二、pytest里的hook实现
-
定义接口类，在接口类添加接口函数 pytest_runtest_makereport
-
定义插件类，插件里添加实现函数 pytest_runtest_makereport
-
实例化插件管理对象pm
-
调用pm.add_hookspecs()，把创建的接口 pytest_runtest_makereport添加到钩子定义中
-
注册实现函数 pytest_runtest_makereport
-
hook.pytest_runtest_makereport 调用钩子函数
-
通过cller类的_multicall方法控制实现执行接口的所有实现函数
参考1：https://blog.csdn.net/redrose2100/article/details/121277958
参考2：https://docs.pytest.org/en/latest/reference/reference.html?highlight=pytest_runtest_makereport#std-hook-pytest_runtest_makereport
标签：__,插件,函数,runtest,hook,pytest,call,源码 来源： https://www.cnblogs.com/congyinew/p/16484005.html