"""日报结构化组装单元测试（_build_report_data，纯逻辑）。""" from __future__ import annotations import json from datetime import date import pytest from scheduler.reporter import _build_report_data def _fake_event(title: str, importance: int, event_type: str = "其他", sentiment: str = "neutral", source_id: str = "cls", url: str = "https://x.com/1") -> dict: return { "title": title, "url": url, "source_id": source_id, "event": { "stock_codes": [], "company_names": [], "industries": [], "sentiment": sentiment, "importance": importance, "event_type": event_type, "summary": f"{title}的摘要", }, } class TestBuildReportData: def test_sections_and_ranks(self) -> None: news = { "total": 2, "hi_threshold": 4, "high": [_fake_event("新闻A", 5), _fake_event("新闻B", 4)], "sentiments": {"neutral": 2}, "importances": {5: 1, 4: 1}, "event_types": {"其他": 2}, } cninfo = { "total": 1, "hi_threshold": 2, "high": [_fake_event("公告C", 3, event_type="公告")], "by_day": {"07月10日": 1}, "announcement": 1, "research": 0, "irm": 0, } pipeline = {"raw_total": 100, "proc": 90} xwlb = {"items": [_fake_event("联播D", 4, event_type="新闻联播", source_id="xwlb")], "date": "07月10日"} r = _build_report_data(news, cninfo, pipeline, "AI摘要", "20260710", xwlb=xwlb) assert r.report_date == date(2026, 7, 10) assert r.report_type == "finance" assert r.file_name == "" assert r.ai_summary == "AI摘要" assert [(e.section, e.rank) for e in r.events] == [ ("news", 1), ("news", 2), ("cninfo", 1), ("xwlb", 1), ] assert r.events[0].source == "cls" assert r.events[3].source == "xwlb" def test_stats_snapshot(self) -> None: news = {"total": 1, "hi_threshold": 4, "high": [], "sentiments": {}, "importances": {}, "event_types": {}} cninfo = {"total": 0, "hi_threshold": 0, "high": [], "by_day": {}, "announcement": 0, "research": 0, "irm": 0} r = _build_report_data(news, cninfo, {"raw_total": 100}, "s", "20260710") # stats 可 JSON 序列化（入库时 json.dumps） json.dumps(r.stats, ensure_ascii=False) assert r.stats["pipeline"] == {"raw_total": 100} assert r.stats["news"]["total"] == 1 assert "xwlb" not in r.stats def test_title_truncated(self) -> None: news = {"total": 1, "hi_threshold": 4, "high": [_fake_event("长" * 600, 4)], "sentiments": {}, "importances": {}, "event_types": {}} cninfo = {"total": 0, "hi_threshold": 0, "high": [], "by_day": {}, "announcement": 0, "research": 0, "irm": 0} r = _build_report_data(news, cninfo, {}, "s", "20260710") assert len(r.events[0].title) == 512 def test_empty_events(self) -> None: news = {"total": 0, "hi_threshold": 0, "high": [], "sentiments": {}, "importances": {}, "event_types": {}} cninfo = {"total": 0, "hi_threshold": 0, "high": [], "by_day": {}, "announcement": 0, "research": 0, "irm": 0} r = _build_report_data(news, cninfo, {}, None, "20260710") assert r.events == [] assert r.ai_summary is None class TestLlmCallRetry: """_llm_call 重试逻辑（纯逻辑，mock client）。""" @staticmethod def _fake_client(failures: int): """构造 mock client：前 failures 次抛 ConnectionError，之后成功。""" from types import SimpleNamespace n = {"count": 0} class Completions: def create(self, **kwargs): n["count"] += 1 if n["count"] <= failures: raise ConnectionError("transient") return SimpleNamespace( choices=[SimpleNamespace( message=SimpleNamespace(content="今日要点摘要"), finish_reason="stop", )] ) return SimpleNamespace(chat=SimpleNamespace(completions=Completions())), n def test_success_first_try(self) -> None: from scheduler.reporter import _llm_call client, n = self._fake_client(0) out = _llm_call(client, "deepseek-v4-flash", "p") assert out == "今日要点摘要" assert n["count"] == 1 def test_retry_then_success(self, monkeypatch) -> None: import scheduler.reporter as rep monkeypatch.setattr(rep, "_LLM_RETRY_TIMES", 3) monkeypatch.setattr(rep, "_LLM_RETRY_BACKOFF_SEC", 0.01) client, n = self._fake_client(2) # 前 2 次失败，第 3 次成功 out = rep._llm_call(client, "deepseek-v4-flash", "p") assert out == "今日要点摘要" assert n["count"] == 3 def test_exhausts_retries_raises(self, monkeypatch) -> None: import scheduler.reporter as rep monkeypatch.setattr(rep, "_LLM_RETRY_TIMES", 2) monkeypatch.setattr(rep, "_LLM_RETRY_BACKOFF_SEC", 0.01) client, n = self._fake_client(99) # 一直失败 with pytest.raises(ConnectionError): rep._llm_call(client, "deepseek-v4-flash", "p") assert n["count"] == 2 # 重试 2 次后放弃 class TestCollectXwlb: """_collect_xwlb 取数逻辑：应查询日报前一日(已播出的联播)，并跳过内容提要。""" def test_queries_previous_day_and_skips_toc(self, monkeypatch) -> None: import json as _json import urllib.request captured: dict[str, str] = {} def fake_urlopen(req, timeout=15): # noqa: ARG001 captured["url"] = req.full_url class Resp: def __enter__(self): return self def __exit__(self, *args): return False def read(self): return _json.dumps({"data": {"news": [ {"daily_sub_id": 1, "news_title": "内容提要", "news_days": "2026-08-04", "news_improve": "开场白"}, {"daily_sub_id": 2, "news_title": "联播要闻A", "news_days": "2026-08-04", "news_improve": "正文A"}, ]}}).encode("utf-8") return Resp() monkeypatch.setattr(urllib.request, "urlopen", fake_urlopen) from scheduler.reporter import _collect_xwlb result = _collect_xwlb("20260805") # 查询的是前一日(20260804)而非当日 assert "start_date=20260804" in captured["url"] assert "end_date=20260804" in captured["url"] # 跳过第 1 条内容提要 assert len(result["items"]) == 1 assert result["items"][0]["title"] == "联播要闻A" assert result["source_date"] == "20260804" assert result["date"] == "08月04日" class TestCollectNewsEventsLookback: """_collect_news_events 30 小时回溯逻辑。""" def test_filters_30h_and_excludes_cninfo(self, monkeypatch) -> None: from datetime import datetime, timedelta import scheduler.reporter as rep now = datetime.now().astimezone() def fake_load(day_str: str) -> list[dict]: # noqa: ARG001 def ev(title: str, hours_ago: float | None, source: str = "cls", importance: int = 5, aware: bool = False) -> dict: pt = None if hours_ago is not None: t = now - timedelta(hours=hours_ago) pt = t.isoformat() if not aware else t.astimezone().isoformat() return { "title": title, "url": "u", "source_id": source, "publish_time": pt, "event": {"importance": importance, "sentiment": "neutral", "event_type": "其他", "summary": "s"}, } return [ ev("窗口内新闻", 10), ev("窗口内新闻带时区", 12, aware=True), ev("窗口外旧闻", 40), ev("无时间戳", None), ev("公告排除", 5, source="cninfo", importance=2), ] monkeypatch.setattr(rep, "_load_events_from_dir", fake_load) result = rep._collect_news_events("20260805") # 两个日期目录各返回 5 条(共 10): 旧闻×2、公告×2 被滤, 保留 6 条 assert result["total"] == 6 titles = {e["title"] for e in result["high"]} assert "窗口内新闻" in titles assert "窗口内新闻带时区" in titles assert "无时间戳" in titles assert "窗口外旧闻" not in titles assert "公告排除" not in titles