| 1 | """Tests for the shared keyless-Reddit throttle (U4).""" |
| 2 | |
| 3 | from unittest import mock |
| 4 | |
| 5 | from lib import http, reddit_rss |
| 6 | |
| 7 | |
| 8 | class TestRateLimiter: |
| 9 | def test_burst_does_not_sleep(self): |
| 10 | # A full bucket lets `burst` calls through immediately. |
| 11 | limiter = http.RateLimiter(rate_per_sec=5.0, burst=3) |
| 12 | with mock.patch.object(http.time, "monotonic", return_value=100.0), \ |
| 13 | mock.patch.object(http.time, "sleep") as slept: |
| 14 | limiter.acquire() |
| 15 | limiter.acquire() |
| 16 | limiter.acquire() |
| 17 | slept.assert_not_called() |
| 18 | |
| 19 | def test_sleeps_when_bucket_empty(self): |
| 20 | # burst=1: first call passes, second (same instant) must wait ~1/rate. |
| 21 | limiter = http.RateLimiter(rate_per_sec=2.0, burst=1) |
| 22 | times = iter([100.0, 100.0, 100.0, 100.5]) |
| 23 | with mock.patch.object(http.time, "monotonic", side_effect=lambda: next(times)), \ |
| 24 | mock.patch.object(http.time, "sleep") as slept: |
| 25 | limiter.acquire() # consumes the one token |
| 26 | limiter.acquire() # bucket empty -> sleep, then refilled token consumed |
| 27 | slept.assert_called() |
| 28 | waited = slept.call_args.args[0] |
| 29 | assert abs(waited - 0.5) < 1e-6 # (1 token deficit) / 2 per sec |
| 30 | |
| 31 | def test_refill_over_time_avoids_sleep(self): |
| 32 | limiter = http.RateLimiter(rate_per_sec=2.0, burst=1) |
| 33 | # Second call 1s later: bucket refilled (2/s * 1s capped at burst=1) -> no sleep. |
| 34 | times = iter([100.0, 101.0]) |
| 35 | with mock.patch.object(http.time, "monotonic", side_effect=lambda: next(times)), \ |
| 36 | mock.patch.object(http.time, "sleep") as slept: |
| 37 | limiter.acquire() |
| 38 | limiter.acquire() |
| 39 | slept.assert_not_called() |
| 40 | |
| 41 | |
| 42 | class TestRedditKeylessGetText: |
| 43 | def test_acquires_limiter_then_delegates(self): |
| 44 | with mock.patch.object(http.REDDIT_KEYLESS_LIMITER, "acquire") as acq, \ |
| 45 | mock.patch.object(http, "get_text", return_value="body") as gt: |
| 46 | out = http.reddit_keyless_get_text("https://www.reddit.com/x.rss", accept="application/atom+xml") |
| 47 | assert out == "body" |
| 48 | acq.assert_called_once() |
| 49 | gt.assert_called_once() |
| 50 | |
| 51 | def test_reddit_rss_routes_through_throttle(self): |
| 52 | # The RSS tier must use the throttled helper, not raw get_text. |
| 53 | with mock.patch.object(reddit_rss.http, "reddit_keyless_get_text", return_value=None) as throttled: |
| 54 | reddit_rss.search_rss("test query") |
| 55 | assert throttled.called |
| 56 |