在过去的一个半小时里，我一直在尝试调试这个测试，但都失败了，我完全被难住了。为了简化测试我正在构建的Flask服务器的过程，我制作了一个相对简单的脚本，启动服务器，然后运行pytest，终止服务器，将输出写入文件，然后使用pytest的退出代码退出。直到今天，这段代码一直运行得很好，从那以后我就没有修改过它(除了调试这个问题(。
问题是：当它在测试中达到某一点时，它就会挂起。奇怪的是，如果我以任何其他方式运行测试，这种情况都不会发生。
- 在VS代码中调试我的服务器，并在终端中运行测试：有效
- 使用测试脚本中使用的相同代码运行我的服务器并手动运行pytest：有效
- 在第二个终端中使用测试脚本运行pytest并通过启动服务器脚本运行服务器(该脚本使用与测试脚本相同的代码来运行服务器(：有效
还有一件有趣的事情：测试总是挂在同一个地方，在安装夹具的中途。它向服务器发送clear命令和echo请求(服务器打印当前测试的名称(。数据库成功清除，服务器回显正确的信息，但回显路由从未退出——我的测试从未得到响应。该回波路径对于在此点之前发生的50次左右的测试表现完美。如果我注释掉导致它失败的测试，那么它将在下一次测试中失败。如果我注释掉对echo的调用，那么它将挂起稍后对不同路由的完全不同请求的测试。当挂起时，不能使用SIGTERM杀死服务器，而是需要SIGKILL。
这是我的回波路线：
@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]tt{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或请求库或其他什么地方发现了一些疯狂的bug。
即使你不知道这是怎么回事，有任何想法我可以进一步调试它也会很有帮助，因为我完全不知道发生了什么。
事实证明，我的服务器输出填满了管道中的所有缓冲区空间，这意味着它将等待缓冲区清空。问题是，我的测试脚本正在等待测试退出，除非服务器处于活动状态，否则测试无法进行。因此，代码达到了三方死锁。我通过将输出重定向到一个文件(缓冲区大小有限不是问题(来解决这个问题。