| 1 | """Run-scoped ScrapeCreators Reddit memo (U5, R9): one run never pays for the |
| 2 | same Reddit query twice, across concurrent streams, the thin-source retry, and |
| 3 | the pinned-backend path. A failed call is not retried within the run.""" |
| 4 | |
| 5 | import threading |
| 6 | from concurrent.futures import ThreadPoolExecutor |
| 7 | from unittest import mock |
| 8 | |
| 9 | import pytest |
| 10 | |
| 11 | from lib import env, http, pipeline, reddit, schema |
| 12 | |
| 13 | FROM, TO = "2026-05-26", "2026-06-25" |
| 14 | |
| 15 | |
| 16 | def _memo(query="kanye", depth="default", subreddits=None, token="k"): |
| 17 | return reddit.search_and_enrich_memo( |
| 18 | query, FROM, TO, depth=depth, token=token, subreddits=subreddits, |
| 19 | ) |
| 20 | |
| 21 | |
| 22 | class TestMemo: |
| 23 | def test_sequential_same_key_calls_once_and_returns_equal_results(self): |
| 24 | with mock.patch("lib.reddit.search_and_enrich", |
| 25 | return_value={"items": [{"id": "a"}]}) as sc: |
| 26 | first = _memo() |
| 27 | second = _memo() |
| 28 | assert sc.call_count == 1 |
| 29 | assert first == second == {"items": [{"id": "a"}]} |
| 30 | |
| 31 | def test_concurrent_same_key_calls_once(self): |
| 32 | release = threading.Event() |
| 33 | calls = [] |
| 34 | |
| 35 | def _slow(*args, **kwargs): |
| 36 | calls.append(args) |
| 37 | release.wait(timeout=5) |
| 38 | return {"items": [{"id": "a"}]} |
| 39 | |
| 40 | with mock.patch("lib.reddit.search_and_enrich", side_effect=_slow): |
| 41 | with ThreadPoolExecutor(max_workers=4) as pool: |
| 42 | futures = [pool.submit(_memo) for _ in range(4)] |
| 43 | # Let every caller reach the memo before the owner finishes. |
| 44 | threading.Timer(0.2, release.set).start() |
| 45 | results = [f.result(timeout=10) for f in futures] |
| 46 | assert len(calls) == 1 |
| 47 | assert all(r == {"items": [{"id": "a"}]} for r in results) |
| 48 | |
| 49 | def test_failure_is_cached_and_not_retried(self): |
| 50 | boom = RuntimeError("HTTP 402: Payment Required") |
| 51 | with mock.patch("lib.reddit.search_and_enrich", side_effect=boom) as sc: |
| 52 | with pytest.raises(RuntimeError) as first: |
| 53 | _memo() |
| 54 | with pytest.raises(RuntimeError) as second: |
| 55 | _memo() |
| 56 | assert sc.call_count == 1 |
| 57 | assert str(first.value) == str(second.value) == "HTTP 402: Payment Required" |
| 58 | |
| 59 | def test_swallowed_failure_is_replayed_to_every_caller(self): |
| 60 | # search_and_enrich swallows a ScrapeCreators HTTP error into the |
| 61 | # caller's failure sink and returns no posts. A memo hit must replay |
| 62 | # that failure, or a later stream reads it as a clean empty result. |
| 63 | def _swallow_429(*_args, **_kwargs): |
| 64 | http._record_failure(http.HTTPError("HTTP 429: Too Many Requests", status_code=429)) |
| 65 | return {"items": []} |
| 66 | |
| 67 | with mock.patch("lib.reddit.search_and_enrich", side_effect=_swallow_429) as sc: |
| 68 | with http.capture_failures() as first_sink: |
| 69 | first = _memo() |
| 70 | with http.capture_failures() as second_sink: |
| 71 | second = _memo() |
| 72 | assert sc.call_count == 1 |
| 73 | assert first == second == {"items": []} |
| 74 | assert [f.status_code for f in first_sink] == [429] |
| 75 | assert [f.status_code for f in second_sink] == [429] |
| 76 | |
| 77 | def test_depth_and_subreddits_are_part_of_the_key(self): |
| 78 | with mock.patch("lib.reddit.search_and_enrich", |
| 79 | return_value={"items": []}) as sc: |
| 80 | _memo(depth="default") |
| 81 | _memo(depth="deep") |
| 82 | _memo(subreddits=["a", "b"]) |
| 83 | _memo(subreddits=["b", "a"]) # same set, different order: same key |
| 84 | _memo(subreddits=["a"]) |
| 85 | assert sc.call_count == 4 |
| 86 | |
| 87 | def test_reset_calls_again(self): |
| 88 | with mock.patch("lib.reddit.search_and_enrich", |
| 89 | return_value={"items": []}) as sc: |
| 90 | _memo() |
| 91 | reddit.reset_scrapecreators_memo() |
| 92 | _memo() |
| 93 | assert sc.call_count == 2 |
| 94 | |
| 95 | def test_callers_get_independent_copies(self): |
| 96 | with mock.patch("lib.reddit.search_and_enrich", |
| 97 | return_value={"items": [{"id": "a"}]}): |
| 98 | first = _memo() |
| 99 | first["items"][0]["id"] = "mutated" |
| 100 | second = _memo() |
| 101 | assert second["items"][0]["id"] == "a" |
| 102 | |
| 103 | |
| 104 | def _runtime(): |
| 105 | return schema.ProviderRuntime(reasoning_provider="mock", planner_model="mock", |
| 106 | rerank_model="mock") |
| 107 | |
| 108 | |
| 109 | def _post(pid): |
| 110 | return { |
| 111 | "id": pid, |
| 112 | "title": f"kanye post {pid}", |
| 113 | "url": f"https://www.reddit.com/r/x/comments/{pid}/t/", |
| 114 | "subreddit": "x", |
| 115 | "date": "2026-06-01", |
| 116 | "score": 10, |
| 117 | "num_comments": 1, |
| 118 | "relevance": 0.9, |
| 119 | } |
| 120 | |
| 121 | |
| 122 | def _run_four_streams(config, depth="default"): |
| 123 | subqueries = [ |
| 124 | schema.SubQuery(label=f"q{i}", search_query=f"kanye angle {i}", |
| 125 | ranking_query="kanye", sources=["reddit"]) |
| 126 | for i in range(4) |
| 127 | ] |
| 128 | |
| 129 | def _one(sq): |
| 130 | return pipeline._retrieve_stream( |
| 131 | topic="kanye", subquery=sq, source="reddit", config=config, |
| 132 | depth=depth, date_range=(FROM, TO), runtime=_runtime(), mock=False, |
| 133 | raw_topic="kanye", |
| 134 | ) |
| 135 | |
| 136 | with ThreadPoolExecutor(max_workers=4) as pool: |
| 137 | results = list(pool.map(_one, subqueries)) |
| 138 | return subqueries, results |
| 139 | |
| 140 | |
| 141 | class TestPipelineIntegration: |
| 142 | KEY = {"SCRAPECREATORS_API_KEY": "k"} |
| 143 | |
| 144 | def test_four_thin_streams_plus_thin_retry_make_one_sc_call(self): |
| 145 | # Free lane returns 1 item (below the default floor of 5), so every |
| 146 | # stream and the thin retry want a backfill. Only the underlying |
| 147 | # ScrapeCreators search may run, once. |
| 148 | with mock.patch("lib.reddit_public.search_reddit_public", |
| 149 | return_value=[_post("free1")]), \ |
| 150 | mock.patch("lib.reddit.search_reddit", |
| 151 | return_value={"items": [_post("sc1")]}) as sc_search, \ |
| 152 | mock.patch("lib.reddit.enrich_with_comments", |
| 153 | side_effect=lambda items, *_a, **_k: items): |
| 154 | subqueries, results = _run_four_streams(self.KEY) |
| 155 | assert all(items for items, _ in results) |
| 156 | |
| 157 | plan = schema.QueryPlan( |
| 158 | intent="breaking_news", freshness_mode="balanced_recent", |
| 159 | cluster_mode="none", raw_topic="kanye", subqueries=subqueries, |
| 160 | source_weights={"reddit": 1.0}, |
| 161 | ) |
| 162 | bundle = schema.RetrievalBundle() # reddit thin -> retried |
| 163 | pipeline._retry_thin_sources( |
| 164 | topic="kanye", bundle=bundle, plan=plan, config=self.KEY, |
| 165 | depth="default", date_range=(FROM, TO), runtime=_runtime(), |
| 166 | mock=False, rate_limited_sources=set(), |
| 167 | rate_limit_lock=threading.Lock(), |
| 168 | settings={"per_stream_limit": 20}, |
| 169 | ) |
| 170 | assert sc_search.call_count == 1 |
| 171 | |
| 172 | def test_pinned_path_four_streams_share_one_call(self): |
| 173 | cfg = {**self.KEY, env.REDDIT_BACKEND_PIN_VAR: "scrapecreators"} |
| 174 | with mock.patch("lib.reddit_public.search_reddit_public", |
| 175 | return_value=[]) as public, \ |
| 176 | mock.patch("lib.reddit.search_reddit", |
| 177 | return_value={"items": [_post("sc1")]}) as sc_search, \ |
| 178 | mock.patch("lib.reddit.enrich_with_comments", |
| 179 | side_effect=lambda items, *_a, **_k: items): |
| 180 | _subqueries, results = _run_four_streams(cfg) |
| 181 | assert sc_search.call_count == 1 |
| 182 | public.assert_not_called() |
| 183 | assert all(items for items, _ in results) |
| 184 |