Reproduce time boundaries without sleeping

Goal

You verify the limit window, the retry time, and per-user isolation with a fake clock.

Why it matters

When sleep was put into a rate limit test, it passed on a developer PC and failed on CI. A slow runner shifted the time boundary, and the test itself took a long time. Time is an input to the program, so you have to make it controllable by the caller and step exactly on the boundary.

Steps

1. In /root/work/test-clock-window-lab/test_service.py, test the following public contract of the provided service.py: validate_limit(limit, window) allows only a positive int limit (excluding bool) and a positive finite int/float window, and returns (limit, float(window)). Everything else is ValueError. It must pass against the correct implementation and be caught, through a failure in the body of an actual test, in an implementation that breaks this contract. Keep the tests from the earlier steps and add a test_ function.

Prepare this once at the start. Existing files are not overwritten.

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

1. In /root/work/test-clock-window-lab/test_service.py, test the following public contract of the provided service.py: active(history, now, window) returns, as a new list in the original order, only the timestamps greater than now-window. history is a sorted, non-decreasing list of timestamps. It must pass against the correct implementation and be caught, through a failure in the body of an actual test, in an implementation that breaks this contract. Keep the tests from the earlier steps and add a test_ function.

2. In /root/work/test-clock-window-lab/test_service.py, test the following public contract of the provided service.py: retry_after(history, now, window) is the larger integer of 0 and the ceil of (the first timestamp + window - now) of a non-empty history that has already been cleaned up. An empty list gives 0. It must pass against the correct implementation and be caught, through a failure in the body of an actual test, in an implementation that breaks this contract. Keep the tests from the earlier steps and add a test_ function.

3. In /root/work/test-clock-window-lab/test_service.py, test the following public contract of the provided service.py: history_for(state, key) returns an empty list for a key that does not exist, and a copy of that record if it does. A lookup alone does not modify state. It must pass against the correct implementation and be caught, through a failure in the body of an actual test, in an implementation that breaks this contract. Keep the tests from the earlier steps and add a test_ function.

4. In /root/work/test-clock-window-lab/test_service.py, test the following public contract of the provided service.py: admit(state, key, now, limit, window) validates the settings and then cleans up the expired records of that key. If there is room, it appends now and returns (True,0); if full, it does not append and returns (False,retry_after). It must pass against the correct implementation and be caught, through a failure in the body of an actual test, in an implementation that breaks this contract. Keep the tests from the earlier steps and add a test_ function.

5. In /root/work/test-clock-window-lab/test_service.py, test the following public contract of the provided service.py: client_key(value) returns a string of 1–40 ASCII letters, digits, and hyphens as it is, and everything else is ValueError. It must pass against the correct implementation and be caught, through a failure in the body of an actual test, in an implementation that breaks this contract. Keep the tests from the earlier steps and add a test_ function.

6. In /root/work/test-clock-window-lab/test_service.py, test the following public contract of the provided service.py: limited_response(wait) is a JSONResponse with status 429, body {error:'rate_limited'}, and a Retry-After header set to wait as a string. It must pass against the correct implementation and be caught, through a failure in the body of an actual test, in an implementation that breaks this contract. Keep the tests from the earlier steps and add a test_ function.

7. In /root/work/test-clock-window-lab/test_service.py, test the following public contract of the provided service.py: create_app(clock, limit=2, window=10) checks X-Client-ID at GET /work: an invalid key gives 400 {error:'invalid_client'}, an allowed request gives 200 {ok:True}, and an excess request gives limited_response. The state is kept separate for each app. It must pass against the correct implementation and be caught, through a failure in the body of an actual test, in an implementation that breaks this contract. Keep the tests from the earlier steps and add a test_ function.

Notes

• You work in the existing lab-dev environment with no internet and no package installation.
• Each step runs within a 45-second grading budget. Do not add real sleeps or network calls.
• The submitted tests are run in a separate temporary folder against the correct and defective implementations. Against the correct one, every test that actually runs must pass, and against a defective one, the body of a test must fail. A collection error, zero tests run, everything skipped, and forced termination are not a pass. Use only the basic pytest features and the provided libraries.
• FastAPI official documentation · pytest official documentation · Python sqlite3
• Limitation: This is an example for a single worker held in process memory. It does not guarantee a global limit shared by several Pods or the identity of a malicious client. X-Client-ID is a key for testing, so in production the key should come from an authenticated principal. A persistent clock going backward should be avoided by using a monotonic clock, and the clock in this lab is non-decreasing. You may read the provided implementation, but grading uses a separate copy. Do not work around a defect by checking the wording of the source or by modifying files; check the execution results of the public interface.

Validate the settings — test

In /root/work/test-clock-window-lab/test_service.py, test the following public contract of the provided service.py: validate_limit(limit, window) allows only a positive int limit (excluding bool) and a positive finite int/float window, and returns (limit, float(window)). Everything else is ValueError. It must pass against the correct implementation and be caught, through a failure in the body of an actual test, in an implementation that breaks this contract. Keep the tests from the earlier steps and add a test_ function.

Prepare this once at the start. Existing files are not overwritten.

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

