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()

其中根据大佬的解析可知：

1. 插件会先注册使得存在这个接口类
2. 调用这个接口会跳到实现函数，也就是我们写的pytest_runtest_makereport

具体来一步步看
一、 实现函数使用装饰器

@pytest.hookimpl(hookwrapper=True, tryfirst=True)
def pytest_runtest_makereport(item, call):

1. 根据pycharm跳转hookimpl的来源，可知

hookimpl = HookimplMarker("pytest")
hookspec = HookspecMarker("pytest")

hookimpl 是HookimplMarker()的实例化

2. 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__魔法方法，也就是类在实例化之后，可以想普通函数一样进行调用。

3. hookimpl = HookimplMarker("pytest")这一步实例化，走__init__魔法方法，即hookimpl 拥有了变量project_name，值为”pytest”

4. 回到@pytest.hookimpl(hookwrapper=True, tryfirst=True)
   也就是说hookimpl这里就进到了__call__里面
   传了两个参数hookwrapper、tryfirst，其他为默认值

   • setattr(object, name, value)
     给object设置属性name的属性值value（不存在name属性就新增）

这段代码简单来说就是给被装饰的函数添加属性值return setattr_hookimpl_opts(function)
属性名为self.project_name + "_impl"，也就是”pytest_impl”
属性值为一个字典，包括hookwrapper、optionalhook、tryfirst、trylast这几个key
最后返回被装饰的函数return func
这个时候pytest_runtest_makereport函数就有了pytest_impl属性值

二、 接下来就是使用PluginManager类创建接口类，并加到钩子定义中，注册实现函数，这部分先略过
简单来说，经过这步这个函数就可以作为钩子调用了

1. 接口方法拥有project_name+”_spec”（即”pytest_spec”)属性，属性值为一个字典，包括firstresult，historic，warn_on_impl这3个key

2. hookwrapper=Ture则把实现函数放到了_wrappers列表中

3. 实例化HookImpl对象，存放实现函数的信息

4. 给self.hook 添加了名为实现方法的函数名的属性，属性值为_HookCaller(name, self._hookexec)

5. _HookCaller(name, self._hookexec)这里依然是调了_HookCaller类的__call__方法，返回了self._hookexec(self, self.get_hookimpls(), kwargs)

6. self.get_hookimpls() 返回的是self._nonwrappers + self._wrappers,也就是实现函数列表

三、跳转到实现函数
应该是触发钩子接口后，跳转到_multicall方法，接下来就是进入实现函数的控制执行了

1. 首先是循环该接口的实现函数
   也就是所有注册的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的实现函数，就不重复贴代码了
进入实现函数都会执行一次各个实现函数的代码

4. 接下来会跑pytest_runtest_logreport、pytest_report_teststatus、pytest_runtest_protocol、pytest_runtest_logstart、pytest_runtest_setup、pytest_fixture_setup等接口的实现函数（可能需要调用这些函数返回什么信息吧）

这块的流程不太清楚，感觉可能在_multicall的上一层应该还有一个控制函数，触发了哪些接口，再调_multicall跑这些接口的实现函数？也有可能debug调试的时候，我点太快跑飞了……

5. 跑完实现函数后，进入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)

6. 跑完实现函数之后，最后会把之前存在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

1. 实例化TestHook这个类，hook_fun为None

2. 调用register_fun2_hook方法，注册self.hook_fun，使得self.hook_fun与传入的参数fun进行关联，这个fun就是我们另外自定义的方法B，self.hook_fun就是钩子函数

3. 执行xxx.fun2()，就会去执行fun1

4. 说回pytest，self.hook_fun 就是 runner.py 定义的接口函数 pytest_runtest_makereport ，fun1 就是我们在 conftest.py 写的实现函数pytest_runtest_makereport

二、pytest里的hook实现

1. 定义接口类，在接口类添加接口函数 pytest_runtest_makereport

2. 定义插件类，插件里添加实现函数 pytest_runtest_makereport

3. 实例化插件管理对象pm

4. 调用pm.add_hookspecs()，把创建的接口 pytest_runtest_makereport添加到钩子定义中

5. 注册实现函数 pytest_runtest_makereport

6. hook.pytest_runtest_makereport 调用钩子函数

7. 通过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