让惰性导出行为可被观察：设计原理

一句话总结

验证行的消费次数、换行转义、公开字段和响应格式。

为什么需要它

用小列表测试过的导出代码看上去一切正常，却在把大输入全部转成 list 时耗尽了内存。结果相同，并不意味着运行方式的契约也相同。必须把惰性消费变成肉眼可见的状态。

工作原理

每当输入生成器被消费，就在列表中留下痕迹。确认刚创建时为 0 次，第一次 next 之后为 1 次，达到上限之后不再继续消费。把含有韩文字符和换行的 JSON Lines 重新解析，并检查 HTTP 响应的 Content-Type 和内部字段的移除。

학생 테스트 → 정상 구현: 실제 시험 모두 통과
           └→ 계약 위반 구현: 해당 동작에서 실패
수집 실패·0개 실행·강제 종료 ≠ 결함 검출

阅读契约并预测失败的工作表

下面不是让你背下整个实现的答案，而是逐步骤的代码评审。每个改动片段都会故意破坏契约。请注意，改动之后正常用例仍然可能通过。运行之前先预测：观察哪些输入、异常和状态，差异才会暴露出来；实现之后，再把预测与结果进行比较。

1. 验证行契约——测试

测试所提供 service.py 的下列公开契约：validate_row(row) 在 row 是 dict、id 是除 bool 之外的正 int、name 是非空 str 时，返回 row。其余抛出 ValueError。允许存在额外的内部字段。在正确实现上应当通过，而在违反该契约的实现上，必须由测试本体的实际失败将其检出。保留前面步骤的测试，并添加 test_ 函数。

判断依据：区分 bool 与数字，并把空名称当作错误处理。不要修改实现文件。用 pytest.raises 确认预期的异常，对正常结果则断言具体的预期值。

待评审的错误改动片段：

not isinstance(row.get("id"), int)

把它与包含该片段的函数的公开契约对照。如果仅凭一个成功用例无法区分，就把应被拒绝的输入或失败之后的状态选作观察对象。

2. 只生成公开行——测试

测试所提供 service.py 的下列公开契约：project(row) 在 validate_row 之后，返回只含 id 和 name 的新 dict。原始的内部字段保持不变。在正确实现上应当通过，而在违反该契约的实现上，必须由测试本体的实际失败将其检出。保留前面步骤的测试，并添加 test_ 函数。

判断依据：导出路径也必须与普通 API 采用相同的公开字段策略。不要修改实现文件。用 pytest.raises 确认预期的异常，对正常结果则断言具体的预期值。

待评审的错误改动片段：

"name":row["name"], "internal_cost":row.get("internal_cost")}

把它与包含该片段的函数的公开契约对照。如果仅凭一个成功用例无法区分，就把应被拒绝的输入或失败之后的状态选作观察对象。

3. 保留行边界地编码——测试

测试所提供 service.py 的下列公开契约：encode_line(row) 把 project 的结果用 ensure_ascii=False、separators=(',',':')、sort_keys=True 编码为 JSON，并在末尾追加一个 '\n'，返回 str。name 中的换行必须是 JSON 转义。在正确实现上应当通过，而在违反该契约的实现上，必须由测试本体的实际失败将其检出。保留前面步骤的测试，并添加 test_ 函数。

判断依据：如果用字符串拼接来构造 JSON，遇到引号和换行时格式就会被破坏。不要修改实现文件。用 pytest.raises 确认预期的异常，对正常结果则断言具体的预期值。

待评审的错误改动片段：

ensure_ascii=True

把它与包含该片段的函数的公开契约对照。如果仅凭一个成功用例无法区分，就把应被拒绝的输入或失败之后的状态选作观察对象。

4. 验证输出数量上限——测试

测试所提供 service.py 的下列公开契约：validate_max(value) 只在 bool 之外的 int 处于 1–1000 范围内时原样返回，其余抛出 ValueError。在正确实现上应当通过，而在违反该契约的实现上，必须由测试本体的实际失败将其检出。保留前面步骤的测试，并添加 test_ 函数。

判断依据：要求调用方给出上限，避免不小心把无限输入一直读到底。不要修改实现文件。用 pytest.raises 确认预期的异常，对正常结果则断言具体的预期值。

待评审的错误改动片段：

<= 1001

把它与包含该片段的函数的公开契约对照。如果仅凭一个成功用例无法区分，就把应被拒绝的输入或失败之后的状态选作观察对象。

5. 只消费需要的行——测试

