Search code examples
pythonunit-testingflaskpytestflask-testing

Pytest hangs when a flask server runs another thread


I am using Python3, Flask 0.12 and Pytest 3.0.7.

I have a flask app similar to this:

class AppInitializer:
    def __init__(self):
        pass

    @staticmethod
    def __function_to_be_refreshed():
        while True:
            try:
                time.sleep(5)
            except Exception as e:
                logger.exception(e)


    def create_app(self):
        daemon_thread = threading.Thread(target=self.__function_to_be_refreshed)
        daemon_thread.daemon = True
        daemon_thread.start()
        atexit.register(daemon_thread.join)
        app_ = Flask(__name__)
        return app_


app_initializer = AppInitializer()
app = app_initializer.create_app()

I am trying to test this app using pytest as follows:

import unittest

import pytest


class TestCheckPriceRequestAPI(unittest.TestCase):
    def setUp(self):
        self.app = api.app.test_client()

    def test_random(self):
        pass

When I run this test using pytest, this test (along with all other tests) runs successfully but pytest hangs. How to stop the running pytest process (or maybe kill the daemon thread)?


Solution

  • The join command only means that threading will wait until thread finishes, but wouldn't finish it. To end thread with endless loop you could do like this:

    class AppInitializer:
        def __init__(self):
            self._run = True
            self._daemon_thread = None
    
        def __function_to_be_refreshed(self):
            while self._run:
                try:
                    time.sleep(5)
                except Exception as e:
                    logger.exception(e)
    
        def __shutdown(self):
            self._run = False
            if self._daemon_thread:
                self._daemon_thread.join()
    
        def create_app(self):
            self._daemon_thread = threading.Thread(target=self.__function_to_be_refreshed)
            ...
            atexit.register(self.__shutdown)
            ...