from __future__ import annotations from pathlib import Path import pytest FIXTURES = Path(__file__).parent / "fixtures" def test_parse_sample_rss_fixture(): from daily.news.feeds import NEWS_CATEGORIES from daily.news.fetch import _parse_feed category = NEWS_CATEGORIES[0] feed_name = category.feeds[0].name xml = (FIXTURES / "sample-rss.xml").read_text(encoding="utf-8") items = _parse_feed(xml, feed_name, category) assert len(items) == 1 assert items[0]["title"] == "Sample AI headline for smoke tests" assert items[0]["link"] == "https://example.com/ai-news/1" assert items[0]["source_name"] == feed_name assert "minimal RSS item" in items[0]["summary"] def test_rss_user_agent_avoids_bot_blocked_feeds(): from daily.news.fetch import RSS_USER_AGENT assert "daily-robots" not in RSS_USER_AGENT def test_fetch_one_retries_transient_errors(monkeypatch): import httpx from daily.news.feeds import NEWS_CATEGORIES from daily.news import fetch as news_fetch monkeypatch.setenv("DAILY_RSS_RETRY", "2") monkeypatch.setattr(news_fetch.time, "sleep", lambda _: None) category = NEWS_CATEGORIES[0] feed = category.feeds[0] calls = {"n": 0} class FakeResponse: def raise_for_status(self): return None @property def text(self): return (FIXTURES / "sample-rss.xml").read_text(encoding="utf-8") class FakeClient: headers = {"User-Agent": "test"} def get(self, url, headers=None): calls["n"] += 1 if calls["n"] < 2: raise httpx.RemoteProtocolError("Server disconnected") return FakeResponse() entries, ok = news_fetch._fetch_one(FakeClient(), category, feed) assert ok is True assert len(entries) == 1 assert calls["n"] == 2 def test_fetch_ai_news_offline(monkeypatch): from daily.news import fetch as news_fetch sample = { "title": "Sample AI headline for smoke tests", "link": "https://example.com/ai-news/1", "summary": "A minimal RSS item used by offline tests.", "published": "Wed, 02 Jul 2026 08:00:00 GMT", "source_name": "OpenAI", "category_id": "official", "category_name": "厂商官方", "category_icon": "🏢", } monkeypatch.setattr( news_fetch, "_fetch_news", lambda categories: { "enabled": True, "hours": 72, "categories": [{"id": "official", "name": "厂商官方", "icon": "🏢", "items": [sample]}], "flat": [sample], "stats": {"feeds_total": 1, "feeds_ok": 1, "items_raw": 1, "feeds_failed": []}, }, ) payload = news_fetch.fetch_ai_news() assert payload["enabled"] is True assert payload["flat"][0]["title"].startswith("Sample AI")