Skip to main content

一个伪造pytest子进程的插件

项目描述

pytest-子进程

PyPI 版本 Python 版本 https://codecov.io/gh/aklajnert/pytest-subprocess/branch/master/graph/badge.svg?token=JAU1cGoYL8 文件状态

一个伪造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"

可选地,stdoutstderr参数可以是要连接在一起的行列表(或元组),每行尾随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 参数 传递,您可以根据输入指定行为。该函数应接受一个参数,即输入数据。如果函数将返回带有stdoutstderr键的字典,则其值将附加到相应的流中。

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()。应该匹配的参数数量可以由minmax参数控制。要使用fake_subprocess.any()您需要将命令定义为tuplelist。即使使用字符串参数调用子进程命令,匹配也将起作用。

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_shellasyncio.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@hackebrotcookiecutter-pytest-plugin模板生成的。

历史

1.4.1 (2022-02-09)

其他变化

  • #74 :为夹具添加fp别名,并注册regisiter_subprocess

1.4.0 (2022-01-23)

特征

  • #71 : 添加对带有 asyncio 的标准输入的支持。

Bug修复

  • #68 :使用 asyncio 函数时将stdoutstderr设为asyncio.StreamReader实例。

  • #63#67:将缺失的项目添加到asyncio.subprocess

其他变化

  • #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)

特征

  • #49#52:添加对asyncio的支持。

其他变化

  • #50:更改文档主题。

1.1.2 (2021-07-17)

Bug修复

  • #47 : 防止allow_unregistered()keep_last_process()影响其他测试。

1.1.1 (2021-06-18)

Bug修复

其他变化

  • #42:修复register_subprocess()的类型注释。

1.1.0 (2021-04-18)

Bug修复

  • #37 : 在proc.args中保留原始命令以防止泄漏内部命令类型。

其他变化

  • #38:将 CI 从 Azure Pipelines 切换到 GitHub Actions。

  • #35 : 放弃对 python 3.4 和 3.5 的支持。将类型注释从.pyi文件移动到源中。

1.0.1 (2021-03-20)

Bug修复

  • #34 : 除非定义为列表/元组,否则防止将换行符附加到输出。

其他变化

  • #32 : 使Command类可迭代。

1.0.0 (2020-08-22)

特征

  • #29:记住子进程调用以检查是否执行了预期的命令。

  • #28 : 允许使用可变参数匹配命令(非精确匹配)。

0.1.5 (2020-06-19)

Bug修复

  • #26 : encodingerrors参数将正确触发文本模式。

0.1.4 (2020-04-28)

Bug修复

  • #22 :使用回调时返回码不会被忽略。

  • #21:回调引发的异常将优先于子进程引发的异常。

  • #20 : 无论命令类型如何,注册过程现在都将保持一致。

  • #19:修复了使用空流定义的 stderr 重定向崩溃。

0.1.3 (2020-03-04)

特征

  • #13 : 允许将关键字参数传递给回调。

Bug修复

  • #12 : 正确地从回调函数中引发异常。

文档更改

  • #15 : 添加关于回调函数的文档章节。

0.1.2 (2020-01-17)

特征

  • #3:添加对流程输入的基本支持。

Bug修复

  • #5 :在期望的时间过去后,让wait()方法引发TimeoutError 。

文档更改

  • #7#8#9:创建 Sphinx 文档。

其他变化

  • #10:从tox切换到nox以运行测试和任务。

  • #4:为 Python 3.9 添加分类器。更新 CI 配置以在该解释器版本上进行测试。

0.1.1 (2019-11-24)

其他变化

  • #1 , #2:启用对 Python 3.4 的支持,为该版本添加 CI 测试。

0.1.0 (2019-11-23)

初始发行

发布历史 发布通知| RSS订阅