在pytest测试中通过多进程启动本地Flask服务后无法访问的问题

• 脚本启动逻辑的分支在多进程导入场景下失效
  常规Flask脚本会把服务启动代码放在if __name__ == "__main__"判断块里，实现“直接运行脚本时启动服务、被其他模块导入时不自动拉起服务”的逻辑。手动执行python flask_file时，模块的__name__值为"__main__"，可以正常进入分支启动服务。但在pytest的多进程场景下，flask_file是被导入的模块，__name__值为模块名，对应判断块里的app.run()逻辑根本不会执行——子进程启动后会执行完空的main函数直接退出，根本没有真正启动Flask服务。而父进程的pytest逻辑感知不到子进程的退出动作，只会按预设完成30秒等待，看起来和“服务正常运行”的状态没有区别。
• 子进程启动异常被完全静默
  即便main函数没有上述分支问题，只要子进程执行过程中抛出任何错误（比如端口被占用、环境变量缺失、Flask启动参数不兼容），子进程会直接退出，但异常栈不会自动回传到父进程的pytest控制台，测试逻辑完全察觉不到服务已经挂掉。
• Flask热重载机制和多进程管理的天然冲突
  如果Flask启动时开启了debug模式/热重载，会自动派生额外的进程做文件监听，和multiprocessing的进程管理逻辑嵌套冲突，要么导致端口绑定失败，要么出现进程残留。
• 就绪检测逻辑完全失效
  固定等待30秒的逻辑没有任何服务状态校验，既无法提前发现服务已经挂掉的问题，也会在服务1秒就启动完成后无意义空等29秒。

• 拆分Flask的初始化和启动逻辑
  把Flask实例创建、路由注册的逻辑抽出来放到独立方法里，不要把启动逻辑包在if __name__ == "__main__"的判断块里。启动逻辑只作为脚本直接运行时的入口，多进程场景下直接调用Flask实例的run方法，避免分支判断失效的问题。
• 多进程启动时关闭热重载
  显式设置use_reloader=False、debug=False，完全关闭Flask的热重载和调试能力，这是多进程下启动Flask服务最容易遗漏的配置，否则会和multiprocessing的进程管理逻辑产生嵌套冲突，要么导致端口绑定失败，要么出现无法清理的残留进程。
• 替换固定睡眠为端口轮询检测
  循环检测服务端口是否可以正常连接，最多等待10秒，一旦检测到服务可访问就继续后续测试逻辑，超时则直接抛出测试错误，不要无意义空等。
• 用pytest fixture封装服务生命周期
  把服务启动、进程清理的逻辑放到fixture中，利用pytest的yield机制自动完成测试结束后的进程销毁，避免测试用例漏写清理逻辑导致端口残留。

import time
import socket
import requests
import pytest
# 从flask_file中导入已经注册好路由的Flask实例，不要直接调用main方法
from flask_file import app


def wait_for_port(port: int, host: str = "127.0.0.1", timeout: int = 10):
    """轮询检测端口连通性，替代固定time.sleep"""
    start_time = time.time()
    while True:
        try:
            with socket.create_connection((host, port), timeout=1):
                return
        except (ConnectionRefusedError, OSError):
            if time.time() - start_time > timeout:
                raise RuntimeError(f"本地Flask服务在{timeout}秒内未完成启动")
            time.sleep(0.2)


@pytest.fixture(scope="function")
def test_endpoint():
    from multiprocessing import Process
    # 启动服务时关闭热重载，避免进程冲突
    server_process = Process(
        target=lambda: app.run(
            host="127.0.0.1",
            port=8888,
            debug=False,
            use_reloader=False
        )
    )
    server_process.start()
    # 等待服务端口就绪
    wait_for_port(8888)
    yield "http://127.0.0.1:8888/"
    # 测试结束后强制清理进程
    server_process.terminate()
    server_process.join(timeout=5)


def test_send_request(test_endpoint: str):
    resp = requests.get(f"{test_endpoint}")
    assert resp.status_code == 200
    assert resp.text == "hello world"

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