| 1 | from __future__ import annotations |
| 2 | |
| 3 | import hashlib |
| 4 | import re |
| 5 | import sys |
| 6 | |
| 7 | import pytest |
| 8 | |
| 9 | import last30days as cli |
| 10 | from lib import env, html_render, registers, render, schema |
| 11 | |
| 12 | |
| 13 | SOURCES = [ |
| 14 | "reddit", |
| 15 | "github", |
| 16 | "youtube", |
| 17 | "tiktok", |
| 18 | "instagram", |
| 19 | "hackernews", |
| 20 | "polymarket", |
| 21 | "grounding", |
| 22 | "x", |
| 23 | "arxiv", |
| 24 | "jobs", |
| 25 | "bluesky", |
| 26 | ] |
| 27 | |
| 28 | |
| 29 | def fixture_report() -> schema.Report: |
| 30 | candidates: list[schema.Candidate] = [] |
| 31 | clusters: list[schema.Cluster] = [] |
| 32 | items_by_source: dict[str, list[schema.SourceItem]] = {} |
| 33 | for index, source in enumerate(SOURCES, start=1): |
| 34 | item = schema.SourceItem( |
| 35 | item_id=f"item-{index}", |
| 36 | source=source, |
| 37 | title=f"{source} signal with a detailed audience-ready headline {index}", |
| 38 | body=f"Evidence body for {source}.", |
| 39 | url=f"https://example.com/{source}/{index}", |
| 40 | author=f"voice{index}", |
| 41 | container="community", |
| 42 | published_at="2026-07-09", |
| 43 | date_confidence="high", |
| 44 | engagement={"score": 1000 - index, "likes": 2000 - index}, |
| 45 | snippet=f"Technical and community evidence from {source}.", |
| 46 | metadata={ |
| 47 | "top_comments": [ |
| 48 | { |
| 49 | "excerpt": f"Memorable community reaction number {index}.", |
| 50 | "score": 1000 - index, |
| 51 | "author": f"commenter{index}", |
| 52 | "url": f"https://example.com/{source}/{index}#comment", |
| 53 | } |
| 54 | ] |
| 55 | }, |
| 56 | ) |
| 57 | candidate = schema.Candidate( |
| 58 | candidate_id=f"candidate-{index}", |
| 59 | item_id=item.item_id, |
| 60 | source=source, |
| 61 | title=item.title, |
| 62 | url=item.url, |
| 63 | snippet=item.snippet, |
| 64 | subquery_labels=["primary"], |
| 65 | native_ranks={f"primary:{source}": index}, |
| 66 | local_relevance=0.95, |
| 67 | freshness=95, |
| 68 | engagement=90, |
| 69 | source_quality=1.0, |
| 70 | rrf_score=0.02, |
| 71 | sources=[source], |
| 72 | source_items=[item], |
| 73 | rerank_score=95, |
| 74 | final_score=101 - index, |
| 75 | fun_score=90, |
| 76 | fun_explanation="high-signal phrasing", |
| 77 | ) |
| 78 | cluster = schema.Cluster( |
| 79 | cluster_id=f"cluster-{index}", |
| 80 | title=f"{source} storyline {index}", |
| 81 | candidate_ids=[candidate.candidate_id], |
| 82 | representative_ids=[candidate.candidate_id], |
| 83 | sources=[source], |
| 84 | score=101 - index, |
| 85 | ) |
| 86 | candidates.append(candidate) |
| 87 | clusters.append(cluster) |
| 88 | items_by_source[source] = [item] |
| 89 | |
| 90 | return schema.Report( |
| 91 | topic="audience register research", |
| 92 | range_from="2026-06-10", |
| 93 | range_to="2026-07-10", |
| 94 | generated_at="2026-07-10T12:00:00Z", |
| 95 | provider_runtime=schema.ProviderRuntime( |
| 96 | reasoning_provider="local", |
| 97 | planner_model="fixture", |
| 98 | rerank_model="fixture", |
| 99 | ), |
| 100 | query_plan=schema.QueryPlan( |
| 101 | intent="general", |
| 102 | freshness_mode="strict_recent", |
| 103 | cluster_mode="story", |
| 104 | raw_topic="audience register research", |
| 105 | subqueries=[ |
| 106 | schema.SubQuery( |
| 107 | label="primary", |
| 108 | search_query="audience register research", |
| 109 | ranking_query="What matters?", |
| 110 | sources=SOURCES, |
| 111 | ) |
| 112 | ], |
| 113 | source_weights={source: 1.0 for source in SOURCES}, |
| 114 | ), |
| 115 | clusters=clusters, |
| 116 | ranked_candidates=candidates, |
| 117 | items_by_source=items_by_source, |
| 118 | errors_by_source={}, |
| 119 | artifacts={"pre_research_flags_present": True}, |
| 120 | ) |
| 121 | |
| 122 | |
| 123 | def _headings(output: str) -> list[str]: |
| 124 | return re.findall(r"^## (.+)$", output, flags=re.MULTILINE) |
| 125 | |
| 126 | |
| 127 | def _cluster_count(output: str) -> int: |
| 128 | evidence = output.split("## Ranked Evidence Clusters", 1)[1] |
| 129 | evidence = evidence.split("\n## ", 1)[0] |
| 130 | return len(re.findall(r"^### \d+\.", evidence, flags=re.MULTILINE)) |
| 131 | |
| 132 | |
| 133 | def _bullet_count(output: str, heading: str) -> int: |
| 134 | section = output.split(f"## {heading}", 1)[1] |
| 135 | section = section.split("\n## ", 1)[0] |
| 136 | return len(re.findall(r'^- "', section, flags=re.MULTILINE)) |
| 137 | |
| 138 | |
| 139 | @pytest.mark.parametrize( |
| 140 | ("name", "expected_order", "cluster_budget", "comment_budget"), |
| 141 | [ |
| 142 | ( |
| 143 | "exec", |
| 144 | ["Stats", "Ranked Evidence Clusters", "Source Coverage", "Best Takes", "Top Community Comments"], |
| 145 | 5, |
| 146 | 3, |
| 147 | ), |
| 148 | ( |
| 149 | "dev", |
| 150 | ["Ranked Evidence Clusters", "Source Coverage", "Stats", "Top Community Comments", "Best Takes"], |
| 151 | 10, |
| 152 | 4, |
| 153 | ), |
| 154 | ( |
| 155 | "creator", |
| 156 | ["Best Takes", "Top Community Comments", "Stats", "Ranked Evidence Clusters", "Source Coverage"], |
| 157 | 6, |
| 158 | 6, |
| 159 | ), |
| 160 | ], |
| 161 | ) |
| 162 | def test_registers_control_section_order_and_budgets( |
| 163 | name: str, |
| 164 | expected_order: list[str], |
| 165 | cluster_budget: int, |
| 166 | comment_budget: int, |
| 167 | ): |
| 168 | output = render.render_compact(fixture_report(), register=name) |
| 169 | |
| 170 | headings = _headings(output) |
| 171 | assert [heading for heading in headings if heading in expected_order] == expected_order |
| 172 | assert _cluster_count(output) == cluster_budget |
| 173 | assert _bullet_count(output, "Top Community Comments") == comment_budget |
| 174 | |
| 175 | |
| 176 | def test_emphasis_weights_promote_audience_specific_sources(): |
| 177 | report = fixture_report() |
| 178 | |
| 179 | dev = render.render_compact(report, register="dev") |
| 180 | creator = render.render_compact(report, register="creator") |
| 181 | |
| 182 | assert "### 1. github storyline" in dev |
| 183 | assert "### 1. tiktok storyline" in creator |
| 184 | |
| 185 | |
| 186 | def test_creator_register_leads_markdown_and_html_with_best_takes(): |
| 187 | report = fixture_report() |
| 188 | |
| 189 | markdown = render.render_compact(report, register="creator") |
| 190 | html = html_render.render_html(report, register="creator") |
| 191 | |
| 192 | assert markdown.index("## Best Takes") < markdown.index("## Ranked Evidence Clusters") |
| 193 | assert html.index("<h2>Best Takes</h2>") < html.index("<h2>Ranked Evidence Clusters</h2>") |
| 194 | |
| 195 | |
| 196 | def test_default_register_is_byte_identical_when_omitted(monkeypatch): |
| 197 | monkeypatch.setattr(render, "_render_badge", lambda: ["fixed badge", ""]) |
| 198 | monkeypatch.setattr(render, "_skill_version", lambda: "fixture") |
| 199 | report = fixture_report() |
| 200 | |
| 201 | implicit = render.render_compact(report) |
| 202 | explicit = render.render_compact(report, register="default") |
| 203 | |
| 204 | assert implicit == explicit |
| 205 | assert hashlib.sha256(implicit.encode()).hexdigest() == ( |
| 206 | # Hash includes #886's linked evidence URLs and #890's Hacker News |
| 207 | # comment-rendering changes from main. |
| 208 | "351089b5c0eae7ef55bcfd35cc23a6eca1008a7f3d8c28e3266fe351a788c985" |
| 209 | ) |
| 210 | |
| 211 | |
| 212 | def test_eli5_is_renderer_equivalent_to_default(): |
| 213 | report = fixture_report() |
| 214 | |
| 215 | assert render.render_compact(report, register="eli5") == render.render_compact( |
| 216 | report, register="default" |
| 217 | ) |
| 218 | |
| 219 | |
| 220 | def test_cli_and_env_register_resolution(): |
| 221 | args = cli.build_parser().parse_args(["topic", "--register", "exec"]) |
| 222 | assert args.register == "exec" |
| 223 | assert cli._audience_register_for_run(args, {}, None).name == "exec" |
| 224 | |
| 225 | args = cli.build_parser().parse_args(["topic"]) |
| 226 | assert cli._audience_register_for_run( |
| 227 | args, {"LAST30DAYS_REGISTER": "creator"}, None |
| 228 | ).name == "creator" |
| 229 | assert cli._audience_register_for_run( |
| 230 | args, {"ELI5_MODE": "true"}, None |
| 231 | ).name == "eli5" |
| 232 | assert cli._audience_register_for_run( |
| 233 | args, {"LAST30DAYS_REGISTER": "default", "ELI5_MODE": "true"}, None |
| 234 | ).name == "default" |
| 235 | |
| 236 | |
| 237 | def test_last30days_register_round_trips_from_process_env(monkeypatch, tmp_path): |
| 238 | monkeypatch.setenv("LAST30DAYS_CONFIG_DIR", str(tmp_path)) |
| 239 | monkeypatch.setenv("LAST30DAYS_REGISTER", "dev") |
| 240 | monkeypatch.setattr(env, "CONFIG_DIR", tmp_path) |
| 241 | monkeypatch.setattr(env, "CONFIG_FILE", tmp_path / "does-not-exist.env") |
| 242 | monkeypatch.setattr(env, "_load_keychain", lambda *args, **kwargs: {}) |
| 243 | monkeypatch.setattr(env, "_load_pass", lambda *args, **kwargs: {}) |
| 244 | |
| 245 | assert env.get_config()["LAST30DAYS_REGISTER"] == "dev" |
| 246 | |
| 247 | |
| 248 | def test_registers_do_not_shape_drill_output(): |
| 249 | args = cli.build_parser().parse_args(["--drill", "cluster 1"]) |
| 250 | |
| 251 | assert cli._audience_register_for_run( |
| 252 | args, {"LAST30DAYS_REGISTER": "creator"}, None |
| 253 | ).name == "default" |
| 254 | |
| 255 | |
| 256 | def test_registers_do_not_shape_comparison_output(): |
| 257 | args = cli.build_parser().parse_args( |
| 258 | ["alpha", "vs", "beta", "--register=creator"] |
| 259 | ) |
| 260 | |
| 261 | assert cli._audience_register_for_run(args, {}, None).name == "default" |
| 262 | |
| 263 | |
| 264 | @pytest.mark.parametrize( |
| 265 | "topic", |
| 266 | [ |
| 267 | "alpha/beta", |
| 268 | "alpha compared to beta", |
| 269 | "difference between alpha and beta", |
| 270 | ], |
| 271 | ) |
| 272 | def test_registers_use_canonical_comparison_detection(topic): |
| 273 | args = cli.build_parser().parse_args([topic]) |
| 274 | |
| 275 | assert cli._audience_register_for_run( |
| 276 | args, {"LAST30DAYS_REGISTER": "board"}, None |
| 277 | ).name == "default" |
| 278 | |
| 279 | |
| 280 | def test_registered_html_excludes_source_failure_diagnostics(): |
| 281 | report = fixture_report() |
| 282 | report.source_status["x"] = schema.SourceOutcome( |
| 283 | source="x", |
| 284 | state=schema.RATE_LIMITED, |
| 285 | detail="HTTP 429 after retry budget", |
| 286 | fix_hint="doctor", |
| 287 | ) |
| 288 | report.errors_by_source["x"] = "private source error diagnostic" |
| 289 | |
| 290 | html = html_render.render_html(report, register="creator") |
| 291 | |
| 292 | assert "Partial Coverage" not in html |
| 293 | assert "Source Errors" not in html |
| 294 | assert "private source error diagnostic" not in html |
| 295 | |
| 296 | |
| 297 | def test_unknown_register_errors_cleanly(): |
| 298 | with pytest.raises(ValueError, match="unknown audience register"): |
| 299 | registers.get_register("board") |
| 300 | |
| 301 | args = cli.build_parser().parse_args(["topic"]) |
| 302 | with pytest.raises(ValueError, match="unknown audience register"): |
| 303 | cli._audience_register_for_run( |
| 304 | args, {"LAST30DAYS_REGISTER": "board"}, None |
| 305 | ) |
| 306 | |
| 307 | with pytest.raises(SystemExit) as exc: |
| 308 | cli.build_parser().parse_args(["topic", "--register", "board"]) |
| 309 | assert exc.value.code == 2 |
| 310 | |
| 311 | |
| 312 | def test_unknown_configured_register_fails_before_retrieval(monkeypatch, capsys): |
| 313 | monkeypatch.setattr( |
| 314 | cli.env, |
| 315 | "get_config", |
| 316 | lambda **_kwargs: {"LAST30DAYS_REGISTER": "board"}, |
| 317 | ) |
| 318 | monkeypatch.setattr( |
| 319 | cli.pipeline, |
| 320 | "diagnose", |
| 321 | lambda *_args, **_kwargs: pytest.fail("retrieval preflight should not run"), |
| 322 | ) |
| 323 | monkeypatch.setattr(sys, "argv", ["last30days.py", "test topic"]) |
| 324 | |
| 325 | assert cli.main() == 2 |
| 326 | assert "unknown audience register 'board'" in capsys.readouterr().err |
| 327 | |
| 328 | |
| 329 | def test_creator_best_takes_honor_source_emphasis(): |
| 330 | from lib import registers, render |
| 331 | |
| 332 | audience = registers.get_register("creator") |
| 333 | assert audience.emphasis_weights, "creator preset must define emphasis weights" |
| 334 | # TikTok emphasis must exceed baseline sources like hackernews. |
| 335 | assert audience.emphasis_for("tiktok") > audience.emphasis_for("hackernews") |
| 336 | |
| 337 | |
| 338 | def test_best_takes_ranking_applies_source_weights(): |
| 339 | from lib import render, schema |
| 340 | |
| 341 | def candidate(cid, source, fun): |
| 342 | item = schema.SourceItem( |
| 343 | item_id=cid, source=source, title=f"take {cid}", body="b", |
| 344 | url=f"https://{source}/{cid}", published_at="2026-07-01", |
| 345 | snippet="s", engagement={"likes": 10}, |
| 346 | ) |
| 347 | return schema.Candidate( |
| 348 | candidate_id=cid, item_id=cid, source=source, title=f"take {cid}", |
| 349 | url=item.url, snippet="s", subquery_labels=["primary"], |
| 350 | native_ranks={f"primary:{source}": 1}, local_relevance=0.9, |
| 351 | freshness=90, engagement=10, source_quality=0.5, rrf_score=0.1, |
| 352 | final_score=90, cluster_id="cl", source_items=[item], |
| 353 | fun_score=80.0, |
| 354 | ) |
| 355 | |
| 356 | hn = candidate("hn1", "hackernews", 80.0) |
| 357 | tt = candidate("tt1", "tiktok", 80.0) |
| 358 | weights = {"tiktok": 1.5, "hackernews": 1.0} |
| 359 | lines = render._render_best_takes( |
| 360 | [hn, tt], limit=2, threshold=70.0, |
| 361 | source_weight=lambda source: weights.get(source, 1.0), |
| 362 | ) |
| 363 | body = "\n".join(lines) |
| 364 | assert body.index("TikTok") < body.index("Hacker News") or body.index("tiktok") < body.index("hackernews") if "tiktok" in body.lower() else True |
| 365 | # Structural assertion: the tiktok take renders before the HN take. |
| 366 | tt_pos = body.lower().find("tiktok") |
| 367 | hn_pos = body.lower().find("hacker") |
| 368 | assert tt_pos != -1 and hn_pos != -1 |
| 369 | assert tt_pos < hn_pos |
| 370 |