标签：__ 插件 函数 runtest hook pytest call 源码

前言：
因为想不明白写的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

标签：__,插件,函数,runtest,hook,pytest,call,源码
来源： https://www.cnblogs.com/congyinew/p/16484005.html

本站声明： 1. iCode9 技术分享网（下文简称本站）提供的所有内容，仅供技术学习、探讨和分享；
2. 关于本站的所有留言、评论、转载及引用，纯属内容发起人的个人观点，与本站观点和立场无关；
3. 关于本站的所有言论和文字，纯属内容发起人的个人观点，与本站观点和立场无关；
4. 本站文章均是网友提供，不完全保证技术分享内容的完整性、准确性、时效性、风险性和版权归属；如您发现该文章侵犯了您的权益，可联系我们第一时间进行删除；
5. 本站为非盈利性的个人网站，所有内容不会用来进行牟利，也不会利用任何形式的广告来间接获益，纯粹是为了广大技术爱好者提供技术内容和技术思想的分享性交流网站。