yumao
6ea2a4e4c6
- 新增常驻调度器 daily/scheduler.py + run-scheduler.ps1（定时生成/推送） - 新增 daily/bridge_manager.py：Windows 兼容的 Cursor SDK 桥接 - 新增 skills/daily-featured-pick 首推 Skill 与叙事轴/去重逻辑 - 新闻抓取窗口、GitHub 搜索、企微 delta 模式等多项改进 - 补充设计文档与 superpowers 计划/规范 - 新增对应测试（scheduler、featured_pick、github_search、news_fetch_window 等） Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
805 lines
30 KiB
Python
805 lines
30 KiB
Python
"""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"], "中文描述")
|