Pytest 源码解读 [5] - pytest 是如何测试 pytest 的

什么叫用 pytest 来测试 pytest 呢?其实就是构建一个 mini 的 pytest 集成测试环境。我们要对一个测试框架做端到端的测试,最基本的就是写一些测试用例,然后用测试框架来执行它们看是否符合预期。另外 pytest 本身是一个插件化的架构,尤其在我们写一些第三方插件的时候,我们也需要构建一个 pytest 集成测试环境,帮助我们来测试自己编写的插件逻辑是否正确。pytest 的作者在实现的时候考虑到了这一点,体用了一个自带的mini 集成测试框架,叫 pytester ,摘抄下 pytester.py 的模块注释上的说明

(disabled by default) support for testing pytest and pytest plugins.”””

这里我们先来看一个例子,看看 pytester 是如何帮助我们构建这个集成测试环境的。我们以 pytest 自身的测试代码 test_marker.py 为例,这个是用来测试 mark (标签) 插件的。我们看其中的一个测试用例

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
def test_marked_class_run_twice(testdir):
"""Test fails file is run twice that contains marked class.
See issue#683.
"""
py_file = testdir.makepyfile(
"""
import pytest
@pytest.mark.parametrize('abc', [1, 2, 3])
class Test1(object):
def test_1(self, abc):
assert abc in [1, 2, 3]
"""
)
file_name = os.path.basename(py_file.strpath)
rec = testdir.inline_run(file_name, file_name)
rec.assertoutcome(passed=6)

这里有几个关键词,我们依次来做下分析介绍,帮助大家理解其中的执行逻辑

  • testdir 这个 fixture
  • 用 “”” 注释的一段测试用例代码
  • rec = testdir.inline_run 这个函数调用

TestDir

testdir 是在 pytester 插件实现中定义的一个 fixture

1
2
3
4
5
6
7
8
9
10
@pytest.fixture
def testdir(request: FixtureRequest, tmpdir_factory) -> "Testdir":
"""
A :class: `TestDir` instance, that can be used to run and test pytest itself.

It is particularly useful for testing plugins. It is similar to the `tmpdir` fixture
but provides methods which aid in testing pytest itself.

"""
return Testdir(request, tmpdir_factory)

它依赖的另外一个叫 tmpdir_factoryfixture , 这个是在 tmpdir 这个插件的实现中定义的

1
2
3
4
5
6
@pytest.fixture(scope="session")
def tmpdir_factory(request: FixtureRequest) -> TempdirFactory:
"""Return a :class:`_pytest.tmpdir.TempdirFactory` instance for the test session.
"""
# Set dynamically by pytest_configure() above.
return request.config._tmpdirhandler # type: ignore

我们最后回来看下 TestDir 这个类型的定义,它是在 pytester.py 中实现的 ,参考 pytester.py 源码 。这里截取一段过去版本的模块注释

This is based on the tmpdir fixture but provides a number of methods which aid with testing pytest itself. Unless :py:meth:chdir is used all methods will use :py:attr:tmpdir as their current working directory.

我们大概可以猜到它的工作逻辑,通过一个临时目录,在里面存放一些测试用例,再用 pytest 来执行它们,来验证 pytest 测试框架或者插件的正确性。

测试代码

我们简单看下这段代码中 testdir.makepyfile 的具体实现, makepyfileTestDir 类型的一个实例方法

1
2
3
def makepyfile(self, *args, **kwargs):
"""Shortcut for .makefile() with a .py extension."""
return self._makefile(".py", args, kwargs)

进一步跟进到 self._makefile 函数

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
def _makefile(self, ext, lines, files, encoding="utf-8"):
items = list(files.items())

def to_text(s):
return s.decode(encoding) if isinstance(s, bytes) else str(s)

if lines:
source = "\n".join(to_text(x) for x in lines)
basename = self._name
items.insert(0, (basename, source))