测试所提供 service.py 的下列公开契约：take_rows(rows, maximum) 是一个 iterator，用 islice 等方式最多惰性地返回 maximum 个。调用时会验证 maximum，每调用一次 next，只消费一次输入。在正确实现上应当通过，而在违反该契约的实现上，必须由测试本体的实际失败将其检出。保留前面步骤的测试，并添加 test_ 函数。

判断依据：一旦转换成 list(rows)，就无法处理无限输入和大容量输入了。不要修改实现文件。用 pytest.raises 确认预期的异常，对正常结果则断言具体的预期值。

待评审的错误改动片段：

islice(list(rows), validate_max(maximum))

把它与包含该片段的函数的公开契约对照。如果仅凭一个成功用例无法区分，就把应被拒绝的输入或失败之后的状态选作观察对象。

6. 惰性序列化行——测试

测试所提供 service.py 的下列公开契约：json_lines(rows, maximum=100) 对从 take_rows 得到的每一行 yield 出 encode_line。不返回把全部拼起来的字符串或列表。在正确实现上应当通过，而在违反该契约的实现上，必须由测试本体的实际失败将其检出。保留前面步骤的测试，并添加 test_ 函数。

判断依据：让对象的选取与表示形式的转换，各自都保持为惰性阶段。不要修改实现文件。用 pytest.raises 确认预期的异常，对正常结果则断言具体的预期值。

待评审的错误改动片段：

for row in list(take_rows(rows, maximum)):

把它与包含该片段的函数的公开契约对照。如果仅凭一个成功用例无法区分，就把应被拒绝的输入或失败之后的状态选作观察对象。

7. 重新验证下载的行——测试

测试所提供 service.py 的下列公开契约：decode_lines(text) 对 splitlines 得到的每个非空行先 json.loads，再 validate_row，并以列表返回。空字符串返回 []，中间出现空行则抛出 ValueError。在正确实现上应当通过，而在违反该契约的实现上，必须由测试本体的实际失败将其检出。保留前面步骤的测试，并添加 test_ 函数。

判断依据：区分空文件与格式被破坏的空记录。不要修改实现文件。用 pytest.raises 确认预期的异常，对正常结果则断言具体的预期值。

待评审的错误改动片段：

continue

把它与包含该片段的函数的公开契约对照。如果仅凭一个成功用例无法区分，就把应被拒绝的输入或失败之后的状态选作观察对象。

8. 完成 HTTP 下载——测试

测试所提供 service.py 的下列公开契约：create_app(rows) 在 GET /export 上，把 json_lines(rows, 100) 作为 application/x-ndjson 的 StreamingResponse 返回。rows 是可再次遍历的列表。不能含有内部字段，并且必须保留每一行的内容和顺序。在正确实现上应当通过，而在违反该契约的实现上，必须由测试本体的实际失败将其检出。保留前面步骤的测试，并添加 test_ 函数。

判断依据：要同时配合生成器测试，确认没有只在 Content-Type 上写流式、内部却把全部内容收集起来。不要修改实现文件。用 pytest.raises 确认预期的异常，对正常结果则断言具体的预期值。

待评审的错误改动片段：

media_type="application/json"

把它与包含该片段的函数的公开契约对照。如果仅凭一个成功用例无法区分，就把应被拒绝的输入或失败之后的状态选作观察对象。

在现场相遇的样子

TestClient 会缓冲响应，所以它无法证明网络首字节的延迟，也无法证明内存上限的全部。是否做到惰性求值，要另用计数生成器来检查。流开始之后如果遇到错误的行，就很难再把状态改成正常的错误 JSON。真实服务必须决定在预先验证、逐行错误格式、中止策略之中采用哪一种。提供的实现可以阅读，但评分使用的是另外的副本。不要通过检查源码措辞或修改文件来绕过缺陷，而要检查公开接口的实际运行结果。

下一项实验要做什么

八个步骤会连成一个可运行的成果。验证行契约——测试 → 只生成公开行——测试 → 保留行边界地编码——测试 → 验证输出数量上限——测试 → 只消费需要的行——测试 → 惰性序列化行——测试 → 重新验证下载的行——测试 → 完成 HTTP 下载——测试。

每一步检查的都不是函数或文件是否存在，而是实际的返回值、异常和状态变化。看过正确答案之后，请故意改动边界比较或清理代码，确认哪些测试会失败。说明为什么前面的测试在后面的步骤中仍然保留，并写出一种本实验不能保证的生产条件。