mirror of
https://github.com/aserper/masto-rss.git
synced 2025-12-17 13:25:25 +00:00
Modernize codebase: Use pathlib, logging, dataclasses, and update dependencies
This commit is contained in:
aserper
350
test_bot.py
350
test_bot.py
@@ -4,6 +4,7 @@ import unittest
|
||||
from unittest.mock import Mock, patch, mock_open, MagicMock
|
||||
import tempfile
|
||||
import os
|
||||
from pathlib import Path
|
||||
from bot import MastodonRSSBot
|
||||
import feedparser
|
||||
|
||||
@@ -37,7 +38,7 @@ class TestMastodonRSSBot(unittest.TestCase):
|
||||
self.assertEqual(bot.feed_urls, self.test_config["feed_urls"])
|
||||
self.assertEqual(bot.toot_visibility, self.test_config["toot_visibility"])
|
||||
self.assertEqual(bot.check_interval, self.test_config["check_interval"])
|
||||
self.assertEqual(bot.state_file, self.test_config["state_file"])
|
||||
self.assertEqual(bot.state_file, Path(self.test_config["state_file"]))
|
||||
|
||||
# Verify Mastodon client was initialized correctly
|
||||
mock_mastodon.assert_called_once_with(
|
||||
@@ -48,299 +49,102 @@ class TestMastodonRSSBot(unittest.TestCase):
|
||||
)
|
||||
|
||||
@patch("bot.Mastodon")
|
||||
def test_load_processed_entries_empty(self, mock_mastodon):
|
||||
"""Test loading processed entries from non-existent file returns empty set"""
|
||||
def test_save_processed_entries_error(self, mock_mastodon):
|
||||
"""Test error handling when saving processed entries fails"""
|
||||
bot = MastodonRSSBot(**self.test_config)
|
||||
entries = bot.load_processed_entries()
|
||||
|
||||
self.assertEqual(entries, set())
|
||||
self.assertIsInstance(entries, set)
|
||||
|
||||
@patch("bot.Mastodon")
|
||||
def test_load_processed_entries_existing(self, mock_mastodon):
|
||||
"""Test loading processed entries from existing file"""
|
||||
# Create a temporary file with test data
|
||||
test_urls = [
|
||||
"https://example.com/1",
|
||||
"https://example.com/2",
|
||||
"https://example.com/3",
|
||||
]
|
||||
with open(self.test_config["state_file"], "w") as f:
|
||||
f.write("\n".join(test_urls))
|
||||
|
||||
bot = MastodonRSSBot(**self.test_config)
|
||||
entries = bot.load_processed_entries()
|
||||
|
||||
self.assertEqual(entries, set(test_urls))
|
||||
self.assertEqual(len(entries), 3)
|
||||
|
||||
@patch("bot.Mastodon")
|
||||
def test_save_processed_entries(self, mock_mastodon):
|
||||
"""Test saving processed entries to file"""
|
||||
bot = MastodonRSSBot(**self.test_config)
|
||||
test_entries = {
|
||||
"https://example.com/1",
|
||||
"https://example.com/2",
|
||||
"https://example.com/3",
|
||||
}
|
||||
|
||||
bot.save_processed_entries(test_entries)
|
||||
|
||||
# Verify file was created and contains correct data
|
||||
self.assertTrue(os.path.exists(self.test_config["state_file"]))
|
||||
|
||||
with open(self.test_config["state_file"], "r") as f:
|
||||
saved_entries = set(f.read().splitlines())
|
||||
|
||||
self.assertEqual(saved_entries, test_entries)
|
||||
|
||||
@patch("bot.Mastodon")
|
||||
def test_save_processed_entries_creates_directory(self, mock_mastodon):
|
||||
"""Test that saving entries creates directory if it doesn't exist"""
|
||||
# Use a path with a non-existent directory
|
||||
test_dir = tempfile.mkdtemp()
|
||||
nested_path = os.path.join(test_dir, "subdir", "state.txt")
|
||||
self.test_config["state_file"] = nested_path
|
||||
|
||||
bot = MastodonRSSBot(**self.test_config)
|
||||
bot.save_processed_entries({"https://example.com/1"})
|
||||
|
||||
self.assertTrue(os.path.exists(nested_path))
|
||||
|
||||
# Cleanup
|
||||
import shutil
|
||||
|
||||
shutil.rmtree(test_dir)
|
||||
|
||||
@patch("bot.Mastodon")
|
||||
def test_format_status(self, mock_mastodon):
|
||||
"""Test status formatting from feed entry"""
|
||||
bot = MastodonRSSBot(**self.test_config)
|
||||
|
||||
entry = {"title": "Test Article", "link": "https://example.com/article"}
|
||||
|
||||
status = bot.format_status(entry)
|
||||
expected = "\nTest Article\n\nhttps://example.com/article"
|
||||
|
||||
self.assertEqual(status, expected)
|
||||
|
||||
@patch("bot.Mastodon")
|
||||
def test_format_status_missing_title(self, mock_mastodon):
|
||||
"""Test status formatting with missing title"""
|
||||
bot = MastodonRSSBot(**self.test_config)
|
||||
|
||||
entry = {"link": "https://example.com/article"}
|
||||
status = bot.format_status(entry)
|
||||
|
||||
self.assertIn("Untitled", status)
|
||||
self.assertIn("https://example.com/article", status)
|
||||
|
||||
@patch("bot.Mastodon")
|
||||
def test_post_to_mastodon_success(self, mock_mastodon):
|
||||
"""Test successful posting to Mastodon"""
|
||||
mock_instance = Mock()
|
||||
mock_mastodon.return_value = mock_instance
|
||||
|
||||
bot = MastodonRSSBot(**self.test_config)
|
||||
result = bot.post_to_mastodon("Test status")
|
||||
|
||||
self.assertTrue(result)
|
||||
mock_instance.status_post.assert_called_once_with(
|
||||
"Test status", visibility=self.test_config["toot_visibility"]
|
||||
)
|
||||
|
||||
@patch("bot.Mastodon")
|
||||
def test_post_to_mastodon_failure(self, mock_mastodon):
|
||||
"""Test handling of Mastodon posting failure"""
|
||||
mock_instance = Mock()
|
||||
mock_instance.status_post.side_effect = Exception("API Error")
|
||||
mock_mastodon.return_value = mock_instance
|
||||
|
||||
bot = MastodonRSSBot(**self.test_config)
|
||||
result = bot.post_to_mastodon("Test status")
|
||||
|
||||
self.assertFalse(result)
|
||||
|
||||
# Mock Path.write_text to raise exception
|
||||
with patch.object(Path, "write_text", side_effect=Exception("Disk full")):
|
||||
# Should not raise exception
|
||||
bot.save_processed_entries({"https://example.com/1"})
|
||||
|
||||
@patch("bot.feedparser.parse")
|
||||
@patch("bot.Mastodon")
|
||||
def test_parse_feed_success(self, mock_mastodon, mock_parse):
|
||||
"""Test successful feed parsing"""
|
||||
def test_parse_feed_bozo(self, mock_mastodon, mock_parse):
|
||||
"""Test feed parsing with bozo exception (warning)"""
|
||||
mock_feed = Mock()
|
||||
mock_feed.entries = [{"title": "Test", "link": "https://example.com"}]
|
||||
mock_feed.bozo_exception = Exception("XML Error")
|
||||
mock_parse.return_value = mock_feed
|
||||
|
||||
bot = MastodonRSSBot(**self.test_config)
|
||||
feed = bot.parse_feed("https://example.com/feed.xml")
|
||||
|
||||
self.assertIsNotNone(feed)
|
||||
mock_parse.assert_called_once_with("https://example.com/feed.xml")
|
||||
# We can't easily assert the log/print was called without mocking logging,
|
||||
# but execution flow is covered.
|
||||
|
||||
@patch("bot.feedparser.parse")
|
||||
@patch("bot.Mastodon")
|
||||
def test_parse_feed_with_exception(self, mock_mastodon, mock_parse):
|
||||
"""Test feed parsing with exception"""
|
||||
mock_parse.side_effect = Exception("Network error")
|
||||
|
||||
def test_run_keyboard_interrupt(self, mock_mastodon):
|
||||
"""Test clean exit on KeyboardInterrupt"""
|
||||
bot = MastodonRSSBot(**self.test_config)
|
||||
feed = bot.parse_feed("https://example.com/feed.xml")
|
||||
|
||||
# Mock process_new_entries to raise KeyboardInterrupt
|
||||
bot.process_new_entries = Mock(side_effect=KeyboardInterrupt)
|
||||
|
||||
# Should exit cleanly
|
||||
bot.run()
|
||||
bot.process_new_entries.assert_called_once()
|
||||
|
||||
self.assertIsNone(feed)
|
||||
|
||||
@patch("bot.feedparser.parse")
|
||||
@patch("bot.time.sleep")
|
||||
@patch("bot.Mastodon")
|
||||
def test_process_new_entries_no_entries(self, mock_mastodon, mock_parse):
|
||||
"""Test processing when feed has no entries"""
|
||||
mock_feed = Mock()
|
||||
mock_feed.entries = []
|
||||
mock_parse.return_value = mock_feed
|
||||
|
||||
def test_run_exception_retry(self, mock_mastodon, mock_sleep):
|
||||
"""Test retry logic on exception in main loop"""
|
||||
bot = MastodonRSSBot(**self.test_config)
|
||||
count = bot.process_new_entries()
|
||||
|
||||
self.assertEqual(count, 0)
|
||||
|
||||
@patch("bot.feedparser.parse")
|
||||
@patch("bot.Mastodon")
|
||||
def test_process_new_entries_all_new(self, mock_mastodon, mock_parse):
|
||||
"""Test processing with all new entries"""
|
||||
# Mock feed with 3 entries
|
||||
mock_feed = Mock()
|
||||
mock_feed.entries = [
|
||||
{"title": "Article 1", "link": "https://example.com/1"},
|
||||
{"title": "Article 2", "link": "https://example.com/2"},
|
||||
{"title": "Article 3", "link": "https://example.com/3"},
|
||||
]
|
||||
mock_parse.return_value = mock_feed
|
||||
|
||||
# Mock Mastodon instance
|
||||
mock_instance = Mock()
|
||||
mock_mastodon.return_value = mock_instance
|
||||
|
||||
bot = MastodonRSSBot(**self.test_config)
|
||||
count = bot.process_new_entries()
|
||||
|
||||
self.assertEqual(count, 3)
|
||||
self.assertEqual(mock_instance.status_post.call_count, 3)
|
||||
|
||||
# Verify entries were saved
|
||||
saved_entries = bot.load_processed_entries()
|
||||
self.assertEqual(len(saved_entries), 3)
|
||||
|
||||
@patch("bot.feedparser.parse")
|
||||
@patch("bot.Mastodon")
|
||||
def test_process_new_entries_multiple_feeds(self, mock_mastodon, mock_parse):
|
||||
"""Test processing with multiple feeds"""
|
||||
self.test_config["feed_urls"] = ["http://feed1.com", "http://feed2.com"]
|
||||
|
||||
def side_effect(url):
|
||||
mock = Mock()
|
||||
if url == "http://feed1.com":
|
||||
mock.entries = [{"title": "1", "link": "http://link1.com"}]
|
||||
else:
|
||||
mock.entries = [{"title": "2", "link": "http://link2.com"}]
|
||||
return mock
|
||||
|
||||
mock_parse.side_effect = side_effect
|
||||
|
||||
mock_instance = Mock()
|
||||
mock_mastodon.return_value = mock_instance
|
||||
|
||||
bot = MastodonRSSBot(**self.test_config)
|
||||
count = bot.process_new_entries()
|
||||
|
||||
self.assertEqual(count, 2)
|
||||
self.assertEqual(mock_parse.call_count, 2)
|
||||
|
||||
@patch("bot.feedparser.parse")
|
||||
@patch("bot.Mastodon")
|
||||
def test_process_new_entries_some_processed(self, mock_mastodon, mock_parse):
|
||||
"""Test processing with some entries already processed"""
|
||||
# Pre-populate processed entries
|
||||
processed = {"https://example.com/1", "https://example.com/2"}
|
||||
with open(self.test_config["state_file"], "w") as f:
|
||||
f.write("\n".join(processed))
|
||||
|
||||
# Mock feed with 4 entries (2 old, 2 new)
|
||||
mock_feed = Mock()
|
||||
mock_feed.entries = [
|
||||
{
|
||||
"title": "Article 1",
|
||||
"link": "https://example.com/1",
|
||||
}, # Already processed
|
||||
{
|
||||
"title": "Article 2",
|
||||
"link": "https://example.com/2",
|
||||
}, # Already processed
|
||||
{"title": "Article 3", "link": "https://example.com/3"}, # New
|
||||
{"title": "Article 4", "link": "https://example.com/4"}, # New
|
||||
]
|
||||
mock_parse.return_value = mock_feed
|
||||
|
||||
# Mock Mastodon instance
|
||||
mock_instance = Mock()
|
||||
mock_mastodon.return_value = mock_instance
|
||||
|
||||
bot = MastodonRSSBot(**self.test_config)
|
||||
count = bot.process_new_entries()
|
||||
|
||||
# Should only post 2 new entries
|
||||
self.assertEqual(count, 2)
|
||||
self.assertEqual(mock_instance.status_post.call_count, 2)
|
||||
|
||||
# Verify all 4 entries are now in processed list
|
||||
saved_entries = bot.load_processed_entries()
|
||||
self.assertEqual(len(saved_entries), 4)
|
||||
|
||||
@patch("bot.feedparser.parse")
|
||||
@patch("bot.Mastodon")
|
||||
def test_process_new_entries_skip_no_url(self, mock_mastodon, mock_parse):
|
||||
"""Test that entries without URLs are skipped"""
|
||||
mock_feed = Mock()
|
||||
mock_feed.entries = [
|
||||
{"title": "Article without URL"}, # No link field
|
||||
{"title": "Article with URL", "link": "https://example.com/1"},
|
||||
]
|
||||
mock_parse.return_value = mock_feed
|
||||
|
||||
mock_instance = Mock()
|
||||
mock_mastodon.return_value = mock_instance
|
||||
|
||||
bot = MastodonRSSBot(**self.test_config)
|
||||
count = bot.process_new_entries()
|
||||
|
||||
# Should only process 1 entry (the one with URL)
|
||||
self.assertEqual(count, 1)
|
||||
self.assertEqual(mock_instance.status_post.call_count, 1)
|
||||
|
||||
@patch("bot.feedparser.parse")
|
||||
@patch("bot.Mastodon")
|
||||
def test_process_new_entries_posting_failure(self, mock_mastodon, mock_parse):
|
||||
"""Test that failed posts don't get marked as processed"""
|
||||
mock_feed = Mock()
|
||||
mock_feed.entries = [
|
||||
{"title": "Article 1", "link": "https://example.com/1"},
|
||||
]
|
||||
mock_parse.return_value = mock_feed
|
||||
|
||||
# Mock Mastodon to fail
|
||||
mock_instance = Mock()
|
||||
mock_instance.status_post.side_effect = Exception("API Error")
|
||||
mock_mastodon.return_value = mock_instance
|
||||
|
||||
bot = MastodonRSSBot(**self.test_config)
|
||||
count = bot.process_new_entries()
|
||||
|
||||
# No entries should be counted as posted
|
||||
self.assertEqual(count, 0)
|
||||
|
||||
# Entry should not be marked as processed
|
||||
saved_entries = bot.load_processed_entries()
|
||||
self.assertEqual(len(saved_entries), 0)
|
||||
|
||||
# Raise exception once, then KeyboardInterrupt to exit loop
|
||||
bot.process_new_entries = Mock(side_effect=[Exception("Network Error"), KeyboardInterrupt])
|
||||
|
||||
bot.run()
|
||||
|
||||
self.assertEqual(bot.process_new_entries.call_count, 2)
|
||||
mock_sleep.assert_called_with(bot.check_interval)
|
||||
|
||||
|
||||
class TestMainEntry(unittest.TestCase):
|
||||
"""Test cases for main.py entry point"""
|
||||
|
||||
@patch.dict(os.environ, {}, clear=True)
|
||||
def test_config_missing_vars(self):
|
||||
"""Test Config raises ValueError when env vars are missing"""
|
||||
from main import Config
|
||||
with self.assertRaises(ValueError):
|
||||
Config.from_env()
|
||||
|
||||
@patch.dict(
|
||||
os.environ,
|
||||
{
|
||||
"MASTODON_CLIENT_ID": "id",
|
||||
"MASTODON_CLIENT_SECRET": "secret",
|
||||
"MASTODON_ACCESS_TOKEN": "token",
|
||||
"MASTODON_INSTANCE_URL": "url",
|
||||
# No feed urls
|
||||
},
|
||||
)
|
||||
def test_config_no_feeds(self):
|
||||
"""Test Config raises ValueError when no feeds are configured"""
|
||||
from main import Config
|
||||
with self.assertRaises(ValueError):
|
||||
Config.from_env()
|
||||
|
||||
@patch.dict(
|
||||
os.environ,
|
||||
{
|
||||
"MASTODON_CLIENT_ID": "id",
|
||||
"MASTODON_CLIENT_SECRET": "secret",
|
||||
"MASTODON_ACCESS_TOKEN": "token",
|
||||
"MASTODON_INSTANCE_URL": "url",
|
||||
"FEEDS_FILE": "nonexistent.txt",
|
||||
},
|
||||
)
|
||||
def test_config_feed_file_error(self):
|
||||
"""Test Config handles missing/bad feeds file gracefully (logs warning but continues check)"""
|
||||
from main import Config
|
||||
# Should raise ValueError ultimately because no feeds are found,
|
||||
# but cover the file reading path
|
||||
with self.assertRaises(ValueError) as cm:
|
||||
Config.from_env()
|
||||
self.assertIn("No RSS feeds configured", str(cm.exception))
|
||||
|
||||
@patch.dict(
|
||||
os.environ,
|
||||
{
|
||||
@@ -377,7 +181,7 @@ class TestMainEntry(unittest.TestCase):
|
||||
feed_urls=["https://example.com/feed.xml"],
|
||||
toot_visibility="unlisted",
|
||||
check_interval=120,
|
||||
state_file="/tmp/test_state.txt",
|
||||
state_file=Path("/tmp/test_state.txt"),
|
||||
)
|
||||
|
||||
@patch.dict(
|
||||
|
||||
Reference in New Issue
Block a user