我正在开发一个 API，当用户上传文件时，它会即时处理该文件以从中提取一些数据。该文件可能非常大（最大 5 GB），并且不会持久保存在服务器上。
我有一堆使用common test facility provided by Flask 的系统测试。我也知道怎么make the test upload a file。
但是，有一个案例我没有被系统测试覆盖。如果用户开始上传文件，然后在中间断开连接，而服务器正忙于从它已经接收到的数据中提取数据，该怎么办？
我使用curl 来启动一个大文件并在中间终止客户端。 API 的行为似乎符合我的预期。但我更喜欢完全自动化的方法。
我该怎么做？ Flask 或 Werkzeug 中有什么东西可以让我进行这样的测试吗？如果没有，有什么可行的方法？
【问题讨论】：
标签： python flask automation system-testing
找了一会儿，似乎Flask的测试功能和Requests library都帮不了我。因此，解决方案是使用 Python's sockets 进行手动 HTTP 请求。
为了节省时间，模拟请求并研究实际发送的内容的最简单方法是使用 curl 和一些附加参数：
--trace-ascii - 可以显示curl 发送和接收的所有内容。非常方便。
--limit-rate 3K 帮助模拟了一个非常慢的连接，结合gunicorn (gunicorn --timeout 2 ...) 中的超时配置允许重现客户端开始 POST 文件，然后挂起太久的情况.
从那里，我只需复制curl 的行为，因为它在终端中报告它到使用socket.socket() 的测试代码。
然后，测试代码通过一个线程将文件流式传输到服务器，同时定期暂停并向主线程报告其进度。当发送文件的一部分但不是全部时，主线程将突然终止正在流式传输文件的线程。然后它将等待 gunicorn 超时，然后与数据库交互以检查处理是否按预期进行。
我不知道的一件事是 Werkzeug 缓冲了请求，这绝对不直观，因为在使用 werkzeug.formparser.parse_form_data() 时，每次换行符时都会调用传递给 Werkzeug 的流的方法 write遭遇。诀窍是它首先被缓冲，并且只有当缓冲区已满时，才会开始为缓冲区中的数据调用write。最初，我只发送包含几千字节的文件，所以看起来 Werkzeug 只是在内存中读取整个文件，然后才让我处理它。当我开始发送大于其 65,536 字节缓冲区的文件时，我注意到在缓冲区被填满后，我会突然收到对 write 的调用。
【讨论】：