一个伪造pytest子进程的插件
项目描述
pytest-子进程
一个伪造pytest子进程的插件
用法
该插件添加了fake_process固定装置(和fp作为别名)。它可用于注册子流程结果,因此您无需依赖真实流程。该插件挂钩subprocess.Popen(),这是其他子流程功能的基础。这使得subprocess.run()、 subprocess.call()、subprocess.check_call()和 subprocess.check_output()方法也可以正常工作。
安装
您可以通过来自PyPI的pip安装pytest-subprocess:
$ pip install pytest-subprocess
基本用法
最重要的方法是fp.register() (或register_subprocess如果您更喜欢更详细),它允许定义假进程行为。
def test_echo_null_byte(fp):
fp.register(["echo", "-ne", "\x00"], stdout=bytes.fromhex("00"))
process = subprocess.Popen(
["echo", "-ne", "\x00"], stdout=subprocess.PIPE,
)
out, _ = process.communicate()
assert process.returncode == 0
assert out == b"\x00"
可选地,stdout和stderr参数可以是要连接在一起的行列表(或元组),每行尾随os.linesep。
def test_git(fp):
fp.register(["git", "branch"], stdout=["* fake_branch", " master"])
process = subprocess.Popen(
["git", "branch"],
stdout=subprocess.PIPE,
universal_newlines=True,
)
out, _ = process.communicate()
assert process.returncode == 0
assert out == "* fake_branch\n master\n"
传递输入
默认情况下,如果你使用Popen.communicate()方法的输入参数 ,它不会崩溃,但也不会做任何有用的事情。通过将函数作为fp.register ()方法的 stdin_callable 参数 传递,您可以根据输入指定行为。该函数应接受一个参数,即输入数据。如果函数将返回带有stdout或 stderr键的字典,则其值将附加到相应的流中。
def test_pass_input(fp):
def stdin_function(input):
return {
"stdout": "This input was added: {data}".format(
data=input.decode()
)
}
fp.register(
["command"],
stdout=[b"Just stdout"],
stdin_callable=stdin_function,
)
process = subprocess.Popen(
["command"], stdin=subprocess.PIPE, stdout=subprocess.PIPE,
)
out, _ = process.communicate(input=b"sample input\n")
assert out.splitlines() == [
b"Just stdout",
b"This input was added: sample input",
]
未注册的命令
默认情况下,当使用fp夹具时,任何尝试运行尚未注册的子进程都会引发ProcessNotRegisteredError异常。要允许它,请使用 fp.allow_unregistered(True) ,它将使用真正的subprocess执行所有未注册的进程,或者使用 fp.pass_command("command")只允许一个命令。
def test_real_process(fp):
with pytest.raises(fp.exceptions.ProcessNotRegisteredError):
# this will fail, as "ls" command is not registered
subprocess.call("ls")
fp.pass_command("ls")
# now it should be fine
assert subprocess.call("ls") == 0
# allow all commands to be called by real subprocess
fp.allow_unregistered(True)
assert subprocess.call(["ls", "-l"]) == 0
不同的结果
每个register()或pass_command()方法调用将只注册一个命令执行。您可以多次调用这些方法,以更改每个子进程运行的伪造输出。当您调用 subprocess 时,将引发更多。为了防止这种情况,请调用fp.keep_last_process(True),它将永远保留最后注册的进程。
def test_different_output(fp):
# register process with output changing each execution
fp.register("test", stdout="first execution")
# the second execution will return non-zero exit code
fp.register("test", stdout="second execution", returncode=1)
assert subprocess.check_output("test") == b"first execution"
second_process = subprocess.run("test", stdout=subprocess.PIPE)
assert second_process.stdout == b"second execution"
assert second_process.returncode == 1
# 3rd time shall raise an exception
with pytest.raises(fp.exceptions.ProcessNotRegisteredError):
subprocess.check_call("test")
# now, register two processes once again,
# but the last one will be kept forever
fp.register("test", stdout="first execution")
fp.register("test", stdout="second execution")
fp.keep_last_process(True)
# now the processes can be called forever
assert subprocess.check_output("test") == b"first execution"
assert subprocess.check_output("test") == b"second execution"
assert subprocess.check_output("test") == b"second execution"
assert subprocess.check_output("test") == b"second execution"
使用回调
您可以将函数作为回调参数传递给register() 方法,该方法将被执行而不是真正的子进程。回调函数可以引发异常,这些异常将在测试中解释为子进程引发的异常。夹具会将FakePopen类实例传递给回调函数,可用于更改返回码或修改输出流。
def callback_function(process):
process.returncode = 1
raise PermissionError("exception raised by subprocess")
def test_raise_exception(fp):
fp.register(["test"], callback=callback_function)
with pytest.raises(
PermissionError, match="exception raised by subprocess"
):
process = subprocess.Popen(["test"])
process.wait()
assert process.returncode == 1
可以使用callback_kwargs参数将其他关键字参数传递给回调:
def callback_function_with_kwargs(process, return_code):
process.returncode = return_code
def test_callback_with_arguments(fp):
return_code = 127
fp.register(
["test"],
callback=callback_function_with_kwargs,
callback_kwargs={"return_code": return_code},
)
process = subprocess.Popen(["test"])
process.wait()
assert process.returncode == return_code
作为上下文管理器
fp夹具提供了context()方法,允许我们将其用作上下文管理器。当某个命令被允许时,它可以用来限制范围,例如确保代码不想在其他地方执行它。
def test_context_manager(fp):
with pytest.raises(fp.exceptions.ProcessNotRegisteredError):
# command not registered, so will raise an exception
subprocess.check_call("test")
with fp.context() as nested_process:
nested_process.register("test", occurrences=3)
# now, we can call the command 3 times without error
assert subprocess.check_call("test") == 0
assert subprocess.check_call("test") == 0
# the command was called 2 times, so one occurrence left, but since the
# context manager has been left, it is not registered anymore
with pytest.raises(fp.exceptions.ProcessNotRegisteredError):
subprocess.check_call("test")
非精确的命令匹配
如果您需要捕获包含一些不可预测元素的命令,例如随机生成的文件名的路径,您可以为此目的使用fake_subprocess.any()。应该匹配的参数数量可以由min和max参数控制。要使用fake_subprocess.any()您需要将命令定义为tuple或list。即使使用字符串参数调用子进程命令,匹配也将起作用。
def test_non_exact_matching(fp):
# define a command that will take any number of arguments
fp.register(["ls", fp.any()])
assert subprocess.check_call("ls -lah") == 0
# `fake_subprocess.any()` is OK even with no arguments
fp.register(["ls", fp.any()])
assert subprocess.check_call("ls") == 0
# but it can force a minimum amount of arguments
fp.register(["cp", fp.any(min=2)])
with pytest.raises(fp.exceptions.ProcessNotRegisteredError):
# only one argument is used, so registered command won't match
subprocess.check_call("cp /source/dir")
# but two arguments will be fine
assert subprocess.check_call("cp /source/dir /tmp/random-dir") == 0
# the `max` argument can be used to limit maximum amount of arguments
fp.register(["cd", fp.any(max=1)])
with pytest.raises(fp.exceptions.ProcessNotRegisteredError):
# cd with two arguments won't match with max=1
subprocess.check_call("cd ~/ /tmp")
# but any single argument is fine
assert subprocess.check_call("cd ~/") == 0
# `min` and `max` can be used together
fp.register(["my_app", fp.any(min=1, max=2)])
assert subprocess.check_call(["my_app", "--help"]) == 0
检查进程是否被调用
您可能只想检查是否调用了某个命令,您可以通过访问fp.calls来做到这一点,所有命令都按调用存储。您还可以使用实用函数fp.call_count()查看调用了多少命令。后者支持fp.any()。
def test_check_if_called(fp):
fp.keep_last_process(True)
# any command can be called
fp.register([fp.any()])
subprocess.check_call(["cp", "/tmp/source", "/source"])
subprocess.check_call(["cp", "/source", "/destination"])
subprocess.check_call(["cp", "/source", "/other/destination"])
# you can check if command is in ``fp.calls``
assert ["cp", "/tmp/source", "/source"] in fp.calls
assert ["cp", "/source", "/destination"] in fp.calls
assert ["cp", "/source", "/other/destination"] in fp.calls
# or check how many it was called, possibly with wildcard arguments
assert fp.call_count(["cp", "/source", "/destination"]) == 1
# with ``call_count()`` you don't need to use the same type as
# the subprocess was called
assert fp.call_count("cp /tmp/source /source") == 1
# can be used with ``fp.any()`` to match more calls
assert fp.call_count(["cp", fp.any()]) == 3
处理信号
您可以在Popen实例中使用标准的kill()、terminate()或send_signal()方法。还有一个附加的received_signals()方法来获取进程接收到的所有信号的元组。也可以为信号设置可选的回调函数。
import signal
def test_signal_callback(fp):
"""Test that signal callbacks work."""
def callback(process, sig):
if sig == signal.SIGTERM:
process.returncode = -1
fp.register("test", signal_callback=callback)
process = subprocess.Popen("test")
process.send_signal(signal.SIGTERM)
process.wait()
assert process.returncode == -1
assert process.received_signals() == (signal.SIGTERM,)
异步支持
该插件现在支持 asyncio 并适用于asyncio.create_subprocess_shell 和asyncio.create_subprocess_exec:
@pytest.mark.asyncio
async def test_basic_usage(fp,):
fp.register(
["some-command-that-is-definitely-unavailable"], returncode=500
)
process = await asyncio.create_subprocess_shell(
"some-command-that-is-definitely-unavailable"
)
returncode = await process.wait()
assert process.returncode == returncode
assert process.returncode == 500
文档
有关完整文档,包括 API 参考,请参阅https://pytest-subprocess.readthedocs.io/en/latest/。
贡献
贡献是非常受欢迎的。可以使用tox运行测试,请确保覆盖率至少保持不变,然后再提交拉取请求。
执照
根据MIT许可条款分发,“pytest-subprocess”是免费的开源软件
问题
如果您遇到任何问题,请提交问题并附上详细说明。
这个pytest插件是使用Cookiecutter和@hackebrot的cookiecutter-pytest-plugin模板生成的。
历史
1.4.1 (2022-02-09)
其他变化
#74 :为夹具添加fp别名,并注册regisiter_subprocess。
1.4.0 (2022-01-23)
特征
#71 : 添加对带有 asyncio 的标准输入的支持。
Bug修复
其他变化
#69 : 将代码提取到单独的文件中以改进导航。
1.3.2 (2021-11-07)
Bug修复
#61:修复了asyncio.create_subproess_exec()的行为。
1.3.1 (2021-11-01)
Bug修复
#58 : 正确处理文件流输出。
1.3.0 (2021-10-24)
特征
#55 : 添加对terminate()、kill()、send_signal()的支持。
1.2.0 (2021-10-09)
特征
其他变化
#50:更改文档主题。
1.1.2 (2021-07-17)
Bug修复
#47 : 防止allow_unregistered()和keep_last_process()影响其他测试。
1.1.1 (2021-06-18)
Bug修复
#43 : 在调用通信()时等待回调线程完成。
其他变化
#42:修复register_subprocess()的类型注释。
1.1.0 (2021-04-18)
Bug修复
#37 : 在proc.args中保留原始命令以防止泄漏内部命令类型。
其他变化
1.0.1 (2021-03-20)
Bug修复
#34 : 除非定义为列表/元组,否则防止将换行符附加到输出。
其他变化
#32 : 使Command类可迭代。
1.0.0 (2020-08-22)
特征
0.1.5 (2020-06-19)
Bug修复
#26 : encoding和errors参数将正确触发文本模式。
0.1.4 (2020-04-28)
Bug修复
0.1.3 (2020-03-04)
特征
#13 : 允许将关键字参数传递给回调。
Bug修复
#12 : 正确地从回调函数中引发异常。
文档更改
#15 : 添加关于回调函数的文档章节。
0.1.2 (2020-01-17)
特征
#3:添加对流程输入的基本支持。
Bug修复
#5 :在期望的时间过去后,让wait()方法引发TimeoutError 。
文档更改
其他变化
0.1.1 (2019-11-24)
其他变化
0.1.0 (2019-11-23)
初始发行