Python CLIck，测试线程应用程序

Question

我正在尝试使用 pytest 测试基于点击的线程应用程序。该应用程序将永远运行并等待键盘事件。

主要.py

#!/usr/bin/python

import threading
import time
import typing
import logging
import click


def test_func(sample_var:str):
    print("Sample is: " + sample_var)

@click.option("--sample", required=True, help="Sample String", default="sample", type=str)
@click.command()
def main(sample: str):

    print("Starting App r")
    interrupt = False
    while not interrupt:
        try:
            print("Start threading ..")    
            my_thread = threading.Thread(
                target=test_func,
                kwargs=dict(sample_var=sample),
                daemon=True)
            my_thread.start()
            my_thread.join(120)
            if not interrupt:
                print("Resting for 60 seconds")
                time.sleep(60)
        except(KeyboardInterrupt, SystemExit):
            print("Received Keyboard Interrupt or system exisying, cleaning all the threads")
            interrupt=True


if __name__ == "__main__":
    main(auto_envvar_prefix="MYAPP")

问题是，在测试时我不知道如何发送Keyboard Interrupt Signal

main_test.py

from click.testing import CliRunner
from myapp.main import main
import pytest
import time
import click


def test_raimonitor_failing(cli_runner):
    """ Tests the Startup off my app"""
    runner = CliRunner()
    params = ["--sample", "notsample"]
    test = runner.invoke(cli = main, args = params)
    expected_msg = "notsample\n"
    print(test.stdout)
    print(test.output)
    assert 0
    assert expected_msg in test.stdout

测试只是挂起，我不知道如何发送信号来停止它。

我怎样才能正确测试这个系统？

Answer 1

KeyboardInterrupt要在单击函数中测试异常处理程序，您可以side_effect使用Mock

from unittest import mock

with mock.patch('test_code.wait_in_loop', side_effect=KeyboardInterrupt):
    result = runner.invoke(cli=main, args=params)

为了使测试更容易，该time.sleep()调用被移至一个单独的函数中，然后对该函数进行模拟。

测试代码

from unittest import mock

def test_raimonitor_failing():
    """ Tests the Startup off my app"""
    runner = CliRunner()
    params = ["--sample", "notsample"]
    with mock.patch('test_code.wait_in_loop', side_effect=KeyboardInterrupt):
        result = runner.invoke(cli=main, args=params)
    expected = '\n'.join(line.strip() for line in """
        Starting App
        Start threading ..
        Sample is: notsample
        Resting for 60 seconds
        Received Keyboard Interrupt or system exiting, cleaning all the threads
        
    """.split('\n')[1:-1])
    assert result.output == expected

被测代码

from click.testing import CliRunner

import click
import threading
import time


def a_test_func(sample_var: str):
    print("Sample is: " + sample_var)


def wait_in_loop():
    time.sleep(60)


@click.option("--sample", required=True, help="Sample String", default="sample", type=str)
@click.command()
def main(sample: str):
    print("Starting App")
    interrupt = False
    while not interrupt:
        try:
            print("Start threading ..")
            my_thread = threading.Thread(
                target=a_test_func,
                kwargs=dict(sample_var=sample),
                daemon=True)
            my_thread.start()
            my_thread.join(120)
            if not interrupt:
                print("Resting for 60 seconds")
                wait_in_loop()
        except(KeyboardInterrupt, SystemExit):
            print("Received Keyboard Interrupt or system exiting, cleaning all the threads")
            interrupt = True