无需 sleep 即可重现时间边界：设计原理

一句话总结

用假时钟验证限制窗口、重试时刻和按用户隔离。

为什么需要它

在请求限流的测试中加入 sleep 之后，开发机上能通过，CI 上却失败了。运行速度慢的执行器改变了时间边界，测试本身也要跑很久。时间是程序的输入，所以应让调用方能够控制它，并且精确踩在边界上。

工作原理

用列表来控制 clock 回调返回的值。把左边界和边界内侧的请求分开，并验证 Retry-After 的向上取整。通过直接读取状态，检查被拒绝之后记录是否没有增加、是否没有占用其他键的配额。

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

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

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

1. 验证配置——测试

测试所提供 service.py 的下列公开契约：validate_limit(limit, window) 只允许除 bool 之外的正 int 作为 limit，以及正的有限 int/float 作为 window，并返回 (limit, float(window))。其余情况抛出 ValueError。在正确实现上应当通过，而在违反该契约的实现上，必须由测试本体的实际失败将其检出。保留前面步骤的测试，并添加 test_ 函数。

判断依据：bool 是 int 的子类型。NaN 和无穷大也必须单独拒绝。不要修改实现文件。用 pytest.raises 确认预期的异常，对正常结果则断言具体的预期值。

待评审的错误改动片段：

not isinstance(limit, int)

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

2. 排除窗口的左边界——测试

测试所提供 service.py 的下列公开契约：active(history, now, window) 只返回大于 now-window 的时刻，并以原有顺序组成新列表。history 是已排序的非递减时刻。在正确实现上应当通过，而在违反该契约的实现上，必须由测试本体的实际失败将其检出。保留前面步骤的测试，并添加 test_ 函数。

判断依据：确认把刚好过期的时刻保留下来的 >= 与 > 之间的差别。不要修改实现文件。用 pytest.raises 确认预期的异常，对正常结果则断言具体的预期值。

待评审的错误改动片段：

stamp >= now - window

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

3. 对等待时间向上取整——测试

测试所提供 service.py 的下列公开契约：retry_after(history, now, window) 对已清理过的非空 history，返回“第一个时刻+window-now 的 ceil 值”与 0 中较大的整数。空列表返回 0。在正确实现上应当通过，而在违反该契约的实现上，必须由测试本体的实际失败将其检出。保留前面步骤的测试，并添加 test_ 函数。

判断依据：如果还剩 0.2 秒却把 Retry-After 设为 0，客户端会立即重新请求。不要修改实现文件。用 pytest.raises 确认预期的异常，对正常结果则断言具体的预期值。

待评审的错误改动片段：

int(history[0] + window - now)

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

4. 按键分开记录——测试

测试所提供 service.py 的下列公开契约：history_for(state, key) 在键不存在时返回空列表，存在时返回该记录的副本。仅仅查询不会修改 state。在正确实现上应当通过，而在违反该契约的实现上，必须由测试本体的实际失败将其检出。保留前面步骤的测试，并添加 test_ 函数。

判断依据：如果返回共享列表，一个请求的清理就可能改掉另一个请求的记录。不要修改实现文件。用 pytest.raises 确认预期的异常，对正常结果则断言具体的预期值。

待评审的错误改动片段：

state.get(key, [])

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

5. 只记录被允许的请求——测试

测试所提供 service.py 的下列公开契约：admit(state, key, now, limit, window) 在验证配置之后，清理该键已过期的记录。有余量就追加 now 并返回 (True,0)，已满则不追加并返回 (False,retry_after)。在正确实现上应当通过，而在违反该契约的实现上，必须由测试本体的实际失败将其检出。保留前面步骤的测试，并添加 test_ 函数。

判断依据：如果把被拒绝的请求也追加进去，每次重试都会把过期时刻往后推。不要修改实现文件。用 pytest.raises 确认预期的异常，对正常结果则断言具体的预期值。

待评审的错误改动片段：

len(history) > limit

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

6. 验证客户端键——测试

测试所提供 service.py 的下列公开契约：client_key(value) 对由 1–40 个 ASCII 字母、数字、连字符组成的字符串原样返回，其余抛出 ValueError。在正确实现上应当通过，而在违反该契约的实现上，必须由测试本体的实际失败将其检出。保留前面步骤的测试，并添加 test_ 函数。

判断依据：限制输入的范围，避免无限制的键大小挤占状态内存。不要修改实现文件。用 pytest.raises 确认预期的异常，对正常结果则断言具体的预期值。

待评审的错误改动片段：

<= 80

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

7. 构造拒绝响应——测试

测试所提供 service.py 的下列公开契约：limited_response(wait) 返回一个 JSONResponse：状态 429，正文 {error:'rate_limited'}，Retry-After 请求头为把 wait 转成字符串后的值。在正确实现上应当通过，而在违反该契约的实现上，必须由测试本体的实际失败将其检出。保留前面步骤的测试，并添加 test_ 函数。

判断依据：同时发送状态和请求头，让客户端知道重试的时间。不要修改实现文件。用 pytest.raises 确认预期的异常，对正常结果则断言具体的预期值。

待评审的错误改动片段：

status_code=503

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

8. 用虚拟时间走完请求流程——测试

测试所提供 service.py 的下列公开契约：create_app(clock, limit=2, window=10) 在 GET /work 上检查 X-Client-ID：无效的键返回 400 {error:'invalid_client'}，允许时返回 200 {ok:True}，超出时返回 limited_response。state 按应用分开。在正确实现上应当通过，而在违反该契约的实现上，必须由测试本体的实际失败将其检出。保留前面步骤的测试，并添加 test_ 函数。

判断依据：不要真的 sleep，而是把放进列表的当前时刻通过 clock 函数传入。不要修改实现文件。用 pytest.raises 确认预期的异常，对正常结果则断言具体的预期值。

待评审的错误改动片段：

admit(state, "shared", clock(), limit, window)

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

在现场相遇的样子

这是基于进程内存、面向单个 worker 的示例。它不保证由多个 Pod 共享的全局限额，也不保证恶意客户端的身份。X-Client-ID 是测试用的键，所以在生产环境中应从已认证的主体获取键。持续的时钟回拨应通过使用单调时钟来避免，本实验的 clock 是非递减的。提供的实现可以阅读，但评分使用的是另外的副本。不要通过检查源码措辞或修改文件来绕过缺陷，而要检查公开接口的实际运行结果。

下一项实验要做什么

八个步骤会连成一个可运行的成果。验证配置——测试 → 排除窗口的左边界——测试 → 对等待时间向上取整——测试 → 按键分开记录——测试 → 只记录被允许的请求——测试 → 验证客户端键——测试 → 构造拒绝响应——测试 → 用虚拟时间走完请求流程——测试。

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