Skip to content

Use freezegun to speed up rate limit tests #14

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
tr4nt0r merged 1 commit into from
Oct 30, 2025
Merged

Use freezegun to speed up rate limit tests #14

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
8 changes: 7 additions & 1 deletion pyproject.toml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -95,14 +95,20 @@ dependencies = [
"httpx==0.28.1",
"ms_cv==0.1.1",
"pydantic==2.12.3",
"pytest==8.4.2",
"pytest-asyncio==1.2.0",
"pytest-cov==7.0.0",
"freezegun==1.5.5",
"respx~=0.22",
]

[tool.hatch.envs.hatch-test]
parallel = true
extra-dependencies = [
"pytest-cov~=6.2",
"pytest-asyncio~=1.1",
"respx~=0.22"
"respx~=0.22",
"freezegun==1.5.5"
]
extra-args = ["--cov=xbox/webapi/", "--cov-report=term-missing", "--cov-report=xml", "-vv"]

Expand Down
123 changes: 64 additions & 59 deletions tests/test_ratelimits.py
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,16 +1,16 @@
import asyncio
from datetime import datetime, timedelta

from freezegun import freeze_time
from freezegun.api import FrozenDateTimeFactory
from httpx import Response
import pytest

from tests.common import get_response_json
from xbox.webapi.api.provider.ratelimitedprovider import RateLimitedProvider
from xbox.webapi.common.exceptions import RateLimitExceededException, XboxException
from xbox.webapi.common.ratelimits import CombinedRateLimit
from xbox.webapi.common.ratelimits.models import TimePeriod

from tests.common import get_response_json


