Python CLick, Testing threading applications

user_7953289 0
Viewed 249

I am trying to test a click based threading application with pytest. The application runs forever and waits for a keyboard event.

main.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")

The problem is that while testing I do not know how to send the 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

And the tests just hangs, and I don't know how to send the signal to stop it.

How can I test this system properly?

python-click pytest python
1 Answers

To test a KeyboardInterrupt exception handler in a click function, you can use side_effect on a Mock

from unittest import mock

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

To make the testing easier, the time.sleep() call was moved into a separate function, and then that function was mocked.

Test Code

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

Code Under Test

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
user_73117670
Related