| 1 | """Tests for digg.py - Digg AI 1000 source via digg-pp-cli.""" |
| 2 | |
| 3 | from __future__ import annotations |
| 4 | |
| 5 | import json |
| 6 | import os |
| 7 | import shutil |
| 8 | from datetime import datetime, timedelta, timezone |
| 9 | from unittest.mock import MagicMock, patch |
| 10 | |
| 11 | import pytest |
| 12 | |
| 13 | from lib import digg |
| 14 | from lib import subproc |
| 15 | |
| 16 | # === Helpers === |
| 17 | |
| 18 | |
| 19 | def _cluster( |
| 20 | cluster_url_id: str = "abc123xy", |
| 21 | title: str = "Sample cluster", |
| 22 | tldr: str = "A short summary of what is happening.", |
| 23 | rank: int = 1, |
| 24 | post_count: int = 5, |
| 25 | unique_authors: int = 3, |
| 26 | first_post_age: str = "5d", |
| 27 | ): |
| 28 | return { |
| 29 | "clusterUrlId": cluster_url_id, |
| 30 | "clusterId": f"uuid-{cluster_url_id}", |
| 31 | "title": title, |
| 32 | "tldr": tldr, |
| 33 | "rank": rank, |
| 34 | "postCount": post_count, |
| 35 | "uniqueAuthors": unique_authors, |
| 36 | "firstPostAge": first_post_age, |
| 37 | } |
| 38 | |
| 39 | |
| 40 | def _post( |
| 41 | username: str = "someone", |
| 42 | body: str = "Some body text about the topic.", |
| 43 | rank: int = 100, |
| 44 | category: str = "Engineer", |
| 45 | post_type: str = "tweet", |
| 46 | x_url: str | None = None, |
| 47 | ): |
| 48 | return { |
| 49 | "author": { |
| 50 | "username": username, |
| 51 | "display_name": username.title(), |
| 52 | "category": category, |
| 53 | "rank": rank, |
| 54 | }, |
| 55 | "body": body, |
| 56 | "post_type": post_type, |
| 57 | "xUrl": x_url or f"https://x.com/{username}/status/1234567890", |
| 58 | "posted_at": "2026-05-01T12:00:00+00:00", |
| 59 | } |
| 60 | |
| 61 | |
| 62 | def _stdout_for(payload: dict) -> subproc.SubprocResult: |
| 63 | return subproc.SubprocResult(returncode=0, stdout=json.dumps(payload), stderr="") |
| 64 | |
| 65 | # === _parse_first_post_age === |
| 66 | |
| 67 | |
| 68 | def test_parse_first_post_age_days(): |
| 69 | today = datetime(2026, 5, 9, tzinfo=timezone.utc) |
| 70 | assert digg._parse_first_post_age("5d", today=today) == "2026-05-04" |
| 71 | |
| 72 | |
| 73 | def test_parse_first_post_age_hours_returns_today(): |
| 74 | today = datetime(2026, 5, 9, 10, 0, tzinfo=timezone.utc) |
| 75 | assert digg._parse_first_post_age("5h", today=today) == "2026-05-09" |
| 76 | |
| 77 | |
| 78 | def test_parse_first_post_age_weeks(): |
| 79 | today = datetime(2026, 5, 9, tzinfo=timezone.utc) |
| 80 | assert digg._parse_first_post_age("2w", today=today) == "2026-04-25" |
| 81 | |
| 82 | |
| 83 | def test_parse_first_post_age_months_inside_window(): |
| 84 | today = datetime(2026, 5, 9, tzinfo=timezone.utc) |
| 85 | # 1 month = 30 days exactly, still inside the 30-day window. |
| 86 | assert digg._parse_first_post_age("1m", today=today) == (today - timedelta(days=30)).date().isoformat() |
| 87 | |
| 88 | |
| 89 | def test_parse_first_post_age_outside_30d_returns_none(): |
| 90 | today = datetime(2026, 5, 9, tzinfo=timezone.utc) |
| 91 | assert digg._parse_first_post_age("2m", today=today) is None |
| 92 | assert digg._parse_first_post_age("31d", today=today) is None |
| 93 | |
| 94 | |
| 95 | def test_parse_first_post_age_invalid(): |
| 96 | assert digg._parse_first_post_age(None) is None |
| 97 | assert digg._parse_first_post_age("") is None |
| 98 | assert digg._parse_first_post_age("garbage") is None |
| 99 | assert digg._parse_first_post_age("5x") is None |
| 100 | assert digg._parse_first_post_age("d") is None |
| 101 | assert digg._parse_first_post_age("-3d") is None |
| 102 | |
| 103 | # === parse_digg_response === |
| 104 | |
| 105 | |
| 106 | def test_parse_response_happy_path(): |
| 107 | response = { |
| 108 | "results": [ |
| 109 | _cluster(cluster_url_id="aaa", title="First", rank=1), |
| 110 | _cluster(cluster_url_id="bbb", title="Second", rank=4), |
| 111 | _cluster(cluster_url_id="ccc", title="Third", rank=12), |
| 112 | ] |
| 113 | } |
| 114 | items = digg.parse_digg_response(response) |
| 115 | assert len(items) == 3 |
| 116 | ids = [i["id"] for i in items] |
| 117 | assert ids == ["aaa", "bbb", "ccc"] |
| 118 | for item in items: |
| 119 | assert item["url"].startswith("https://di.gg/ai/") |
| 120 | assert item["engagement"]["postCount"] == 5 |
| 121 | assert item["engagement"]["uniqueAuthors"] == 3 |
| 122 | assert item["engagement"]["rank"] in (1, 4, 12) |
| 123 | assert item["posts"] == [] |
| 124 | assert item["date"] is not None |
| 125 | |
| 126 | |
| 127 | def test_parse_response_empty(): |
| 128 | assert digg.parse_digg_response({"results": []}) == [] |
| 129 | assert digg.parse_digg_response({}) == [] |
| 130 | assert digg.parse_digg_response({"results": "not-a-list"}) == [] |
| 131 | |
| 132 | |
| 133 | def test_parse_response_drops_missing_id(): |
| 134 | response = { |
| 135 | "results": [ |
| 136 | {"title": "no id", "tldr": "x", "postCount": 1, "uniqueAuthors": 1, "firstPostAge": "1d"}, |
| 137 | _cluster(cluster_url_id="ok", title="ok"), |
| 138 | ] |
| 139 | } |
| 140 | items = digg.parse_digg_response(response) |
| 141 | assert [i["id"] for i in items] == ["ok"] |
| 142 | |
| 143 | |
| 144 | def test_parse_response_drops_clusters_outside_30d(): |
| 145 | response = { |
| 146 | "results": [ |
| 147 | _cluster(cluster_url_id="recent", first_post_age="2d"), |
| 148 | _cluster(cluster_url_id="ancient", first_post_age="2m"), |
| 149 | ] |
| 150 | } |
| 151 | items = digg.parse_digg_response(response) |
| 152 | assert [i["id"] for i in items] == ["recent"] |
| 153 | |
| 154 | |
| 155 | def test_parse_response_keeps_cluster_when_age_missing(): |
| 156 | # When firstPostAge is absent or empty, we don't have evidence to drop; |
| 157 | # keep the cluster with date=None and let date-confidence downgrade it. |
| 158 | response = { |
| 159 | "results": [ |
| 160 | {**_cluster(cluster_url_id="noage"), "firstPostAge": None}, |
| 161 | ] |
| 162 | } |
| 163 | items = digg.parse_digg_response(response) |
| 164 | assert len(items) == 1 |
| 165 | assert items[0]["date"] is None |
| 166 | |
| 167 | |
| 168 | def test_parse_response_relevance_with_query(): |
| 169 | response = { |
| 170 | "results": [ |
| 171 | _cluster(cluster_url_id="match", title="OpenClaw launch", tldr="OpenClaw shipped today"), |
| 172 | _cluster(cluster_url_id="nomatch", title="Cricket scores", tldr="Mumbai vs Delhi"), |
| 173 | ] |
| 174 | } |
| 175 | items = digg.parse_digg_response(response, query="OpenClaw") |
| 176 | by_id = {i["id"]: i for i in items} |
| 177 | assert by_id["match"]["relevance"] > by_id["nomatch"]["relevance"] |
| 178 | |
| 179 | |
| 180 | def test_parse_response_engagement_rank_score(): |
| 181 | response = { |
| 182 | "results": [ |
| 183 | _cluster(cluster_url_id="top", rank=1), |
| 184 | _cluster(cluster_url_id="off-leaderboard", rank=999), |
| 185 | ] |
| 186 | } |
| 187 | items = digg.parse_digg_response(response) |
| 188 | by_id = {i["id"]: i for i in items} |
| 189 | assert by_id["top"]["engagement"]["rank_score"] == 50.0 |
| 190 | assert by_id["off-leaderboard"]["engagement"]["rank_score"] == 0.0 |
| 191 | |
| 192 | # === _parse_post === |
| 193 | |
| 194 | |
| 195 | def test_parse_post_happy(): |
| 196 | out = digg._parse_post(_post(username="adam", body="Hello world")) |
| 197 | assert out is not None |
| 198 | assert out["username"] == "adam" |
| 199 | assert out["body"] == "Hello world" |
| 200 | assert out["x_url"].startswith("https://x.com/") |
| 201 | |
| 202 | |
| 203 | def test_parse_post_drops_missing_body_or_handle_or_url(): |
| 204 | assert digg._parse_post({"author": {"username": "x"}, "body": "", "xUrl": "u"}) is None |
| 205 | assert digg._parse_post({"author": {}, "body": "txt", "xUrl": "u"}) is None |
| 206 | assert digg._parse_post({"author": {"username": "x"}, "body": "txt", "xUrl": ""}) is None |
| 207 | assert digg._parse_post(None) is None # type: ignore[arg-type] |
| 208 | |
| 209 | |
| 210 | def test_parse_post_drops_non_http_scheme_xurl(): |
| 211 | """Any non-http(s) scheme on xUrl yields no post. |
| 212 | |
| 213 | A malicious Digg API response (or compromised upstream) could set xUrl to |
| 214 | javascript:, data:text/html;..., file:, vbscript:, etc. The HTML report |
| 215 | renders the xUrl into an <a href> attribute, so any non-web scheme is a |
| 216 | stored-XSS or local-file vector when the user clicks the attribution. |
| 217 | """ |
| 218 | for bad_url in ( |
| 219 | "javascript:alert(1)", |
| 220 | "data:text/html;base64,PHNjcmlwdD5hbGVydCgxKTwvc2NyaXB0Pg==", |
| 221 | "vbscript:msgbox(1)", |
| 222 | "file:///etc/passwd", |
| 223 | "about:blank", |
| 224 | ): |
| 225 | out = digg._parse_post(_post(x_url=bad_url)) |
| 226 | assert out is None, f"expected non-http xUrl to be dropped, got: {out}" |
| 227 | |
| 228 | # http and https remain accepted. |
| 229 | assert digg._parse_post(_post(x_url="https://x.com/a/status/1")) is not None |
| 230 | assert digg._parse_post(_post(x_url="http://x.com/a/status/1")) is not None |
| 231 | |
| 232 | |
| 233 | def test_parse_post_logs_unsafe_xurl_rejection_even_in_non_tty(capsys): |
| 234 | """Security-class drops must be observable in non-interactive runs. |
| 235 | |
| 236 | The default ``log.source_log`` path is TTY-gated; without forcing |
| 237 | ``tty_only=False`` the rejection is invisible in Claude Code runs, |
| 238 | which is exactly the attack surface. Guard against silent regression. |
| 239 | """ |
| 240 | digg._parse_post(_post(x_url="javascript:alert(1)")) |
| 241 | err = capsys.readouterr().err |
| 242 | assert "[Digg] dropped post with unsafe xUrl scheme" in err |
| 243 | assert "javascript:alert(1)" in err |
| 244 | |
| 245 | |
| 246 | # === _run_cli / search_digg with stubbed subprocess === |
| 247 | |
| 248 | |
| 249 | def test_search_digg_binary_missing_returns_empty(monkeypatch): |
| 250 | monkeypatch.setattr(digg.shutil, "which", lambda _: None) |
| 251 | out = digg.search_digg("anything", "2026-04-09", "2026-05-09") |
| 252 | assert out["results"] == [] |
| 253 | assert "error" in out |
| 254 | |
| 255 | |
| 256 | def test_search_digg_passes_since_30d(monkeypatch): |
| 257 | captured: dict = {} |
| 258 | |
| 259 | def fake_run(cmd, *, timeout, env=None, on_pid=None): |
| 260 | captured["cmd"] = list(cmd) |
| 261 | return _stdout_for({"results": []}) |
| 262 | |
| 263 | monkeypatch.setattr(digg.shutil, "which", lambda _: "/fake/path") |
| 264 | monkeypatch.setattr(digg.subproc, "run_with_timeout", fake_run) |
| 265 | digg.search_digg("openclaw", "2026-04-09", "2026-05-09") |
| 266 | assert "--since" in captured["cmd"] |
| 267 | assert captured["cmd"][captured["cmd"].index("--since") + 1] == "30d" |
| 268 | assert "--agent" in captured["cmd"] |
| 269 | assert captured["cmd"][:3] == [digg.CLI_BIN, "search", "openclaw"] |
| 270 | |
| 271 | |
| 272 | def test_search_digg_subproc_timeout_returns_empty(monkeypatch): |
| 273 | monkeypatch.setattr(digg.shutil, "which", lambda _: "/fake/path") |
| 274 | |
| 275 | def fake_run(*_a, **_kw): |
| 276 | raise subproc.SubprocTimeout("boom") |
| 277 | |
| 278 | monkeypatch.setattr(digg.subproc, "run_with_timeout", fake_run) |
| 279 | out = digg.search_digg("openclaw", "2026-04-09", "2026-05-09") |
| 280 | assert out["results"] == [] |
| 281 | assert "error" in out |
| 282 | |
| 283 | |
| 284 | def test_search_digg_nonzero_exit_returns_empty(monkeypatch): |
| 285 | monkeypatch.setattr(digg.shutil, "which", lambda _: "/fake/path") |
| 286 | monkeypatch.setattr( |
| 287 | digg.subproc, |
| 288 | "run_with_timeout", |
| 289 | lambda *a, **k: subproc.SubprocResult(returncode=2, stdout="", stderr="cluster not found"), |
| 290 | ) |
| 291 | out = digg.search_digg("openclaw", "2026-04-09", "2026-05-09") |
| 292 | assert out["results"] == [] |
| 293 | assert "error" in out |
| 294 | |
| 295 | |
| 296 | def test_search_digg_invalid_json_returns_empty(monkeypatch): |
| 297 | monkeypatch.setattr(digg.shutil, "which", lambda _: "/fake/path") |
| 298 | monkeypatch.setattr( |
| 299 | digg.subproc, |
| 300 | "run_with_timeout", |
| 301 | lambda *a, **k: subproc.SubprocResult(returncode=0, stdout="not json", stderr=""), |
| 302 | ) |
| 303 | out = digg.search_digg("openclaw", "2026-04-09", "2026-05-09") |
| 304 | assert out["results"] == [] |
| 305 | assert "error" in out |
| 306 | |
| 307 | |
| 308 | def test_search_digg_empty_query_short_circuits(monkeypatch): |
| 309 | called = MagicMock() |
| 310 | monkeypatch.setattr(digg.shutil, "which", lambda _: "/fake/path") |
| 311 | monkeypatch.setattr(digg.subproc, "run_with_timeout", called) |
| 312 | out = digg.search_digg("", "2026-04-09", "2026-05-09") |
| 313 | assert out["results"] == [] |
| 314 | called.assert_not_called() |
| 315 | |
| 316 | # === enrich_with_top_posts === |
| 317 | |
| 318 | |
| 319 | def test_enrich_with_top_posts_attaches_posts(monkeypatch): |
| 320 | monkeypatch.setattr(digg.shutil, "which", lambda _: "/fake/path") |
| 321 | |
| 322 | def fake_run(cmd, *, timeout, env=None, on_pid=None): |
| 323 | # cmd = ['digg-pp-cli', 'posts', '<urlId>', '--agent', '--by', 'rank', '--limit', '3'] |
| 324 | cluster_url_id = cmd[2] |
| 325 | return _stdout_for( |
| 326 | { |
| 327 | "results": [ |
| 328 | _post(username=f"u_{cluster_url_id}", body=f"body for {cluster_url_id}"), |
| 329 | _post(username=f"v_{cluster_url_id}", body=f"second for {cluster_url_id}"), |
| 330 | ] |
| 331 | } |
| 332 | ) |
| 333 | |
| 334 | monkeypatch.setattr(digg.subproc, "run_with_timeout", fake_run) |
| 335 | |
| 336 | items = [ |
| 337 | {"id": "aaa", "engagement": {"postCount": 5}, "posts": []}, |
| 338 | {"id": "bbb", "engagement": {"postCount": 3}, "posts": []}, |
| 339 | {"id": "ccc", "engagement": {"postCount": 9}, "posts": []}, |
| 340 | {"id": "ddd", "engagement": {"postCount": 1}, "posts": []}, |
| 341 | ] |
| 342 | digg.enrich_with_top_posts(items, top_k=2, posts_per=3) |
| 343 | assert len(items[0]["posts"]) == 2 |
| 344 | assert items[0]["posts"][0]["username"] == "u_aaa" |
| 345 | assert len(items[1]["posts"]) == 2 |
| 346 | assert items[2]["posts"] == [] # not enriched (top_k=2) |
| 347 | |
| 348 | |
| 349 | def test_enrich_skips_zero_postcount(monkeypatch): |
| 350 | monkeypatch.setattr(digg.shutil, "which", lambda _: "/fake/path") |
| 351 | fake = MagicMock(return_value=_stdout_for({"results": [_post()]})) |
| 352 | monkeypatch.setattr(digg.subproc, "run_with_timeout", fake) |
| 353 | |
| 354 | items = [ |
| 355 | {"id": "no-posts", "engagement": {"postCount": 0}, "posts": []}, |
| 356 | {"id": "ok", "engagement": {"postCount": 7}, "posts": []}, |
| 357 | ] |
| 358 | digg.enrich_with_top_posts(items, top_k=2, posts_per=3) |
| 359 | assert items[0]["posts"] == [] |
| 360 | assert len(items[1]["posts"]) == 1 |
| 361 | |
| 362 | |
| 363 | def test_enrich_partial_timeout_does_not_break_others(monkeypatch): |
| 364 | monkeypatch.setattr(digg.shutil, "which", lambda _: "/fake/path") |
| 365 | call_count = {"n": 0} |
| 366 | |
| 367 | def fake_run(cmd, *, timeout, env=None, on_pid=None): |
| 368 | call_count["n"] += 1 |
| 369 | if call_count["n"] == 2: |
| 370 | raise subproc.SubprocTimeout("boom") |
| 371 | return _stdout_for({"results": [_post(username=f"u{call_count['n']}")]}) |
| 372 | |
| 373 | monkeypatch.setattr(digg.subproc, "run_with_timeout", fake_run) |
| 374 | |
| 375 | items = [ |
| 376 | {"id": "a", "engagement": {"postCount": 3}, "posts": []}, |
| 377 | {"id": "b", "engagement": {"postCount": 3}, "posts": []}, |
| 378 | {"id": "c", "engagement": {"postCount": 3}, "posts": []}, |
| 379 | ] |
| 380 | digg.enrich_with_top_posts(items, top_k=3, posts_per=3) |
| 381 | assert len(items[0]["posts"]) == 1 |
| 382 | assert items[1]["posts"] == [] # timed out |
| 383 | assert len(items[2]["posts"]) == 1 |
| 384 | |
| 385 | |
| 386 | def test_enrich_top_k_zero_skips_all(monkeypatch): |
| 387 | fake = MagicMock() |
| 388 | monkeypatch.setattr(digg.subproc, "run_with_timeout", fake) |
| 389 | items = [{"id": "a", "engagement": {"postCount": 5}, "posts": []}] |
| 390 | digg.enrich_with_top_posts(items, top_k=0) |
| 391 | fake.assert_not_called() |
| 392 | |
| 393 | # === enrich_source_items (post-dedupe path) === |
| 394 | |
| 395 | |
| 396 | class _FakeSourceItem: |
| 397 | def __init__(self, source, item_id, engagement, metadata): |
| 398 | self.source = source |
| 399 | self.item_id = item_id |
| 400 | self.engagement = engagement |
| 401 | self.metadata = metadata |
| 402 | |
| 403 | |
| 404 | def test_enrich_source_items_attaches_to_survivors(monkeypatch): |
| 405 | monkeypatch.setattr(digg.shutil, "which", lambda _: "/fake/path") |
| 406 | monkeypatch.setattr( |
| 407 | digg.subproc, |
| 408 | "run_with_timeout", |
| 409 | lambda cmd, *, timeout, env=None, on_pid=None: _stdout_for( |
| 410 | {"results": [_post(username=f"u_{cmd[2]}")]} |
| 411 | ), |
| 412 | ) |
| 413 | items = [ |
| 414 | _FakeSourceItem("digg", "ID1", {"postCount": 4}, {"clusterUrlId": "ID1", "posts": []}), |
| 415 | _FakeSourceItem("digg", "ID2", {"postCount": 6}, {"clusterUrlId": "ID2", "posts": []}), |
| 416 | _FakeSourceItem("digg", "ID3", {"postCount": 8}, {"clusterUrlId": "ID3", "posts": []}), |
| 417 | ] |
| 418 | digg.enrich_source_items(items, top_k=2) |
| 419 | assert items[0].metadata["posts"][0]["username"] == "u_ID1" |
| 420 | assert items[1].metadata["posts"][0]["username"] == "u_ID2" |
| 421 | assert items[2].metadata["posts"] == [] |
| 422 | |
| 423 | |
| 424 | def test_enrich_source_items_skips_non_digg(monkeypatch): |
| 425 | fake = MagicMock() |
| 426 | monkeypatch.setattr(digg.shutil, "which", lambda _: "/fake/path") |
| 427 | monkeypatch.setattr(digg.subproc, "run_with_timeout", fake) |
| 428 | items = [_FakeSourceItem("hackernews", "HN1", {"points": 100}, {"posts": []})] |
| 429 | digg.enrich_source_items(items, top_k=3) |
| 430 | fake.assert_not_called() |
| 431 | |
| 432 | |
| 433 | def test_enrich_source_items_falls_back_to_item_id(monkeypatch): |
| 434 | monkeypatch.setattr(digg.shutil, "which", lambda _: "/fake/path") |
| 435 | captured = {} |
| 436 | |
| 437 | def fake_run(cmd, *, timeout, env=None, on_pid=None): |
| 438 | captured["cluster_id"] = cmd[2] |
| 439 | return _stdout_for({"results": [_post()]}) |
| 440 | |
| 441 | monkeypatch.setattr(digg.subproc, "run_with_timeout", fake_run) |
| 442 | items = [_FakeSourceItem("digg", "fallbackid", {"postCount": 3}, {"posts": []})] |
| 443 | digg.enrich_source_items(items, top_k=1) |
| 444 | assert captured["cluster_id"] == "fallbackid" |
| 445 | |
| 446 | # === Live tests (opt-in) === |
| 447 | |
| 448 | LIVE = os.environ.get("LAST30DAYS_DIGG_LIVE", "").lower() in ("1", "true", "yes") |
| 449 | HAVE_BIN = shutil.which(digg.CLI_BIN) is not None |
| 450 | |
| 451 | @pytest.mark.skipif(not (LIVE and HAVE_BIN), reason="LAST30DAYS_DIGG_LIVE not set or digg-pp-cli missing") |
| 452 | |
| 453 | |
| 454 | class TestLiveDigg: |
| 455 | def test_search_returns_clusters(self): |
| 456 | out = digg.search_digg("claude code", "2026-04-09", "2026-05-09", depth="quick") |
| 457 | assert "results" in out |
| 458 | assert isinstance(out["results"], list) |
| 459 | # Topic should produce at least 1 cluster in the last 30d. |
| 460 | assert len(out["results"]) >= 1 |
| 461 | sample = out["results"][0] |
| 462 | for key in ("clusterUrlId", "title", "firstPostAge", "postCount"): |
| 463 | assert key in sample |
| 464 | |
| 465 | def test_parse_then_enrich_roundtrip(self): |
| 466 | raw = digg.search_digg("claude code", "2026-04-09", "2026-05-09", depth="quick") |
| 467 | items = digg.parse_digg_response(raw, query="claude code") |
| 468 | assert items, "expected at least one parsed cluster" |
| 469 | digg.enrich_with_top_posts(items, top_k=1, posts_per=2) |
| 470 | # Either the top cluster was successfully enriched, or it was a 0-post |
| 471 | # cluster and posts stayed empty. Both are valid; we just want no crash. |
| 472 | assert isinstance(items[0]["posts"], list) |
| 473 | |
| 474 | def test_off_topic_returns_list(self): |
| 475 | # Digg's live search uses fuzzy/popularity fallback so an impossible |
| 476 | # token may return clusters Digg considers loosely related rather |
| 477 | # than an empty list. The contract we depend on is shape: results |
| 478 | # must always be a list. Token-overlap relevance later in the |
| 479 | # pipeline filters off-topic noise. |
| 480 | out = digg.search_digg("ksdjflksjdflkjsdf-impossible-token", "2026-04-09", "2026-05-09", depth="quick") |
| 481 | assert isinstance(out.get("results"), list) |
| 482 | |
| 483 | def test_missing_cluster_id_graceful(self): |
| 484 | posts = digg.fetch_top_posts("notarealclusterid", posts_per=2) |
| 485 | assert posts == [] |
| 486 | |
| 487 | if __name__ == "__main__": |
| 488 | pytest.main([__file__, "-v"]) |
| 489 |