Reputation: 706
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
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