249 lines
8.7 KiB
Python
249 lines
8.7 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")
|
|
|
|
|
|
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.assertIn("[新入 #3]", 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"}
|
|
],
|
|
"github_emerging_moves": [],
|
|
"github_topic_moves": [],
|
|
}
|
|
text = build_github_delta_sections(movement, topic_name="llm")
|
|
self.assertIn("GitHub Trending 变化", text)
|
|
self.assertNotIn("新兴", text)
|
|
|
|
|
|
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",
|
|
)
|
|
self.assertIn("Skills Trending 变化", out)
|
|
self.assertNotIn("Skills Trending Top", out)
|