sleepを使わず時刻境界を再現する：設計原理
一言でいうと
フェイククロックで、制限ウィンドウ・リトライ時刻・ユーザーごとの分離を検証します。
なぜ必要なのか
リクエスト制限のテストにsleepを入れたところ、開発PCでは成功し、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で期待する例外を確認し、正常な結果には具体的な期待値をassertしてください。
レビューする誤った変更の断片:
not isinstance(limit, int)
この断片が入った関数の公開契約と比較してください。成功するケースが1つだけでは区別できない場合は、拒否されるべき入力や、失敗した後の状態を観測の対象に選びます。
2. ウィンドウの左端の境界を除外する(テスト)
提供されたservice.pyの次の公開契約をテストしてください: active(history, now, window)は、now-windowより大きい時刻だけを、元の順序の新しいリストとして返します。historyはソート済みの非減少の時刻です。正常な実装では成功し、この契約に違反する実装では、実際のテスト本文の失敗として検出する必要があります。前のステップのテストを維持したまま、test_関数を追加してください。
判断の根拠: ちょうど期限切れになった時刻を残す>=と、>の違いを確認します。実装ファイルは修正しません。pytest.raisesで期待する例外を確認し、正常な結果には具体的な期待値をassertしてください。
レビューする誤った変更の断片:
stamp >= now - window
この断片が入った関数の公開契約と比較してください。成功するケースが1つだけでは区別できない場合は、拒否されるべき入力や、失敗した後の状態を観測の対象に選びます。
3. 待機時間を切り上げる(テスト)
提供されたservice.pyの次の公開契約をテストしてください: retry_after(history, now, window)は、すでに整理された空でないhistoryの最初の時刻+window-nowをceilした値と、0のうち、大きいほうの整数です。空のリストは0です。正常な実装では成功し、この契約に違反する実装では、実際のテスト本文の失敗として検出する必要があります。前のステップのテストを維持したまま、test_関数を追加してください。
判断の根拠: 0.2秒残っているからといってRetry-Afterを0にすると、クライアントがすぐに再リクエストします。実装ファイルは修正しません。pytest.raisesで期待する例外を確認し、正常な結果には具体的な期待値をassertしてください。
レビューする誤った変更の断片:
int(history[0] + window - now)
この断片が入った関数の公開契約と比較してください。成功するケースが1つだけでは区別できない場合は、拒否されるべき入力や、失敗した後の状態を観測の対象に選びます。
4. キーごとに記録を分ける(テスト)
提供されたservice.pyの次の公開契約をテストしてください: history_for(state, key)は、存在しないキーなら空のリスト、存在すればその記録のコピーを返します。参照しただけではstateを変更しません。正常な実装では成功し、この契約に違反する実装では、実際のテスト本文の失敗として検出する必要があります。前のステップのテストを維持したまま、test_関数を追加してください。
判断の根拠: 共有されたリストを返すと、あるリクエストの整理が、別のリクエストの記録を変えてしまうことがあります。実装ファイルは修正しません。pytest.raisesで期待する例外を確認し、正常な結果には具体的な期待値をassertしてください。
レビューする誤った変更の断片:
state.get(key, [])
この断片が入った関数の公開契約と比較してください。成功するケースが1つだけでは区別できない場合は、拒否されるべき入力や、失敗した後の状態を観測の対象に選びます。
5. 許可したリクエストだけを記録する(テスト)
提供されたservice.pyの次の公開契約をテストしてください: admit(state, key, now, limit, window)は、設定を検証した後、該当するキーの期限切れの記録を整理します。余裕があればnowを追加して(True,0)を返し、満杯なら追加せずに(False,retry_after)を返します。正常な実装では成功し、この契約に違反する実装では、実際のテスト本文の失敗として検出する必要があります。前のステップのテストを維持したまま、test_関数を追加してください。
判断の根拠: 拒否したリクエストを追加すると、リトライするたびに有効期限が後ろへずれていきます。実装ファイルは修正しません。pytest.raisesで期待する例外を確認し、正常な結果には具体的な期待値をassertしてください。
レビューする誤った変更の断片:
len(history) > limit
この断片が入った関数の公開契約と比較してください。成功するケースが1つだけでは区別できない場合は、拒否されるべき入力や、失敗した後の状態を観測の対象に選びます。
6. クライアントキーを検証する(テスト)
提供されたservice.pyの次の公開契約をテストしてください: client_key(value)は、1–40文字のASCIIの英数字・ハイフンの文字列をそのまま返し、それ以外はValueErrorです。正常な実装では成功し、この契約に違反する実装では、実際のテスト本文の失敗として検出する必要があります。前のステップのテストを維持したまま、test_関数を追加してください。
判断の根拠: 無制限のキーサイズで状態のメモリを圧迫しないように、入力の範囲を制限します。実装ファイルは修正しません。pytest.raisesで期待する例外を確認し、正常な結果には具体的な期待値をassertしてください。
レビューする誤った変更の断片:
<= 80
この断片が入った関数の公開契約と比較してください。成功するケースが1つだけでは区別できない場合は、拒否されるべき入力や、失敗した後の状態を観測の対象に選びます。
7. 拒否レスポンスを作る(テスト)
提供されたservice.pyの次の公開契約をテストしてください: limited_response(wait)は、ステータス429、本文{error:'rate_limited'}、Retry-Afterヘッダーはwaitを文字列にした値のJSONResponseです。正常な実装では成功し、この契約に違反する実装では、実際のテスト本文の失敗として検出する必要があります。前のステップのテストを維持したまま、test_関数を追加してください。
判断の根拠: クライアントがリトライする時間を知れるように、ステータスとヘッダーを一緒に送ります。実装ファイルは修正しません。pytest.raisesで期待する例外を確認し、正常な結果には具体的な期待値をassertしてください。
レビューする誤った変更の断片:
status_code=503
この断片が入った関数の公開契約と比較してください。成功するケースが1つだけでは区別できない場合は、拒否されるべき入力や、失敗した後の状態を観測の対象に選びます。
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で期待する例外を確認し、正常な結果には具体的な期待値をassertしてください。
レビューする誤った変更の断片:
admit(state, "shared", clock(), limit, window)
この断片が入った関数の公開契約と比較してください。成功するケースが1つだけでは区別できない場合は、拒否されるべき入力や、失敗した後の状態を観測の対象に選びます。
現場での姿
プロセスのメモリ上にある、シングルワーカー向けの例です。複数のPodが共有するグローバルな上限や、悪意のあるクライアントの身元は保証しません。X-Client-IDはテスト用のキーなので、本番では認証されたプリンシパルからキーを得る必要があります。継続的な時計の逆行は、単調クロックを使って避ける必要があり、このラボのclockは非減少です。提供された実装は読んでもかまいませんが、採点は別のコピーを使用します。ソースの文言の検査やファイルの修正で欠陥を回避せず、公開インターフェースの実行結果を検査してください。
次のラボですること
8つのステップが、1つの実行可能な成果物につながります。設定を検証する(テスト) → ウィンドウの左端の境界を除外する(テスト) → 待機時間を切り上げる(テスト) → キーごとに記録を分ける(テスト) → 許可したリクエストだけを記録する(テスト) → クライアントキーを検証する(テスト) → 拒否レスポンスを作る(テスト) → 仮想時間でリクエストの流れを完結させる(テスト)。
各ステップでは、関数やファイルが存在するという事実ではなく、実際の戻り値・例外・状態の変化を検査します。正解を見たあとは、わざと境界の比較や後始末のコードを変えて、どのテストが失敗するかを確認してください。前のテストが次のステップでも維持される理由を説明し、このラボが保証しない本番の条件を1つ書いてみてください。