pythonunit-testingtestingpytestinfinite-loop

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


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.
    1. 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.

Solution

  • 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(...) == ...