Is it possible to test a while True loop with pytest (I try with a timeout)?

Question

I have a python function foo with a while True loop inside. For background: It is expected do stream info from the web, do some writing and run indefinitely. The asserts test if the writing was done correctly.

Clearly I need it to stop sometime, in order to test.

What I did was to run via multirpocessing and introduce a timeout there, however when I see the test coverage, the function which ran through the multiprocessing, are not marked as covered.

Question 1: Why does pytest now work this way?

Question 2: How can I make this work?

I was thinking it's probably because I technically exit the loop, so maybe pytest does not mark this as tested....

import time
import multiprocessing

def test_a_while_loop():
    # Start through multiprocessing in order to have a timeout.
    p = multiprocessing.Process(
        target=foo
        name="Foo",
    )
    try:
        p.start()
        # my timeout
        time.sleep(10)
        p.terminate()
    finally:
        # Cleanup.
        p.join()

    # Asserts below
    ...

More info

1. I looked into adding a decorator such as @pytest.mark.timeout(5), but that did not work and it stops the whole function, so I never get to the asserts. (as suggested here).
2. If I don't find a way, I will just test the parts, but ideally I would like to find a way to test by breaking the loop.
3. I know I can re-write my code in order to make it have a timeout, but that would mean changing the code to make it testable, which I don't think is a good design.
   4. Mocks I have not tried (as suggested here), because I don't believe I can mock what I do, since it writes info from the web. I need to actually see the "original" working.

Answer 1

Break out the functionality you want to test into a helper method. Test the helper method.

def scrape_web_info(url):
  data = get_it(url)
  return data

# In production:

while True:
  scrape_web_info(...)

# During test:

def test_web_info():
  assert scrape_web_info(...) == ...

Answer 2

Yes, it is possible and the code above shows one way to do it (run through a multiprocessing with a timeout).

Since the asserts were running fine, I found out that the issue was not the pytest, but the coverage report not accounting for the multiprocessing properly.

I describe how I fix this (now separate) issue question here.

Answer 3

Actually, I had the same problem with an endless task to test and coverage. However, In my code, there is a .run_forever() method which runs a .run_once() method inside in an infinite loop. So, I can write a unit test for the .run_once() method to test its functionality. Nevertheless, if you want to test your forever function despite the Halting Problem for getting more extent code coverage, I propose the following approach using a timeout regardless of tools you've mentioned including multiprocessing or @pytest.mark.timeout(5) which didn't work for me either:

• First, install the interruptingcow PyPI package to have a nice timeout for raising an optional exception: pip install interruptingcow
• Then:

import pytest
import asyncio
from interruptingcow import timeout
from <path-to-loop-the-module> import EventLoop

class TestCase:
    @pytest.mark.parametrize("test_case", ['none'])
    def test_events(self, test_case: list):
        assert EventLoop().run_once()  # It's usual

    @pytest.mark.parametrize("test_case", ['none'])
    def test_events2(self, test_case: list):
        try:
            with timeout(10, exception=asyncio.CancelledError):
                EventLoop().run_forever()
                assert False
        except asyncio.CancelledError:
            assert True