Compare commits

21 Commits

Author SHA1 Message Date
Amit Serper
e4e5a22a0f Fix repository link in README.md
Updated the repository link in README.md from piko-system to aserper.
2025-12-13 23:07:31 -05:00
unroll3677
fc5a4bec00 Update docker-compose.yml 2025-07-16 14:36:55 +02:00
unroll3677
6ff72f7f97 Update README.md 2025-07-16 14:35:20 +02:00
unroll3677
c42d230053 Update Dockerfile 2025-07-16 14:24:29 +02:00
unroll3677
22677de465 Update docker-publish.yml 2025-07-16 14:18:13 +02:00
unroll3677
760d84d104 Delete .github/workflows/masto-rss.yml 2025-07-16 14:14:48 +02:00
unroll3677
faf443b5b7 Create docker-publish.yml 2025-07-16 14:14:04 +02:00
unroll3677
f795520ad6 Create .dockerignore 2025-07-15 15:22:56 +02:00
unroll3677
6ae6978d68 Update Dockerfile 2025-07-15 15:22:17 +02:00
unroll3677
4c043a596b Update README.md 2025-07-15 14:07:44 +02:00
unroll3677
d3c02fddac Create docker-compose.yml 2025-07-15 10:12:06 +02:00
unroll3677
0860fa555d Update README.md 2025-07-15 10:08:40 +02:00
unroll3677
7a16534593 Update README.md 2025-07-15 09:58:43 +02:00
unroll3677
b4810530ab Update main.py
added toot template and translated to english
2025-07-15 09:55:28 +02:00
unroll3677
83fe193d67 Update main.py
added hashtag fix
2025-07-14 14:55:16 +02:00
unroll3677
6e947528cb Update README.md 2025-07-14 14:45:21 +02:00
unroll3677
2e11ea076a Update README.md 2025-07-14 14:39:16 +02:00
unroll3677
43c9ab7bdb Update README.md 2025-07-14 14:38:51 +02:00
unroll3677
ec657cb5af Update main.py
added state file max lines
2025-07-14 14:37:41 +02:00
unroll3677
a96fcf9ca6 Update README.md 2025-07-14 14:25:23 +02:00
unroll3677
8ef903a720 Update main.py
added hashtag support
2025-07-14 14:23:13 +02:00
18 changed files with 334 additions and 1933 deletions

23
.dockerignore Normal file
View File

@@ -0,0 +1,23 @@
# Git-related files
.git
.gitignore
# Docker-related files
Dockerfile
.dockerignore
# Python-specific files and directories
__pycache__/
*.pyc
*.pyo
*.pyd
.venv/
venv/
env/
# IDE / Editor-specific directories
.vscode/
.idea/
# Documentation
README.md

25
.github/workflows/docker-publish.yml vendored Normal file
View File

@@ -0,0 +1,25 @@
name: Push to Docker Hub
on:
push:
branches: [ "main" ]
jobs:
build-and-push:
runs-on: ubuntu-latest
steps:
- name: Checkout repository
uses: actions/checkout@v4
- name: Login to Docker Hub
uses: docker/login-action@v2
with:
username: ${{ secrets.DOCKERHUB_USERNAME }}
password: ${{ secrets.DOCKERHUB_TOKEN }}
- name: Build and push Docker image
uses: docker/build-push-action@v4
with:
context: .
push: true
tags: ${{ secrets.DOCKERHUB_USERNAME }}/masto-rss-advanced:latest

View File

@@ -1,59 +0,0 @@
name: Masto-rss
on:
push:
branches: [ main ]
pull_request:
branches: [ main ]
jobs:
build:
runs-on: ubuntu-latest
permissions:
contents: read
packages: write
steps:
- uses: actions/checkout@v4
- name: Set up QEMU
uses: docker/setup-qemu-action@v3
- name: Set up Docker Buildx
uses: docker/setup-buildx-action@v3
- name: Login to Docker Hub
uses: docker/login-action@v3
with:
username: amitserper
password: ${{ secrets.DH_PASSWORD }}
- name: Login to GitHub Container Registry
uses: docker/login-action@v3
with:
registry: ghcr.io
username: ${{ github.actor }}
password: ${{ secrets.GITHUB_TOKEN }}
- name: Extract metadata
id: meta
uses: docker/metadata-action@v5
with:
images: |
amitserper/masto-rss
ghcr.io/aserper/masto-rss
tags: |
type=ref,event=branch
type=sha,prefix={{branch}}-
type=raw,value=latest,enable={{is_default_branch}}
- name: Build and push multiarch image
uses: docker/build-push-action@v5
with:
context: .
platforms: linux/amd64,linux/arm64
push: ${{ github.event_name != 'pull_request' }}
tags: ${{ steps.meta.outputs.tags }}
labels: ${{ steps.meta.outputs.labels }}
cache-from: type=gha
cache-to: type=gha,mode=max

View File

@@ -1,162 +0,0 @@
name: Tests
on:
push:
branches: [ main ]
pull_request:
branches: [ main ]
jobs:
unit-tests:
name: Unit Tests
runs-on: ubuntu-latest
strategy:
matrix:
python-version: ['3.10', '3.11', '3.12']
steps:
- name: Checkout code
uses: actions/checkout@v4
- name: Set up Python ${{ matrix.python-version }}
uses: actions/setup-python@v5
with:
python-version: ${{ matrix.python-version }}
cache: 'pip'
- name: Install dependencies
run: |
python -m pip install --upgrade pip
pip install -r requirements-test.txt
- name: Run unit tests
run: |
pytest test_bot.py -m "not integration" --cov=bot --cov=main --cov-report=xml --cov-report=term
- name: Upload coverage to Codecov
uses: codecov/codecov-action@v4
with:
file: ./coverage.xml
flags: unittests
name: codecov-${{ matrix.python-version }}
fail_ci_if_error: false
integration-tests:
name: Integration Tests
runs-on: ubuntu-latest
strategy:
matrix:
python-version: ['3.10', '3.11', '3.12']
steps:
- name: Checkout code
uses: actions/checkout@v4
- name: Set up Python ${{ matrix.python-version }}
uses: actions/setup-python@v5
with:
python-version: ${{ matrix.python-version }}
cache: 'pip'
- name: Install dependencies
run: |
python -m pip install --upgrade pip
pip install -r requirements-test.txt
- name: Run integration tests
run: |
pytest test_integration.py --cov=bot --cov-report=xml --cov-report=term --cov-fail-under=70
- name: Upload coverage to Codecov
uses: codecov/codecov-action@v4
with:
file: ./coverage.xml
flags: integration
name: codecov-integration-${{ matrix.python-version }}
fail_ci_if_error: false
code-quality:
name: Code Quality
runs-on: ubuntu-latest
steps:
- name: Checkout code
uses: actions/checkout@v4
- name: Set up Python
uses: actions/setup-python@v5
with:
python-version: '3.11'
cache: 'pip'
- name: Install dependencies
run: |
python -m pip install --upgrade pip
pip install -r requirements-test.txt
- name: Run flake8
run: |
flake8 bot.py main.py --count --select=E9,F63,F7,F82 --show-source --statistics
flake8 bot.py main.py --count --max-complexity=10 --max-line-length=127 --statistics
- name: Run black check
run: |
black --check bot.py main.py test_bot.py test_integration.py
- name: Run mypy
run: |
mypy bot.py main.py --ignore-missing-imports
continue-on-error: true
docker-build-test:
name: Docker Build Test
runs-on: ubuntu-latest
steps:
- name: Checkout code
uses: actions/checkout@v4
- name: Set up Docker Buildx
uses: docker/setup-buildx-action@v3
- name: Build Docker image
uses: docker/build-push-action@v5
with:
context: .
load: true
tags: masto-rss:test
cache-from: type=gha
cache-to: type=gha,mode=max
- name: Test Docker image structure
run: |
docker run --rm masto-rss:test python --version
docker run --rm masto-rss:test pip list | grep feedparser
docker run --rm masto-rss:test pip list | grep Mastodon
all-tests-pass:
name: All Tests Passed
needs: [unit-tests, integration-tests, code-quality, docker-build-test]
runs-on: ubuntu-latest
if: always()
steps:
- name: Check test results
run: |
if [ "${{ needs.unit-tests.result }}" != "success" ]; then
echo "Unit tests failed"
exit 1
fi
if [ "${{ needs.integration-tests.result }}" != "success" ]; then
echo "Integration tests failed"
exit 1
fi
if [ "${{ needs.code-quality.result }}" != "success" ]; then
echo "Code quality checks failed"
exit 1
fi
if [ "${{ needs.docker-build-test.result }}" != "success" ]; then
echo "Docker build test failed"
exit 1
fi
echo "All tests passed successfully!"

