Flask unit tests not closing port after each test

Question

I'm doing some unit testing for a flask application. A part of this includes restarting the flask application for each test. To do this, I'm creating my flask application in the setUp() function of my unitest.TestCase, so that I get the application in its fresh state for each run. Also, I'm starting the application in a separate thread so the tests can run without the flask application blocking.

Example below:

import requests
import unittest
from threading import Thread

class MyTest(unittest.TestCase):

    def setUp(self):
        test_port = 8000
        self.test_url = f"http://0.0.0.0:{str(test_port)}"
        self.app_thread = Thread(target=app.run, kwargs={"host": "0.0.0.0", "port": test_port, "debug": False})
        self.app_thread.start()

    def test_a_test_that_contacts_the_server(self):
        response = requests.post(
            f"{self.test_url}/dosomething",
            json={"foo": "bar"},
            headers=foo_bar
        )
        is_successful = json.loads(response.text)["isSuccessful"]
        self.assertTrue(is_successful, msg=json.loads(response.text)["message"])

    def tearDown(self):
        # what should I do here???
        pass

This becomes problematic because when the tests that come after the initial test run, they run into an issue with port 8000 being used. This raises OSError: [Errno 98] Address already in use.

(For now, I've built a workaround, where I generate a list of high ranged ports, and another list of ports used per test, so that I never select a port used by a previous test. This work around works, but I'd really like to know the proper way to shut down this flask application, ultimately closing the connection and releasing/freeing that port.)

I'm hopeful that there is a specific way to shutdown this flask application in the tearDown() function.

How should I go about shutting down the flask application in my tearDown() method?

Answer 1

I found the solution to my own question while writing it, and since it's encouraged to answer your own question on Stack Overflow, I'd like to still share this for anyone else with the same issue.

The solution to this problem is to treat the flask application as another process instead of a thread. This is accomplished using Process from the multiprocessing module en lieu of Thread from the threading module.

I came to this conclusion after reading this Stack Overflow answer regarding stopping flask without using CTRL + C. Reading that answer then lead me to read about the differences between multiprocessing and threading in this Stack Overflow answer. Of course, after that, I moved on to the official documentation on the multiprocessing module, found here. More specifically, this link will take you straight to the Process class.

I'm not able to fully articulate why the multiprocessing module serves this purpose better than threading, but I do feel that it makes more sense for this application. After all, the flask application is acting as its own API server that is separate from my test, and my test is testing the calls to it/responses it gets back. For this reason, I think it makes the most sense for my flask application to be its own process.

tl;dr

Use multiprocessing.Process en lieu of threading.Thread, and then call Process.terminate() to kill the process, followed by Process.join() to block until the process is terminated.

example:

import requests
import unittest
from multiprocessing import Process

class MyTest(unittest.TestCase):

    def setUp(self):
        test_port = 8000
        self.test_url = f"http://0.0.0.0:{str(test_port)}"
        self.app_process = Process(target=app.run, kwargs={"host": "0.0.0.0", "port": test_port, "debug": False})
        self.app_process.start()

    def test_a_test_that_contacts_the_server(self):
        response = requests.post(
            f"{self.test_url}/dosomething",
            json={"foo": "bar"},
            headers=foo_bar
        )
        is_successful = json.loads(response.text)["isSuccessful"]
        self.assertTrue(is_successful, msg=json.loads(response.text)["message"])

    def tearDown(self):
        self.app_process.terminate()
        self.app_process.join()

Test early, and test often!