A community based topic aggregation platform built on atproto
bretton.dev
1"""
2Tests for RSS feed fetching functionality.
3"""
4import pytest
5import responses
6from pathlib import Path
7
8from src.rss_fetcher import RSSFetcher
9
10
11@pytest.fixture
12def sample_rss_feed():
13 """Load sample RSS feed from fixtures."""
14 fixture_path = Path(__file__).parent / "fixtures" / "world.xml"
15 # For now, use a minimal test feed
16 return """<?xml version='1.0' encoding='UTF-8'?>
17<rss version="2.0">
18 <channel>
19 <title>Kagi News - World</title>
20 <item>
21 <title>Test Story</title>
22 <link>https://kite.kagi.com/test/world/1</link>
23 <guid>https://kite.kagi.com/test/world/1</guid>
24 <pubDate>Fri, 24 Oct 2025 12:00:00 +0000</pubDate>
25 <category>World</category>
26 </item>
27 </channel>
28</rss>"""
29
30
31class TestRSSFetcher:
32 """Test suite for RSSFetcher."""
33
34 @responses.activate
35 def test_fetch_feed_success(self, sample_rss_feed):
36 """Test successful RSS feed fetch."""
37 url = "https://news.kagi.com/world.xml"
38 responses.add(responses.GET, url, body=sample_rss_feed, status=200)
39
40 fetcher = RSSFetcher()
41 feed = fetcher.fetch_feed(url)
42
43 assert feed is not None
44 assert feed.feed.title == "Kagi News - World"
45 assert len(feed.entries) == 1
46 assert feed.entries[0].title == "Test Story"
47
48 @responses.activate
49 def test_fetch_feed_timeout(self):
50 """Test fetch with timeout."""
51 url = "https://news.kagi.com/world.xml"
52 responses.add(responses.GET, url, body="timeout", status=408)
53
54 fetcher = RSSFetcher(timeout=5)
55
56 with pytest.raises(Exception): # Should raise on timeout
57 fetcher.fetch_feed(url)
58
59 @responses.activate
60 def test_fetch_feed_with_retry(self, sample_rss_feed):
61 """Test fetch with retry on failure then success."""
62 url = "https://news.kagi.com/world.xml"
63
64 # First call fails, second succeeds
65 responses.add(responses.GET, url, body="error", status=500)
66 responses.add(responses.GET, url, body=sample_rss_feed, status=200)
67
68 fetcher = RSSFetcher(max_retries=2)
69 feed = fetcher.fetch_feed(url)
70
71 assert feed is not None
72 assert len(feed.entries) == 1
73
74 @responses.activate
75 def test_fetch_feed_invalid_xml(self):
76 """Test handling of invalid XML."""
77 url = "https://news.kagi.com/world.xml"
78 responses.add(responses.GET, url, body="Not valid XML!", status=200)
79
80 fetcher = RSSFetcher()
81 feed = fetcher.fetch_feed(url)
82
83 # feedparser is lenient, but should have bozo flag set
84 assert feed.bozo == 1 # feedparser uses 1 for True
85
86 def test_fetch_feed_requires_url(self):
87 """Test that fetch_feed requires a URL."""
88 fetcher = RSSFetcher()
89
90 with pytest.raises((ValueError, TypeError)):
91 fetcher.fetch_feed("")