Flask модульные тесты не закрывают порт после каждого теста

Question

Я провожу модульное тестирование для приложения flask. Часть этого включает перезапуск приложения flask для каждого теста. Для этого я создаю свое приложение flask в функции setUp() моего unitest.TestCase, так что я получаю приложение в его состоянии fre sh для каждого запуска. Кроме того, я запускаю приложение в отдельном потоке, чтобы тесты могли выполняться без блокировки приложения flask.

Пример ниже:

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

Это становится проблематичным c, поскольку когда тесты, которые приходят после первоначального запуска теста, сталкиваются с проблемой использования порта 8000. Это поднимает OSError: [Errno 98] Address already in use.

(На данный момент я создал обходной путь, где я генерирую список портов высокого ранга и другой список портов, используемых для каждого теста, так что я никогда не выбираю используемый порт предыдущим тестом. Это обходное решение работает, но я действительно хотел бы знать, как правильно закрыть это приложение flask, в конечном итоге закрыть соединение и освободить / освободить этот порт.)

I ' Я надеюсь, что есть определенный c способ выключения этого flask приложения в функции tearDown().

Как мне go сообщить о закрытии приложения flask в моем tearDown() метод?

Answer 1

Я нашел решение для своего собственного вопроса, когда писал его, и, поскольку рекомендуется ответить на ваш собственный вопрос о переполнении стека , я бы хотел поделиться этим с кем-то еще с той же проблемой.

Решением этой проблемы является обработка приложения flask как другого процесса вместо потока. Это достигается с помощью Process из multiprocessing модуля вместо Thread из threading модуля.

Я пришел к такому выводу после прочтения этого ответа переполнения стека относительно остановки flask без использования CTRL + C. После прочтения этого ответа я узнаю о различиях между multiprocessing и threading в этом ответе переполнения стека . Конечно, после этого я перешел к официальной документации по multiprocessing модулю, , найденному здесь . Более конкретно, эта ссылка приведет вас прямо к классу Process.

Я не могу полностью сформулировать , почему модуль multiprocessing служит этому цель лучше, чем threading, но я чувствую, что это имеет больше смысла для этого приложения. В конце концов, приложение flask действует как собственный сервер API, который отделен от моего теста, и мой тест проверяет вызовы к нему / ответы, которые он получает. По этой причине, я думаю, что для моего flask приложения имеет смысл иметь собственный процесс.

tl; dr

Использовать multiprocessing.Process вместо threading.Thread, и затем вызовите Process.terminate(), чтобы завершить процесс, а затем Process.join(), чтобы заблокировать, пока процесс не будет завершен.

пример:

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

Проверяйте рано и часто проверяйте!