Flask 单元测试在每次测试后不关闭端口

Question

我正在对烧瓶应用程序进行一些单元测试。其中一部分包括为每个测试重新启动烧瓶应用程序。为此，我在setUp()my 的函数中创建 Flask 应用程序unitest.TestCase，以便每次运行时应用程序都处于新鲜状态。另外，我在单独的线程中启动应用程序，以便测试可以在 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

这会成为问题，因为当初始测试运行后进行的测试时，它们会遇到端口8000使用的问题。这引发了OSError: [Errno 98] Address already in use.

（目前，我已经构建了一个解决方法，在其中生成一个高范围端口列表，以及每个测试使用的另一个端口列表，这样我就不会选择先前测试使用的端口。这个解决方法有效，但我我真的很想知道关闭此烧瓶应用程序的正确方法，最终关闭连接并释放/释放该端口。）

我希望有一种特定的方法可以在函数中关闭此烧瓶应用程序tearDown()。

我应该如何关闭我的tearDown()方法中的烧瓶应用程序？

Answer 1

我在编写它时找到了自己问题的解决方案，并且由于鼓励在 Stack Overflow 上回答您自己的问题，因此我仍然想与其他有相同问题的人分享这个问题。

解决这个问题的方法是将flask应用程序视为另一个进程而不是线程。这是使用Processfrom the multiprocessingmodule 代替Threadfrom the threadingmodule 来完成的。

在阅读了有关在不使用 CTRL + C 的情况下停止 Flask 的Stack Overflow 答案后，我得出了这个结论。阅读该答案然后让我了解了Stack Overflow 答案中multiprocessing和之间的差异。当然，在那之后，我继续阅读该模块的官方文档，可以在此处找到。更具体地说，此链接将带您直接进入课程。threading multiprocessingProcess

我无法完全阐明为什么该multiprocessing模块比 更好地满足此目的threading，但我确实认为它对于该应用程序更有意义。毕竟，Flask 应用程序充当其自己的 API 服务器，与我的测试分开，而我的测试正在测试对其的调用/它返回的响应。出于这个原因，我认为我的烧瓶应用程序成为它自己的进程是最有意义的。

太长了；博士

使用multiprocessing.Processen 代替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()

尽早测试，并经常测试！