| 1 | """U1 - nominate stage: river/listing candidate discovery. |
| 2 | |
| 3 | Covers the two behaviors the nominate stage adds over the old inline sweep: |
| 4 | a ``keyword_gate`` toggle (domain scoping vs global trending) and |
| 5 | fault-tolerant per-source failure recording that never raises. |
| 6 | """ |
| 7 | |
| 8 | from unittest import mock |
| 9 | |
| 10 | from lib import pipeline, reddit_listing, schema |
| 11 | |
| 12 | |
| 13 | def _plan(domain: str, sources: list[str], subreddits: list[str] | None = None) -> schema.DiscoveryPlan: |
| 14 | return schema.DiscoveryPlan( |
| 15 | domain=domain, |
| 16 | category=None, |
| 17 | subreddits=subreddits or ["all"], |
| 18 | sources=sources, |
| 19 | ) |
| 20 | |
| 21 | |
| 22 | def test_fetch_discovery_source_reddit_gate_filters_off_domain(): |
| 23 | """With the keyword gate on, off-domain listing items are dropped; with it |
| 24 | off (global trending), the feed's own hot ranking is kept verbatim.""" |
| 25 | plan = _plan("AI agents", ["reddit"]) |
| 26 | payload = { |
| 27 | "items": [ |
| 28 | {"title": "New AI agents framework launched", "selftext": ""}, |
| 29 | {"title": "Gardening tips for spring", "selftext": ""}, |
| 30 | ], |
| 31 | "errors": [], |
| 32 | } |
| 33 | with mock.patch.object(reddit_listing, "fetch_discovery_listings", return_value=payload): |
| 34 | gated, _ = pipeline._fetch_discovery_source( |
| 35 | "reddit", plan, |
| 36 | from_date="2026-06-10", to_date="2026-07-10", |
| 37 | depth="default", mock=False, config={}, keyword_gate=True, |
| 38 | ) |
| 39 | ungated, _ = pipeline._fetch_discovery_source( |
| 40 | "reddit", plan, |
| 41 | from_date="2026-06-10", to_date="2026-07-10", |
| 42 | depth="default", mock=False, config={}, keyword_gate=False, |
| 43 | ) |
| 44 | |
| 45 | assert [item["title"] for item in gated] == ["New AI agents framework launched"] |
| 46 | assert len(ungated) == 2 |
| 47 | |
| 48 | |
| 49 | def test_nominate_candidates_threads_keyword_gate(): |
| 50 | """nominate_candidates forwards its keyword_gate to each source fetch, so a |
| 51 | global (no-domain) run really does disable the gate.""" |
| 52 | plan = _plan("", ["reddit"]) |
| 53 | seen: dict[str, bool] = {} |
| 54 | |
| 55 | def fake_fetch(source, plan, *, from_date, to_date, depth, mock, config, keyword_gate=True): |
| 56 | seen["keyword_gate"] = keyword_gate |
| 57 | return [], None |
| 58 | |
| 59 | with mock.patch.object(pipeline, "_fetch_discovery_source", side_effect=fake_fetch): |
| 60 | pipeline.nominate_candidates( |
| 61 | plan, |
| 62 | from_date="2026-06-10", to_date="2026-07-10", |
| 63 | depth="default", mock=False, config={}, lookback_days=30, |
| 64 | keyword_gate=False, |
| 65 | ) |
| 66 | |
| 67 | assert seen["keyword_gate"] is False |
| 68 | |
| 69 | |
| 70 | def test_nominate_candidates_records_source_failure_without_raising(): |
| 71 | """One dead feed is recorded on the bundle as a failure; the surviving feed |
| 72 | still yields candidates, and the call never raises.""" |
| 73 | plan = _plan("AI agents", ["reddit", "hackernews"]) |
| 74 | |
| 75 | def fake_fetch(source, plan, *, from_date, to_date, depth, mock, config, keyword_gate=True): |
| 76 | if source == "hackernews": |
| 77 | raise TimeoutError("hn listing timed out") |
| 78 | return pipeline._mock_discovery_items(source, plan.domain, to_date), None |
| 79 | |
| 80 | with mock.patch.object(pipeline, "_fetch_discovery_source", side_effect=fake_fetch): |
| 81 | bundle = pipeline.nominate_candidates( |
| 82 | plan, |
| 83 | from_date="2026-06-10", to_date="2026-07-10", |
| 84 | depth="default", mock=False, config={}, lookback_days=30, |
| 85 | ) |
| 86 | |
| 87 | # Surviving source produced candidates. |
| 88 | assert bundle.items_by_source.get("reddit") |
| 89 | # Dead source recorded as a failure, not silently dropped or raised. |
| 90 | hackernews = bundle.source_status["hackernews"] |
| 91 | assert hackernews.state not in (schema.NO_RESULTS,) |
| 92 | assert "ok" != hackernews.state |
| 93 |