Fjerkroa/discord_bot
1
1
Fork 0
Code Issues Pull Requests Releases Wiki Activity

news memory + get_news tool (news-07..12): feed summaries, deduped rolling store (schema v6), on-demand filtered retrieval

Browse Source
This commit is contained in:
Oleksandr Kozachuk
2026-07-14 11:34:23 +02:00
parent 891fbdc101
commit 5d01400638
6 changed files with 390 additions and 15 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/test_spec_news.py
+146 -4
View File
@@ -1,9 +1,24 @@
"""Unit coverage for SPEC-013 news digest (NEWS-01..03)."""
"""Unit coverage for SPEC-013 news digest + memory + tool (NEWS-01..12)."""
import tempfile
import unittest
from pathlib import Path
from unittest.mock import AsyncMock
from fjerkroa_bot.news import NewsFetcher, NewsPoster, load_seen, parse_feed, render_digest, save_seen
from fjerkroa_bot.news import (
GET_NEWS_TOOL,
NewsFetcher,
NewsPoster,
load_seen,
parse_feed,
query_news,
render_digest,
save_seen,
)
from fjerkroa_bot.openai_responder import OpenAIResponder
from fjerkroa_bot.persistence import PersistentStore
CONFIG = {"openai-token": "t", "model": "m", "system": "s", "history-limit": 5}
RSS = b"""<?xml version="1.0"?><rss><channel>
<item><title>Game X released</title><link>https://ex.com/x</link></item>
@@ -166,10 +181,137 @@ class TestSeenState(unittest.TestCase):
def test_cap_bounds_state(self):
"""NEWS-05: save keeps at most `cap` keys."""
import json
import tempfile
from pathlib import Path
with tempfile.TemporaryDirectory() as tmp:
path = str(Path(tmp) / "state.json")
save_seen(path, {f"k{i}" for i in range(100)}, cap=10)
self.assertEqual(len(json.load(open(path))), 10)
RSS_DESC = b"""<?xml version="1.0"?><rss><channel>
<item><title>Storm hits coast</title><link>https://ex.com/s</link>
<description>&lt;p&gt;Heavy &lt;b&gt;wind&lt;/b&gt; expected&lt;/p&gt;</description></item>
</channel></rss>"""
ATOM_SUM = b"""<?xml version="1.0"?><feed xmlns="http://www.w3.org/2005/Atom">
<entry><title>Atom T</title><link href="https://ex.com/a"/><summary>Short gist here</summary></entry>
</feed>"""
class TestSummaries(unittest.TestCase):
def test_rss_description_stripped(self):
"""NEWS-07: RSS description parsed, HTML stripped, entities unescaped, whitespace collapsed."""
items = parse_feed(RSS_DESC, "S")
self.assertEqual(items[0]["summary"], "Heavy wind expected")
def test_atom_summary(self):
"""NEWS-07: Atom summary collapsed to clean text."""
items = parse_feed(ATOM_SUM, "A")
self.assertEqual(items[0]["summary"], "Short gist here")
def test_missing_description_is_empty(self):
"""NEWS-07: no description -> empty summary, never an error."""
self.assertEqual(parse_feed(RSS, "S")[0]["summary"], "")
def test_digest_carries_summary(self):
"""NEWS-08: digest appends the sanitized capped summary; zero cap = title only."""
items = [{"title": "T", "link": "https://ex.com/x", "source": "NRK", "summary": "the gist of it"}]
digest = render_digest(items, 10, 100)
self.assertIn("[NRK]", digest)
self.assertIn("the gist of it", digest)
self.assertNotIn("the gist", render_digest(items, 10, 0)) # zero cap -> title only
class NewsStoreBase(unittest.TestCase):
def setUp(self):
self.tmp = tempfile.TemporaryDirectory()
self.addCleanup(self.tmp.cleanup)
self.store = PersistentStore(Path(self.tmp.name) / "bot.db")
class TestNewsStore(NewsStoreBase):
def test_dedup_and_rolling_window(self):
"""NEWS-09: items deduped by link; prune keeps the newest N."""
first = [
{"title": "A", "link": "L1", "source": "S", "summary": "sa"},
{"title": "B", "link": "L2", "source": "S", "summary": "sb"},
]
self.assertEqual(self.store.add_news_items(first), 2)
self.assertEqual(self.store.add_news_items([dict(first[0])]), 0) # dup link ignored
self.assertEqual(self.store.news_count(), 2)
self.store.prune_news(1)
self.assertEqual(self.store.news_count(), 1)
self.assertEqual(self.store.recent_news(5)[0]["title"], "B") # newest survives
def test_dedup_by_title_when_no_link(self):
"""NEWS-09: linkless items dedup on title."""
self.store.add_news_items([{"title": "Same", "link": "", "source": "S", "summary": ""}])
self.store.add_news_items([{"title": "Same", "link": "", "source": "S", "summary": ""}])
self.assertEqual(self.store.news_count(), 1)
class TestQueryNews(NewsStoreBase):
def seed(self):
self.store.add_news_items(
[
{"title": "Nordland storm", "link": "L1", "source": "Nordland", "summary": "strong wind on the coast"},
{"title": "Oslo budget", "link": "L2", "source": "NRK", "summary": "@everyone spending plan"},
{"title": "Sport result", "link": "L3", "source": "Sport", "summary": "the match ended"},
]
)
def test_topic_filter(self):
"""NEWS-11: topic keywords must appear in title or summary."""
self.seed()
res = query_news(self.store, topic="storm")
self.assertEqual([r["title"] for r in res["results"]], ["Nordland storm"])
def test_topic_matches_source_label(self):
"""NEWS-11: topic also matches the source label, so 'Nordland' finds regional items."""
self.store.add_news_items([{"title": "Ferry delayed", "link": "LX", "source": "Nordland", "summary": "boat late"}])
res = query_news(self.store, topic="Nordland")
self.assertTrue(any(r["link"] == "LX" for r in res["results"])) # matched via source, not title/summary
def test_source_filter_and_sanitize(self):
"""NEWS-11: source narrows results; title/summary are sanitized."""
self.seed()
res = query_news(self.store, source="NRK")
self.assertTrue(res["results"] and all(r["source"] == "NRK" for r in res["results"]))
self.assertNotIn("@everyone", res["results"][0]["summary"])
def test_limit_clamped_and_no_store(self):
"""NEWS-11: limit clamps to 1..30; a missing store returns an error."""
self.seed()
self.assertLessEqual(len(query_news(self.store, limit=999)["results"]), 30)
self.assertGreaterEqual(len(query_news(self.store, limit=0)["results"]), 1)
self.assertIn("error", query_news(None))
class TestNewsTool(unittest.IsolatedAsyncioTestCase):
def setUp(self):
self.tmp = tempfile.TemporaryDirectory()
self.addCleanup(self.tmp.cleanup)
def _responder(self, **extra):
cfg = dict(CONFIG, **{"history-directory": self.tmp.name}, **extra)
return OpenAIResponder(cfg, "chat")
def test_tool_offered_needs_flag_and_store(self):
"""NEWS-10: get_news offered only with enable-news-tool AND a store."""
no_store = OpenAIResponder(dict(CONFIG, **{"enable-news-tool": True}), "chat")
self.assertIsNone(no_store.store)
self.assertNotIn("get_news", [f["name"] for f in no_store._available_tools()])
flag_off = self._responder()
self.assertNotIn("get_news", [f["name"] for f in flag_off._available_tools()])
on = self._responder(**{"enable-news-tool": True})
self.assertIn("get_news", [f["name"] for f in on._available_tools()])
self.assertEqual(GET_NEWS_TOOL["name"], "get_news")
async def test_dispatch_caps_news(self):
"""NEWS-12: over news-daily-per-user, get_news refuses without querying."""
responder = self._responder(**{"enable-news-tool": True, "news-daily-per-user": 2})
responder.store.add_news_items([{"title": "x", "link": "l", "source": "s", "summary": "y"}])
for _ in range(2):
self.assertIn("results", await responder._dispatch_tool("get_news", {}, "alice"))
blocked = await responder._dispatch_tool("get_news", {}, "alice")
self.assertIn("error", blocked)