ret = None
for basename, value in items:
p = self.tmpdir.join(basename).new(ext=ext)
p.dirpath().ensure_dir()
source = Source(value)
source = "\n".join(to_text(line) for line in source.lines)
p.write(source.strip().encode(encoding), "wb")
if ret is None:
ret = p
return ret

我们看到第14行,就是在之前创建的临时目录里创建一个 python 文件,并写入相关的内容。

inline_run 的执行方式

最后在来看下 inline_run是如何来运行前面存放在临时目录里的一些测试用例的。首先我们知道 pytest 的运行方式是在当前要执行的测试目录下运行 pytest 的,这里怎么做目录切换呢?

我们看到 TestDir 类的初始化函数里就有一行这个 self.chdir() , 看了下具体的实现,原来在初始化 TestDir 实例的时候,就已经完成了当前目录的切换了。

1
2
3
4
5
6
7
def chdir(self):
"""Cd into the temporary directory.

This is done automatically upon instantiation.

"""
self.tmpdir.chdir(

我们最后回来看下 inline_run 的实现,它前面一大部分逻辑是用来处理 plugin 的加载和环境的 snapshot 的,这里先不啰嗦介绍了,最关键的代码就这行, 用来执行测试,看起来到这里就完成整体逻辑了

1
ret = pytest.main(list(args), plugins=plugins)

这有另外一个问题

我们看下 main这个函数签名,

1
def main(args=None, plugins=None) -> Union[int, ExitCode]:

main 就只返回一个退出码,那我们的测试代码怎么拿这个结果来做一个比较详细的断言分析呢?这里我我们再仔细挖掘下 inline_run 中的实现,我们看到这段代码

1
2
3
4
5
6
rec = []
class Collect:
def pytest_configure(x, config):
rec.append(self.make_hook_recorder(config.pluginmanager))

plugins.append(Collect())

此处定义了一个新的 plugin 叫 Collect , 它的作用是在 pytest_configure 阶段注入了一段 recorder 逻辑,我们再来看下 self.make_hook_recorder 的实现

1
2
3
4
5
def make_hook_recorder(self, pluginmanager):
"""Create a new :py:class:`HookRecorder` for a PluginManager."""
pluginmanager.reprec = reprec = HookRecorder(pluginmanager)
self.request.addfinalizer(reprec.finish_recording)
return repre

这里提到了,它给 PluginManager 创建了一个 HookRecorder 对象,我们最后看下 HookRecorder 对象是做什么的

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
class HookRecorder:
"""Record all hooks called in a plugin manager.

This wraps all the hook calls in the plugin manager, recording each call
before propagating the normal calls.

"""

def __init__(self, pluginmanager) -> None:
self._pluginmanager = pluginmanager
self.calls = [] # type: List[ParsedCall]

def before(hook_name: str, hook_impls, kwargs) -> None:
self.calls.append(ParsedCall(hook_name, kwargs))

def after(outcome, hook_name: str, hook_impls, kwargs) -> None:
pass

self._undo_wrapping = pluginmanager.add_hookcall_monitoring(before, after

看到最后一段代码,它给 PluginManager 添加了一个 monitoring ,用来记录所有的 hookcall 的执行上下文,通过把测试执行过程的所有上下文 snapshot 下来,给到最终测试代码来做校验,所以 inline_run 没有返回错误码,而是返回了 HookRecorder 对象 ,最后看一下 inline_run 的返回逻辑,是返回了一个 List[HookRecorder] 中的对象。

1
2
3
4
5
6
7
8
9
10
11
12
13
rec = []
class Collect:
def pytest_configure(x, config):
rec.append(self.make_hook_recorder(config.pluginmanager))

plugins.append(Collect())

...
if len(rec) == 1:
reprec = rec.pop()
...

return reprec

好了,今天的介绍就到这里,希望可以帮助大家更好的编写和测试自己的 pytest 插件