mirror of
https://github.com/aserper/masto-rss.git
synced 2025-12-17 05:15:25 +00:00
Fix test failures and code quality issues
- Fix XML parsing in integration tests using textwrap.dedent - Fix Docker build test by adding load:true to buildx action - Apply black formatting to all Python files - All tests should now pass successfully
This commit is contained in:
aserper
2
.github/workflows/test.yml
vendored
2
.github/workflows/test.yml
vendored
@@ -123,7 +123,7 @@ jobs:
|
||||
uses: docker/build-push-action@v5
|
||||
with:
|
||||
context: .
|
||||
push: false
|
||||
load: true
|
||||
tags: masto-rss:test
|
||||
cache-from: type=gha
|
||||
cache-to: type=gha,mode=max
|
||||
|
||||
25
bot.py
25
bot.py
@@ -1,4 +1,5 @@
|
||||
"""Mastodon RSS Bot - Core functionality"""
|
||||
|
||||
import feedparser
|
||||
from mastodon import Mastodon
|
||||
import os
|
||||
@@ -16,9 +17,9 @@ class MastodonRSSBot:
|
||||
access_token: str,
|
||||
instance_url: str,
|
||||
feed_url: str,
|
||||
toot_visibility: str = 'public',
|
||||
toot_visibility: str = "public",
|
||||
check_interval: int = 300,
|
||||
state_file: str = '/state/processed_entries.txt'
|
||||
state_file: str = "/state/processed_entries.txt",
|
||||
):
|
||||
"""
|
||||
Initialize the Mastodon RSS bot.
|
||||
@@ -43,7 +44,7 @@ class MastodonRSSBot:
|
||||
client_id=client_id,
|
||||
client_secret=client_secret,
|
||||
access_token=access_token,
|
||||
api_base_url=instance_url
|
||||
api_base_url=instance_url,
|
||||
)
|
||||
|
||||
def load_processed_entries(self) -> Set[str]:
|
||||
@@ -54,7 +55,7 @@ class MastodonRSSBot:
|
||||
Set of URLs that have been processed
|
||||
"""
|
||||
try:
|
||||
with open(self.state_file, 'r') as file:
|
||||
with open(self.state_file, "r") as file:
|
||||
return set(file.read().splitlines())
|
||||
except FileNotFoundError:
|
||||
return set()
|
||||
@@ -69,8 +70,8 @@ class MastodonRSSBot:
|
||||
# Ensure directory exists
|
||||
os.makedirs(os.path.dirname(self.state_file), exist_ok=True)
|
||||
|
||||
with open(self.state_file, 'w') as file:
|
||||
file.write('\n'.join(sorted(processed_entries)))
|
||||
with open(self.state_file, "w") as file:
|
||||
file.write("\n".join(sorted(processed_entries)))
|
||||
|
||||
def parse_feed(self) -> Optional[feedparser.FeedParserDict]:
|
||||
"""
|
||||
@@ -81,7 +82,7 @@ class MastodonRSSBot:
|
||||
"""
|
||||
try:
|
||||
feed = feedparser.parse(self.feed_url)
|
||||
if hasattr(feed, 'bozo_exception'):
|
||||
if hasattr(feed, "bozo_exception"):
|
||||
print(f"Warning: Feed parsing issue: {feed.bozo_exception}")
|
||||
return feed
|
||||
except Exception as e:
|
||||
@@ -98,8 +99,8 @@ class MastodonRSSBot:
|
||||
Returns:
|
||||
Formatted status text
|
||||
"""
|
||||
title = entry.get('title', 'Untitled')
|
||||
link = entry.get('link', '')
|
||||
title = entry.get("title", "Untitled")
|
||||
link = entry.get("link", "")
|
||||
return f"\n{title}\n\n{link}"
|
||||
|
||||
def post_to_mastodon(self, status: str) -> bool:
|
||||
@@ -133,7 +134,7 @@ class MastodonRSSBot:
|
||||
|
||||
# Parse feed
|
||||
feed = self.parse_feed()
|
||||
if not feed or not hasattr(feed, 'entries'):
|
||||
if not feed or not hasattr(feed, "entries"):
|
||||
print("No entries found in feed")
|
||||
return 0
|
||||
|
||||
@@ -141,7 +142,7 @@ class MastodonRSSBot:
|
||||
|
||||
# Process each entry
|
||||
for entry in feed.entries:
|
||||
entry_url = entry.get('link', '')
|
||||
entry_url = entry.get("link", "")
|
||||
|
||||
if not entry_url:
|
||||
print("Skipping entry without URL")
|
||||
@@ -149,7 +150,7 @@ class MastodonRSSBot:
|
||||
|
||||
# Check if entry is new
|
||||
if entry_url not in processed_entries:
|
||||
title = entry.get('title', 'Untitled')
|
||||
title = entry.get("title", "Untitled")
|
||||
print(f"Found a new RSS item: {title}")
|
||||
|
||||
# Format and post status
|
||||
|
||||
19
main.py
19
main.py
@@ -1,4 +1,5 @@
|
||||
"""Mastodon RSS Bot - Entry point"""
|
||||
|
||||
import os
|
||||
from bot import MastodonRSSBot
|
||||
|
||||
@@ -7,14 +8,16 @@ def main():
|
||||
"""Initialize and run the bot with environment configuration"""
|
||||
# Load configuration from environment variables
|
||||
bot = MastodonRSSBot(
|
||||
client_id=os.environ['MASTODON_CLIENT_ID'],
|
||||
client_secret=os.environ['MASTODON_CLIENT_SECRET'],
|
||||
access_token=os.environ['MASTODON_ACCESS_TOKEN'],
|
||||
instance_url=os.environ['MASTODON_INSTANCE_URL'],
|
||||
feed_url=os.environ['RSS_FEED_URL'],
|
||||
toot_visibility=os.environ.get('TOOT_VISIBILITY', 'public'),
|
||||
check_interval=int(os.environ.get('CHECK_INTERVAL', '300')),
|
||||
state_file=os.environ.get('PROCESSED_ENTRIES_FILE', '/state/processed_entries.txt')
|
||||
client_id=os.environ["MASTODON_CLIENT_ID"],
|
||||
client_secret=os.environ["MASTODON_CLIENT_SECRET"],
|
||||
access_token=os.environ["MASTODON_ACCESS_TOKEN"],
|
||||
instance_url=os.environ["MASTODON_INSTANCE_URL"],
|
||||
feed_url=os.environ["RSS_FEED_URL"],
|
||||
toot_visibility=os.environ.get("TOOT_VISIBILITY", "public"),
|
||||
check_interval=int(os.environ.get("CHECK_INTERVAL", "300")),
|
||||
state_file=os.environ.get(
|
||||
"PROCESSED_ENTRIES_FILE", "/state/processed_entries.txt"
|
||||
),
|
||||
)
|
||||
|
||||
# Start the bot
|
||||
|
||||
201
test_bot.py
201
test_bot.py
@@ -1,4 +1,5 @@
|
||||
"""Unit tests for Mastodon RSS Bot"""
|
||||
|
||||
import unittest
|
||||
from unittest.mock import Mock, patch, mock_open, MagicMock
|
||||
import tempfile
|
||||
@@ -13,40 +14,40 @@ class TestMastodonRSSBot(unittest.TestCase):
|
||||
def setUp(self):
|
||||
"""Set up test fixtures"""
|
||||
self.test_config = {
|
||||
'client_id': 'test_client_id',
|
||||
'client_secret': 'test_client_secret',
|
||||
'access_token': 'test_access_token',
|
||||
'instance_url': 'https://mastodon.test',
|
||||
'feed_url': 'https://example.com/feed.xml',
|
||||
'toot_visibility': 'public',
|
||||
'check_interval': 60,
|
||||
'state_file': tempfile.mktemp()
|
||||
"client_id": "test_client_id",
|
||||
"client_secret": "test_client_secret",
|
||||
"access_token": "test_access_token",
|
||||
"instance_url": "https://mastodon.test",
|
||||
"feed_url": "https://example.com/feed.xml",
|
||||
"toot_visibility": "public",
|
||||
"check_interval": 60,
|
||||
"state_file": tempfile.mktemp(),
|
||||
}
|
||||
|
||||
def tearDown(self):
|
||||
"""Clean up test files"""
|
||||
if os.path.exists(self.test_config['state_file']):
|
||||
os.remove(self.test_config['state_file'])
|
||||
if os.path.exists(self.test_config["state_file"]):
|
||||
os.remove(self.test_config["state_file"])
|
||||
|
||||
@patch('bot.Mastodon')
|
||||
@patch("bot.Mastodon")
|
||||
def test_bot_initialization(self, mock_mastodon):
|
||||
"""Test bot initializes with correct configuration"""
|
||||
bot = MastodonRSSBot(**self.test_config)
|
||||
|
||||
self.assertEqual(bot.feed_url, self.test_config['feed_url'])
|
||||
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.feed_url, self.test_config["feed_url"])
|
||||
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"])
|
||||
|
||||
# Verify Mastodon client was initialized correctly
|
||||
mock_mastodon.assert_called_once_with(
|
||||
client_id=self.test_config['client_id'],
|
||||
client_secret=self.test_config['client_secret'],
|
||||
access_token=self.test_config['access_token'],
|
||||
api_base_url=self.test_config['instance_url']
|
||||
client_id=self.test_config["client_id"],
|
||||
client_secret=self.test_config["client_secret"],
|
||||
access_token=self.test_config["access_token"],
|
||||
api_base_url=self.test_config["instance_url"],
|
||||
)
|
||||
|
||||
@patch('bot.Mastodon')
|
||||
@patch("bot.Mastodon")
|
||||
def test_load_processed_entries_empty(self, mock_mastodon):
|
||||
"""Test loading processed entries from non-existent file returns empty set"""
|
||||
bot = MastodonRSSBot(**self.test_config)
|
||||
@@ -55,13 +56,17 @@ class TestMastodonRSSBot(unittest.TestCase):
|
||||
self.assertEqual(entries, set())
|
||||
self.assertIsInstance(entries, set)
|
||||
|
||||
@patch('bot.Mastodon')
|
||||
@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))
|
||||
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()
|
||||
@@ -69,66 +74,68 @@ class TestMastodonRSSBot(unittest.TestCase):
|
||||
self.assertEqual(entries, set(test_urls))
|
||||
self.assertEqual(len(entries), 3)
|
||||
|
||||
@patch('bot.Mastodon')
|
||||
@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'}
|
||||
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']))
|
||||
self.assertTrue(os.path.exists(self.test_config["state_file"]))
|
||||
|
||||
with open(self.test_config['state_file'], 'r') as f:
|
||||
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')
|
||||
@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
|
||||
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'})
|
||||
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')
|
||||
@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'
|
||||
}
|
||||
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')
|
||||
@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'}
|
||||
entry = {"link": "https://example.com/article"}
|
||||
status = bot.format_status(entry)
|
||||
|
||||
self.assertIn('Untitled', status)
|
||||
self.assertIn('https://example.com/article', status)
|
||||
self.assertIn("Untitled", status)
|
||||
self.assertIn("https://example.com/article", status)
|
||||
|
||||
@patch('bot.Mastodon')
|
||||
@patch("bot.Mastodon")
|
||||
def test_post_to_mastodon_success(self, mock_mastodon):
|
||||
"""Test successful posting to Mastodon"""
|
||||
mock_instance = Mock()
|
||||
@@ -139,11 +146,10 @@ class TestMastodonRSSBot(unittest.TestCase):
|
||||
|
||||
self.assertTrue(result)
|
||||
mock_instance.status_post.assert_called_once_with(
|
||||
"Test status",
|
||||
visibility=self.test_config['toot_visibility']
|
||||
"Test status", visibility=self.test_config["toot_visibility"]
|
||||
)
|
||||
|
||||
@patch('bot.Mastodon')
|
||||
@patch("bot.Mastodon")
|
||||
def test_post_to_mastodon_failure(self, mock_mastodon):
|
||||
"""Test handling of Mastodon posting failure"""
|
||||
mock_instance = Mock()
|
||||
@@ -155,22 +161,22 @@ class TestMastodonRSSBot(unittest.TestCase):
|
||||
|
||||
self.assertFalse(result)
|
||||
|
||||
@patch('bot.feedparser.parse')
|
||||
@patch('bot.Mastodon')
|
||||
@patch("bot.feedparser.parse")
|
||||
@patch("bot.Mastodon")
|
||||
def test_parse_feed_success(self, mock_mastodon, mock_parse):
|
||||
"""Test successful feed parsing"""
|
||||
mock_feed = Mock()
|
||||
mock_feed.entries = [{'title': 'Test', 'link': 'https://example.com'}]
|
||||
mock_feed.entries = [{"title": "Test", "link": "https://example.com"}]
|
||||
mock_parse.return_value = mock_feed
|
||||
|
||||
bot = MastodonRSSBot(**self.test_config)
|
||||
feed = bot.parse_feed()
|
||||
|
||||
self.assertIsNotNone(feed)
|
||||
mock_parse.assert_called_once_with(self.test_config['feed_url'])
|
||||
mock_parse.assert_called_once_with(self.test_config["feed_url"])
|
||||
|
||||
@patch('bot.feedparser.parse')
|
||||
@patch('bot.Mastodon')
|
||||
@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")
|
||||
@@ -180,8 +186,8 @@ class TestMastodonRSSBot(unittest.TestCase):
|
||||
|
||||
self.assertIsNone(feed)
|
||||
|
||||
@patch('bot.feedparser.parse')
|
||||
@patch('bot.Mastodon')
|
||||
@patch("bot.feedparser.parse")
|
||||
@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()
|
||||
@@ -193,16 +199,16 @@ class TestMastodonRSSBot(unittest.TestCase):
|
||||
|
||||
self.assertEqual(count, 0)
|
||||
|
||||
@patch('bot.feedparser.parse')
|
||||
@patch('bot.Mastodon')
|
||||
@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'},
|
||||
{"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
|
||||
|
||||
@@ -220,22 +226,28 @@ class TestMastodonRSSBot(unittest.TestCase):
|
||||
saved_entries = bot.load_processed_entries()
|
||||
self.assertEqual(len(saved_entries), 3)
|
||||
|
||||
@patch('bot.feedparser.parse')
|
||||
@patch('bot.Mastodon')
|
||||
@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))
|
||||
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
|
||||
{
|
||||
"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
|
||||
|
||||
@@ -254,14 +266,14 @@ class TestMastodonRSSBot(unittest.TestCase):
|
||||
saved_entries = bot.load_processed_entries()
|
||||
self.assertEqual(len(saved_entries), 4)
|
||||
|
||||
@patch('bot.feedparser.parse')
|
||||
@patch('bot.Mastodon')
|
||||
@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'},
|
||||
{"title": "Article without URL"}, # No link field
|
||||
{"title": "Article with URL", "link": "https://example.com/1"},
|
||||
]
|
||||
mock_parse.return_value = mock_feed
|
||||
|
||||
@@ -275,13 +287,13 @@ class TestMastodonRSSBot(unittest.TestCase):
|
||||
self.assertEqual(count, 1)
|
||||
self.assertEqual(mock_instance.status_post.call_count, 1)
|
||||
|
||||
@patch('bot.feedparser.parse')
|
||||
@patch('bot.Mastodon')
|
||||
@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'},
|
||||
{"title": "Article 1", "link": "https://example.com/1"},
|
||||
]
|
||||
mock_parse.return_value = mock_feed
|
||||
|
||||
@@ -304,17 +316,20 @@ class TestMastodonRSSBot(unittest.TestCase):
|
||||
class TestMainEntry(unittest.TestCase):
|
||||
"""Test cases for main.py entry point"""
|
||||
|
||||
@patch.dict(os.environ, {
|
||||
'MASTODON_CLIENT_ID': 'test_id',
|
||||
'MASTODON_CLIENT_SECRET': 'test_secret',
|
||||
'MASTODON_ACCESS_TOKEN': 'test_token',
|
||||
'MASTODON_INSTANCE_URL': 'https://mastodon.test',
|
||||
'RSS_FEED_URL': 'https://example.com/feed.xml',
|
||||
'TOOT_VISIBILITY': 'unlisted',
|
||||
'CHECK_INTERVAL': '120',
|
||||
'PROCESSED_ENTRIES_FILE': '/tmp/test_state.txt'
|
||||
})
|
||||
@patch('main.MastodonRSSBot')
|
||||
@patch.dict(
|
||||
os.environ,
|
||||
{
|
||||
"MASTODON_CLIENT_ID": "test_id",
|
||||
"MASTODON_CLIENT_SECRET": "test_secret",
|
||||
"MASTODON_ACCESS_TOKEN": "test_token",
|
||||
"MASTODON_INSTANCE_URL": "https://mastodon.test",
|
||||
"RSS_FEED_URL": "https://example.com/feed.xml",
|
||||
"TOOT_VISIBILITY": "unlisted",
|
||||
"CHECK_INTERVAL": "120",
|
||||
"PROCESSED_ENTRIES_FILE": "/tmp/test_state.txt",
|
||||
},
|
||||
)
|
||||
@patch("main.MastodonRSSBot")
|
||||
def test_main_loads_environment_config(self, mock_bot_class):
|
||||
"""Test that main() loads configuration from environment"""
|
||||
from main import main
|
||||
@@ -330,16 +345,16 @@ class TestMainEntry(unittest.TestCase):
|
||||
|
||||
# Verify bot was created with correct config
|
||||
mock_bot_class.assert_called_once_with(
|
||||
client_id='test_id',
|
||||
client_secret='test_secret',
|
||||
access_token='test_token',
|
||||
instance_url='https://mastodon.test',
|
||||
feed_url='https://example.com/feed.xml',
|
||||
toot_visibility='unlisted',
|
||||
client_id="test_id",
|
||||
client_secret="test_secret",
|
||||
access_token="test_token",
|
||||
instance_url="https://mastodon.test",
|
||||
feed_url="https://example.com/feed.xml",
|
||||
toot_visibility="unlisted",
|
||||
check_interval=120,
|
||||
state_file='/tmp/test_state.txt'
|
||||
state_file="/tmp/test_state.txt",
|
||||
)
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
|
||||
@@ -1,9 +1,11 @@
|
||||
"""Integration tests for Mastodon RSS Bot"""
|
||||
|
||||
import unittest
|
||||
from unittest.mock import Mock, patch
|
||||
import tempfile
|
||||
import os
|
||||
import time
|
||||
from textwrap import dedent
|
||||
from bot import MastodonRSSBot
|
||||
import responses
|
||||
import feedparser
|
||||
@@ -15,27 +17,28 @@ class TestRSSFeedIntegration(unittest.TestCase):
|
||||
def setUp(self):
|
||||
"""Set up test fixtures"""
|
||||
self.test_config = {
|
||||
'client_id': 'test_client_id',
|
||||
'client_secret': 'test_client_secret',
|
||||
'access_token': 'test_access_token',
|
||||
'instance_url': 'https://mastodon.test',
|
||||
'feed_url': 'https://example.com/feed.xml',
|
||||
'toot_visibility': 'public',
|
||||
'check_interval': 1,
|
||||
'state_file': tempfile.mktemp()
|
||||
"client_id": "test_client_id",
|
||||
"client_secret": "test_client_secret",
|
||||
"access_token": "test_access_token",
|
||||
"instance_url": "https://mastodon.test",
|
||||
"feed_url": "https://example.com/feed.xml",
|
||||
"toot_visibility": "public",
|
||||
"check_interval": 1,
|
||||
"state_file": tempfile.mktemp(),
|
||||
}
|
||||
|
||||
def tearDown(self):
|
||||
"""Clean up test files"""
|
||||
if os.path.exists(self.test_config['state_file']):
|
||||
os.remove(self.test_config['state_file'])
|
||||
if os.path.exists(self.test_config["state_file"]):
|
||||
os.remove(self.test_config["state_file"])
|
||||
|
||||
@responses.activate
|
||||
@patch('bot.Mastodon')
|
||||
@patch("bot.Mastodon")
|
||||
def test_end_to_end_rss_to_mastodon(self, mock_mastodon):
|
||||
"""Test complete flow from RSS feed to Mastodon post"""
|
||||
# Mock RSS feed response
|
||||
rss_feed = """<?xml version="1.0" encoding="UTF-8"?>
|
||||
rss_feed = dedent(
|
||||
"""<?xml version="1.0" encoding="UTF-8"?>
|
||||
<rss version="2.0">
|
||||
<channel>
|
||||
<title>Test Feed</title>
|
||||
@@ -53,13 +56,14 @@ class TestRSSFeedIntegration(unittest.TestCase):
|
||||
</item>
|
||||
</channel>
|
||||
</rss>"""
|
||||
).strip()
|
||||
|
||||
responses.add(
|
||||
responses.GET,
|
||||
'https://example.com/feed.xml',
|
||||
"https://example.com/feed.xml",
|
||||
body=rss_feed,
|
||||
status=200,
|
||||
content_type='application/xml'
|
||||
content_type="application/xml",
|
||||
)
|
||||
|
||||
# Mock Mastodon instance
|
||||
@@ -76,16 +80,17 @@ class TestRSSFeedIntegration(unittest.TestCase):
|
||||
|
||||
# Verify the content of posts
|
||||
calls = mock_instance.status_post.call_args_list
|
||||
self.assertIn('First Article', calls[0][0][0])
|
||||
self.assertIn('https://example.com/article1', calls[0][0][0])
|
||||
self.assertIn('Second Article', calls[1][0][0])
|
||||
self.assertIn('https://example.com/article2', calls[1][0][0])
|
||||
self.assertIn("First Article", calls[0][0][0])
|
||||
self.assertIn("https://example.com/article1", calls[0][0][0])
|
||||
self.assertIn("Second Article", calls[1][0][0])
|
||||
self.assertIn("https://example.com/article2", calls[1][0][0])
|
||||
|
||||
@responses.activate
|
||||
@patch('bot.Mastodon')
|
||||
@patch("bot.Mastodon")
|
||||
def test_atom_feed_parsing(self, mock_mastodon):
|
||||
"""Test parsing Atom feeds"""
|
||||
atom_feed = """<?xml version="1.0" encoding="UTF-8"?>
|
||||
atom_feed = dedent(
|
||||
"""<?xml version="1.0" encoding="UTF-8"?>
|
||||
<feed xmlns="http://www.w3.org/2005/Atom">
|
||||
<title>Test Atom Feed</title>
|
||||
<link href="https://example.com"/>
|
||||
@@ -98,13 +103,14 @@ class TestRSSFeedIntegration(unittest.TestCase):
|
||||
<summary>This is an atom article</summary>
|
||||
</entry>
|
||||
</feed>"""
|
||||
).strip()
|
||||
|
||||
responses.add(
|
||||
responses.GET,
|
||||
'https://example.com/feed.xml',
|
||||
"https://example.com/feed.xml",
|
||||
body=atom_feed,
|
||||
status=200,
|
||||
content_type='application/atom+xml'
|
||||
content_type="application/atom+xml",
|
||||
)
|
||||
|
||||
mock_instance = Mock()
|
||||
@@ -115,13 +121,14 @@ class TestRSSFeedIntegration(unittest.TestCase):
|
||||
|
||||
self.assertEqual(count, 1)
|
||||
calls = mock_instance.status_post.call_args_list
|
||||
self.assertIn('Atom Article', calls[0][0][0])
|
||||
self.assertIn("Atom Article", calls[0][0][0])
|
||||
|
||||
@responses.activate
|
||||
@patch('bot.Mastodon')
|
||||
@patch("bot.Mastodon")
|
||||
def test_persistence_across_runs(self, mock_mastodon):
|
||||
"""Test that processed entries persist across multiple bot runs"""
|
||||
rss_feed = """<?xml version="1.0" encoding="UTF-8"?>
|
||||
rss_feed = dedent(
|
||||
"""<?xml version="1.0" encoding="UTF-8"?>
|
||||
<rss version="2.0">
|
||||
<channel>
|
||||
<title>Test Feed</title>
|
||||
@@ -131,13 +138,14 @@ class TestRSSFeedIntegration(unittest.TestCase):
|
||||
</item>
|
||||
</channel>
|
||||
</rss>"""
|
||||
).strip()
|
||||
|
||||
responses.add(
|
||||
responses.GET,
|
||||
'https://example.com/feed.xml',
|
||||
"https://example.com/feed.xml",
|
||||
body=rss_feed,
|
||||
status=200,
|
||||
content_type='application/xml'
|
||||
content_type="application/xml",
|
||||
)
|
||||
|
||||
mock_instance = Mock()
|
||||
@@ -157,11 +165,12 @@ class TestRSSFeedIntegration(unittest.TestCase):
|
||||
self.assertEqual(mock_instance.status_post.call_count, 1)
|
||||
|
||||
@responses.activate
|
||||
@patch('bot.Mastodon')
|
||||
@patch("bot.Mastodon")
|
||||
def test_incremental_feed_updates(self, mock_mastodon):
|
||||
"""Test handling of new entries added to feed over time"""
|
||||
# Initial feed with 2 articles
|
||||
initial_feed = """<?xml version="1.0" encoding="UTF-8"?>
|
||||
initial_feed = dedent(
|
||||
"""<?xml version="1.0" encoding="UTF-8"?>
|
||||
<rss version="2.0">
|
||||
<channel>
|
||||
<title>Test Feed</title>
|
||||
@@ -175,13 +184,14 @@ class TestRSSFeedIntegration(unittest.TestCase):
|
||||
</item>
|
||||
</channel>
|
||||
</rss>"""
|
||||
).strip()
|
||||
|
||||
responses.add(
|
||||
responses.GET,
|
||||
'https://example.com/feed.xml',
|
||||
"https://example.com/feed.xml",
|
||||
body=initial_feed,
|
||||
status=200,
|
||||
content_type='application/xml'
|
||||
content_type="application/xml",
|
||||
)
|
||||
|
||||
mock_instance = Mock()
|
||||
@@ -194,7 +204,8 @@ class TestRSSFeedIntegration(unittest.TestCase):
|
||||
|
||||
# Update feed with 1 new article
|
||||
responses.reset()
|
||||
updated_feed = """<?xml version="1.0" encoding="UTF-8"?>
|
||||
updated_feed = dedent(
|
||||
"""<?xml version="1.0" encoding="UTF-8"?>
|
||||
<rss version="2.0">
|
||||
<channel>
|
||||
<title>Test Feed</title>
|
||||
@@ -212,13 +223,14 @@ class TestRSSFeedIntegration(unittest.TestCase):
|
||||
</item>
|
||||
</channel>
|
||||
</rss>"""
|
||||
).strip()
|
||||
|
||||
responses.add(
|
||||
responses.GET,
|
||||
'https://example.com/feed.xml',
|
||||
"https://example.com/feed.xml",
|
||||
body=updated_feed,
|
||||
status=200,
|
||||
content_type='application/xml'
|
||||
content_type="application/xml",
|
||||
)
|
||||
|
||||
# Second run - should only post the new article
|
||||
@@ -229,14 +241,14 @@ class TestRSSFeedIntegration(unittest.TestCase):
|
||||
self.assertEqual(mock_instance.status_post.call_count, 3)
|
||||
|
||||
@responses.activate
|
||||
@patch('bot.Mastodon')
|
||||
@patch("bot.Mastodon")
|
||||
def test_network_error_handling(self, mock_mastodon):
|
||||
"""Test handling of network errors when fetching feed"""
|
||||
responses.add(
|
||||
responses.GET,
|
||||
'https://example.com/feed.xml',
|
||||
"https://example.com/feed.xml",
|
||||
body="Network error",
|
||||
status=500
|
||||
status=500,
|
||||
)
|
||||
|
||||
mock_instance = Mock()
|
||||
@@ -250,7 +262,7 @@ class TestRSSFeedIntegration(unittest.TestCase):
|
||||
self.assertEqual(mock_instance.status_post.call_count, 0)
|
||||
|
||||
@responses.activate
|
||||
@patch('bot.Mastodon')
|
||||
@patch("bot.Mastodon")
|
||||
def test_malformed_xml_handling(self, mock_mastodon):
|
||||
"""Test handling of malformed XML feeds"""
|
||||
malformed_feed = """<?xml version="1.0" encoding="UTF-8"?>
|
||||
@@ -263,10 +275,10 @@ class TestRSSFeedIntegration(unittest.TestCase):
|
||||
|
||||
responses.add(
|
||||
responses.GET,
|
||||
'https://example.com/feed.xml',
|
||||
"https://example.com/feed.xml",
|
||||
body=malformed_feed,
|
||||
status=200,
|
||||
content_type='application/xml'
|
||||
content_type="application/xml",
|
||||
)
|
||||
|
||||
mock_instance = Mock()
|
||||
@@ -287,40 +299,40 @@ class TestMastodonAPIIntegration(unittest.TestCase):
|
||||
def setUp(self):
|
||||
"""Set up test fixtures"""
|
||||
self.test_config = {
|
||||
'client_id': 'test_client_id',
|
||||
'client_secret': 'test_client_secret',
|
||||
'access_token': 'test_access_token',
|
||||
'instance_url': 'https://mastodon.test',
|
||||
'feed_url': 'https://example.com/feed.xml',
|
||||
'toot_visibility': 'public',
|
||||
'check_interval': 1,
|
||||
'state_file': tempfile.mktemp()
|
||||
"client_id": "test_client_id",
|
||||
"client_secret": "test_client_secret",
|
||||
"access_token": "test_access_token",
|
||||
"instance_url": "https://mastodon.test",
|
||||
"feed_url": "https://example.com/feed.xml",
|
||||
"toot_visibility": "public",
|
||||
"check_interval": 1,
|
||||
"state_file": tempfile.mktemp(),
|
||||
}
|
||||
|
||||
def tearDown(self):
|
||||
"""Clean up test files"""
|
||||
if os.path.exists(self.test_config['state_file']):
|
||||
os.remove(self.test_config['state_file'])
|
||||
if os.path.exists(self.test_config["state_file"]):
|
||||
os.remove(self.test_config["state_file"])
|
||||
|
||||
@patch('bot.Mastodon')
|
||||
@patch("bot.Mastodon")
|
||||
def test_different_visibility_levels(self, mock_mastodon):
|
||||
"""Test posting with different visibility levels"""
|
||||
mock_instance = Mock()
|
||||
mock_mastodon.return_value = mock_instance
|
||||
|
||||
visibility_levels = ['public', 'unlisted', 'private', 'direct']
|
||||
visibility_levels = ["public", "unlisted", "private", "direct"]
|
||||
|
||||
for visibility in visibility_levels:
|
||||
self.test_config['toot_visibility'] = visibility
|
||||
self.test_config["toot_visibility"] = visibility
|
||||
bot = MastodonRSSBot(**self.test_config)
|
||||
bot.post_to_mastodon("Test status")
|
||||
|
||||
# Verify all visibility levels were used
|
||||
calls = mock_instance.status_post.call_args_list
|
||||
for idx, visibility in enumerate(visibility_levels):
|
||||
self.assertEqual(calls[idx][1]['visibility'], visibility)
|
||||
self.assertEqual(calls[idx][1]["visibility"], visibility)
|
||||
|
||||
@patch('bot.Mastodon')
|
||||
@patch("bot.Mastodon")
|
||||
def test_retry_on_rate_limit(self, mock_mastodon):
|
||||
"""Test that rate limit errors are handled appropriately"""
|
||||
from mastodon import MastodonRatelimitError
|
||||
@@ -329,7 +341,7 @@ class TestMastodonAPIIntegration(unittest.TestCase):
|
||||
# First call raises rate limit error, second succeeds
|
||||
mock_instance.status_post.side_effect = [
|
||||
MastodonRatelimitError("Rate limited"),
|
||||
None
|
||||
None,
|
||||
]
|
||||
mock_mastodon.return_value = mock_instance
|
||||
|
||||
@@ -344,5 +356,5 @@ class TestMastodonAPIIntegration(unittest.TestCase):
|
||||
self.assertTrue(result2)
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
|
||||
Reference in New Issue
Block a user