时间戳

首页 > 解决方案 > click.testing.CliRunner 和处理 SIGINT/SIGTERM 信号

python - click.testing.CliRunner 和处理 SIGINT/SIGTERM 信号

问题描述

我想添加一些关于我的 cli 应用程序如何处理不同信号（SIGTERM等）的测试。我正在使用本机测试解决方案click.testing.CliRunner以及 pytest。

测试看起来非常标准和简单

def test_breaking_process(server, runner):

    address = server.router({'^/$': Page("").exists().slow()})

    runner = CliRunner(mix_stderr=True)
    args = [address, '--no-colors', '--no-progress']
    result = runner.invoke(main, args)
    assert result.exit_code == 0

在这里我被卡住了，我怎么能发送SIGTERM到进程中runner.invoke？如果我使用 e2e 测试（调用可执行文件而不是 CLIrunner），我认为这样做没有问题，但我想尝试实现这一点（至少能够发送 os.kill）

有办法吗？

标签： pythonpytestpython-click

解决方案

因此，如果您想测试您的点击驱动应用程序以处理不同的信号，您可以执行下一个过程。

def test_breaking_process(server, runner):

    from multiprocessing import Queue, Process
    from threading import Timer
    from time import sleep
    from os import kill, getpid
    from signal import SIGINT

    url = server.router({'^/$': Page("").slow().exists()})
    args = [url, '--no-colors', '--no-progress']

    q = Queue()

    # Running out app in SubProcess and after a while using signal sending 
    # SIGINT, results passed back via channel/queue  
    def background():
        Timer(0.2, lambda: kill(getpid(), SIGINT)).start()
        result = runner.invoke(main, args)
        q.put(('exit_code', result.exit_code))
        q.put(('output', result.output))

    p = Process(target=background)
    p.start()

    results = {}

    while p.is_alive():
        sleep(0.1)
    else:
        while not q.empty():
            key, value = q.get()
            results[key] = value

    assert results['exit_code'] == 0
    assert "Results can be inconsistent, as execution was terminated" in results['output']

推荐阅读