注入故障以发现资源泄漏：设计原理

一句话总结

观察正常关闭、重复关闭、失败路径，以及应用 lifespan 的启动与清理。

为什么需要它

请求成功时有关闭连接的代码，但一旦发生异常，已打开的连接就留了下来。只确认成功响应的测试，发现不了这种泄漏。生命周期测试应当关注的是资源何时打开、何时关闭，而不是返回值。

工作原理

为每个测试创建独立的资源对象。确认 start 和 stop 的次数以及 open 状态，并在 context 内故意抛出异常。用 with 使用 TestClient 来真正运行 lifespan，并断言 with 结束之后的状态。

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

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

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

1. 让资源状态相互独立——测试

测试所提供 service.py 的下列公开契约：new_resource() 返回 {open:False, events:[]} 形式的新字典，多次调用之间不共享 events。在正确实现上应当通过，而在违反该契约的实现上，必须由测试本体的实际失败将其检出。保留前面步骤的测试，并添加 test_ 函数。

判断依据：不要把可变列表作为全局变量或默认参数来共享。不要修改实现文件。用 pytest.raises 确认预期的异常，对正常结果则断言具体的预期值。

待评审的错误改动片段：

"open":True

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

2. 拒绝重复启动——测试

测试所提供 service.py 的下列公开契约：start(resource) 在资源已经打开时抛出 ValueError，否则把 open 改为 True，并在 events 中追加 'open'。在正确实现上应当通过，而在违反该契约的实现上，必须由测试本体的实际失败将其检出。保留前面步骤的测试，并添加 test_ 函数。

判断依据：拒绝“启动两次而丢掉其中一个资源”的行为。不要修改实现文件。用 pytest.raises 确认预期的异常，对正常结果则断言具体的预期值。

待评审的错误改动片段：

if False:
        raise

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

3. 让关闭具备幂等性——测试

测试所提供 service.py 的下列公开契约：stop(resource) 仅在资源处于打开状态时，才把 open 改为 False，并在 events 中追加 'close'。如果已经关闭，则保持原样。在正确实现上应当通过，而在违反该契约的实现上，必须由测试本体的实际失败将其检出。保留前面步骤的测试，并添加 test_ 函数。

判断依据：即使多条清理路径重叠，也不应出现重复的 close 事件。不要修改实现文件。用 pytest.raises 确认预期的异常，对正常结果则断言具体的预期值。

待评审的错误改动片段：

resource["events"].append("closed")

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

4. 禁止使用已关闭的资源——测试

测试所提供 service.py 的下列公开契约：read(resource) 在资源已关闭时抛出 RuntimeError，处于打开状态时返回 {ready:True}。在正确实现上应当通过，而在违反该契约的实现上，必须由测试本体的实际失败将其检出。保留前面步骤的测试，并添加 test_ 函数。

判断依据：就绪状态与对象是否存在是两回事。对象存在，也可能是关闭的。不要修改实现文件。用 pytest.raises 确认预期的异常，对正常结果则断言具体的预期值。

待评审的错误改动片段：

if False:

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

5. 在异常路径中使用 finally——测试

测试所提供 service.py 的下列公开契约：scope(resource) 是一个 contextmanager。进入时执行 start，在块内 yield 出 resource，无论块成功还是失败，都用 stop 关闭。块内的异常要继续向上传播。在正确实现上应当通过，而在违反该契约的实现上，必须由测试本体的实际失败将其检出。保留前面步骤的测试，并添加 test_ 函数。

判断依据：如果只在 yield 之后写 close，一旦出现异常，就到不了那一行。不要修改实现文件。用 pytest.raises 确认预期的异常，对正常结果则断言具体的预期值。

待评审的错误改动片段：

pass

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

6. 把应用生命周期与资源连接起来——测试

测试所提供 service.py 的下列公开契约：lifespan_for(resource) 返回 asynccontextmanager 函数 lifespan(app)。它在 scope(resource) 内设置 app.state.resource 并 yield。在正确实现上应当通过，而在违反该契约的实现上，必须由测试本体的实际失败将其检出。保留前面步骤的测试，并添加 test_ 函数。

判断依据：不是调用 lifespan 函数本身，而是把它传给 FastAPI 构造函数。不要修改实现文件。用 pytest.raises 确认预期的异常，对正常结果则断言具体的预期值。

待评审的错误改动片段：

app.state.resource = dict(resource)

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

7. 通过真实请求读取就绪状态——测试

测试所提供 service.py 的下列公开契约：create_app(resource) 使用 lifespan_for。GET /ready 返回对 app.state.resource 执行 read 的结果。context 结束时必须关闭资源。在正确实现上应当通过，而在违反该契约的实现上，必须由测试本体的实际失败将其检出。保留前面步骤的测试，并添加 test_ 函数。

判断依据：必须使用 with TestClient，才会同时执行 lifespan 的启动和关闭。不要修改实现文件。用 pytest.raises 确认预期的异常，对正常结果则断言具体的预期值。

待评审的错误改动片段：

FastAPI()

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

8. 请求之后的失败也要清理——测试

测试所提供 service.py 的下列公开契约：exercise(resource, fail=False) 在 with TestClient(create_app(resource)) 内调用 GET /ready。如果 fail=True，就在其中抛出 RuntimeError，否则返回响应 JSON。两种情况下资源都必须被关闭。在正确实现上应当通过，而在违反该契约的实现上，必须由测试本体的实际失败将其检出。保留前面步骤的测试，并添加 test_ 函数。

判断依据：把正常路径和异常路径纳入同一种清理结构，就能减少遗漏的关闭路径。不要修改实现文件。用 pytest.raises 确认预期的异常，对正常结果则断言具体的预期值。

待评审的错误改动片段：

if False:

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

在现场相遇的样子

教学用的资源字典只是代替真实数据库连接池的观测装置。在生产环境中，还必须设计部分初始化失败、连接池的并发、取消处理以及关闭的超时时间。检查的不是写进报告里的事件字符串，而是学习者的代码在运行时改变的对象状态。提供的实现可以阅读，但评分使用的是另外的副本。不要通过检查源码措辞或修改文件来绕过缺陷，而要检查公开接口的实际运行结果。

下一项实验要做什么

八个步骤会连成一个可运行的成果。让资源状态相互独立——测试 → 拒绝重复启动——测试 → 让关闭具备幂等性——测试 → 禁止使用已关闭的资源——测试 → 在异常路径中使用 finally——测试 → 把应用生命周期与资源连接起来——测试 → 通过真实请求读取就绪状态——测试 → 请求之后的失败也要清理——测试。

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