CODEWITHSUNDEEP
pythonunit-testingflaskpytestflask-testing
tusharmakkar08
tusharmakkar08

Reputation: 706

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)?

Upvotes: 1

Views: 1694

Answers (1)

Fine
Fine

Reputation: 2154

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

Upvotes: 3

Related Questions