bool is a subtype of int. NaN and infinity also have to be rejected separately. Do not modify the implementation file. Use pytest.raises to check the expected exception, and assert a concrete expected value for the normal result.

After saving, check with bash /opt/lab/checks/test-clock-window-lab/01-contract.sh.

Exclude the left edge of the window — test

In /root/work/test-clock-window-lab/test_service.py, test the following public contract of the provided service.py: active(history, now, window) returns, as a new list in the original order, only the timestamps greater than now-window. history is a sorted, non-decreasing list of timestamps. It must pass against the correct implementation and be caught, through a failure in the body of an actual test, in an implementation that breaks this contract. Keep the tests from the earlier steps and add a test_ function.

Check the difference between >=, which keeps a timestamp that has exactly expired, and >. Do not modify the implementation file. Use pytest.raises to check the expected exception, and assert a concrete expected value for the normal result.

After saving, check with bash /opt/lab/checks/test-clock-window-lab/02-contract.sh.

Round the waiting time up — test

In /root/work/test-clock-window-lab/test_service.py, test the following public contract of the provided service.py: retry_after(history, now, window) is the larger integer of 0 and the ceil of (the first timestamp + window - now) of a non-empty history that has already been cleaned up. An empty list gives 0. It must pass against the correct implementation and be caught, through a failure in the body of an actual test, in an implementation that breaks this contract. Keep the tests from the earlier steps and add a test_ function.

If you send Retry-After as 0 because 0.2 seconds remain, the client requests again immediately. Do not modify the implementation file. Use pytest.raises to check the expected exception, and assert a concrete expected value for the normal result.

After saving, check with bash /opt/lab/checks/test-clock-window-lab/03-contract.sh.

Split the records per key — test

In /root/work/test-clock-window-lab/test_service.py, test the following public contract of the provided service.py: history_for(state, key) returns an empty list for a key that does not exist, and a copy of that record if it does. A lookup alone does not modify state. It must pass against the correct implementation and be caught, through a failure in the body of an actual test, in an implementation that breaks this contract. Keep the tests from the earlier steps and add a test_ function.

If you return a shared list, the cleanup in one request can change the record of another request. Do not modify the implementation file. Use pytest.raises to check the expected exception, and assert a concrete expected value for the normal result.

After saving, check with bash /opt/lab/checks/test-clock-window-lab/04-contract.sh.

Record only allowed requests — test

In /root/work/test-clock-window-lab/test_service.py, test the following public contract of the provided service.py: admit(state, key, now, limit, window) validates the settings and then cleans up the expired records of that key. If there is room, it appends now and returns (True,0); if full, it does not append and returns (False,retry_after). It must pass against the correct implementation and be caught, through a failure in the body of an actual test, in an implementation that breaks this contract. Keep the tests from the earlier steps and add a test_ function.

If you append a rejected request, the expiry time is pushed back on every retry. Do not modify the implementation file. Use pytest.raises to check the expected exception, and assert a concrete expected value for the normal result.

After saving, check with bash /opt/lab/checks/test-clock-window-lab/05-contract.sh.

Validate the client key — test

In /root/work/test-clock-window-lab/test_service.py, test the following public contract of the provided service.py: client_key(value) returns a string of 1–40 ASCII letters, digits, and hyphens as it is, and everything else is ValueError. It must pass against the correct implementation and be caught, through a failure in the body of an actual test, in an implementation that breaks this contract. Keep the tests from the earlier steps and add a test_ function.

Limit the range of the input so that an unbounded key size cannot put pressure on the memory of the state. Do not modify the implementation file. Use pytest.raises to check the expected exception, and assert a concrete expected value for the normal result.

After saving, check with bash /opt/lab/checks/test-clock-window-lab/06-contract.sh.

Build the rejection response — test

In /root/work/test-clock-window-lab/test_service.py, test the following public contract of the provided service.py: limited_response(wait) is a JSONResponse with status 429, body {error:'rate_limited'}, and a Retry-After header set to wait as a string. It must pass against the correct implementation and be caught, through a failure in the body of an actual test, in an implementation that breaks this contract. Keep the tests from the earlier steps and add a test_ function.

Send the status and the header together so that the client knows when to retry. Do not modify the implementation file. Use pytest.raises to check the expected exception, and assert a concrete expected value for the normal result.

After saving, check with bash /opt/lab/checks/test-clock-window-lab/07-contract.sh.

Finish the request flow with virtual time — test

In /root/work/test-clock-window-lab/test_service.py, test the following public contract of the provided service.py: create_app(clock, limit=2, window=10) checks X-Client-ID at GET /work: an invalid key gives 400 {error:'invalid_client'}, an allowed request gives 200 {ok:True}, and an excess request gives limited_response. The state is kept separate for each app. It must pass against the correct implementation and be caught, through a failure in the body of an actual test, in an implementation that breaks this contract. Keep the tests from the earlier steps and add a test_ function.

Do not actually sleep; pass the current time, held in a list, through the clock function. Do not modify the implementation file. Use pytest.raises to check the expected exception, and assert a concrete expected value for the normal result.

After saving, check with bash /opt/lab/checks/test-clock-window-lab/08-contract.sh.