Matt Van Horn
8d3a9e4368
* test(reddit): add live RSS + shreddit comment fixtures Captured from reddit.com on 2026-05-29 (search.rss listing + the /svc/shreddit/comments partial), trimmed to a representative subset plus two synthetic edge cases (deleted author, negative score) for offline parser tests. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com> * feat(http): add keyless get_text helper Browser-UA text fetch for RSS/HTML endpoints; returns None on any HTTP or network failure so tiered callers fall through cleanly. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com> * feat(reddit): keyless RSS discovery (search.rss + listing feeds) Replaces the now-403 search.json with keyless Atom feeds, normalized to the existing reddit_public post shape. Scores are placeholder zeros, backfilled during shreddit enrichment. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com> * feat(reddit): keyless shreddit comment scraper Parses <shreddit-comment> elements from /svc/shreddit/comments/r/{sub}/t3_{id} (score/author/created/permalink + thingId-anchored body) into top comments, matching reddit_enrich output. Replaces the dead {thread}.json enrichment. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com> * feat(reddit): tiered keyless orchestrator Tier 0 one-shot .json (residential bonus) -> Tier 1 RSS discovery -> Tier 2 shreddit enrichment. Returns [] never raises, so the SC backup still engages when every keyless tier is empty. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com> * fix(reddit): route free path through keyless pipeline (.json is dead) search_reddit_public is now a thin shim over reddit_keyless, so pipeline.py and other callers need no change. Removes the dead .json enrichment helpers; search/_parse_posts remain as the demoted Tier 0 attempt. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com> * fix(reddit): request sort=top so true top comments land on page 1 Guarantees the highest-scored comments are captured even on large threads, independent of Reddit's default comment sort. Local score re-sort remains. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com> * feat(reddit): recover post upvote scores via keyless listing partials The shreddit community-more-posts partial server-renders each post's score and comment count (works for normal users, not IP-gated), unlike RSS or the comments endpoint. Use it as a scored discovery source and to backfill scores onto RSS-discovered posts (subreddits derived from results when not provided). Ranking now uses real upvote score. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com> * fix(reddit): listings backfill scores only on bare queries, not discovery Caught running the full pipeline on a bare topic: deriving subreddits from noisy RSS results and merging their top/hot listings flooded results with high-upvote off-topic posts. Now derived-subreddit listings are used only to backfill scores onto keyword-matched RSS posts; listing cards are merged as discovery only when the caller explicitly provides subreddits (on-topic). Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com> --------- Co-authored-by: Matt Van Horn <455140+mvanhorn@users.noreply.github.com> Co-authored-by: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
104 lines
3.9 KiB
Python
104 lines
3.9 KiB
Python
"""Tests for scripts/lib/reddit_shreddit.py — keyless shreddit comment scrape."""
|
|
|
|
from pathlib import Path
|
|
from unittest import mock
|
|
|
|
from lib import reddit_shreddit as rs
|
|
|
|
FIXTURE = Path(__file__).resolve().parent.parent / "fixtures" / "reddit_shreddit_comments_sample.html"
|
|
|
|
|
|
def _html():
|
|
return FIXTURE.read_text(encoding="utf-8")
|
|
|
|
|
|
class TestExtractPostRef:
|
|
def test_extracts_sub_and_id(self):
|
|
ref = rs.extract_post_ref("https://www.reddit.com/r/Rakuten/comments/1taeiw0/title/")
|
|
assert ref == ("Rakuten", "1taeiw0")
|
|
|
|
def test_non_thread_url_returns_none(self):
|
|
assert rs.extract_post_ref("https://www.reddit.com/r/Rakuten/") is None
|
|
assert rs.extract_post_ref("") is None
|
|
|
|
def test_svc_url_shape(self):
|
|
# sort=top guarantees the highest-scored comments land on page 1.
|
|
assert rs._svc_url("Rakuten", "1taeiw0") == (
|
|
"https://www.reddit.com/svc/shreddit/comments/r/Rakuten/t3_1taeiw0?sort=top"
|
|
)
|
|
|
|
|
|
class TestParseComments:
|
|
"""parse_comments reads <shreddit-comment> elements into scored dicts."""
|
|
|
|
def test_happy_path(self):
|
|
comments = rs.parse_comments(_html())
|
|
assert len(comments) >= 1
|
|
for c in comments:
|
|
assert isinstance(c["score"], int)
|
|
assert c["author"] and c["author"] not in ("[deleted]", "[removed]")
|
|
assert c["body"]
|
|
|
|
def test_sorted_by_score_desc(self):
|
|
scores = [c["score"] for c in rs.parse_comments(_html())]
|
|
assert scores == sorted(scores, reverse=True)
|
|
|
|
def test_deleted_and_removed_filtered(self):
|
|
authors = [c["author"] for c in rs.parse_comments(_html())]
|
|
assert "[deleted]" not in authors and "[removed]" not in authors
|
|
|
|
def test_negative_score_retained(self):
|
|
scores = [c["score"] for c in rs.parse_comments(_html())]
|
|
assert -7 in scores # synthetic downvoted-but-real comment
|
|
|
|
def test_limit_honored(self):
|
|
assert len(rs.parse_comments(_html(), limit=2)) == 2
|
|
|
|
def test_body_text_extracted(self):
|
|
bodies = [c["body"] for c in rs.parse_comments(_html())]
|
|
assert any("$750" in b or "pending" in b for b in bodies)
|
|
|
|
def test_comment_url_built(self):
|
|
for c in rs.parse_comments(_html()):
|
|
if c["url"]:
|
|
assert c["url"].startswith("https://reddit.com/r/")
|
|
|
|
def test_empty_html_returns_empty(self):
|
|
assert rs.parse_comments("") == []
|
|
assert rs.parse_comments("<html>no comments here</html>") == []
|
|
|
|
|
|
class TestTotalComments:
|
|
def test_reads_total(self):
|
|
assert rs._total_comments(_html()) == 14
|
|
|
|
def test_missing_returns_none(self):
|
|
assert rs._total_comments("<html></html>") is None
|
|
|
|
|
|
class TestFetchComments:
|
|
"""fetch_comments wires URL -> svc fetch -> parse, never raising."""
|
|
|
|
def test_happy_path(self):
|
|
url = "https://www.reddit.com/r/Rakuten/comments/1taeiw0/title/"
|
|
with mock.patch.object(rs.http, "get_text", return_value=_html()) as m:
|
|
out = rs.fetch_comments(url)
|
|
# svc endpoint, not .json
|
|
assert "/svc/shreddit/comments/" in m.call_args[0][0]
|
|
assert ".json" not in m.call_args[0][0]
|
|
assert out["num_comments"] == 14
|
|
assert len(out["top_comments"]) >= 1
|
|
first = out["top_comments"][0]
|
|
assert {"score", "date", "author", "excerpt", "url"} <= set(first.keys())
|
|
assert isinstance(out["comment_insights"], list)
|
|
|
|
def test_bad_url_returns_empty(self):
|
|
out = rs.fetch_comments("https://www.reddit.com/r/Rakuten/")
|
|
assert out["top_comments"] == [] and out["num_comments"] is None
|
|
|
|
def test_fetch_failure_returns_empty(self):
|
|
url = "https://www.reddit.com/r/Rakuten/comments/1taeiw0/title/"
|
|
with mock.patch.object(rs.http, "get_text", return_value=None):
|
|
out = rs.fetch_comments(url)
|
|
assert out["top_comments"] == [] and out["num_comments"] is None
|