"""Tests for WeCom delta mode.""" from __future__ import annotations import os import tempfile import unittest from pathlib import Path from unittest.mock import patch from daily.config import ( delta_baseline_fallback, env_bool, force_push, news_dedup_days, skip_push_when_silent, wecom_mode, ) from daily.news.pushed_links import filter_unpushed_items, record_pushed_links class ConfigHelpersTests(unittest.TestCase): def test_wecom_mode_defaults_delta(self): with patch.dict(os.environ, {}, clear=True): self.assertEqual(wecom_mode(), "delta") def test_wecom_mode_full(self): with patch.dict(os.environ, {"DAILY_WECOM_MODE": "full"}, clear=True): self.assertEqual(wecom_mode(), "full") def test_env_bool_truthy(self): with patch.dict(os.environ, {"DAILY_FORCE_PUSH": "1"}, clear=True): self.assertTrue(env_bool("DAILY_FORCE_PUSH", False)) def test_skip_push_when_silent_default(self): with patch.dict(os.environ, {}, clear=True): self.assertTrue(skip_push_when_silent()) def test_news_dedup_days_default(self): with patch.dict(os.environ, {}, clear=True): self.assertEqual(news_dedup_days(), 7) def test_delta_baseline_fallback_default(self): with patch.dict(os.environ, {}, clear=True): self.assertEqual(delta_baseline_fallback(), "full") def test_wecom_delta_pad_default_true(self): from daily.config import wecom_delta_pad with patch.dict(os.environ, {}, clear=True): self.assertTrue(wecom_delta_pad()) class NewsPushedLinksTests(unittest.TestCase): def test_filter_and_record_roundtrip(self): with tempfile.TemporaryDirectory() as tmp: cache = Path(tmp) / "pushed-news-links.json" items = [ {"title": "A", "link": "https://example.com/a?utm_source=x"}, {"title": "B", "link": "https://example.com/b"}, ] with patch("daily.news.pushed_links._cache_path", return_value=cache): with patch("daily.news.pushed_links.news_dedup_days", return_value=7): record_pushed_links("2026-07-08", ["https://example.com/a"]) out = filter_unpushed_items(items, date_str="2026-07-09") self.assertEqual(len(out), 1) self.assertEqual(out[0]["link"], "https://example.com/b") class SkillMovePartitionTests(unittest.TestCase): def test_partition_dedupes_across_boards(self): trending = [ { "id": "a/b/foo", "rank": 4, "title": "foo", "source": "a/b", "installs": 1, "link": "", "description": "", } ] hot = [ { "id": "a/b/foo", "rank": 2, "title": "foo", "source": "a/b", "installs": 1, "link": "", "description": "", } ] from daily.delta import partition_skill_moves_for_wecom t_out, h_out = partition_skill_moves_for_wecom(trending, hot) self.assertEqual(len(t_out), 1) self.assertEqual(len(h_out), 0) self.assertIn("Trending #4", t_out[0]["badge"]) self.assertIn("Hot #2", t_out[0]["badge"]) def test_effective_mode_fallback_full_without_baseline(self): from daily.delta import effective_wecom_mode with patch("daily.delta.find_previous_data", return_value=None): with patch("daily.delta.wecom_mode", return_value="delta"): with patch("daily.delta.delta_baseline_fallback", return_value="full"): self.assertEqual(effective_wecom_mode(date_str="2026-07-10"), "full") class DeltaFormatTests(unittest.TestCase): def test_skills_delta_omits_empty_board(self): from daily.format_wecom import build_skills_delta_sections moves = [ { "id": "x/y/z", "rank": 3, "title": "z", "source": "x/y", "installs": 10, "installs_fmt": "10", "link": "https://skills.sh/x/y/z", "description": "d", "badge": "Trending #3", } ] text = build_skills_delta_sections(moves, []) self.assertIn("Skills Trending 变化", text) self.assertNotIn("[新入 #", text) self.assertNotIn("Skills Hot 变化", text) def test_github_delta_omits_stable_board(self): from daily.format_wecom import build_github_delta_sections movement = { "github_trending_moves": [ { "repo": "a/b", "url": "https://github.com/a/b", "rank": 1, "language": "Go", "description": "open-source codebase and curriculum", } ], "github_emerging_moves": [], "github_topic_moves": [], } with patch( "daily.format_wecom.localize_brief_descriptions", return_value={"github:a/b": "开源代码库与课程体系"}, ): text = build_github_delta_sections(movement, topic_name="llm") self.assertIn("开源代码库", text) self.assertIn("GitHub Trending 变化", text) self.assertNotIn("新兴", text) self.assertNotIn("[新入 #", text) self.assertNotIn("\n > ", text) def test_delta_pad_groups_same_source_moves(self): from daily.format_wecom import build_skills_delta_sections moves = [ { "id": f"lllllllama/rigorpilot-skills/s{i}", "title": f"s{i}", "source": "lllllllama/rigorpilot-skills", "installs": 250 - i, "installs_fmt": str(250 - i), "link": f"https://www.skills.sh/lllllllama/rigorpilot-skills/s{i}", "description": f"skill {i}", } for i in range(1, 11) ] with patch("daily.format_wecom.localize_brief_descriptions", return_value={}): with patch("daily.format_wecom.needs_chinese", return_value=False): text = build_skills_delta_sections(moves, [], trending_limit=10, pad=True) self.assertIn("10 skills", text) self.assertNotIn("[**s2**]", text) def test_delta_pad_fills_skills_to_limit(self): from daily.format_wecom import build_skills_delta_sections moves = [ { "id": "a/b/new", "rank": 3, "title": "new", "source": "a/b", "installs": 99, "installs_fmt": "99", "link": "https://skills.sh/a/b/new", "description": "new skill", } ] full = [ { "id": "a/b/new", "title": "new", "source": "a/b", "installs": 99, "installs_fmt": "99", "link": "https://skills.sh/a/b/new", "description": "new skill", }, *[ { "id": f"src{i}/skill", "title": "skill", "source": f"src{i}/pkg", "installs": 100 - i, "installs_fmt": str(100 - i), "link": f"https://skills.sh/src{i}/pkg/skill", "description": f"skill from src{i}", } for i in range(1, 12) ], ] with patch("daily.format_wecom.localize_brief_descriptions", return_value={}): with patch("daily.format_wecom.needs_chinese", return_value=False): text = build_skills_delta_sections( moves, [], trending_full=full, hot_full=[], trending_limit=10, pad=True, ) self.assertIn("Skills Trending Top 10", text) self.assertNotIn("Skills Trending 变化", text) def test_delta_pad_keeps_large_clusters_merged_and_fills_limit(self): """补榜按 source 合并态取条，大 cluster 不得撑爆 flat 预算导致短榜。""" from daily.format_wecom import build_skills_delta_sections def cluster(source: str, n: int, installs: int) -> dict: titles = [f"t{i}" for i in range(n)] return { "id": f"{source}/{titles[0]}", "title": titles[0], "source": source, "installs": installs, "installs_fmt": str(installs), "cluster": True, "cluster_count": n, "cluster_skills": titles, "cluster_titles": ", ".join(titles[:4]) + "…", "link": f"https://skills.sh/{source}/{titles[0]}", "description": f"{source} cluster", } full = [cluster(f"big{i}/pkg", 20, 1000 - i) for i in range(1, 5)] + [ { "id": f"other{n}/pkg/skill", "title": "skill", "source": f"other{n}/pkg", "installs": 50 - n, "installs_fmt": str(50 - n), "link": f"https://skills.sh/other{n}/pkg/skill", "description": f"other {n}", } for n in range(1, 12) ] with patch("daily.format_wecom.localize_brief_descriptions", return_value={}): with patch("daily.format_wecom.needs_chinese", return_value=False): text = build_skills_delta_sections( [], [], trending_full=full, hot_full=[], trending_limit=10, pad=True, ) self.assertIn("Skills Trending Top 10", text) self.assertIn("20 skills", text) self.assertIn("other6/pkg", text) def test_delta_pad_recent_blocks_same_source_not_just_primary_id(self): """周去重按 source：换同仓另一个 skill id 不得再上榜。""" from daily.format_wecom import build_skills_delta_sections titles = [f"t{i}" for i in range(20)] full = [ { "id": f"big/pkg/{titles[0]}", "title": titles[0], "source": "big/pkg", "installs": 999, "installs_fmt": "999", "cluster": True, "cluster_count": 20, "cluster_skills": titles, "cluster_titles": ", ".join(titles[:4]) + "…", "link": f"https://skills.sh/big/pkg/{titles[0]}", "description": "big cluster", }, *[ { "id": f"other{n}/pkg/skill", "title": "skill", "source": f"other{n}/pkg", "installs": 50 - n, "installs_fmt": str(50 - n), "link": f"https://skills.sh/other{n}/pkg/skill", "description": f"other {n}", } for n in range(1, 12) ], ] # 昨日展示的是同 source 另一 skill id（非今日 primary） recent = {f"big/pkg/{titles[5]}"} with patch("daily.format_wecom.localize_brief_descriptions", return_value={}): with patch("daily.format_wecom.needs_chinese", return_value=False): text = build_skills_delta_sections( [], [], trending_full=full, hot_full=[], trending_limit=10, pad=True, recent_trending=recent, ) self.assertIn("Skills Trending Top 10", text) self.assertNotIn("big/pkg", text) self.assertIn("other1/pkg", text) def test_delta_pad_hot_recent_unions_trending_history_by_source(self): """Skills Hot 周去重合并 Trending 历史：隔日换榜也不能同 source 再出现。""" from daily.format_wecom import build_skills_delta_sections hot_full = [ { "id": "101-skills/skills/ai-music", "title": "ai-music", "source": "101-skills/skills", "installs": 200, "installs_fmt": "200", "link": "https://skills.sh/101-skills/skills/ai-music", "description": "hot candidate", }, { "id": "fresh/src/skill", "title": "skill", "source": "fresh/src", "installs": 100, "installs_fmt": "100", "link": "https://skills.sh/fresh/src/skill", "description": "fresh", }, ] with patch("daily.format_wecom.localize_brief_descriptions", return_value={}): with patch("daily.format_wecom.needs_chinese", return_value=False): text = build_skills_delta_sections( [], [], trending_full=[], hot_full=hot_full, trending_limit=10, hot_limit=10, pad=True, recent_trending={"101-skills/skills/ai-video-generation"}, recent_hot=set(), ) self.assertIn("Skills Hot Top 1", text) self.assertIn("fresh/src", text) self.assertNotIn("101-skills", text) def test_delta_pad_hot_excludes_trending_by_source(self): """同日 Hot 补榜按 source 避开 Trending，而非展开全部 cluster skill id。""" from daily.format_wecom import build_skills_delta_sections trending_full = [ { "id": "same/src/a", "title": "a", "source": "same/src", "installs": 100, "installs_fmt": "100", "link": "https://skills.sh/same/src/a", "description": "trending item", } ] hot_full = [ { "id": "same/src/b", "title": "b", "source": "same/src", "installs": 90, "installs_fmt": "90", "link": "https://skills.sh/same/src/b", "description": "hot twin", }, { "id": "fresh/src/skill", "title": "skill", "source": "fresh/src", "installs": 80, "installs_fmt": "80", "link": "https://skills.sh/fresh/src/skill", "description": "fresh hot", }, ] with patch("daily.format_wecom.localize_brief_descriptions", return_value={}): with patch("daily.format_wecom.needs_chinese", return_value=False): text = build_skills_delta_sections( [], [], trending_full=trending_full, hot_full=hot_full, trending_limit=10, hot_limit=10, pad=True, ) self.assertIn("Skills Hot Top 1", text) self.assertIn("fresh/src", text) self.assertNotIn("same/src/b", text) def test_delta_pad_uses_large_pool_when_recent_excludes_top(self): from daily.format_wecom import build_skills_delta_sections full_small = [ { "id": f"seen/src/s{i}", "title": f"s{i}", "source": "seen/src", "installs": 100 - i, "installs_fmt": str(100 - i), "link": f"https://skills.sh/seen/src/s{i}", "description": f"seen {i}", } for i in range(1, 11) ] full_large = [ { "id": f"fresh/src{n}/skill", "title": "skill", "source": f"fresh/src{n}", "installs": 50 - n, "installs_fmt": str(50 - n), "link": f"https://skills.sh/fresh/src{n}/skill", "description": f"fresh {n}", } for n in range(1, 11) ] recent = {f"seen/src/s{i}" for i in range(1, 11)} with patch("daily.format_wecom.localize_brief_descriptions", return_value={}): with patch("daily.format_wecom.needs_chinese", return_value=False): small = build_skills_delta_sections( [], [], trending_full=full_small, trending_limit=10, pad=True, recent_trending=recent, ) large = build_skills_delta_sections( [], [], trending_full=full_large, trending_limit=10, pad=True, recent_trending=recent, ) self.assertNotIn("Skills Trending Top 10", small) self.assertIn("Skills Trending Top 10", large) self.assertIn("fresh/src1", large) def test_delta_pad_fills_github_to_limit(self): from daily.format_wecom import build_github_delta_sections movement = {"github_trending_moves": [], "github_emerging_moves": [], "github_topic_moves": []} full = [ { "repo": f"org/r{i}", "url": f"https://github.com/org/r{i}", "language": "Go", "stars_today_fmt": "100", "total_stars_fmt": "1K", "description": f"repo {i}", "desc_short": f"repo {i}", } for i in range(1, 12) ] with patch("daily.format_wecom.localize_brief_descriptions", return_value={}): text = build_github_delta_sections( movement, topic_name="llm", github_trending=full, trending_limit=10, pad=True, ) self.assertIn("GitHub Trending Top 10", text) self.assertNotIn("GitHub Trending 变化", text) def test_delta_pad_github_unions_recent_across_boards(self): """GitHub 三榜共用周去重：Trending 出过的 repo，新兴/Topic 不得再出。""" from daily.format_wecom import build_github_delta_sections movement = {"github_trending_moves": [], "github_emerging_moves": [], "github_topic_moves": []} shared = { "repo": "seen/repo", "url": "https://github.com/seen/repo", "language": "Go", "stars_today_fmt": "100", "total_stars_fmt": "1K", "created_at": "2026-07-01", "description": "already shown", "desc_short": "already shown", } fresh = { "repo": "fresh/repo", "url": "https://github.com/fresh/repo", "language": "Go", "stars_today_fmt": "90", "total_stars_fmt": "900", "created_at": "2026-07-02", "description": "fresh", "desc_short": "fresh", } with patch("daily.format_wecom.localize_brief_descriptions", return_value={}): text = build_github_delta_sections( movement, topic_name="llm", github_trending=[], github_emerging=[shared, fresh], github_topic=[shared], emerging_limit=5, topic_limit=5, pad=True, recent_board_keys={"github_trending": {"seen/repo"}}, ) self.assertIn("fresh/repo", text) self.assertNotIn("seen/repo", text) def test_delta_pad_skips_recent_skills(self): from daily.format_wecom import build_skills_delta_sections full = [ { "id": f"x/y/s{i}", "title": f"s{i}", "source": "x/y", "installs": 100 - i, "installs_fmt": str(100 - i), "link": f"https://skills.sh/x/y/s{i}", "description": f"skill {i}", } for i in range(4, 6) ] + [ { "id": "fresh/src/skill", "title": "skill", "source": "fresh/src", "installs": 50, "installs_fmt": "50", "link": "https://skills.sh/fresh/src/skill", "description": "fresh skill", } ] with patch("daily.format_wecom.localize_brief_descriptions", return_value={}): with patch("daily.format_wecom.needs_chinese", return_value=False): text = build_skills_delta_sections( [], [], trending_full=full, hot_full=[], trending_limit=10, pad=True, # 同仓历史 skill id → 整仓 source 去重；仅保留其它 source recent_trending={f"x/y/s{i}" for i in range(1, 4)}, ) self.assertIn("Skills Trending Top 1", text) self.assertIn("fresh/src", text) self.assertNotIn("x/y", text) def test_delta_pad_skips_recent_github(self): from daily.format_wecom import build_github_delta_sections movement = {"github_trending_moves": [], "github_emerging_moves": [], "github_topic_moves": []} full = [ { "repo": f"org/r{i}", "url": f"https://github.com/org/r{i}", "language": "Go", "total_stars_fmt": "1K", "description": f"repo {i}", "desc_short": f"repo {i}", } for i in range(1, 6) ] with patch("daily.format_wecom.localize_brief_descriptions", return_value={}): text = build_github_delta_sections( movement, topic_name="llm", github_trending=full, trending_limit=10, pad=True, recent_board_keys={"github_trending": {f"org/r{i}" for i in range(1, 4)}}, ) self.assertIn("GitHub Trending Top 2", text) self.assertIn("org/r4", text) self.assertNotIn("org/r1", text) def test_delta_pad_skips_recent_github_moves(self): """异动新入榜若昨日企微已展示，pad 时仍应排除（不只滤补榜）。""" from daily.format_wecom import build_github_delta_sections movement = { "github_trending_moves": [ { "repo": "vinta/awesome-python", "url": "https://github.com/vinta/awesome-python", "language": "Python", "total_stars_fmt": "308K", "description": "list", } ], "github_emerging_moves": [], "github_topic_moves": [], } full = [ { "repo": "fresh/repo", "url": "https://github.com/fresh/repo", "language": "Go", "total_stars_fmt": "1K", "description": "fresh", "desc_short": "fresh", } ] with patch("daily.format_wecom.localize_brief_descriptions", return_value={}): text = build_github_delta_sections( movement, topic_name="llm", github_trending=full, trending_limit=10, pad=True, recent_board_keys={"github_trending": {"vinta/awesome-python"}}, ) self.assertIn("fresh/repo", text) self.assertNotIn("vinta/awesome-python", text) def test_load_recent_board_keys_from_data_json(self): import json import tempfile from pathlib import Path from daily.delta import load_recent_board_keys with tempfile.TemporaryDirectory() as tmp: out = Path(tmp) payload = { "data": { "date": "2026-07-09", "movement_baseline": { "skills_trending": [ {"id": "a/b/raw", "title": "raw", "source": "a/b"}, ], "skills_hot": [], "github_trending": [{"repo": "org/raw"}], "github_emerging": [], "github_topic": [], }, "wecom_shown_keys": { "skills_trending": ["a/b/foo"], "github_trending": ["org/bar"], }, } } (out / "2026-07-09.data.json").write_text(json.dumps(payload), encoding="utf-8") with patch("daily.board_history.OUTPUT_DIR", out): recent = load_recent_board_keys("2026-07-10", lookback_days=7) self.assertIn("a/b/foo", recent["skills_trending"]) self.assertIn("org/bar", recent["github_trending"]) self.assertNotIn("a/b/raw", recent["skills_trending"]) self.assertNotIn("org/raw", recent["github_trending"]) class PushGateTests(unittest.TestCase): def test_push_when_board_has_moves(self): from daily.push_gate import evaluate_push_gate gate = evaluate_push_gate( movement={"skills_trending_moves": [{"id": "a/b/c"}], "skills_hot_moves": [], "github_trending_moves": [], "github_emerging_moves": [], "github_topic_moves": []}, ai_news_items=[], cn_ai_news_items=[], featured_pick=None, ) self.assertTrue(gate.should_push) self.assertIn("board_moves", gate.reasons) def test_silent_when_all_empty(self): from daily.push_gate import evaluate_push_gate gate = evaluate_push_gate( movement={ "skills_trending_moves": [], "skills_hot_moves": [], "github_trending_moves": [], "github_emerging_moves": [], "github_topic_moves": [], }, ai_news_items=[], cn_ai_news_items=[], featured_pick=None, ) self.assertFalse(gate.should_push) self.assertTrue(gate.silent) def test_force_push_overrides_silent(self): from daily.push_gate import evaluate_push_gate with patch("daily.push_gate.force_push", return_value=True): gate = evaluate_push_gate( movement={ "skills_trending_moves": [], "skills_hot_moves": [], "github_trending_moves": [], "github_emerging_moves": [], "github_topic_moves": [], }, ai_news_items=[], cn_ai_news_items=[], featured_pick=None, ) self.assertTrue(gate.should_push) self.assertIn("force_push", gate.reasons) class WebhookSilentTests(unittest.TestCase): def test_skip_when_silent(self): import json import tempfile from pathlib import Path from daily.webhook import should_skip_push with tempfile.TemporaryDirectory() as tmp: root = Path(tmp) data = root / "2026-07-10.data.json" data.write_text( json.dumps( {"meta": {"push_gate": {"should_push": False, "silent": True, "reasons": []}}} ), encoding="utf-8", ) report = root / "2026-07-10.wecom.md" report.write_text("📰 test", encoding="utf-8") self.assertTrue(should_skip_push(report)) class E2ESmokeTests(unittest.TestCase): def test_delta_mode_no_full_top_label(self): from daily.format_wecom import replace_wecom_board_sections md = "📰 **早报 · 2026-07-10**\n\n🌍 **国际 AI · 精选 1**\n1. [x](https://a.com)\n" movement = { "skills_trending_moves": [ { "id": "a/b/c", "rank": 2, "title": "c", "source": "a/b", "installs": 1, "link": "https://skills.sh/a/b/c", "description": "d", } ], "skills_hot_moves": [], "github_trending_moves": [], "github_emerging_moves": [], "github_topic_moves": [], } with patch("daily.format_wecom.localize_brief_descriptions", return_value={}): with patch("daily.format_wecom.needs_chinese", return_value=False): out = replace_wecom_board_sections( md, mode="delta", movement=movement, trending=[], hot=[], topic_name="llm", pad=False, ) self.assertIn("Skills Trending 变化", out) self.assertNotIn("Skills Trending Top", out) class SyncMovementGithubTests(unittest.TestCase): def test_sync_copies_localized_description(self): from daily.generate import _sync_movement_github_descriptions movement = { "github_trending_moves": [{"repo": "a/b", "description": "english"}], "github_emerging_moves": [], "github_topic_moves": [], } _sync_movement_github_descriptions( movement, github_trending=[{"repo": "a/b", "description": "中文描述"}], github_emerging=[], github_topic=[], ) self.assertEqual(movement["github_trending_moves"][0]["description"], "中文描述")