如何测试Python中的阻塞式CLI应用？以Flask启动工具为例

你的问题很典型——当CLI命令会启动一个阻塞服务（比如Flask的开发服务器）时，直接用CliRunner.invoke()会卡住后续测试代码。下面给你几个更正规的解决方案，按推荐程度排序：

1. 优先使用Flask自带的测试客户端（无需启动真实服务器） #

其实根本不需要启动真实的HTTP服务器来测试Flask应用。Flask提供了测试客户端，可以直接模拟请求，完全在进程内完成，既高效又不会有阻塞问题。

修改你的test_manage.py如下：

from app import app
import unittest

class TestManage(unittest.TestCase):
    def setUp(self):
        # 创建测试客户端
        self.client = app.test_client()
        # 开启测试模式，捕获异常
        app.config['TESTING'] = True

    def test_runserver_functionality(self):
        # 直接用客户端模拟GET请求
        response = self.client.get('/')
        self.assertEqual(response.status_code, 200)
        self.assertEqual(response.data.decode('utf-8'), 'Hello, World!')

if __name__ == '__main__':
    unittest.main()

这个方案的优势：

• 不需要启动真实服务器，测试速度极快
• 没有进程/线程管理的额外复杂度
• 完全符合Flask官方推荐的测试方式

2. 让CLI命令启动非阻塞的服务器 #

如果你确实需要测试真实的CLI启动流程（比如验证命令参数、Click的配置等），可以修改manage.py中的runserver命令，让它在后台启动服务器，避免阻塞：

修改manage.py：

import click
from app import app
import threading

@click.command()
def runserver():
    # 用线程启动Flask服务器，避免阻塞CLI进程
    server_thread = threading.Thread(target=app.run, kwargs={'use_reloader': False})
    server_thread.daemon = True  # 主线程结束时自动销毁子线程
    server_thread.start()

if __name__ == "__main__":
    runserver()

然后你的测试代码就可以用CliRunner.invoke()正常执行了：

from manage import runserver
from click.testing import CliRunner
import requests
import time
import unittest

class TestManage(unittest.TestCase):
    def test_runserver(self):
        runner = CliRunner()
        runner.invoke(runserver)
        # 给服务器一点启动时间
        time.sleep(0.5)
        r = requests.get('http://127.0.0.1:5000')
        self.assertEqual(r.status_code, 200)

if __name__ == '__main__':
    unittest.main()

注意：use_reloader=False必须加上，否则Flask的自动重载会启动另一个进程，导致线程管理失效。

3. 优化多进程测试方案（如果必须用真实阻塞服务器） #

如果你的场景必须保留app.run()的阻塞特性（比如测试生产环境的启动逻辑），可以优化你现有的多进程方案，比如增加服务器启动等待逻辑，避免请求时服务器还没就绪：

from manage import runserver
from multiprocessing import Process
from click.testing import CliRunner
import requests
import time
import unittest

class TestManage(unittest.TestCase):
    def setUp(self):
        self.runner = CliRunner()
        self.server = Process(target=self.runner.invoke, args=(runserver,))
        self.server.start()
        # 等待服务器启动，最多等5秒
        max_wait = 5
        start_time = time.time()
        while time.time() - start_time < max_wait:
            try:
                requests.get('http://127.0.0.1:5000')
                break
            except requests.exceptions.ConnectionError:
                time.sleep(0.1)
        else:
            raise TimeoutError("服务器启动超时")

    def test_runserver(self):
        r = requests.get('http://127.0.0.1:5000')
        self.assertEqual(r.status_code, 200)

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

if __name__ == '__main__':
    unittest.main()

这个方案解决了原代码中可能出现的“服务器还没启动就发请求”的问题，让测试更稳定。

内容的提问来源于stack exchange，提问作者Bitto