pytest断言流式响应需用requests的stream=true配合iter_lines()或iter_content()边收边验，禁用timeout、避免sleep，mock时须构造可迭代对象，推荐fixture封装读取逻辑并确保close()。
pytest怎么断言流式响应的分块数据
流式响应（如 text/event-stream 或分块传输 Transfer-Encoding: chunked）不会一次性返回完整 body，而是持续推送多个小片段。直接用 response.text 或 response.json() 会卡住或报错——因为连接未关闭，读取阻塞。
关键不是“等它结束”，而是边收边验。pytest 本身不处理流，得靠 requests 的 stream=True + 手动迭代 response.iter_lines() 或 response.iter_content()。
- 用
response.iter_lines(decode_unicode=True)处理 SSE（server-sent events），每行是独立事件 - 用
response.iter_content(chunk_size=1024)处理原始二进制流（如分块 JSON、CSV 流），需自行拼接/解析 - 务必加超时：
timeout=(3, 10)（连接 3s，读取单块最多 10s），否则测试可能永久挂起 - 别在测试里用
time.sleep()等流，不可靠；用for chunk in response.iter_lines(): ... break控制读取数量
如何在pytest中模拟流式响应做单元测试
真实调用后端成本高、不稳定，单元测试应隔离依赖。不能 mock 整个 requests.get 返回普通 Response 对象——它不支持流式迭代。
必须 mock 一个带流式行为的对象：给 response.iter_lines() 返回生成器，或 patch response.raw 并设 response._content_consumed = False。
调用 Cutout.Pro 视觉处理 API 进行背景移除、人像抠图和照片增强，支持文件上传与图片 URL 输入。
- 推荐方式：用
unittest.mock.Mock构造response，手动设置response.iter_lines = Mock(return_value=["data: hello", "data: world"]) - 更贴近真实：mock
response.raw为io.BytesIO，再让iter_content按需 yield 字节块 - 注意 pytest 参数化时，每个 case 要新建 mock 对象，否则生成器耗尽后后续测试失败
- 避免 mock
requests.Session.send——太底层，容易漏掉 headers / encoding 等细节
pytest fixture 怎么复用流式测试逻辑
多个测试都要读流、校验前 N 条、检查 event type、提取 data 字段……这些逻辑不该重复写。用 fixture 封装流读取和断言骨架最稳妥。
定义一个 streaming_response fixture，接收 URL 和可选参数（如 max_events=5, event_type="message"），内部完成请求、流读取、超时控制，并返回结构化结果（如 {"events": [...], "raw_chunks": [...]}）。
- fixture 中统一加
try/finally确保response.close()，防止连接泄漏拖慢整个测试套件 - 把解析逻辑（如正则提取
data:后内容）放进 fixture，测试函数只写业务断言：assert first_event["id"] == "1" - 不要在 fixture 里做 heavy 断言（如全文本 diff），留到 test 函数里，便于定位失败点
- 如果接口需认证，把 token 注入 fixture 的
headers参数，而非硬编码在 fixture 内部
为什么 pytest-xdist 并行跑流式测试会出错
常见现象：ConnectionResetError、IncompleteRead、甚至 mock 响应被多个 worker 覆盖。根本原因是流式响应依赖长连接和顺序读取，而 xdist 默认共享 session 或复用 socket。
- 禁用连接复用：在测试请求中显式加
headers={"Connection": "close"} - 避免全局
requests.Session()实例；每个测试用独立requests.get，或用 fixture 创建新 session - mock 流式响应时，确保每个测试拿到独立生成器实例（例如用
lambda: iter([...])而非iter([...])） - 如果必须用 session，加
@pytest.mark.xfail(reason="flaky under xdist")并单独跑这类测试
流式测试真正的难点不在语法，而在对「连接生命周期」和「读取边界」的精确控制——少一个 response.close()，多一次 iter_lines() 调用，都可能让后续测试静默失败。
Python免费学习笔记（深入）：立即使用
在学习笔记中，你将探索 Python 的核心概念和高级技巧！
