mirror of
https://github.com/aserper/masto-rss.git
synced 2025-12-17 05:15:25 +00:00
Compare commits
21 Commits
0719b713e4
...
patch-1
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
e4e5a22a0f | ||
|
|
fc5a4bec00 | ||
|
|
6ff72f7f97 | ||
|
|
c42d230053 | ||
|
|
22677de465 | ||
|
|
760d84d104 | ||
|
|
faf443b5b7 | ||
|
|
f795520ad6 | ||
|
|
6ae6978d68 | ||
|
|
4c043a596b | ||
|
|
d3c02fddac | ||
|
|
0860fa555d | ||
|
|
7a16534593 | ||
|
|
b4810530ab | ||
|
|
83fe193d67 | ||
|
|
6e947528cb | ||
|
|
2e11ea076a | ||
|
|
43c9ab7bdb | ||
|
|
ec657cb5af | ||
|
|
a96fcf9ca6 | ||
|
|
8ef903a720 |
23
.dockerignore
Normal file
23
.dockerignore
Normal 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
25
.github/workflows/docker-publish.yml
vendored
Normal 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
|
||||
59
.github/workflows/masto-rss.yml
vendored
59
.github/workflows/masto-rss.yml
vendored
@@ -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
|
||||
162
.github/workflows/test.yml
vendored
162
.github/workflows/test.yml
vendored
@@ -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
47
.gitignore
vendored
@@ -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
|
||||
|
||||
24
Dockerfile
24
Dockerfile
@@ -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"]
|
||||
|
||||
301
README.md
301
README.md
@@ -1,233 +1,158 @@
|
||||

|
||||
# 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.
|
||||
|
||||
[](https://github.com/aserper/masto-rss/actions/workflows/masto-rss.yml)
|
||||
[](https://github.com/aserper/masto-rss/actions/workflows/test.yml)
|
||||
[](https://hub.docker.com/r/amitserper/masto-rss)
|
||||
[](https://github.com/aserper/masto-rss/pkgs/container/masto-rss)
|
||||
[](LICENSE)
|
||||
[](https://www.python.org/downloads/)
|
||||
[](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
|
||||
|
||||
```bash
|
||||
docker-compose up -d
|
||||
```
|
||||
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` |
|
||||
|
||||
### 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)
|
||||
### 4. Customizing the Toot Format
|
||||
|
||||
#### Run with Multiple Feeds (Docker)
|
||||
You can change the layout of your posts using the `TOOT_TEMPLATE` variable. Use the following placeholders:
|
||||
|
||||
```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
|
||||
```
|
||||
* `{title}`: The title of the RSS entry.
|
||||
* `{link}`: The URL of the RSS entry.
|
||||
* `{hashtags}`: The configured hashtags.
|
||||
|
||||
## Configuration
|
||||
**Examples in `docker-compose.yml`:**
|
||||
|
||||
All configuration is done via environment variables:
|
||||
* **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}'
|
||||
```
|
||||
|
||||
| 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` |
|
||||
## Running the Bot
|
||||
|
||||
\* At least one of `RSS_FEED_URL`, `RSS_FEEDS`, or `FEEDS_FILE` must be provided.
|
||||
1. **Create the `state` directory.** This is required for the bot to remember which links it has already posted.
|
||||
```bash
|
||||
mkdir ./state
|
||||
```
|
||||
|
||||
### Getting Mastodon API Credentials
|
||||
2. **Start the container** in detached (background) mode:
|
||||
```bash
|
||||
docker-compose up -d
|
||||
```
|
||||
|
||||
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
|
||||
The bot is now running!
|
||||
|
||||
## Building from Source
|
||||
## Managing the Bot
|
||||
|
||||
### Build Locally
|
||||
* **View logs in real-time:**
|
||||
```bash
|
||||
docker-compose logs -f
|
||||
```
|
||||
|
||||
```bash
|
||||
git clone https://github.com/aserper/masto-rss.git
|
||||
cd masto-rss
|
||||
docker build -t masto-rss .
|
||||
```
|
||||
* **Stop the container:**
|
||||
```bash
|
||||
docker-compose down
|
||||
```
|
||||
|
||||
### Build Multiarch Images
|
||||
* **Restart the container:**
|
||||
```bash
|
||||
docker-compose restart
|
||||
```
|
||||
|
||||
```bash
|
||||
# Set up buildx
|
||||
docker buildx create --use
|
||||
## Updating the Bot
|
||||
|
||||
# Build for both architectures
|
||||
docker buildx build \
|
||||
--platform linux/amd64,linux/arm64 \
|
||||
-t yourusername/masto-rss:latest \
|
||||
--push \
|
||||
.
|
||||
```
|
||||
To update to the latest version of the image from Docker Hub:
|
||||
|
||||
## Running Without Docker
|
||||
1. **Pull the latest image:**
|
||||
```bash
|
||||
docker-compose pull
|
||||
```
|
||||
2. **Restart the container** to apply the update:
|
||||
```bash
|
||||
docker-compose up -d
|
||||
```
|
||||
|
||||
If you prefer to run the bot directly with Python:
|
||||
## Acknowledgements
|
||||
|
||||
```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
|
||||
```
|
||||
|
||||
> **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.
|
||||
|
||||
## 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.
|
||||
|
||||
235
TESTING.md
235
TESTING.md
@@ -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
|
||||
|
||||
[](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/)
|
||||
214
TEST_SUMMARY.md
214
TEST_SUMMARY.md
@@ -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
212
bot.py
@@ -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
30
docker-compose.yml
Normal 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
|
||||
BIN
header.jpg
BIN
header.jpg
Binary file not shown.
|
Before Width: | Height: | Size: 178 KiB |
188
main.py
188
main.py
@@ -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):
|
||||
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)
|
||||
except Exception as e:
|
||||
print(f"Error reading feeds file {feeds_file}: {e}")
|
||||
|
||||
# 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)
|
||||
|
||||
if not unique_feed_urls:
|
||||
print(
|
||||
"Error: No RSS feeds configured. Please set RSS_FEED_URL, RSS_FEEDS, or FEEDS_FILE."
|
||||
)
|
||||
return
|
||||
|
||||
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"
|
||||
),
|
||||
# --- Mastodon Initialization ---
|
||||
try:
|
||||
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 initializing Mastodon client: {e}")
|
||||
exit(1)
|
||||
|
||||
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()
|
||||
# --- Functions ---
|
||||
|
||||
# Start the bot
|
||||
bot.run()
|
||||
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)
|
||||
|
||||
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))
|
||||
|
||||
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])
|
||||
|
||||
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()
|
||||
|
||||
46
pytest.ini
46
pytest.ini
@@ -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:
|
||||
@@ -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
|
||||
@@ -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
|
||||
|
||||
418
test_bot.py
418
test_bot.py
@@ -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()
|
||||
@@ -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()
|
||||
Reference in New Issue
Block a user