def helper_test_combinedratelimit(
crl: CombinedRateLimit, burstLimit: int, sustainLimit: int
Expand Down Expand Up @@ -143,7 +143,11 @@ async def make_request():


async def helper_reach_and_wait_for_burst(
make_request, start_time, burst_limit: int, expected_counter: int
make_request,
start_time,
burst_limit: int,
expected_counter: int,
frozen_datetime: FrozenDateTimeFactory
):
# Make as many requests as possible without exceeding the BURST limit.
for _ in range(burst_limit):
Expand All @@ -164,79 +168,80 @@ async def helper_reach_and_wait_for_burst(
burst_resets_after = ex.rate_limit.get_reset_after()

# Wait for the burst limit timeout to elapse.
await asyncio.sleep(TimePeriod.BURST.value) # 15 seconds
frozen_datetime.tick(timedelta(seconds=TimePeriod.BURST.value))

# Assert that the reset_after value has passed.
assert burst_resets_after < datetime.now()
assert burst_resets_after == datetime.now()
frozen_datetime.tick(timedelta(seconds=1))


@pytest.mark.asyncio
async def test_ratelimits_exceeded_sustain_only(respx_mock, xbl_client):
async def make_request():
route = respx_mock.get("https://social.xboxlive.com").mock(
return_value=Response(200, json=get_response_json("people_summary_own"))
)
await xbl_client.people.get_friends_summary_own()

assert route.called
with freeze_time("2025-10-30T00:00:00-00:00") as frozen_datetime:
async def make_request():
route = respx_mock.get("https://social.xboxlive.com").mock(
return_value=Response(200, json=get_response_json("people_summary_own"))
)
await xbl_client.people.get_friends_summary_own()

# Record the start time to ensure that the timeouts are the correct length
start_time = datetime.now()
assert route.called

# Get the max requests for this route.
max_request_num = xbl_client.people.RATE_LIMITS["sustain"] # 30
burst_max_request_num = xbl_client.people.RATE_LIMITS["burst"] # 10
# Record the start time to ensure that the timeouts are the correct length
start_time = datetime.now()

# In this case, the BURST limit is three times that of SUSTAIN, so we need to exceed the burst limit three times.
# Get the max requests for this route.
max_request_num = xbl_client.people.RATE_LIMITS["sustain"] # 30
burst_max_request_num = xbl_client.people.RATE_LIMITS["burst"] # 10

# Exceed the burst limit and wait for it to reset (10 requests)
await helper_reach_and_wait_for_burst(
make_request, start_time, burst_limit=burst_max_request_num, expected_counter=10
)
# In this case, the BURST limit is three times that of SUSTAIN, so we need to exceed the burst limit three times.

# Repeat: Exceed the burst limit and wait for it to reset (10 requests)
# Counter (the sustain one will be returned)
# For (CombinedRateLimit).get_counter(), the highest counter is returned. (sustain in this case)
await helper_reach_and_wait_for_burst(
make_request, start_time, burst_limit=burst_max_request_num, expected_counter=20
)
# Exceed the burst limit and wait for it to reset (10 requests)
await helper_reach_and_wait_for_burst(
make_request, start_time, burst_limit=burst_max_request_num, expected_counter=10, frozen_datetime=frozen_datetime
)

# Now, make the rest of the requests (10 left, 20/30 done!)
for _ in range(10):
await make_request()
# Repeat: Exceed the burst limit and wait for it to reset (10 requests)
# Counter (the sustain one will be returned)
# For (CombinedRateLimit).get_counter(), the highest counter is returned. (sustain in this case)
await helper_reach_and_wait_for_burst(
make_request, start_time, burst_limit=burst_max_request_num, expected_counter=20, frozen_datetime=frozen_datetime
)

# Wait for the burst limit to 'reset'.
await asyncio.sleep(TimePeriod.BURST.value) # 15 seconds
# Now, make the rest of the requests (10 left, 20/30 done!)
for _ in range(10):
await make_request()

# Now, we have made 30 requests.
# The counters should be as follows:
# - BURST: 0* (will reset on next check)
# - SUSTAIN: 30
# The next request we make should exceed the SUSTAIN rate limit.
# Wait for the burst limit to 'reset'.
frozen_datetime.tick(timedelta(seconds=TimePeriod.BURST.value+1))
# Now, we have made 30 requests.
# The counters should be as follows:
# - BURST: 0* (will reset on next check)
# - SUSTAIN: 30
# The next request we make should exceed the SUSTAIN rate limit.

# Make another request, ensure that it raises the exception.
with pytest.raises(RateLimitExceededException) as exception:
await make_request()
# Make another request, ensure that it raises the exception.
with pytest.raises(RateLimitExceededException) as exception:
await make_request()

# Get the error instance from pytest
ex: RateLimitExceededException = exception.value
# Get the error instance from pytest
ex: RateLimitExceededException = exception.value

# Get the SingleRateLimit objects from the exception
rl: CombinedRateLimit = ex.rate_limit
burst = rl.get_limits_by_period(TimePeriod.BURST)[0]
sustain = rl.get_limits_by_period(TimePeriod.SUSTAIN)[0]
# Get the SingleRateLimit objects from the exception
rl: CombinedRateLimit = ex.rate_limit
burst = rl.get_limits_by_period(TimePeriod.BURST)[0]
sustain = rl.get_limits_by_period(TimePeriod.SUSTAIN)[0]

# Assert that we have only exceeded the sustain limit.
assert not burst.is_exceeded()
assert sustain.is_exceeded()
# Assert that we have only exceeded the sustain limit.
assert not burst.is_exceeded()
assert sustain.is_exceeded()

# Assert that the counter matches the max request num (should not have incremented above max value)
assert ex.rate_limit.get_counter() == max_request_num
# Assert that the counter matches the max request num (should not have incremented above max value)
assert ex.rate_limit.get_counter() == max_request_num

# Get the timeout we were issued
try_again_in = ex.rate_limit.get_reset_after()
# Get the timeout we were issued
try_again_in = ex.rate_limit.get_reset_after()

# Assert that the timeout is the correct length
# The SUSTAIN counter has not been reset during this test, so the try again in should be 300 seconds since we started this test.
delta: timedelta = try_again_in - start_time
assert delta.seconds == TimePeriod.SUSTAIN.value # 300 seconds (5 minutes)
# Assert that the timeout is the correct length
# The SUSTAIN counter has not been reset during this test, so the try again in should be 300 seconds since we started this test.
delta: timedelta = try_again_in - start_time
assert delta.seconds == TimePeriod.SUSTAIN.value # 300 seconds (5 minutes)
Loading