How-to test an infinite loop

Let's say you have some code like this that you want to test:

import time


def runner():
    while True:
        print("Iterating")
        time.sleep(0.1)

There are several approaches if you want to (and can) modify the loop itself. For example, we can add a flag, so that the actual loop depends on it. And that flag can be set using some stop() method or by patching it during tests.

But if you really can't modify the loop, there is an option using signals:

import contextlib
import signal


class TimeoutException(Exception):
    pass


def timeout_handler(signum, frame):
    raise TimeoutException


@contextlib.contextmanager
def timeout(timeout):
    signal.signal(signal.SIGALRM, timeout_handler)
    signal.alarm(timeout)

    with contextlib.suppress(TimeoutException):
        yield

        # Cancel signal if not yet triggered
        signal.alarm(0)

You can use timeout in your tests like:

def test_runner(capsys):
    with timeout(1):
        runner()

    captured = capsys.readouterr()
    assert captured.out.startswith("Iterating")

But keep in mind that SIGALRM is only available in Unix systems.

If you want to include the test code in the coverage report generated by coverage you'll need to add the following setting to the config:

[tool.coverage.run]
sigterm = true

From the docs:

[run] sigterm (boolean, default False) if true, register a SIGTERM signal handler to capture data when the process ends due to a SIGTERM signal. This includes Process.terminate, and other ways to terminate a process. This can help when collecting data in usual situations, but can also introduce problems (see issue 1310).