bvandeusen
260103d533
Add _build_briefing_article_context() helper to llm.py that reads rss_item_ids from briefing message metadata and injects article content into the system prompt. Pass conv_id through build_context() and generation_task.py. Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
109 lines
3.4 KiB
Python
109 lines
3.4 KiB
Python
"""Tests for news API — retention constant and endpoint formatting."""
|
|
import pytest
|
|
from unittest.mock import AsyncMock, MagicMock
|
|
from datetime import datetime, timezone
|
|
|
|
|
|
def test_rss_item_max_age_is_90_days():
|
|
"""Retention window should be 90 days."""
|
|
from fabledassistant.services.rss import ITEM_MAX_AGE_DAYS
|
|
|
|
assert ITEM_MAX_AGE_DAYS == 90
|
|
|
|
|
|
def _make_mock_session(rows):
|
|
"""Return a mock async context manager whose session.execute returns rows."""
|
|
mock_result = MagicMock()
|
|
mock_result.mappings.return_value.all.return_value = rows
|
|
|
|
mock_session = AsyncMock()
|
|
mock_session.execute = AsyncMock(return_value=mock_result)
|
|
|
|
mock_cm = MagicMock()
|
|
mock_cm.__aenter__ = AsyncMock(return_value=mock_session)
|
|
mock_cm.__aexit__ = AsyncMock(return_value=False)
|
|
return mock_cm
|
|
|
|
|
|
def test_list_news_item_serialisation():
|
|
"""News item serialisation should truncate content to 300 chars and include reaction."""
|
|
pub = datetime(2026, 3, 28, 8, 0, tzinfo=timezone.utc)
|
|
item = {
|
|
"id": 1,
|
|
"title": "EU AI Act deadline",
|
|
"url": "https://example.com/ai",
|
|
"content": "x" * 500,
|
|
"published_at": pub,
|
|
"topics": ["tech"],
|
|
"feed_title": "TechCrunch",
|
|
"reaction": "up",
|
|
}
|
|
|
|
result = {
|
|
"id": item["id"],
|
|
"title": item["title"],
|
|
"url": item["url"],
|
|
"snippet": (item["content"] or "")[:300],
|
|
"published_at": item["published_at"].isoformat() if item["published_at"] else None,
|
|
"topics": item["topics"] or [],
|
|
"source": item["feed_title"],
|
|
"reaction": item["reaction"],
|
|
}
|
|
|
|
assert result["snippet"] == "x" * 300
|
|
assert result["source"] == "TechCrunch"
|
|
assert result["reaction"] == "up"
|
|
assert result["published_at"] == pub.isoformat()
|
|
|
|
|
|
|
|
def test_build_briefing_article_context_metadata_extraction():
|
|
"""Verify the rss_item_ids extraction logic from message metadata."""
|
|
import json
|
|
|
|
# Simulates the logic in _build_briefing_article_context
|
|
def extract_ids(msg_metadata):
|
|
meta = msg_metadata or {}
|
|
if isinstance(meta, str):
|
|
try:
|
|
meta = json.loads(meta)
|
|
except Exception:
|
|
return []
|
|
return meta.get("rss_item_ids") or []
|
|
|
|
assert extract_ids({}) == []
|
|
assert extract_ids(None) == []
|
|
assert extract_ids({"rss_item_ids": [1, 2, 3]}) == [1, 2, 3]
|
|
assert extract_ids(json.dumps({"rss_item_ids": [4, 5]})) == [4, 5]
|
|
assert extract_ids("not json") == []
|
|
|
|
|
|
def test_list_news_null_content_serialisation():
|
|
"""Null content should produce empty snippet without error."""
|
|
item = {
|
|
"id": 2,
|
|
"title": "No content article",
|
|
"url": "https://example.com/b",
|
|
"content": None,
|
|
"published_at": None,
|
|
"topics": None,
|
|
"feed_title": "Hacker News",
|
|
"reaction": None,
|
|
}
|
|
|
|
result = {
|
|
"id": item["id"],
|
|
"title": item["title"],
|
|
"url": item["url"],
|
|
"snippet": (item["content"] or "")[:300],
|
|
"published_at": item["published_at"].isoformat() if item["published_at"] else None,
|
|
"topics": item["topics"] or [],
|
|
"source": item["feed_title"],
|
|
"reaction": item["reaction"],
|
|
}
|
|
|
|
assert result["snippet"] == ""
|
|
assert result["topics"] == []
|
|
assert result["published_at"] is None
|
|
assert result["reaction"] is None
|