| 1 | """Keyless Reddit discovery via Reddit's site search fragment (reddit_search). |
| 2 | |
| 3 | Fixtures under fixtures/reddit_search_*.html are trimmed live captures of |
| 4 | /svc/shreddit/search/ and /svc/shreddit/r/{sub}/search/, except the challenge |
| 5 | page, which is synthetic (see its provenance line). |
| 6 | """ |
| 7 | |
| 8 | import itertools |
| 9 | import json |
| 10 | import re |
| 11 | import urllib.error |
| 12 | from datetime import datetime, timezone |
| 13 | from pathlib import Path |
| 14 | from unittest import mock |
| 15 | from urllib.parse import parse_qs, urlsplit |
| 16 | |
| 17 | import pytest |
| 18 | |
| 19 | from lib import dates, health, http, reddit, reddit_search, schema |
| 20 | |
| 21 | FIXTURES = Path(__file__).resolve().parent.parent / "fixtures" |
| 22 | |
| 23 | PAGE1_URL = ( |
| 24 | "https://www.reddit.com/svc/shreddit/search/" |
| 25 | "?q=ButcherBox&type=posts&t=month&sort=relevance" |
| 26 | ) |
| 27 | PAGE1_CURSOR = ( |
| 28 | "eyJjYW5kaWRhdGVzX3JldHVybmVkIjoie1wic2VjdGlvbl8xX3BpcGVsaW5lXzBfZ2xvYmFsX21vZGlmaWVyc1wiOlwiMVwiLFwic2VjdGlvbl8xX3BpcGVsaW5lXzFfbG9jYWxfbW9kaWZpZXJzXCI6XCIyXCIsXCJzZWN0aW9uXzJfcGlwZWxpbmVfMTBfdXh0c191bml0XCI6XCIwXCIsXCJzZWN0aW9uXzJfcGlwZWxpbmVfMTFfcXVlcnlfc3VnZ2VzdGlvbnNcIjpcIjZcIixcInNlY3Rpb25fMl9waXBlbGluZV84X3Bvc3Rfc2VhcmNoXCI6XCI3XCIsXCJzZWN0aW9uXzNfcGlwZWxpbmVfMF9zdWJyZWRkaXRfc2VhcmNoXCI6XCI1XCIsXCJzZWN0aW9uXzNfcGlwZWxpbmVfMV9hdXRob3Jfc2VhcmNoXCI6XCI1XCJ9IiwiZXhwZXJpZW5jZV9zZWxlY3Rpb24iOiJwb3N0X3dpdGhfc2lkZWJhciIsImV4cGVyaWVuY2VfdmVyc2lvbiI6ImRlZmF1bHQiLCJzZWN0aW9uXzJfcGlwZWxpbmVfOF9wb3N0X3NlYXJjaCI6IjcifQ==" |
| 29 | ) |
| 30 | PAGE2_URL = PAGE1_URL + "&cursor=" + PAGE1_CURSOR.replace("=", "%3D") |
| 31 | |
| 32 | PAGE1_POSTS = [ |
| 33 | # (post id, subreddit, title, date, score, comments) |
| 34 | ("1w6swt2", "referralcodes", "Looking for a ButcherBox referral code", "2026-09-04", 1, 27), |
| 35 | ("1wkybur", "TimDillon", "Today’s Ad Read for ButcherBox", "2026-09-19", 144, 26), |
| 36 | ("1w9s403", "Netherlands", "Best online butchers in the Netherlands?\U0001f969", "2026-09-07", 0, 62), |
| 37 | ("1wbvi5s", "findareferralcode", "Anyone has a butcher box referral code?", "2026-09-09", 1, 1), |
| 38 | ("1w7w7k3", "ButcherHero_HooknChew", "Found the free gems", "2026-09-05", 7, 8), |
| 39 | ("1wmumud", "rawprimal", "What! Butcher’s website", "2026-09-22", 3, 7), |
| 40 | ("1w4tbqb", "findareferralcode", "Anybody have a referral code or link for Butcher Box?", "2026-09-01", 1, 2), |
| 41 | ] |
| 42 | LAST_PAGE_IDS = ["1wtoesn", "1wns3fq"] |
| 43 | |
| 44 | |
| 45 | def _fixture(name: str) -> str: |
| 46 | return (FIXTURES / name).read_text(encoding="utf-8") |
| 47 | |
| 48 | |
| 49 | def _post_ids(posts): |
| 50 | return [p["metadata"]["post_id"] for p in posts] |
| 51 | |
| 52 | |
| 53 | def _synthetic_page(post_ids, cursor=None, sub="example"): |
| 54 | """A minimal search fragment with the live markup's result-unit shape.""" |
| 55 | units = [] |
| 56 | for pid in post_ids: |
| 57 | ctx = json.dumps({ |
| 58 | "action_info": {"type": "post"}, |
| 59 | "post": {"id": f"t3_{pid}", "title": f"Topic post {pid}"}, |
| 60 | "subreddit": {"name": sub}, |
| 61 | }).replace('"', """) |
| 62 | units.append( |
| 63 | f'<search-telemetry-tracker data-faceplate-tracking-context="{ctx}" ' |
| 64 | f'view-events="search/view/post"><h2><a data-testid="post-title" ' |
| 65 | f'href="/r/{sub}/comments/{pid}/topic_post/">Topic post {pid}</a></h2>' |
| 66 | f'</search-telemetry-tracker>' |
| 67 | f'<faceplate-timeago ts="2026-09-20T10:00:00.000000+0000"></faceplate-timeago>' |
| 68 | f'<div data-testid="search-counter-row"><span><faceplate-number number="5" pretty>' |
| 69 | f'</faceplate-number> votes</span><span><faceplate-number number="2" pretty>' |
| 70 | f'</faceplate-number> comments</span></div>' |
| 71 | ) |
| 72 | if cursor: |
| 73 | units.append( |
| 74 | f'<faceplate-partial loading="lazy" src="/svc/shreddit/search/?q=x&' |
| 75 | f'type=posts&cursor={cursor}"></faceplate-partial>' |
| 76 | ) |
| 77 | return "<html><body>" + "".join(units) + "</body></html>" |
| 78 | |
| 79 | |
| 80 | class FakeReddit: |
| 81 | """Stands in for get_text: serves bodies by URL and records every request.""" |
| 82 | |
| 83 | def __init__(self, routes=None, default=None): |
| 84 | self.routes = dict(routes or {}) |
| 85 | self.default = default |
| 86 | self.requests = [] |
| 87 | |
| 88 | def __call__(self, url, **_kwargs): |
| 89 | self.requests.append(url) |
| 90 | body = self.routes.get(url, self.default) |
| 91 | if callable(body): |
| 92 | return body(url) |
| 93 | return body |
| 94 | |
| 95 | |
| 96 | @pytest.fixture |
| 97 | def fake_reddit(): |
| 98 | fake = FakeReddit() |
| 99 | with mock.patch.object(http, "get_text", side_effect=fake), \ |
| 100 | mock.patch.object(http.REDDIT_KEYLESS_LIMITER, "acquire"), \ |
| 101 | mock.patch.object(http.time, "sleep"): |
| 102 | yield fake |
| 103 | |
| 104 | |
| 105 | def _status(code, reason): |
| 106 | def _respond(_url): |
| 107 | http._record_failure(http.HTTPError(f"HTTP {code}: {reason}", code)) |
| 108 | return None |
| 109 | return _respond |
| 110 | |
| 111 | |
| 112 | class TestParse: |
| 113 | def test_page1_yields_post_units_with_dates_scores_and_comments(self): |
| 114 | posts, _cursor = reddit_search.parse_page(_fixture("reddit_search_page1.html"), "ButcherBox") |
| 115 | got = [ |
| 116 | (p["metadata"]["post_id"], p["subreddit"], p["title"], p["date"], p["score"], p["num_comments"]) |
| 117 | for p in posts |
| 118 | ] |
| 119 | # Seven post units; the page's community links and six "People also |
| 120 | # search for" suggestions are not posts. |
| 121 | assert got == PAGE1_POSTS |
| 122 | first = posts[0] |
| 123 | assert first["engagement"] == {"score": 1, "num_comments": 27, "upvote_ratio": None} |
| 124 | assert first["created_utc"] == datetime(2026, 9, 4, 3, 26, 52, 161000, tzinfo=timezone.utc).timestamp() |
| 125 | assert first["author"] == "" |
| 126 | assert first["selftext"] == "" |
| 127 | assert first["why_relevant"] == "Reddit search" |
| 128 | assert first["relevance"] > 0 |
| 129 | assert posts[4]["relevance"] == 0.0 # "Found the free gems" shares no query token |
| 130 | |
| 131 | def test_url_is_canonical_permalink_matching_post_id(self): |
| 132 | posts, _ = reddit_search.parse_page(_fixture("reddit_search_page1.html"), "ButcherBox") |
| 133 | assert posts[1]["url"] == ( |
| 134 | "https://www.reddit.com/r/TimDillon/comments/1wkybur/todays_ad_read_for_butcherbox/" |
| 135 | ) |
| 136 | for p in posts: |
| 137 | assert p["url"].startswith(f"https://www.reddit.com/r/{p['subreddit']}/comments/") |
| 138 | assert p["url"].split("/comments/")[1].split("/")[0] == p["metadata"]["post_id"] |
| 139 | |
| 140 | def test_cursor_read_from_next_page_partial(self): |
| 141 | _, cursor = reddit_search.parse_page(_fixture("reddit_search_page1.html")) |
| 142 | assert cursor == PAGE1_CURSOR |
| 143 | _, last_cursor = reddit_search.parse_page(_fixture("reddit_search_last_page.html")) |
| 144 | assert last_cursor is None |
| 145 | |
| 146 | |
| 147 | class TestPartialDrift: |
| 148 | def test_skipped_post_does_not_leak_counts_into_its_neighbour(self): |
| 149 | body = _fixture("reddit_search_page1.html") |
| 150 | clean, _ = reddit_search.parse_page(body) |
| 151 | assert len(clean) >= 3 |
| 152 | # Corrupt only the second post's tracker (post field no longer an object). |
| 153 | victim = clean[1]["metadata"]["post_id"] |
| 154 | tracker = next( |
| 155 | m for m in reddit_search._TRACKER.finditer(body) |
| 156 | if "t3_" + victim in m.group(0) and ""type":"post"" in m.group(0) |
| 157 | ) |
| 158 | start = body.index(""post":{", tracker.start(), tracker.end()) |
| 159 | corrupted = body[:start] + ""post":"x","old_post":{" + body[start + len(""post":{"):] |
| 160 | parsed, _ = reddit_search.parse_page(corrupted) |
| 161 | by_id = {p["metadata"]["post_id"]: p for p in parsed} |
| 162 | assert victim not in by_id |
| 163 | for post in clean: |
| 164 | pid = post["metadata"]["post_id"] |
| 165 | if pid == victim: |
| 166 | continue |
| 167 | assert (by_id[pid]["score"], by_id[pid]["num_comments"]) == (post["score"], post["num_comments"]) |
| 168 | |
| 169 | |
| 170 | class TestUrls: |
| 171 | def test_global_and_cursor_urls(self): |
| 172 | assert reddit_search.search_url("ButcherBox", "month") == PAGE1_URL |
| 173 | assert reddit_search.search_url("ButcherBox", "month", cursor=PAGE1_CURSOR) == PAGE2_URL |
| 174 | |
| 175 | def test_per_sub_url_uses_sub_search_path(self): |
| 176 | assert reddit_search.search_url("Tubi TV", "week", subreddit="r/TubiTV") == ( |
| 177 | "https://www.reddit.com/svc/shreddit/r/TubiTV/search/" |
| 178 | "?q=Tubi+TV&type=posts&t=week&sort=relevance" |
| 179 | ) |
| 180 | |
| 181 | @pytest.mark.parametrize("days,expected", [(7, "month"), (90, "year")]) |
| 182 | def test_window_sets_time_filter_like_scrapecreators_path(self, fake_reddit, days, expected): |
| 183 | fake_reddit.default = _fixture("reddit_search_no_results.html") |
| 184 | from_date, to_date = dates.get_date_range(days) |
| 185 | reddit_search.search("Tubi", depth="quick", from_date=from_date, to_date=to_date) |
| 186 | sent = parse_qs(urlsplit(fake_reddit.requests[0]).query)["t"] |
| 187 | assert sent == [expected] |
| 188 | assert sent == [reddit._window_to_time_filter(from_date, to_date)] |
| 189 | |
| 190 | |
| 191 | class TestPaging: |
| 192 | def test_follows_cursor_to_last_page_and_stops(self, fake_reddit): |
| 193 | fake_reddit.routes = { |
| 194 | PAGE1_URL: _fixture("reddit_search_page1.html"), |
| 195 | PAGE2_URL: _fixture("reddit_search_last_page.html"), |
| 196 | } |
| 197 | posts = reddit_search.search("ButcherBox", depth="default") |
| 198 | assert fake_reddit.requests == [PAGE1_URL, PAGE2_URL] |
| 199 | assert _post_ids(posts) == [row[0] for row in PAGE1_POSTS] + LAST_PAGE_IDS |
| 200 | assert [p["id"] for p in posts] == [f"R{i}" for i in range(1, 10)] |
| 201 | |
| 202 | def test_stops_on_repeated_cursor(self, fake_reddit): |
| 203 | fake_reddit.routes = { |
| 204 | PAGE1_URL: _fixture("reddit_search_page1.html"), |
| 205 | # New posts, but the page points back at the cursor already used. |
| 206 | PAGE2_URL: _synthetic_page(["aaa111", "bbb222"], cursor=PAGE1_CURSOR), |
| 207 | } |
| 208 | posts = reddit_search.search("ButcherBox", depth="deep") |
| 209 | assert fake_reddit.requests == [PAGE1_URL, PAGE2_URL] |
| 210 | assert _post_ids(posts)[-2:] == ["aaa111", "bbb222"] |
| 211 | |
| 212 | def test_stops_on_page_with_no_new_ids(self, fake_reddit): |
| 213 | repeat = _fixture("reddit_search_page1.html").replace(PAGE1_CURSOR.replace("=", "%3D"), "freshcursor") |
| 214 | fake_reddit.routes = { |
| 215 | PAGE1_URL: _fixture("reddit_search_page1.html"), |
| 216 | PAGE2_URL: repeat, |
| 217 | } |
| 218 | posts = reddit_search.search("ButcherBox", depth="deep") |
| 219 | assert fake_reddit.requests == [PAGE1_URL, PAGE2_URL] |
| 220 | assert _post_ids(posts) == [row[0] for row in PAGE1_POSTS] |
| 221 | |
| 222 | @pytest.mark.parametrize("depth,pages", [("quick", 2), ("default", 4), ("deep", 8)]) |
| 223 | def test_global_page_cap_by_depth(self, fake_reddit, depth, pages): |
| 224 | served = [] |
| 225 | |
| 226 | def endless(_url): |
| 227 | n = len(served) |
| 228 | served.append(n) |
| 229 | return _synthetic_page([f"p{n}x{i}" for i in range(7)], cursor=f"c{n}") |
| 230 | |
| 231 | fake_reddit.default = endless |
| 232 | posts = reddit_search.search("topic", depth=depth) |
| 233 | assert len(fake_reddit.requests) == pages |
| 234 | assert len(set(fake_reddit.requests)) == pages |
| 235 | # Existing depth post caps still apply on top of the page cap. |
| 236 | assert len(posts) == {"quick": 10, "default": 25, "deep": 50}[depth] |
| 237 | |
| 238 | def test_failed_second_page_keeps_first_page_and_records_failure(self, fake_reddit): |
| 239 | fake_reddit.routes = { |
| 240 | PAGE1_URL: _fixture("reddit_search_page1.html"), |
| 241 | PAGE2_URL: _status(500, "Internal Server Error"), |
| 242 | } |
| 243 | with http.capture_failures() as failures: |
| 244 | posts = reddit_search.search("ButcherBox", depth="default") |
| 245 | assert _post_ids(posts) == [row[0] for row in PAGE1_POSTS] |
| 246 | assert [f.status_code for f in failures] == [500] |
| 247 | |
| 248 | |
| 249 | class TestTargetedSubs: |
| 250 | def test_targeted_sub_gets_one_page_of_sub_search(self, fake_reddit): |
| 251 | sub_url = ( |
| 252 | "https://www.reddit.com/svc/shreddit/r/TubiTV/search/" |
| 253 | "?q=Tubi&type=posts&t=month&sort=relevance" |
| 254 | ) |
| 255 | global_url = ( |
| 256 | "https://www.reddit.com/svc/shreddit/search/" |
| 257 | "?q=Tubi&type=posts&t=month&sort=relevance" |
| 258 | ) |
| 259 | fake_reddit.routes = { |
| 260 | global_url: _fixture("reddit_search_no_results.html"), |
| 261 | sub_url: _synthetic_page(["s1", "s2"], cursor="more", sub="TubiTV"), |
| 262 | } |
| 263 | posts = reddit_search.search("Tubi", depth="deep", subreddits=["r/TubiTV"]) |
| 264 | assert sorted(fake_reddit.requests) == sorted([global_url, sub_url]) |
| 265 | assert _post_ids(posts) == ["s1", "s2"] |
| 266 | assert {p["subreddit"] for p in posts} == {"TubiTV"} |
| 267 | |
| 268 | def test_targeted_sub_posts_survive_the_depth_cap(self, fake_reddit): |
| 269 | sub_url = ( |
| 270 | "https://www.reddit.com/svc/shreddit/r/TubiTV/search/" |
| 271 | "?q=Tubi&type=posts&t=month&sort=relevance" |
| 272 | ) |
| 273 | pages = itertools.count() |
| 274 | |
| 275 | def endless(_url): |
| 276 | n = next(pages) |
| 277 | return _synthetic_page([f"g{n}x{i}" for i in range(7)], cursor=f"c{n}") |
| 278 | |
| 279 | fake_reddit.default = endless |
| 280 | fake_reddit.routes = {sub_url: _synthetic_page(["s1", "s2"], sub="TubiTV")} |
| 281 | posts = reddit_search.search("Tubi", depth="quick", subreddits=["TubiTV"]) |
| 282 | assert len(posts) == 10 |
| 283 | assert {"s1", "s2"} <= set(_post_ids(posts)) |
| 284 | |
| 285 | def test_full_targeted_pages_do_not_crowd_out_global_results(self, fake_reddit): |
| 286 | subs = ["SubA", "SubB", "SubC", "SubD"] |
| 287 | fake_reddit.routes = { |
| 288 | "https://www.reddit.com/svc/shreddit/r/" + sub + "/search/" |
| 289 | "?q=Tubi&type=posts&t=month&sort=relevance": _synthetic_page( |
| 290 | [f"{sub}{i}" for i in range(7)], sub=sub |
| 291 | ) |
| 292 | for sub in subs |
| 293 | } |
| 294 | fake_reddit.default = _synthetic_page([f"g{i}" for i in range(7)]) |
| 295 | posts = reddit_search.search("Tubi", depth="default", subreddits=subs) |
| 296 | ids = _post_ids(posts) |
| 297 | assert len(ids) == 25 |
| 298 | # 28 targeted posts alone would fill all 25 slots; global keeps a share. |
| 299 | assert {f"g{i}" for i in range(5)} <= set(ids) |
| 300 | assert all(any(i.startswith(sub) for i in ids) for sub in subs) |
| 301 | |
| 302 | |
| 303 | class TestOutcomes: |
| 304 | def test_challenge_page_records_failure_and_is_not_memoized(self, fake_reddit): |
| 305 | fake_reddit.default = _fixture("reddit_search_challenge.html") |
| 306 | with http.capture_failures() as failures: |
| 307 | first = reddit_search.search("ButcherBox", depth="quick") |
| 308 | second = reddit_search.search("ButcherBox", depth="quick") |
| 309 | assert first == [] and second == [] |
| 310 | assert fake_reddit.requests == [PAGE1_URL, PAGE1_URL] |
| 311 | assert len(failures) == 2 |
| 312 | assert all(f.outcome_state == health.SCHEMA_DRIFT for f in failures) |
| 313 | |
| 314 | @pytest.mark.parametrize("drift", [ |
| 315 | # Tracking context attribute gone entirely. |
| 316 | lambda body: re.sub(r'\sdata-faceplate-tracking-context="[^"]*"', "", body), |
| 317 | # Attribute kept, JSON shape changed (action_info renamed). |
| 318 | lambda body: body.replace(""action_info"", ""action""), |
| 319 | # Field kept but no longer an object, so .get() on it would raise. |
| 320 | lambda body: body.replace(""post":{", ""post":"x","old_post":{"), |
| 321 | ], ids=["stripped", "reshaped", "post-not-object"]) |
| 322 | def test_results_marker_with_no_parseable_units_is_schema_drift(self, fake_reddit, drift): |
| 323 | body = drift(_fixture("reddit_search_page1.html")) |
| 324 | assert reddit_search.RESULTS_MARKER in body |
| 325 | assert reddit_search.parse_page(body)[0] == [] |
| 326 | assert reddit_search.unrecognized_body(body) |
| 327 | fake_reddit.default = body |
| 328 | with http.capture_failures() as failures: |
| 329 | first = reddit_search.search("ButcherBox", depth="quick") |
| 330 | second = reddit_search.search("ButcherBox", depth="quick") |
| 331 | assert first == [] and second == [] |
| 332 | # Not memoized: the second run fetches again. |
| 333 | assert fake_reddit.requests == [PAGE1_URL, PAGE1_URL] |
| 334 | assert len(failures) == 2 |
| 335 | assert all(f.outcome_state == health.SCHEMA_DRIFT for f in failures) |
| 336 | |
| 337 | def test_recognized_page_is_memoized(self, fake_reddit): |
| 338 | fake_reddit.routes = {PAGE1_URL: _fixture("reddit_search_no_results.html")} |
| 339 | reddit_search.search("ButcherBox", depth="quick") |
| 340 | reddit_search.search("ButcherBox", depth="quick") |
| 341 | assert fake_reddit.requests == [PAGE1_URL] |
| 342 | |
| 343 | def test_no_results_page_returns_empty_without_failure(self, fake_reddit): |
| 344 | fake_reddit.default = _fixture("reddit_search_no_results.html") |
| 345 | with http.capture_failures() as failures: |
| 346 | posts = reddit_search.search("zxqvbnmplkqqwerty7731", depth="default") |
| 347 | assert posts == [] |
| 348 | assert failures == [] |
| 349 | assert len(fake_reddit.requests) == 1 |
| 350 | |
| 351 | def test_429_then_success_retries_once_through_limiter(self): |
| 352 | bodies = [None, _fixture("reddit_search_page1.html"), _fixture("reddit_search_last_page.html")] |
| 353 | |
| 354 | def fake_get(*_args, **_kwargs): |
| 355 | body = bodies.pop(0) |
| 356 | if body is None: |
| 357 | http._record_failure(http.HTTPError("HTTP 429: Too Many Requests", 429)) |
| 358 | return body |
| 359 | |
| 360 | with mock.patch.object(http, "get_text", side_effect=fake_get) as gt, \ |
| 361 | mock.patch.object(http.REDDIT_KEYLESS_LIMITER, "acquire") as acq, \ |
| 362 | mock.patch.object(http.time, "sleep"), \ |
| 363 | http.capture_failures() as failures: |
| 364 | posts = reddit_search.search("ButcherBox", depth="quick") |
| 365 | assert _post_ids(posts)[:7] == [row[0] for row in PAGE1_POSTS] |
| 366 | assert [c.args[0] for c in gt.call_args_list] == [PAGE1_URL, PAGE1_URL, PAGE2_URL] |
| 367 | assert acq.call_count == 3 |
| 368 | assert failures == [] |
| 369 | |
| 370 | @mock.patch("lib.http.time.sleep") |
| 371 | @mock.patch("lib.http.urllib.request.urlopen") |
| 372 | def test_fanout_429_reaches_pipeline_failure_sink(self, mock_urlopen, _sleep): |
| 373 | # get_text launders the 429 into None; the sink must survive the |
| 374 | # ThreadPoolExecutor hop into the search workers (issue #899). |
| 375 | mock_urlopen.side_effect = urllib.error.HTTPError( |
| 376 | "https://www.reddit.com/svc/shreddit/search/", 429, "Too Many Requests", {}, None |
| 377 | ) |
| 378 | with http.capture_failures() as failures: |
| 379 | posts = reddit_search.search("test topic", depth="quick", subreddits=["example"]) |
| 380 | assert posts == [] |
| 381 | assert failures[-1].outcome_state == schema.RATE_LIMITED |
| 382 | |
| 383 | def test_unexpected_error_never_raises(self): |
| 384 | with mock.patch.object(http, "reddit_keyless_get_text_retry_429", side_effect=RuntimeError("boom")): |
| 385 | assert reddit_search.search("ButcherBox") == [] |
| 386 | |
| 387 | |
| 388 | def test_result_timeout_includes_keyless_wait_allowance(monkeypatch): |
| 389 | limiter = http.RateLimiter(rate_per_sec=1.0, burst=2) |
| 390 | monkeypatch.delenv(http.REDDIT_KEYLESS_RATE_ENV, raising=False) |
| 391 | with mock.patch.object(http, "REDDIT_KEYLESS_LIMITER", limiter): |
| 392 | pad = http.REDDIT_KEYLESS_CONTENTION_SECONDS |
| 393 | timeout = reddit_search.SEARCH_TIMEOUT |
| 394 | assert reddit_search._result_timeout(13) == timeout + 5 + 13.0 + pad |
| 395 | # A paging stream waits for each of its sequential pages. |
| 396 | assert reddit_search._result_timeout(13, pages=4) == 4 * (timeout + 5) + 13.0 + pad |
| 397 |