131 lines
4.9 KiB
Python
131 lines
4.9 KiB
Python
import pytest
|
|
from unittest.mock import patch, MagicMock
|
|
|
|
|
|
def test_extract_item_fields():
|
|
"""extract_item() should pull title, link, id, summary from a feedparser entry."""
|
|
from fabledassistant.services.rss import extract_item
|
|
entry = MagicMock()
|
|
entry.get = lambda k, d="": {"title": "Test Post", "link": "https://example.com/1",
|
|
"id": "guid-1", "summary": "A summary"}.get(k, d)
|
|
entry.published_parsed = None
|
|
item = extract_item(entry)
|
|
assert item["title"] == "Test Post"
|
|
assert item["url"] == "https://example.com/1"
|
|
assert item["guid"] == "guid-1"
|
|
assert item["content"] == "A summary"
|
|
|
|
|
|
def test_extract_item_truncates_content():
|
|
"""extract_item() should truncate content to 2000 chars."""
|
|
from fabledassistant.services.rss import extract_item
|
|
entry = MagicMock()
|
|
entry.get = lambda k, d="": {"summary": "x" * 3000, "title": "", "link": "", "id": "g"}.get(k, d)
|
|
entry.published_parsed = None
|
|
item = extract_item(entry)
|
|
assert len(item["content"]) == 2000
|
|
|
|
|
|
def test_extract_item_prefers_content_over_summary():
|
|
"""extract_item() should prefer 'content' field over 'summary' when present."""
|
|
from fabledassistant.services.rss import extract_item
|
|
entry = MagicMock()
|
|
content_obj = MagicMock()
|
|
content_obj.value = "Full content here"
|
|
entry.get = lambda k, d="": {
|
|
"title": "T", "link": "http://x.com", "id": "g",
|
|
"summary": "Short summary",
|
|
}.get(k, d)
|
|
entry.content = [content_obj]
|
|
entry.published_parsed = None
|
|
item = extract_item(entry)
|
|
assert item["content"] == "Full content here"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_classify_items_batch_returns_topic_map():
|
|
"""classify_items_batch should return a dict mapping item_id to topic list."""
|
|
from unittest.mock import AsyncMock
|
|
from fabledassistant.services.rss_classifier import classify_items_batch
|
|
|
|
fake_response = '{"1": ["technology", "ai"], "2": ["politics"]}'
|
|
|
|
with patch(
|
|
"fabledassistant.services.rss_classifier._llm_classify",
|
|
new_callable=AsyncMock,
|
|
return_value=fake_response,
|
|
):
|
|
items = [
|
|
{"id": 1, "title": "OpenAI releases GPT-5", "content": "..."},
|
|
{"id": 2, "title": "EU passes new law", "content": "..."},
|
|
]
|
|
result = await classify_items_batch(items, user_include_topics=[])
|
|
assert result[1] == ["technology", "ai"]
|
|
assert result[2] == ["politics"]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_classify_items_batch_handles_llm_failure():
|
|
"""classify_items_batch should return empty dict on LLM error."""
|
|
from unittest.mock import AsyncMock
|
|
from fabledassistant.services.rss_classifier import classify_items_batch
|
|
|
|
with patch(
|
|
"fabledassistant.services.rss_classifier._llm_classify",
|
|
new_callable=AsyncMock,
|
|
side_effect=Exception("LLM unavailable"),
|
|
):
|
|
items = [{"id": 5, "title": "Some news", "content": ""}]
|
|
result = await classify_items_batch(items, user_include_topics=[])
|
|
assert result == {}
|
|
|
|
|
|
def test_score_rss_items_excludes_blacklisted_topics():
|
|
"""Items with excluded topics should be removed."""
|
|
from fabledassistant.services.briefing_preferences import score_and_filter_items
|
|
|
|
items = [
|
|
{"id": 1, "title": "Tech news", "topics": ["technology"], "published_at": "2026-03-25T08:00:00"},
|
|
{"id": 2, "title": "Sports score", "topics": ["sports"], "published_at": "2026-03-25T08:00:00"},
|
|
]
|
|
result = score_and_filter_items(
|
|
items,
|
|
include_topics=["technology"],
|
|
exclude_topics=["sports"],
|
|
topic_scores={},
|
|
max_items=10,
|
|
)
|
|
ids = [r["id"] for r in result]
|
|
assert 1 in ids
|
|
assert 2 not in ids
|
|
|
|
|
|
def test_score_rss_items_boosts_included_topics():
|
|
"""Items matching include_topics should rank higher than neutral items."""
|
|
from fabledassistant.services.briefing_preferences import score_and_filter_items
|
|
|
|
items = [
|
|
{"id": 1, "title": "Random news", "topics": ["other"], "published_at": "2026-03-25T07:00:00"},
|
|
{"id": 2, "title": "Tech news", "topics": ["technology"], "published_at": "2026-03-25T06:00:00"},
|
|
]
|
|
result = score_and_filter_items(
|
|
items,
|
|
include_topics=["technology"],
|
|
exclude_topics=[],
|
|
topic_scores={},
|
|
max_items=10,
|
|
)
|
|
assert result[0]["id"] == 2
|
|
|
|
|
|
def test_score_rss_items_no_preferences_returns_all():
|
|
"""With no preferences, all items should be returned sorted by recency."""
|
|
from fabledassistant.services.briefing_preferences import score_and_filter_items
|
|
|
|
items = [
|
|
{"id": 1, "title": "A", "topics": [], "published_at": "2026-03-24T10:00:00"},
|
|
{"id": 2, "title": "B", "topics": [], "published_at": "2026-03-25T10:00:00"},
|
|
]
|
|
result = score_and_filter_items(items, [], [], {}, max_items=10)
|
|
assert result[0]["id"] == 2 # Newer first
|