47
.gitignore vendored
View File

@@ -1,50 +1,3 @@
# State files
processed_entries.txt
/state/
# IDEs
.idea/
.vscode/
*.swp
*.swo
# Python
venv/
.venv/
__pycache__/
*.py[cod]
*$py.class
*.so
.Python
build/
develop-eggs/
dist/
downloads/
eggs/
.eggs/
lib/
lib64/
parts/
sdist/
var/
wheels/
*.egg-info/
.installed.cfg
*.egg
# Testing
.pytest_cache/
.coverage
coverage.xml
htmlcov/
.tox/
.hypothesis/
# Type checking
.mypy_cache/
.dmypy.json
dmypy.json
# Environment
.env
.env.local

View File

@@ -1,18 +1,22 @@
# Use an appropriate base image with Python pre-installed
FROM alpine:3.18
# Step 1: Use an official, slim Python image as a base
# 'slim' is a good balance between size and functionality.
FROM python:3.11-slim
# Set the working directory inside the container
WORKDIR /app
# Install Python dependencies in a single layer
RUN apk add --no-cache python3 py3-pip
# Step 2: Copy only the requirements file
# This makes optimal use of Docker's layer caching.
COPY requirements.txt .
# Copy requirements first for better layer caching
COPY requirements.txt /app/
# Step 3: Install the dependencies
# The --no-cache-dir option keeps the image smaller.
RUN pip install --no-cache-dir -r requirements.txt
# Copy the application code
COPY . /app
# Step 4: Now, copy the rest of the application code
# Because the code changes more often than the dependencies, this step is placed later.
COPY . .
# Run Python script with unbuffered output for container logs
CMD ["python", "-u", "main.py"]
# Step 5: Execute the correct Python script
# Note the correct filename.
CMD ["python3", "main.py"]

297
README.md
View File

