Flask与pytest同作为子进程运行时挂起问题排查求助

问题背景 #

我花了一个半小时调试测试代码却毫无进展，完全陷入困境。为简化Flask服务器的测试流程，我编写了一个简易脚本：启动服务器、运行pytest、终止服务器、将输出写入文件，最后以pytest的退出码结束。该代码此前运行正常，但今日突然出现问题（除调试操作外未修改过代码）。

问题表现 #

测试执行到特定环节时会挂起，但其他运行方式均无此问题：

• 在VS Code中调试服务器，终端运行测试：正常
• 用测试脚本中的代码启动服务器，手动运行pytest：正常
• 用测试脚本运行pytest，在另一终端用启动脚本启动服务器：正常

另一个奇怪现象：测试总是在前置fixture的执行过程中挂起。执行数据库清空命令后，向服务器发送echo请求（用于打印当前测试名称），数据库清空成功，服务器也打印出了正确信息，但echo路由始终不返回响应，测试无法继续。此前的50余次测试中，该echo路由均正常工作。若注释掉导致挂起的测试，会在后续测试的其他路由请求时挂起。挂起时，服务器无法通过SIGTERM终止，必须使用SIGKILL。

相关代码片段 #

Echo路由代码 #

@debug.get('/echo')
def echo() -> IEcho:
    """
    Echo an input. This returns the given value, but also prints it to stdout
    on the server. Useful for debugging tests.

    ## Params:
    * `value` (`str`): value to echo
    """
    try:
        value = request.args['value']
    except KeyError:
        raise http_errors.BadRequest('echo route requires a `value` argument')

    to_print = f'{Fore.MAGENTA}[ECHO]		{value}{Fore.RESET}'
    # Print it to both stdout and stderr to ensure it is seen across all logs
    # Otherwise it could be more difficult to figure out what's up with server
    # output
    print(to_print)
    print(to_print, file=sys.stderr)
    return {'value': value}

请求发送代码 #

def get(token: JWT | None, url: str, params: dict) -> dict:
    """
    Returns the response to a GET web request

    This also parses the response to help with error checking

    ### Args:
    * `url` (`str`): URL to request to

    * `params` (`dict`): parameters to send

    ### Returns:
    * `dict`: response data
    """
    return handle_response(requests.get(
        url,
        params=params,
        headers=encode_headers(token),
        timeout=3
    ))

def echo(value: str) -> IEcho:
    """
    Echo an input. This returns the given value, but also prints it to stdout
    on the server. Useful for debugging tests.

    ## Params:
    * `value` (`str`): value to echo
    """
    return cast(IEcho, get(None, f"{URL}/echo", {"value": value}))

@pytest.fixture(autouse=True)
def before_each(request: pytest.FixtureRequest):
    """Clear the database between tests"""
    clear()
    echo(f"{request.module.__name__}.{request.function.__name__}")
    print("After echo")  # This never prints

测试脚本中运行pytest的代码 #

def pytest():
    pytest = subprocess.Popen(
        [sys.executable, '-u', '-m', 'pytest', '-v', '-s'],
    )

    # Wait for tests to finish
    print("🔨 Running tests...")
    try:
        ret = pytest.wait()
    except KeyboardInterrupt:
        print("❗ Testing cancelled")
        pytest.terminate()
        # write_outputs(pytest, None)
        # write_outputs(pytest, "pytest")
        raise
    # write_outputs(pytest, "pytest")
    if ret == 0:
        print("✅ It works!")
    else:
        print("❌ Tests failed")
    return bool(ret)

测试脚本中启动服务器的代码 #

def backend(debug=False, live_output=False):
    env = os.environ.copy()
    if debug:
        env.update({"ENSEMBLE_DEBUG": "TRUE"})
        debug_flag = ["--debug"]
    else:
        debug_flag = []
    if live_output is False:
        outputs = subprocess.PIPE
    else:
        outputs = None
    flask = subprocess.Popen(
        [sys.executable, '-u', '-m', 'flask'] + debug_flag + ['run'],
        env=env,
        stderr=outputs,
        stdout=outputs,
    )
    if outputs is not None and (flask.stderr is None or flask.stdout is None):
        print("❗ Can't read flask output", file=sys.stderr)
        flask.kill()
        sys.exit(1)

    # Request until we get a success, but crash if we failed to start in 10
    # seconds
    start_time = time.time()
    started = False
    while time.time() - start_time < 10:
        try:
            requests.get(
                f'http://localhost:{os.getenv("FLASK_RUN_PORT")}/debug/echo',
                params={'value': 'Test script startup...'},
            )
        except requests.ConnectionError:
            continue
        started = True
        break

    if not started:
        print("❗ Server failed to start in time")
        flask.kill()
        if outputs is not None:
            write_outputs(flask, None)
        sys.exit(1)
    else:
        if flask.poll() is not None:
            print("❗ Server crashed during startup")
            if outputs is not None:
                write_outputs(flask, None)
            sys.exit(1)
        print("✅ Server started")
        return flask

疑问与调试请求 #

这么简单的路由居然会挂起，这让我非常担忧。我怀疑这可能是Flask或requests库的某个罕见bug。

即使不知道具体原因，也恳请提供进一步调试的思路。

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