Oleksandr Kozachuk
d4eec4088d
Replaces the broken pre-1.0-openai news_feed.py. Stdlib parsing with defusedxml (feeds are untrusted XML), titles sanitized (SAF-03), feed URLs SSRF-guarded. CLI: python -m fjerkroa_bot.news --config <cfg>.
76 lines
3.0 KiB
Python
76 lines
3.0 KiB
Python
"""Unit coverage for SPEC-013 news digest (NEWS-01..03)."""
|
|
|
|
import unittest
|
|
from unittest.mock import AsyncMock
|
|
|
|
from fjerkroa_bot.news import NewsFetcher, parse_feed, render_digest
|
|
|
|
RSS = b"""<?xml version="1.0"?><rss><channel>
|
|
<item><title>Game X released</title><link>https://ex.com/x</link></item>
|
|
<item><title>Patch Y notes</title><link>https://ex.com/y</link></item>
|
|
</channel></rss>"""
|
|
|
|
ATOM = b"""<?xml version="1.0"?><feed xmlns="http://www.w3.org/2005/Atom">
|
|
<entry><title>Atom headline</title><link href="https://ex.com/a"/></entry>
|
|
</feed>"""
|
|
|
|
|
|
class TestParse(unittest.TestCase):
|
|
def test_rss(self):
|
|
"""NEWS-01: RSS items parsed with title + link."""
|
|
items = parse_feed(RSS, "Src")
|
|
self.assertEqual([i["title"] for i in items], ["Game X released", "Patch Y notes"])
|
|
self.assertEqual(items[0]["link"], "https://ex.com/x")
|
|
self.assertEqual(items[0]["source"], "Src")
|
|
|
|
def test_atom(self):
|
|
"""NEWS-01: Atom entries parsed with href link."""
|
|
items = parse_feed(ATOM, "A")
|
|
self.assertEqual(items[0]["title"], "Atom headline")
|
|
self.assertEqual(items[0]["link"], "https://ex.com/a")
|
|
|
|
def test_malformed_never_raises(self):
|
|
"""NEWS-01: garbage XML returns [] without raising."""
|
|
self.assertEqual(parse_feed(b"<not xml", "bad"), [])
|
|
self.assertEqual(parse_feed(b"", "empty"), [])
|
|
|
|
|
|
class TestDigest(unittest.TestCase):
|
|
def test_sanitized_and_capped(self):
|
|
"""NEWS-02: headlines sanitized, item count capped."""
|
|
items = [{"title": "@everyone big news \x00", "link": "", "source": "S"} for _ in range(20)]
|
|
digest = render_digest(items, max_items=5)
|
|
self.assertEqual(digest.count("\n"), 4) # 5 lines
|
|
self.assertNotIn("@everyone", digest)
|
|
self.assertNotIn("\x00", digest)
|
|
|
|
|
|
class TestCollect(unittest.IsolatedAsyncioTestCase):
|
|
async def test_ssrf_skip_and_dedup(self):
|
|
"""NEWS-03: guarded feed skipped, dup titles dropped, bad fetch survived."""
|
|
|
|
def guard(url):
|
|
return "refused" if "internal" in url else None
|
|
|
|
async def fetch(url):
|
|
if "boom" in url:
|
|
raise ValueError("boom")
|
|
return RSS # same content from two feeds -> dedup
|
|
|
|
fetcher = NewsFetcher(guard, fetch)
|
|
feeds = [
|
|
("https://a.com/feed", "A"),
|
|
("https://internal/feed", "Internal"), # SSRF-skipped
|
|
("https://boom.com/feed", "Boom"), # fetch fails
|
|
("https://b.com/feed", "B"), # same RSS -> dup titles dropped
|
|
]
|
|
items = await fetcher.collect(feeds, per_feed=5)
|
|
titles = [i["title"] for i in items]
|
|
self.assertEqual(titles, ["Game X released", "Patch Y notes"]) # deduped, internal+boom skipped
|
|
|
|
async def test_per_feed_limit(self):
|
|
"""NEWS-03: per-feed cap honored."""
|
|
fetcher = NewsFetcher(lambda u: None, AsyncMock(return_value=RSS))
|
|
items = await fetcher.collect([("https://a.com", "A")], per_feed=1)
|
|
self.assertEqual(len(items), 1)
|