注入故障以发现资源泄漏

目标

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

为什么重要

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

步骤

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

先做一次准备。已有的文件不会被覆盖。

mkdir -p /root/work/test-lifespan-cleanup-lab
test -e /root/work/test-lifespan-cleanup-lab/service.py || cp /opt/fixtures/ten_labs/test-lifespan-cleanup-lab/service.py /root/work/test-lifespan-cleanup-lab/service.py
test -e /root/work/test-lifespan-cleanup-lab/test_service.py || cp /opt/fixtures/ten_labs/test-lifespan-cleanup-lab/test_service.py /root/work/test-lifespan-cleanup-lab/test_service.py
cd /root/work/test-lifespan-cleanup-lab

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

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

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

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

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

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

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

参考

• 无需联网，也无需安装软件包，在现有的 lab-dev 环境中进行。
• 每一步都在 45 秒的评分预算内运行。不要添加真实的 sleep 或网络调用。
• 提交的测试会在另外的临时文件夹中，分别在正确实现和有缺陷的实现上运行。在正确实现上，实际运行的测试必须全部通过；在有缺陷的实现上，测试本体必须失败。收集错误、运行 0 个、全部跳过、被强制终止都不算通过。只使用 pytest 的基本功能和提供的库。
• FastAPI 官方文档 · pytest 官方文档 · Python sqlite3
• 局限：教学用的资源字典只是代替真实数据库连接池的观测装置。在生产环境中，还必须设计部分初始化失败、连接池的并发、取消处理以及关闭的超时时间。检查的不是写进报告里的事件字符串，而是学习者的代码在运行时改变的对象状态。提供的实现可以阅读，但评分使用的是另外的副本。不要通过检查源码措辞或修改文件来绕过缺陷，而要检查公开接口的实际运行结果。

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

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

先做一次准备。已有的文件不会被覆盖。

mkdir -p /root/work/test-lifespan-cleanup-lab
test -e /root/work/test-lifespan-cleanup-lab/service.py || cp /opt/fixtures/ten_labs/test-lifespan-cleanup-lab/service.py /root/work/test-lifespan-cleanup-lab/service.py
test -e /root/work/test-lifespan-cleanup-lab/test_service.py || cp /opt/fixtures/ten_labs/test-lifespan-cleanup-lab/test_service.py /root/work/test-lifespan-cleanup-lab/test_service.py
cd /root/work/test-lifespan-cleanup-lab

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

保存后用 bash /opt/lab/checks/test-lifespan-cleanup-lab/01-contract.sh 确认。

拒绝重复启动——测试

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

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

保存后用 bash /opt/lab/checks/test-lifespan-cleanup-lab/02-contract.sh 确认。

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

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

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

保存后用 bash /opt/lab/checks/test-lifespan-cleanup-lab/03-contract.sh 确认。

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

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

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

保存后用 bash /opt/lab/checks/test-lifespan-cleanup-lab/04-contract.sh 确认。

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

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

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

保存后用 bash /opt/lab/checks/test-lifespan-cleanup-lab/05-contract.sh 确认。

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

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

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

保存后用 bash /opt/lab/checks/test-lifespan-cleanup-lab/06-contract.sh 确认。

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

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

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

保存后用 bash /opt/lab/checks/test-lifespan-cleanup-lab/07-contract.sh 确认。

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

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

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

保存后用 bash /opt/lab/checks/test-lifespan-cleanup-lab/08-contract.sh 确认。