169 lines
6.2 KiB
Python
169 lines
6.2 KiB
Python
"""Tests for us2kr.collectors.news (no network: requests.get is mocked)."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from datetime import UTC, datetime, timedelta
|
|
|
|
import pytest
|
|
import requests
|
|
|
|
from us2kr.collectors import news
|
|
from us2kr.collectors.news import fetch_news, normalize_title, strip_html
|
|
|
|
SINCE = datetime(2026, 8, 25, 12, 0, tzinfo=UTC)
|
|
|
|
|
|
def _rss(*items: str) -> bytes:
|
|
body = "".join(items)
|
|
return (
|
|
'<?xml version="1.0" encoding="UTF-8"?><rss version="2.0"><channel><title>Feed</title>' + body
|
|
).encode() + b"</channel></rss>"
|
|
|
|
|
|
def _item(title: str, link: str, when: datetime, summary: str = "") -> str:
|
|
pub = when.strftime("%a, %d %b %Y %H:%M:%S GMT")
|
|
return (
|
|
f"<item><title>{title}</title><link>{link}</link>"
|
|
f"<description><![CDATA[{summary}]]></description><pubDate>{pub}</pubDate></item>"
|
|
)
|
|
|
|
|
|
class _Resp:
|
|
def __init__(self, content: bytes, status: int = 200):
|
|
self.content = content
|
|
self.status_code = status
|
|
|
|
def raise_for_status(self) -> None:
|
|
if self.status_code >= 400:
|
|
raise requests.HTTPError(f"{self.status_code} error")
|
|
|
|
|
|
@pytest.fixture
|
|
def two_feeds(monkeypatch):
|
|
feeds = [("Feed A", "https://a.example/rss"), ("Feed B", "https://b.example/rss")]
|
|
monkeypatch.setattr(news, "FEEDS", feeds)
|
|
|
|
|
|
def test_since_filter_dedup_sort_and_tz(two_feeds, mocker):
|
|
fresh = SINCE + timedelta(hours=3)
|
|
fresher = SINCE + timedelta(hours=5)
|
|
stale = SINCE - timedelta(hours=1)
|
|
feed_a = _rss(
|
|
_item(
|
|
"Fed holds rates steady!",
|
|
"https://a.example/1",
|
|
fresh,
|
|
"<p>Rates <b>unchanged</b> & steady</p>",
|
|
),
|
|
_item("Old story", "https://a.example/old", stale),
|
|
_item("Nvidia jumps 5%", "https://a.example/2", fresher),
|
|
)
|
|
feed_b = _rss(
|
|
_item("Fed Holds Rates Steady", "https://b.example/9", fresh), # dup by normalised title
|
|
_item("Different title, same link", "https://a.example/2/", fresh), # dup by link (trailing slash)
|
|
_item("Oil slips", "https://b.example/3", fresh + timedelta(minutes=1)),
|
|
)
|
|
mocker.patch.object(
|
|
news.requests,
|
|
"get",
|
|
side_effect=lambda url, **kw: _Resp(feed_a if "a.example" in url else feed_b),
|
|
)
|
|
|
|
items = fetch_news(SINCE, max_items=40)
|
|
|
|
titles = [i.title for i in items]
|
|
assert titles == ["Nvidia jumps 5%", "Oil slips", "Fed holds rates steady!"]
|
|
assert all(i.published.tzinfo is not None and i.published.utcoffset() == timedelta(0) for i in items)
|
|
assert items[0].published > items[1].published > items[2].published
|
|
fed = items[-1]
|
|
assert fed.summary == "Rates unchanged & steady"
|
|
assert fed.source == "Feed A"
|
|
# requests called with UA header + timeout for each feed
|
|
for call in news.requests.get.call_args_list:
|
|
assert call.kwargs["timeout"] == news.FEED_TIMEOUT_SECONDS
|
|
assert "User-Agent" in call.kwargs["headers"]
|
|
|
|
|
|
def test_max_items_cap(two_feeds, mocker):
|
|
items_xml = [
|
|
_item(f"Story {i}", f"https://a.example/{i}", SINCE + timedelta(minutes=i)) for i in range(10)
|
|
]
|
|
mocker.patch.object(news.requests, "get", return_value=_Resp(_rss(*items_xml)))
|
|
items = fetch_news(SINCE, max_items=3)
|
|
assert len(items) == 3
|
|
assert [i.title for i in items] == ["Story 9", "Story 8", "Story 7"]
|
|
|
|
|
|
def test_single_feed_failure_is_isolated(two_feeds, mocker):
|
|
good = _rss(_item("Survivor", "https://b.example/1", SINCE + timedelta(hours=1)))
|
|
|
|
def fake_get(url, **kw):
|
|
if "a.example" in url:
|
|
raise requests.ConnectionError("boom")
|
|
return _Resp(good)
|
|
|
|
mocker.patch.object(news.requests, "get", side_effect=fake_get)
|
|
items = fetch_news(SINCE)
|
|
assert [i.title for i in items] == ["Survivor"]
|
|
|
|
|
|
def test_http_error_and_garbage_are_isolated(two_feeds, mocker):
|
|
good = _rss(_item("Survivor", "https://b.example/1", SINCE + timedelta(hours=1)))
|
|
mocker.patch.object(
|
|
news.requests,
|
|
"get",
|
|
side_effect=lambda url, **kw: _Resp(b"<<<not xml at all", 200) if "a.example" in url else _Resp(good),
|
|
)
|
|
assert [i.title for i in fetch_news(SINCE)] == ["Survivor"]
|
|
|
|
mocker.patch.object(
|
|
news.requests,
|
|
"get",
|
|
side_effect=lambda url, **kw: _Resp(b"", 503) if "a.example" in url else _Resp(good),
|
|
)
|
|
assert [i.title for i in fetch_news(SINCE)] == ["Survivor"]
|
|
|
|
|
|
def test_all_feeds_failing_raises(two_feeds, mocker):
|
|
mocker.patch.object(news.requests, "get", side_effect=requests.Timeout("slow"))
|
|
with pytest.raises(RuntimeError, match="all news feeds failed"):
|
|
fetch_news(SINCE)
|
|
|
|
|
|
def test_entries_without_date_or_link_are_skipped(two_feeds, mocker):
|
|
xml = _rss(
|
|
"<item><title>No date</title><link>https://a.example/nd</link></item>",
|
|
"<item><title>No link</title><pubDate>Tue, 25 Aug 2026 20:00:00 GMT</pubDate></item>",
|
|
_item("Kept", "https://a.example/k", SINCE + timedelta(hours=1)),
|
|
)
|
|
mocker.patch.object(news.requests, "get", return_value=_Resp(xml))
|
|
assert [i.title for i in fetch_news(SINCE)] == ["Kept"]
|
|
|
|
|
|
def test_naive_since_is_treated_as_utc(two_feeds, mocker):
|
|
xml = _rss(_item("Kept", "https://a.example/k", SINCE + timedelta(hours=1)))
|
|
mocker.patch.object(news.requests, "get", return_value=_Resp(xml))
|
|
assert len(fetch_news(SINCE.replace(tzinfo=None))) == 1
|
|
assert len(fetch_news((SINCE + timedelta(hours=2)).replace(tzinfo=None))) == 0
|
|
|
|
|
|
def test_feedparser_receives_bytes(two_feeds, mocker):
|
|
xml = _rss(_item("Kept", "https://a.example/k", SINCE + timedelta(hours=1)))
|
|
mocker.patch.object(news.requests, "get", return_value=_Resp(xml))
|
|
spy = mocker.spy(news.feedparser, "parse")
|
|
fetch_news(SINCE)
|
|
assert spy.call_count == 2
|
|
assert all(isinstance(c.args[0], bytes) for c in spy.call_args_list)
|
|
|
|
|
|
def test_strip_html_truncates_and_unescapes():
|
|
assert strip_html("<p>A & B</p><script>x()</script><br/>C") == "A & B C"
|
|
long = "word " * 100
|
|
out = strip_html(f"<div>{long}</div>", max_chars=50)
|
|
assert len(out) <= 50 and out.endswith("…")
|
|
assert strip_html(None) == ""
|
|
|
|
|
|
def test_normalize_title():
|
|
assert normalize_title("Fed Holds Rates, Steady!") == normalize_title("fed holds rates steady")
|