@@ -1,233 +1,158 @@
![Masto-RSS Header](header.jpg)
# Masto-RSS Advanced Bot
# Masto-RSS
A flexible and robust Python script to automatically post the latest entry from an RSS feed to Mastodon, designed to run continuously in a Docker container.
[![Build Status](https://img.shields.io/github/actions/workflow/status/aserper/masto-rss/masto-rss.yml?style=for-the-badge&logo=github&label=Build)](https://github.com/aserper/masto-rss/actions/workflows/masto-rss.yml)
[![Tests](https://img.shields.io/github/actions/workflow/status/aserper/masto-rss/test.yml?style=for-the-badge&logo=github&label=Tests)](https://github.com/aserper/masto-rss/actions/workflows/test.yml)
[![Docker Hub](https://img.shields.io/badge/docker%20hub-amitserper%2Fmasto--rss-blue?style=for-the-badge&logo=docker&logoColor=white)](https://hub.docker.com/r/amitserper/masto-rss)
[![GHCR](https://img.shields.io/badge/ghcr.io-masto--rss-blue?style=for-the-badge&logo=docker&logoColor=white)](https://github.com/aserper/masto-rss/pkgs/container/masto-rss)
[![License: GPL v3](https://img.shields.io/badge/License-GPLv3-yellow.svg?style=for-the-badge)](LICENSE)
[![Python](https://img.shields.io/badge/python-3.10%2B-blue.svg?style=for-the-badge&logo=python&logoColor=white)](https://www.python.org/downloads/)
[![GitHub stars](https://img.shields.io/github/stars/aserper/masto-rss.svg?style=social)](https://github.com/aserper/masto-rss)
A simple, lightweight Mastodon bot that automatically posts updates from RSS feeds to the Fediverse. Built with Python and designed to run seamlessly in Docker with multiarch support (amd64 & arm64).
This project is based on the original concept of [masto-rss](https://github.com/aserper/masto-rss) but has been significantly rewritten and extended with modern features for stability and customization.
## Features
- Automatically monitors RSS/Atom feeds and posts new items to Mastodon
- Persistent state tracking to avoid duplicate posts
- Configurable post visibility (public, unlisted, private, direct)
- Lightweight Alpine-based Docker image
- Multiarch support (amd64 & arm64) for broad compatibility
- Continuous monitoring with configurable check intervals
* **Posts Only the Latest Item**: Checks an RSS feed periodically and posts only the single most recent entry, preventing feed spam.
* **Prevents Duplicates**: Keeps a history of posted items to ensure the same link is never posted twice.
* **History Limit**: The history file is automatically pruned to a configurable size (`MAX_HISTORY_SIZE`) to prevent it from growing indefinitely.
* **Custom Toot Format**: Fully customize the look of your posts using a template string with placeholders (`{title}`, `{link}`, `{hashtags}`).
* **Hashtag Support**: Automatically append a configurable list of hashtags to every post.
* **Dockerized**: Runs as a pre-built container image for easy, reliable, and isolated deployment. You don't need to manage Python or dependencies.
* **Robust Error Handling**: Designed to run forever, it handles connection errors to the RSS feed or Mastodon gracefully without crashing.
## Quick Start
## Getting Started
### Using Docker (Recommended)
The recommended and simplest way to run this bot is by using the pre-built Docker image from Docker Hub with Docker Compose.
The easiest way to run Masto-RSS is using the pre-built multiarch Docker images available on both Docker Hub and GitHub Container Registry.
### 1. Prerequisites
#### Pull from Docker Hub
* Docker and Docker Compose installed on your system.
* A Mastodon account and an application set up via **Preferences > Development**. You will need the **Client Key**, **Client Secret**, and **Access Token**.
### 2. Configuration
You only need to create one file: `docker-compose.yml`.
First, create a directory for your project and navigate into it:
```bash
docker pull amitserper/masto-rss:latest
mkdir my-masto-bot
cd my-masto-bot
```
#### Pull from GitHub Container Registry
```bash
docker pull ghcr.io/aserper/masto-rss:latest
```
#### Run the Bot
```bash
docker run -d \
--name masto-rss-bot \
-e MASTODON_CLIENT_ID="your_client_id" \
-e MASTODON_CLIENT_SECRET="your_client_secret" \
-e MASTODON_ACCESS_TOKEN="your_access_token" \
-e MASTODON_INSTANCE_URL="https://mastodon.social" \
-e RSS_FEED_URL="https://example.com/feed.xml" \
-e TOOT_VISIBILITY="public" \
-e CHECK_INTERVAL="300" \
-v /path/to/state:/state \
amitserper/masto-rss:latest
```
> **Important:** Use a bind mount for `/state` to persist the list of processed feed items across container restarts.
### Using Docker Compose
Create a `docker-compose.yml`:
Next, create a file named `docker-compose.yml` and paste the following content into it. **You must edit the `environment` section with your own credentials and settings.**
```yaml
version: '3.8'
services:
masto-rss:
image: amitserper/masto-rss:latest
# Or use GHCR: ghcr.io/aserper/masto-rss:latest
# Use the pre-built image from Docker Hub
image: doorknob2947/masto-rss-advanced:latest
container_name: masto-rss-bot
restart: unless-stopped
environment:
MASTODON_CLIENT_ID: "your_client_id"
MASTODON_CLIENT_SECRET: "your_client_secret"
MASTODON_ACCESS_TOKEN: "your_access_token"
MASTODON_INSTANCE_URL: "https://mastodon.social"
RSS_FEED_URL: "https://example.com/feed.xml"
TOOT_VISIBILITY: "public"
CHECK_INTERVAL: "300"
volumes:
# This volume persists the history of posted links
- ./state:/state
environment:
# --- Mastodon API Credentials (Required) ---
- MASTODON_CLIENT_ID=YOUR_CLIENT_KEY_HERE
- MASTODON_CLIENT_SECRET=YOUR_CLIENT_SECRET_HERE
- MASTODON_ACCESS_TOKEN=YOUR_ACCESS_TOKEN_HERE
- MASTODON_INSTANCE_URL=[https://mastodon.social](https://mastodon.social)
# --- RSS Feed Configuration (Required) ---
- RSS_FEED_URL=[https://www.theverge.com/rss/index.xml](https://www.theverge.com/rss/index.xml)
# --- Bot Behavior (Optional) ---
- CHECK_INTERVAL=3600 # Time in seconds between checks (e.g., 1 hour)
- MAX_HISTORY_SIZE=500 # Max number of posted links to remember
- TOOT_VISIBILITY=public # public, unlisted, private, or direct
# --- Toot Content (Optional) ---
- MASTO_RSS_HASHTAGS="news tech python" # Space-separated list of hashtags
- TOOT_TEMPLATE='{title}\n\n{link}\n\n{hashtags}'
# --- System (Do not change) ---
- PYTHONUNBUFFERED=1
```
Then run:
### 3. Environment Variables
All configuration is handled through the `environment` section in your `docker-compose.yml` file.
| Variable | Description | Example |
| ------------------------ | ------------------------------------------------------------------------ | --------------------------------------------------- |
| `MASTODON_CLIENT_ID` | Your Mastodon application's Client Key. | `abc...` |
| `MASTODON_CLIENT_SECRET` | Your Mastodon application's Client Secret. | `def...` |
| `MASTODON_ACCESS_TOKEN` | Your Mastodon application's Access Token. | `ghi...` |
| `MASTODON_INSTANCE_URL` | The base URL of your Mastodon instance. | `https://mastodon.social` |
| `RSS_FEED_URL` | The full URL of the RSS feed you want to monitor. | `https://www.theverge.com/rss/index.xml` |
| `CHECK_INTERVAL` | (Optional) The time in seconds between checks. | `3600` (for 1 hour) |
| `MASTO_RSS_HASHTAGS` | (Optional) A space-separated list of hashtags to add to each post. | `news tech python` |
| `MAX_HISTORY_SIZE` | (Optional) The maximum number of post URLs to remember. | `500` |
| `TOOT_VISIBILITY` | (Optional) The visibility of the toot (`public`, `unlisted`, `private`, `direct`). | `public` |
| `TOOT_TEMPLATE` | (Optional) A string to format the toot. See "Customizing the Toot Format" below. | `'{title}\n\n{link}\n\n{hashtags}'` |
| `PYTHONUNBUFFERED` | Should be kept at `1` to ensure logs appear in real-time in Docker. | `1` |
### 4. Customizing the Toot Format
You can change the layout of your posts using the `TOOT_TEMPLATE` variable. Use the following placeholders:
* `{title}`: The title of the RSS entry.
* `{link}`: The URL of the RSS entry.
* `{hashtags}`: The configured hashtags.
**Examples in `docker-compose.yml`:**
* **Compact Format:**
```yaml
- TOOT_TEMPLATE='{title} - {link} {hashtags}'
```
* **Personalized Format:**
```yaml
- TOOT_TEMPLATE='New on the blog: {title}\nRead it here: {link}\n\n{hashtags}'
```
## Running the Bot
1. **Create the `state` directory.** This is required for the bot to remember which links it has already posted.
```bash
mkdir ./state
```
2. **Start the container** in detached (background) mode:
```bash
docker-compose up -d
```
The bot is now running!
### Multiple Feeds
To monitor multiple feeds, you can either:
- Use the `RSS_FEEDS` environment variable (comma-separated list)
- Use the `FEEDS_FILE` environment variable (path to file with one URL per line)
#### Run with Multiple Feeds (Docker)
## Managing the Bot
* **View logs in real-time:**
```bash
docker run -d \
--name masto-rss-bot \
-e MASTODON_CLIENT_ID="your_client_id" \
-e MASTODON_CLIENT_SECRET="your_client_secret" \
-e MASTODON_ACCESS_TOKEN="your_access_token" \
-e MASTODON_INSTANCE_URL="https://mastodon.social" \
-e RSS_FEEDS="https://feed1.com/rss,https://feed2.com/rss" \
-e TOOT_VISIBILITY="public" \
-e CHECK_INTERVAL="300" \
-v /path/to/state:/state \
amitserper/masto-rss:latest
docker-compose logs -f
```
## Configuration
All configuration is done via environment variables:
| Variable | Description | Required | Example |
|----------|-------------|----------|---------|
| `MASTODON_CLIENT_ID` | Mastodon application client ID | Yes | `abc123...` |
| `MASTODON_CLIENT_SECRET` | Mastodon application client secret | Yes | `xyz789...` |
| `MASTODON_ACCESS_TOKEN` | Mastodon access token | Yes | `token123...` |
| `MASTODON_INSTANCE_URL` | URL of your Mastodon instance | Yes | `https://mastodon.social` |
| `RSS_FEED_URL` | Single RSS/Atom feed URL (Legacy) | No* | `https://example.com/feed.xml` |
| `RSS_FEEDS` | Comma-separated list of feed URLs | No* | `https://site1.com,https://site2.com` |
| `FEEDS_FILE` | Path to file containing list of feed URLs | No* | `/config/feeds.txt` |
| `TOOT_VISIBILITY` | Post visibility level | Yes | `public`, `unlisted`, `private`, or `direct` |
| `CHECK_INTERVAL` | Seconds between feed checks | Yes | `300` (5 minutes) |
| `PROCESSED_ENTRIES_FILE`| Custom path for state file | No | `/state/processed.txt` |
\* At least one of `RSS_FEED_URL`, `RSS_FEEDS`, or `FEEDS_FILE` must be provided.
### Getting Mastodon API Credentials
1. Log into your Mastodon instance
2. Go to **Settings****Development****New Application**
3. Give it a name (e.g., "RSS Bot")
4. Set scopes to `write:statuses`
5. Save and copy the client ID, client secret, and access token
## Building from Source
### Build Locally
* **Stop the container:**
```bash
git clone https://github.com/aserper/masto-rss.git
cd masto-rss
docker build -t masto-rss .
docker-compose down
```
### Build Multiarch Images
* **Restart the container:**
```bash
# Set up buildx
docker buildx create --use
# Build for both architectures
docker buildx build \
--platform linux/amd64,linux/arm64 \
-t yourusername/masto-rss:latest \
--push \
.
docker-compose restart
```
## Running Without Docker
## Updating the Bot
If you prefer to run the bot directly with Python:
To update to the latest version of the image from Docker Hub:
1. **Pull the latest image:**
```bash
# Clone the repository
git clone https://github.com/aserper/masto-rss.git
cd masto-rss
# Install dependencies
pip install -r requirements.txt
# Set environment variables
export MASTODON_CLIENT_ID="your_client_id"
export MASTODON_CLIENT_SECRET="your_client_secret"
export MASTODON_ACCESS_TOKEN="your_access_token"
export MASTODON_INSTANCE_URL="https://mastodon.social"
export RSS_FEED_URL="https://example.com/feed.xml"
export TOOT_VISIBILITY="public"
export CHECK_INTERVAL="300"
# Run the bot
python main.py
docker-compose pull
```
2. **Restart the container** to apply the update:
```bash
docker-compose up -d
```
> **Note:** When running without Docker, the bot stores its state in `/state/processed_entries.txt`. Make sure this directory exists or modify [main.py](main.py#L15) to use a different path.
## Acknowledgements
## How It Works
1. The bot fetches the RSS feed at regular intervals (defined by `CHECK_INTERVAL`)
2. For each feed item, it checks if the item's URL has been processed before
3. If the item is new, it posts to Mastodon with the format: `{title}\n\n{link}`
4. The item URL is saved to prevent duplicate posts
5. The process repeats indefinitely
## Architecture
- **Base Image:** Alpine Linux 3.18 (minimal footprint)
- **Python Version:** 3.10+
- **Platforms:** linux/amd64, linux/arm64
- **Dependencies:** feedparser, mastodon.py (see [requirements.txt](requirements.txt))
## State Persistence
The bot maintains state in `/state/processed_entries.txt` to track which feed items have already been posted. This prevents duplicate posts across restarts.
**Important:** Always mount `/state` as a volume to preserve this state file.
## Contributing
Contributions are welcome! Feel free to:
- Report bugs by opening an issue
- Submit pull requests for improvements
- Suggest new features or enhancements
## License
This project is licensed under the GNU General Public License v3.0 - see the [LICENSE](LICENSE) file for details.
## Support
If you find this project useful, please consider giving it a star on GitHub!
## Links
- [Docker Hub Repository](https://hub.docker.com/r/amitserper/masto-rss)
- [GitHub Container Registry](https://github.com/aserper/masto-rss/pkgs/container/masto-rss)
- [Source Code](https://github.com/aserper/masto-rss)
- [Issues](https://github.com/aserper/masto-rss/issues)
* This project is heavily inspired by and based on the original [masto-rss](https://github.com/aserper/masto-rss). This version aims to add more robustness, flexibility, and ease of deployment using modern practices.
* The Python script, Docker configuration, and this README were written and modified with the assistance of AI.

View File

@@ -1,235 +0,0 @@
# Testing Guide for Masto-RSS
This document describes the testing strategy and how to run tests for the Masto-RSS bot.
## Test Architecture
The test suite is organized into two main categories:
### Unit Tests ([test_bot.py](test_bot.py))
- Test individual functions and methods in isolation
- Use mocks and stubs for external dependencies
- Fast execution time
- High code coverage
- Test edge cases and error handling
### Integration Tests ([test_integration.py](test_integration.py))
- Test interactions between components
- Mock external services (RSS feeds, Mastodon API)
- Test end-to-end workflows
- Verify data persistence
- Test error recovery
## Running Tests Locally
### Prerequisites
```bash
# Install test dependencies
pip install -r requirements-test.txt
```
### Run All Tests
```bash
# Run all tests with coverage
pytest
# Run with verbose output
pytest -v
# Run with coverage report
pytest --cov=bot --cov=main --cov-report=html
```
### Run Specific Test Categories
```bash
# Run only unit tests
pytest test_bot.py
# Run only integration tests
pytest test_integration.py
# Run tests matching a pattern
pytest -k "test_parse_feed"
```
### Run with Markers
```bash
# Run only unit tests (using markers)
pytest -m unit
# Run only integration tests
pytest -m integration
# Skip slow tests
pytest -m "not slow"
```
### Coverage Reports
```bash
# Generate HTML coverage report
pytest --cov=bot --cov=main --cov-report=html
# View report
open htmlcov/index.html # macOS
xdg-open htmlcov/index.html # Linux
```
## GitHub Actions CI/CD
Tests run automatically on every push to `main` and on all pull requests via [.github/workflows/test.yml](.github/workflows/test.yml).
### Test Jobs
1. **Unit Tests**
- Runs on Python 3.10, 3.11, 3.12
- Executes all unit tests
- Uploads coverage to Codecov
2. **Integration Tests**
- Runs on Python 3.10, 3.11, 3.12
- Executes all integration tests with mocked external services
- Uploads coverage to Codecov
3. **Code Quality**
- Runs flake8 for linting
- Runs black for code formatting checks
- Runs mypy for type checking
4. **Docker Build Test**
- Builds the Docker image
- Verifies Python and dependencies are installed
- Ensures the image can run
5. **All Tests Pass**
- Final job that requires all previous jobs to succeed
- Provides a single status check for PR requirements
## Test Coverage Requirements
- **Minimum coverage**: 80%
- Coverage is measured for `bot.py` and `main.py`
- Test files are excluded from coverage metrics
## Code Quality Standards
### Flake8
- Maximum line length: 127 characters
- Maximum cyclomatic complexity: 10
- Critical error codes checked: E9, F63, F7, F82
### Black
- Line length: 88 characters (default)
- All Python files must pass black formatting
### Mypy
- Type hints encouraged but not required
- Runs in non-strict mode with missing imports ignored
## Test Data and Fixtures
### Mock RSS Feeds
Integration tests use realistic RSS 2.0 and Atom feed XML for testing feed parsing.
### Mock Mastodon API
The Mastodon API is mocked using `unittest.mock` to avoid making real API calls.
### Temporary State Files
Tests use `tempfile.mktemp()` to create temporary state files that are cleaned up after each test.
## Writing New Tests
### Unit Test Template
```python
import unittest
from unittest.mock import Mock, patch
from bot import MastodonRSSBot
class TestNewFeature(unittest.TestCase):
def setUp(self):
"""Set up test fixtures"""
self.bot = MastodonRSSBot(
client_id='test',
client_secret='test',
access_token='test',
instance_url='https://test.com',
feed_url='https://feed.test/rss.xml',
state_file='/tmp/test_state.txt'
)
def test_feature(self):
"""Test description"""
result = self.bot.some_method()
self.assertEqual(result, expected_value)
```
### Integration Test Template
```python
import unittest
import responses
from bot import MastodonRSSBot
class TestNewIntegration(unittest.TestCase):
@responses.activate
@patch('bot.Mastodon')
def test_integration(self, mock_mastodon):
"""Test description"""
# Mock HTTP responses
responses.add(
responses.GET,
'https://example.com/feed.xml',
body=rss_xml,
status=200
)
# Run test
bot = MastodonRSSBot(...)
result = bot.process_new_entries()
self.assertEqual(result, expected)
```
## Continuous Integration Status
[![Tests](https://img.shields.io/github/actions/workflow/status/aserper/masto-rss/test.yml?style=for-the-badge&logo=github&label=Tests)](https://github.com/aserper/masto-rss/actions/workflows/test.yml)
## Troubleshooting
### Tests Fail Locally But Pass in CI
- Ensure you're using the same Python version
- Check that all dependencies are installed: `pip install -r requirements-test.txt`
- Clear pytest cache: `pytest --cache-clear`
### Coverage Below 80%
- Identify untested code: `pytest --cov=bot --cov-report=term-missing`
- Add tests for the missing lines
- Some error handling paths may be acceptable to skip
### Import Errors
- Ensure the project root is in PYTHONPATH
- Run tests from the project root directory
- Check virtual environment is activated
## Best Practices
1. **Test One Thing**: Each test should verify one specific behavior
2. **Clear Names**: Test names should describe what they're testing
3. **Arrange-Act-Assert**: Structure tests with setup, execution, and verification
4. **Mock External Services**: Never make real HTTP requests or API calls
5. **Clean Up**: Always clean up temporary files and state
6. **Test Edge Cases**: Test both happy paths and error conditions
7. **Keep Tests Fast**: Unit tests should run in milliseconds
8. **Document Complex Tests**: Add comments explaining non-obvious test logic
## Resources
- [pytest documentation](https://docs.pytest.org/)
- [unittest.mock documentation](https://docs.python.org/3/library/unittest.mock.html)
- [responses library](https://github.com/getsentry/responses)
- [Coverage.py documentation](https://coverage.readthedocs.io/)

View File

@@ -1,214 +0,0 @@
# Test Implementation Summary
## Overview
Comprehensive test suite designed for the Masto-RSS bot with unit tests, integration tests, and automated CI/CD testing via GitHub Actions.
## Files Created
### 1. Core Refactoring
- **[bot.py](bot.py)** - Refactored core functionality into testable `MastodonRSSBot` class
- Separated concerns (parsing, posting, state management)
- Dependency injection for easier testing
- Error handling and logging
- Type hints for better code clarity
- **[main.py](main.py)** - Simplified entry point
- Loads configuration from environment
- Instantiates and runs the bot
- Clean separation from core logic
### 2. Test Files
#### [test_bot.py](test_bot.py) - Unit Tests
Contains **20+ unit tests** covering:
- Bot initialization with configuration
- Loading/saving processed entries (with and without existing files)
- Directory creation for state files
- Status formatting from feed entries
- Mastodon posting (success and failure cases)
- Feed parsing (success, errors, malformed data)
- Processing new entries (all new, partially processed, no entries)
- Handling entries without URLs
- Failed posts don't get marked as processed
- Main entry point configuration loading
**Coverage:** Core business logic, edge cases, error handling
#### [test_integration.py](test_integration.py) - Integration Tests
Contains **10+ integration tests** covering:
- End-to-end RSS to Mastodon flow
- Real RSS 2.0 feed parsing
- Real Atom feed parsing
- State persistence across bot runs
- Incremental feed updates (new entries added over time)
- Network error handling (500 errors, timeouts)
- Malformed XML handling
- Different visibility levels (public, unlisted, private, direct)
- Rate limiting error handling
- Retry mechanisms
**Coverage:** Component integration, external service mocking, data flow
### 3. Configuration Files
#### [pytest.ini](pytest.ini)
- Test discovery patterns
- Coverage settings (80% minimum)
- Output formatting
- Test markers (unit, integration, slow)
- Coverage exclusions
#### [requirements-test.txt](requirements-test.txt)
Test dependencies:
- `pytest` - Testing framework
- `pytest-cov` - Coverage plugin
- `pytest-mock` - Mocking utilities
- `responses` - HTTP mocking for integration tests
- `flake8` - Linting
- `black` - Code formatting
- `mypy` - Type checking
- `coverage` - Coverage reporting
#### [.gitignore](.gitignore)
Updated to exclude:
- Test artifacts (`.pytest_cache/`, `htmlcov/`, `coverage.xml`)
- Python cache files
- Virtual environments
- IDE configurations
### 4. CI/CD Pipeline
#### [.github/workflows/test.yml](.github/workflows/test.yml)
Comprehensive GitHub Actions workflow with **5 jobs**:
1. **Unit Tests**
- Runs on Python 3.10, 3.11, 3.12 (matrix)
- Executes unit tests with coverage
- Uploads coverage to Codecov
2. **Integration Tests**
- Runs on Python 3.10, 3.11, 3.12 (matrix)
- Executes integration tests
- Uploads coverage to Codecov
3. **Code Quality**
- Flake8 linting (error detection + complexity)
- Black formatting verification
- Mypy type checking
4. **Docker Build Test**
- Builds Docker image
- Verifies Python installation
- Checks dependencies are installed
5. **All Tests Pass**
- Requires all previous jobs to succeed
- Provides single status check for PRs
### 5. Documentation
#### [TESTING.md](TESTING.md)
Comprehensive testing guide covering:
- Test architecture explanation
- Running tests locally (all, specific, by marker)
- Coverage report generation
- GitHub Actions CI/CD workflow details
- Test coverage requirements (80% minimum)
- Code quality standards
- Writing new tests (templates provided)
- Troubleshooting common issues
- Best practices
#### [README.md](README.md)
Updated with:
- Test status badge
- Link to testing documentation
## Test Statistics
### Coverage Targets
- **Minimum:** 80% code coverage
- **Measured:** `bot.py` and `main.py`
- **Excluded:** Test files, virtual environments
### Test Count
- **Unit Tests:** 20+ tests
- **Integration Tests:** 10+ tests
- **Total:** 30+ tests
### Python Versions Tested
- Python 3.10
- Python 3.11
- Python 3.12
## Key Testing Features
### Mocking Strategy
- **Mastodon API:** Mocked using `unittest.mock` to avoid real API calls
- **RSS Feeds:** Mocked using `responses` library with realistic XML
- **File System:** Uses temporary files that are cleaned up automatically
### Test Data
- Realistic RSS 2.0 and Atom feed examples
- Multiple entry scenarios (new, processed, malformed)
- Error conditions (network failures, API errors, rate limits)
### Continuous Integration
- Runs on every push to `main`
- Runs on all pull requests
- Parallel test execution across Python versions
- Automatic coverage reporting
- Docker image validation
## Running Tests
### Locally
```bash
# Install dependencies
pip install -r requirements-test.txt
# Run all tests
pytest
# Run with coverage
pytest --cov=bot --cov=main --cov-report=html
# Run specific tests
pytest test_bot.py # Unit tests only
pytest test_integration.py # Integration tests only
```
### In CI/CD
Tests run automatically via GitHub Actions:
- **Trigger:** Push to main or pull request
- **Duration:** ~5-10 minutes
- **Matrix:** 3 Python versions × 2 test types = 6 parallel jobs
- **Plus:** Code quality and Docker build validation
## Benefits
1. **Code Quality:** Ensures all changes are tested before merging
2. **Regression Prevention:** Catches bugs before they reach production
3. **Documentation:** Tests serve as executable documentation
4. **Confidence:** Safe to refactor with comprehensive test coverage
5. **Type Safety:** Mypy catches type-related bugs early
6. **Code Style:** Black and flake8 enforce consistency
## Future Enhancements
Potential additions:
- Performance/load testing
- End-to-end tests with real Mastodon test instance
- Security scanning (Bandit, Safety)
- Mutation testing (mutmut)
- Property-based testing (Hypothesis)
- Contract testing for API interactions
## Maintenance
- **Add tests** for all new features
- **Update tests** when changing behavior
- **Keep coverage above 80%**
- **Run tests before committing**
- **Review test failures** in CI before merging

212
bot.py
View File

@@ -1,212 +0,0 @@
"""Mastodon RSS Bot - Core functionality"""
import feedparser
from mastodon import Mastodon
import os
import time
from typing import Set, Optional
class MastodonRSSBot:
"""Bot that monitors RSS feeds and posts updates to Mastodon"""
def __init__(
self,
client_id: str,
client_secret: str,
access_token: str,
instance_url: str,
feed_urls: list[str],
toot_visibility: str = "public",
check_interval: int = 300,
state_file: str = "/state/processed_entries.txt",
):
"""
Initialize the Mastodon RSS bot.
Args:
client_id: Mastodon application client ID
client_secret: Mastodon application client secret
access_token: Mastodon access token
instance_url: URL of the Mastodon instance
feed_urls: List of URLs of the RSS/Atom feeds to monitor
toot_visibility: Visibility level for posts ('public', 'unlisted', 'private', 'direct')
check_interval: Seconds between feed checks
state_file: Path to file storing processed entry URLs
"""
self.feed_urls = feed_urls
self.toot_visibility = toot_visibility
self.check_interval = check_interval
self.state_file = state_file
# Initialize Mastodon client
self.mastodon = Mastodon(
client_id=client_id,
client_secret=client_secret,
access_token=access_token,
api_base_url=instance_url,
)
def load_processed_entries(self) -> Set[str]:
"""
Load the set of processed entry URLs from the state file.
Returns:
Set of URLs that have been processed
"""
try:
with open(self.state_file, "r") as file:
return set(file.read().splitlines())
except FileNotFoundError:
return set()
def save_processed_entries(self, processed_entries: Set[str]) -> None:
"""
Save the set of processed entry URLs to the state file.
Args:
processed_entries: Set of processed entry URLs
"""
# 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)))
def parse_feed(self, feed_url: str) -> Optional[feedparser.FeedParserDict]:
"""
Parse the RSS feed.
Args:
feed_url: URL of the feed to parse
Returns:
Parsed feed object or None if parsing fails
"""
try:
feed = feedparser.parse(feed_url)
if hasattr(feed, "bozo_exception"):
print(
f"Warning: Feed parsing issue for {feed_url}: {feed.bozo_exception}"
)
return feed
except Exception as e:
print(f"Error parsing feed {feed_url}: {e}")
return None
def format_status(self, entry: feedparser.FeedParserDict) -> str:
"""
Format a feed entry as a Mastodon status.
Args:
entry: Feed entry from feedparser
Returns:
Formatted status text
"""
title = entry.get("title", "Untitled")
link = entry.get("link", "")
return f"\n{title}\n\n{link}"
def post_to_mastodon(self, status: str) -> bool:
"""
Post a status to Mastodon.
Args:
status: Status text to post
Returns:
True if successful, False otherwise
"""
try:
self.mastodon.status_post(status, visibility=self.toot_visibility)
return True
except Exception as e:
print(f"Error posting to Mastodon: {e}")
return False
def process_feed(self, feed_url: str, processed_entries: Set[str]) -> int:
"""
Process a single feed for new entries.
Args:
feed_url: URL of the feed to process
processed_entries: Set of already processed entry URLs
Returns:
Number of new entries posted
"""
print(f"Checking feed: {feed_url}")
feed = self.parse_feed(feed_url)
if not feed or not hasattr(feed, "entries"):
print(f"No entries found in feed: {feed_url}")
return 0
new_entries_count = 0
# Process each entry
for entry in feed.entries:
entry_url = entry.get("link", "")
if not entry_url:
print("Skipping entry without URL")
continue
# Check if entry is new
if entry_url not in processed_entries:
title = entry.get("title", "Untitled")
print(f"Found a new RSS item: {title}")
# Format and post status
status = self.format_status(entry)
if self.post_to_mastodon(status):
processed_entries.add(entry_url)
new_entries_count += 1
else:
print(f"Failed to post entry: {title}")
return new_entries_count
def process_new_entries(self) -> int:
"""
Check for new feed entries in all feeds and post them to Mastodon.
Returns:
Total number of new entries posted across all feeds
"""
print("Checking for new RSS items...")
# Load processed entries
processed_entries = self.load_processed_entries()
total_new_entries = 0
for feed_url in self.feed_urls:
total_new_entries += self.process_feed(feed_url, processed_entries)
# Save updated state
if total_new_entries > 0:
self.save_processed_entries(processed_entries)
return total_new_entries
def run(self) -> None:
"""
Main loop: continuously monitor the feed and post new entries.
"""
while True:
try:
count = self.process_new_entries()
if count > 0:
print(f"Posted {count} new entries")
print(f"Sleeping for {self.check_interval} seconds...")
time.sleep(self.check_interval)
except KeyboardInterrupt:
print("\nBot stopped by user")
break
except Exception as e:
print(f"Error in main loop: {e}")
print(f"Retrying in {self.check_interval} seconds...")
time.sleep(self.check_interval)

30
docker-compose.yml Normal file
View File

@@ -0,0 +1,30 @@
services:
masto-rss:
# Use the pre-built image from Docker Hub
image: doorknob2947/masto-rss-advanced:latest
container_name: masto-rss-bot
restart: unless-stopped
volumes:
# This volume persists the history of posted links
- ./state:/state
environment:
# --- Mastodon API Credentials (Required) ---
- MASTODON_CLIENT_ID=YOUR_CLIENT_KEY_HERE
- MASTODON_CLIENT_SECRET=YOUR_CLIENT_SECRET_HERE
- MASTODON_ACCESS_TOKEN=YOUR_ACCESS_TOKEN_HERE
- MASTODON_INSTANCE_URL=[https://mastodon.social](https://mastodon.social)
# --- RSS Feed Configuration (Required) ---
- RSS_FEED_URL=[https://www.theverge.com/rss/index.xml](https://www.theverge.com/rss/index.xml)
# --- Bot Behavior (Optional) ---
- CHECK_INTERVAL=3600 # Time in seconds between checks (e.g., 1 hour)
- MAX_HISTORY_SIZE=500 # Max number of posted links to remember
- TOOT_VISIBILITY=public # public, unlisted, private, or direct
# --- Toot Content (Optional) ---
- MASTO_RSS_HASHTAGS="news tech python" # Space-separated list of hashtags
- TOOT_TEMPLATE='{title}\n\n{link}\n\n{hashtags}'
# --- System (Do not change) ---
- PYTHONUNBUFFERED=1

Binary file not shown.

Before

Width:  |  Height:  |  Size: 178 KiB

180
main.py
View File

@@ -1,83 +1,131 @@
"""Mastodon RSS Bot - Entry point"""
import feedparser
from mastodon import Mastodon
import os
from bot import MastodonRSSBot
import time
from collections import deque
# --- Configuration ---
# Get configuration from environment variables
MASTODON_CLIENT_ID = os.environ.get('MASTODON_CLIENT_ID')
MASTODON_CLIENT_SECRET = os.environ.get('MASTODON_CLIENT_SECRET')
MASTODON_ACCESS_TOKEN = os.environ.get('MASTODON_ACCESS_TOKEN')
MASTODON_INSTANCE_URL = os.environ.get('MASTODON_INSTANCE_URL')
RSS_FEED_URL = os.environ.get('RSS_FEED_URL')
def main():
"""Initialize and run the bot with environment configuration"""
print("Starting Mastodon RSS Bot...")
# Optional configuration with default values
TOOT_VISIBILITY = os.environ.get('TOOT_VISIBILITY', 'public')
CHECK_INTERVAL = int(os.environ.get('CHECK_INTERVAL', 3600))
HASHTAGS = os.environ.get('MASTO_RSS_HASHTAGS', '')
MAX_HISTORY_SIZE = int(os.environ.get('MAX_HISTORY_SIZE', 500))
# Load configuration from environment variables
feed_urls = []
# NEW: Template for the toot format.
# Use {title}, {link}, and {hashtags} as placeholders.
DEFAULT_TEMPLATE = '{title}\n\n{link}\n\n{hashtags}'
TOOT_TEMPLATE = os.environ.get('TOOT_TEMPLATE', DEFAULT_TEMPLATE)
# 1. Legacy single feed URL
if os.environ.get("RSS_FEED_URL"):
feed_urls.append(os.environ["RSS_FEED_URL"])
# File to store the processed entry URLs.
PROCESSED_ENTRIES_FILE = '/state/processed_entries.txt'
# 2. Comma-separated list of feeds
if os.environ.get("RSS_FEEDS"):
feeds = [
url.strip() for url in os.environ["RSS_FEEDS"].split(",") if url.strip()
]
feed_urls.extend(feeds)
# Check if all required configuration is present
if not all([MASTODON_CLIENT_ID, MASTODON_CLIENT_SECRET, MASTODON_ACCESS_TOKEN, MASTODON_INSTANCE_URL, RSS_FEED_URL]):
print("Error: Not all required environment variables are set.")
exit(1)
# 3. File containing list of feeds
feeds_file = os.environ.get("FEEDS_FILE")
if feeds_file and os.path.exists(feeds_file):
# --- Mastodon Initialization ---
try:
with open(feeds_file, "r") as f:
file_feeds = [
line.strip()
for line in f
if line.strip() and not line.startswith("#")
]
feed_urls.extend(file_feeds)
mastodon = Mastodon(
client_id=MASTODON_CLIENT_ID,
client_secret=MASTODON_CLIENT_SECRET,
access_token=MASTODON_ACCESS_TOKEN,
api_base_url=MASTODON_INSTANCE_URL
)
mastodon.account_verify_credentials()
print("Successfully logged in to Mastodon.")
except Exception as e:
print(f"Error reading feeds file {feeds_file}: {e}")
print(f"Error initializing Mastodon client: {e}")
exit(1)
# Deduplicate while preserving order
unique_feed_urls = []
seen = set()
for url in feed_urls:
if url not in seen:
unique_feed_urls.append(url)
seen.add(url)
# --- Functions ---
if not unique_feed_urls:
print(
"Error: No RSS feeds configured. Please set RSS_FEED_URL, RSS_FEEDS, or FEEDS_FILE."
)
return
def load_processed_entries():
"""Loads processed entry URLs from a file into a deque."""
os.makedirs(os.path.dirname(PROCESSED_ENTRIES_FILE), exist_ok=True)
try:
with open(PROCESSED_ENTRIES_FILE, 'r') as file:
lines = file.read().splitlines()
return deque(lines, maxlen=MAX_HISTORY_SIZE)
except FileNotFoundError:
return deque(maxlen=MAX_HISTORY_SIZE)
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_urls=unique_feed_urls,
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"
),
)
def save_processed_entries(processed_entries_deque):
"""Saves the processed entry URLs from the deque to a file."""
with open(PROCESSED_ENTRIES_FILE, 'w') as file:
file.write('\n'.join(processed_entries_deque))
print("Bot configured successfully:")
print(f" Instance: {os.environ['MASTODON_INSTANCE_URL']}")
print(f" Monitoring {len(unique_feed_urls)} feed(s):")
for url in unique_feed_urls:
print(f" - {url}")
print(f" Visibility: {os.environ.get('TOOT_VISIBILITY', 'public')}")
print(f" Check interval: {os.environ.get('CHECK_INTERVAL', '300')} seconds")
print(
f" State file: {os.environ.get('PROCESSED_ENTRIES_FILE', '/state/processed_entries.txt')}"
)
print()
def format_hashtags(hashtag_string):
"""Formats a string of hashtags into a correct list."""
if not hashtag_string:
return ""
clean_string = hashtag_string.strip(' "\'')
tags = filter(None, clean_string.split(' '))
return " ".join([f"#{tag.lstrip('#')}" for tag in tags])
# Start the bot
bot.run()
def check_and_post_new_items():
"""Checks the RSS feed and only posts the latest item if it's new."""
formatted_hashtags = format_hashtags(HASHTAGS)
if formatted_hashtags:
print(f"Hashtags configured: {formatted_hashtags}")
else:
print("INFO: No hashtags configured.")
print(f"INFO: Using toot template: {TOOT_TEMPLATE.replace(chr(10), ' ')}")
processed_entries = load_processed_entries()
while True:
print(f"Checking for new RSS items from: {RSS_FEED_URL}")
feed = feedparser.parse(RSS_FEED_URL)
if feed.bozo:
print(f"Warning: RSS feed may be malformed. Error: {feed.bozo_exception}")
if not feed.entries:
print("No items found in the RSS feed.")
else:
latest_entry = feed.entries[0]
entry_url = latest_entry.get('link')
entry_title = latest_entry.get('title', 'No title')
if not entry_url:
print(f"Skipping latest item '{entry_title}' (no link).")
elif entry_url not in processed_entries:
print(f"Found new latest item: {entry_title}")
# Compose the Mastodon status based on the template
status = TOOT_TEMPLATE.format(
title=entry_title,
link=entry_url,
hashtags=formatted_hashtags
).strip()
try:
print(f"Posting: {status.replace(chr(10), ' ')}")
mastodon.status_post(status, visibility=TOOT_VISIBILITY)
print("Post successful.")
processed_entries.append(entry_url)
save_processed_entries(processed_entries)
except Exception as e:
print(f"Error posting to Mastodon: {e}")
else:
print("The latest item has already been posted.")
print(f"Waiting for {CHECK_INTERVAL} seconds...")
time.sleep(int(CHECK_INTERVAL))
# --- Main Program ---
if __name__ == "__main__":
main()
check_and_post_new_items()

View File

@@ -1,46 +0,0 @@
[pytest]
# Pytest configuration for Masto-RSS
# Test discovery patterns
python_files = test_*.py
python_classes = Test*
python_functions = test_*
# Output options
addopts =
-v
--strict-markers
--tb=short
--cov=bot
--cov=main
--cov-report=term-missing
--cov-report=html
--cov-report=xml
--cov-fail-under=80
# Test paths
testpaths = .
# Markers
markers =
unit: Unit tests that don't require external services
integration: Integration tests that may use mocked external services
slow: Tests that take a long time to run
# Coverage options
[coverage:run]
source = .
omit =
test_*.py
.venv/*
*/__pycache__/*
*/site-packages/*
[coverage:report]
exclude_lines =
pragma: no cover
def __repr__
raise AssertionError
raise NotImplementedError
if __name__ == .__main__.:
if TYPE_CHECKING:

View File

@@ -1,19 +0,0 @@
# Test dependencies for Masto-RSS bot
# Updated to latest versions (2025-12-12)
# Install with: pip install -r requirements-test.txt
# Base requirements
-r requirements.txt
# Testing framework
pytest==9.0.2
pytest-cov==6.0.0
pytest-mock==3.14.0
# Code quality
flake8==7.1.1
black==25.12.0
mypy==1.19.0
# Coverage reporting
coverage==7.6.12

View File

@@ -1,26 +1,28 @@
#
# Updated dependencies to latest versions (2025-12-12)
# Major updates: Mastodon.py 1.8.1 -> 2.1.4, requests 2.32.2 -> 2.32.5
# This file is autogenerated by pip-compile with Python 3.10
# by the following command:
#
# pip-compile
#
blurhash==1.1.4
# via mastodon-py
certifi==2025.11.12
certifi==2024.7.4
# via requests
charset-normalizer==3.3.2
# via requests
decorator==5.1.1
# via mastodon-py
feedparser==6.0.12
feedparser==6.0.11
# via -r requirements.in
idna==3.11
idna==3.7
# via requests
mastodon-py==2.1.4
mastodon-py==1.8.1
# via -r requirements.in
python-dateutil==2.9.0.post0
python-dateutil==2.8.2
# via mastodon-py
python-magic==0.4.27
# via mastodon-py
requests==2.32.5
requests==2.32.2
# via mastodon-py
sgmllib3k==1.0.0
# via feedparser
@@ -28,5 +30,5 @@ six==1.16.0
# via
# mastodon-py
# python-dateutil
urllib3==2.6.2
urllib3==2.2.2
# via requests

View File

@@ -1,418 +0,0 @@
"""Unit tests for Mastodon RSS Bot"""
import unittest
from unittest.mock import Mock, patch, mock_open, MagicMock
import tempfile
import os
from bot import MastodonRSSBot
import feedparser
class TestMastodonRSSBot(unittest.TestCase):
"""Test cases for MastodonRSSBot class"""
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_urls": ["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"])
@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_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"])
# 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"],
)
@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)
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)
@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_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")
@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")
bot = MastodonRSSBot(**self.test_config)
feed = bot.parse_feed("https://example.com/feed.xml")
self.assertIsNone(feed)
@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()
mock_feed.entries = []
mock_parse.return_value = mock_feed
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)
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")
def test_main_loads_legacy_environment_config(self, mock_bot_class):
"""Test that main() loads configuration from legacy environment variable"""
from main import main
mock_bot_instance = Mock()
mock_bot_class.return_value = mock_bot_instance
# Call main (but it will try to run, so we need to handle that)
try:
main()
except Exception:
pass # Expected since we're mocking
# 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_urls=["https://example.com/feed.xml"],
toot_visibility="unlisted",
check_interval=120,
state_file="/tmp/test_state.txt",
)
@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_FEEDS": "http://feed1.com, http://feed2.com",
# No RSS_FEED_URL
"TOOT_VISIBILITY": "public",
},
)
@patch("main.MastodonRSSBot")
def test_main_loads_multiple_feeds_env(self, mock_bot_class):
"""Test that main() loads multiple feeds from environment variable"""
# Ensure RSS_FEED_URL is not set from previous tests or env
if "RSS_FEED_URL" in os.environ:
del os.environ["RSS_FEED_URL"]
from main import main
mock_bot_instance = Mock()
mock_bot_class.return_value = mock_bot_instance
try:
main()
except Exception:
pass
mock_bot_class.assert_called_once()
_, kwargs = mock_bot_class.call_args
self.assertEqual(kwargs["feed_urls"], ["http://feed1.com", "http://feed2.com"])
if __name__ == "__main__":
unittest.main()

View File

@@ -1,244 +0,0 @@
"""Integration tests for Mastodon RSS Bot"""
import unittest
from unittest.mock import Mock, patch
import tempfile
import os
from bot import MastodonRSSBot
class TestRSSFeedIntegration(unittest.TestCase):
"""Integration tests for RSS feed parsing"""
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_urls": ["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"])
@patch("bot.feedparser.parse")
@patch("bot.Mastodon")
def test_end_to_end_rss_to_mastodon(self, mock_mastodon, mock_parse):
"""Test complete flow from RSS feed to Mastodon post"""
# Create mock feed object
mock_feed = Mock()
mock_feed.entries = [
{"title": "First Article", "link": "https://example.com/article1"},
{"title": "Second Article", "link": "https://example.com/article2"},
]
mock_parse.return_value = mock_feed
# Mock Mastodon instance
mock_instance = Mock()
mock_mastodon.return_value = mock_instance
# Create bot and process entries
bot = MastodonRSSBot(**self.test_config)
count = bot.process_new_entries()
# Verify results
self.assertEqual(count, 2)
self.assertEqual(mock_instance.status_post.call_count, 2)
# 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])
@patch("bot.feedparser.parse")
@patch("bot.Mastodon")
def test_atom_feed_parsing(self, mock_mastodon, mock_parse):
"""Test parsing Atom feeds"""
# Create mock Atom feed object
mock_feed = Mock()
mock_feed.entries = [
{"title": "Atom Article", "link": "https://example.com/atom1"}
]
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()
self.assertEqual(count, 1)
calls = mock_instance.status_post.call_args_list
self.assertIn("Atom Article", calls[0][0][0])
@patch("bot.feedparser.parse")
@patch("bot.Mastodon")
def test_persistence_across_runs(self, mock_mastodon, mock_parse):
"""Test that processed entries persist across multiple bot runs"""
# Create mock feed object
mock_feed = Mock()
mock_feed.entries = [{"title": "Article 1", "link": "https://example.com/1"}]
mock_parse.return_value = mock_feed
mock_instance = Mock()
mock_mastodon.return_value = mock_instance
# First run - should post the article
bot1 = MastodonRSSBot(**self.test_config)
count1 = bot1.process_new_entries()
self.assertEqual(count1, 1)
# Second run - should NOT post again (already processed)
bot2 = MastodonRSSBot(**self.test_config)
count2 = bot2.process_new_entries()
self.assertEqual(count2, 0)
# Total posts should be 1
self.assertEqual(mock_instance.status_post.call_count, 1)
@patch("bot.feedparser.parse")
@patch("bot.Mastodon")
def test_incremental_feed_updates(self, mock_mastodon, mock_parse):
"""Test handling of new entries added to feed over time"""
mock_instance = Mock()
mock_mastodon.return_value = mock_instance
# First run - initial feed with 2 articles
mock_feed = Mock()
mock_feed.entries = [
{"title": "Article 1", "link": "https://example.com/1"},
{"title": "Article 2", "link": "https://example.com/2"},
]
mock_parse.return_value = mock_feed
bot = MastodonRSSBot(**self.test_config)
count1 = bot.process_new_entries()
self.assertEqual(count1, 2)
# Second run - updated feed with 1 new article
mock_feed.entries = [
{"title": "Article 3", "link": "https://example.com/3"},
{"title": "Article 2", "link": "https://example.com/2"},
{"title": "Article 1", "link": "https://example.com/1"},
]
# Second run - should only post the new article
count2 = bot.process_new_entries()
self.assertEqual(count2, 1)
# Verify only 3 total posts
self.assertEqual(mock_instance.status_post.call_count, 3)
@patch("bot.feedparser.parse")
@patch("bot.Mastodon")
def test_network_error_handling(self, mock_mastodon, mock_parse):
"""Test handling of network errors when fetching feed"""
# Simulate network error by returning None
mock_parse.return_value = None
mock_instance = Mock()
mock_mastodon.return_value = mock_instance
bot = MastodonRSSBot(**self.test_config)
count = bot.process_new_entries()
# Should handle error gracefully
self.assertEqual(count, 0)
self.assertEqual(mock_instance.status_post.call_count, 0)
@patch("bot.feedparser.parse")
@patch("bot.Mastodon")
def test_malformed_xml_handling(self, mock_mastodon, mock_parse):
"""Test handling of malformed XML feeds"""
# Create mock feed with bozo_exception (feedparser's error indicator)
mock_feed = Mock()
mock_feed.entries = []
mock_feed.bozo_exception = Exception("XML parsing error")
mock_parse.return_value = mock_feed
mock_instance = Mock()
mock_mastodon.return_value = mock_instance
bot = MastodonRSSBot(**self.test_config)
# Should handle malformed feed gracefully
count = bot.process_new_entries()
# feedparser is lenient and may parse some entries
# The important thing is it doesn't crash
self.assertIsInstance(count, int)
class TestMastodonAPIIntegration(unittest.TestCase):
"""Integration tests for Mastodon API interaction"""
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_urls": ["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"])
@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"]
for visibility in visibility_levels:
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)
@patch("bot.Mastodon")
def test_retry_on_rate_limit(self, mock_mastodon):
"""Test that rate limit errors are handled appropriately"""
from mastodon import MastodonRatelimitError
mock_instance = Mock()
# First call raises rate limit error, second succeeds
mock_instance.status_post.side_effect = [
MastodonRatelimitError("Rate limited"),
None,
]
mock_mastodon.return_value = mock_instance
bot = MastodonRSSBot(**self.test_config)
# First attempt should fail
result1 = bot.post_to_mastodon("Test status")
self.assertFalse(result1)
# Second attempt should succeed
result2 = bot.post_to_mastodon("Test status")
self.assertTrue(result2)
if __name__ == "__main__":
unittest.main()