Skip to content

Testing Guide

FenLiu has a comprehensive test suite with 305+ tests covering all major functionality.

Running Tests

Quick Test

pytest

With Coverage

pytest --cov=src/fenliu tests/

Specific Test File

pytest tests/test_spam_scoring.py

Specific Test

pytest tests/test_spam_scoring.py::TestSpamScoring::test_high_link_count

Watch Mode

pytest-watch

Test Structure

Tests organized by module:

  • test_spam_scoring.py - Spam detection rules
  • test_models.py - Database models
  • test_api_*.py - API endpoints
  • test_services.py - Business logic
  • test_export_eligibility.py - Export filters
  • test_curated.py - Queue API

Writing Tests

Test Pattern

import pytest
from sqlalchemy.orm import Session
from fenliu.models import Post
from fenliu.services.spam_scoring import score_post

def test_high_link_count_increases_spam_score(db_session: Session):
    """Test that posts with many links score higher for spam."""
    post = Post(
        post_id="https://example.com/123",
        content="Check these links: https://a.com https://b.com https://c.com",
        author_username="user",
        instance="mastodon.social"
    )
    db_session.add(post)
    db_session.flush()

    score = score_post(post, db_session)

    assert score > 50  # High link count means higher spam score

Key Fixtures

db_session: Fresh database session for each test 

@pytest.fixture
def db_session():
    """Provide a clean database session."""

sample_post: Pre-created test post 

@pytest.fixture
def sample_post(db_session: Session) -> Post:
    """Create a sample post for testing."""
    post = Post(
        post_id="https://example.com/123",
        content="Sample post content",
        author_username="testuser",
        instance="mastodon.social"
    )
    db_session.add(post)
    db_session.flush()
    return post

sample_stream: Pre-created test stream 

@pytest.fixture
def sample_stream(db_session: Session) -> HashtagStream:
    """Create a sample stream for testing."""
    stream = HashtagStream(
        hashtag="python",
        instance="mastodon.social",
        active=True
    )
    db_session.add(stream)
    db_session.flush()
    return stream

Testing Services

Spam Scoring

def test_spam_scoring_rule_detection(db_session: Session):
    """Test individual spam scoring rules."""
    post = Post(
        post_id="https://example.com/1",
        content="BUY NOW!!! BUY NOW!!! BUY NOW!!!",  # Repetition
        author_username="spammer",
        instance="mastodon.social"
    )
    db_session.add(post)
    db_session.flush()

    score = score_post(post, db_session)
    assert 70 <= score <= 100  # High spam confidence

Fediverse Client

Mock external Mastodon API calls:

from unittest.mock import AsyncMock, patch

@pytest.mark.asyncio
async def test_fetch_posts_with_hashtag(db_session: Session):
    """Test fetching posts from Mastodon instance."""
    with patch('fenliu.services.fediverse.httpx.AsyncClient') as mock_client:
        mock_response = AsyncMock()
        mock_response.json.return_value = {
            "value": [
                {
                    "id": "123",
                    "content": "Test post",
                    "attributedTo": "https://mastodon.social/users/user"
                }
            ]
        }
        mock_client.return_value.__aenter__.return_value.get = AsyncMock(
            return_value=mock_response
        )

        # Test your fetch logic

Export Eligibility

def test_blocked_user_excluded_from_export(db_session: Session):
    """Test that blocked users are excluded from export."""
    # Create blocked user
    blocked = BlockedUser(account_identifier="@spam@mastodon.social")
    db_session.add(blocked)

    # Create post from blocked user
    post = Post(
        post_id="https://example.com/1",
        content="Spam post",
        author_username="spam",
        instance="mastodon.social"
    )
    db_session.add(post)
    db_session.flush()

    # Check eligibility
    result = check_reblog_filters(post, db_session)

    assert not result.eligible
    assert "blocked user" in result.reason.lower()

Testing API Endpoints

Using Test Client

from fastapi.testclient import TestClient
from fenliu.main import app

@pytest.fixture
def client():
    return TestClient(app)

def test_list_streams(client, db_session, sample_stream):
    """Test listing hashtag streams."""
    response = client.get("/api/v1/hashtags")

    assert response.status_code == 200
    data = response.json()
    assert len(data) > 0
    assert data[0]["hashtag"] == "python"

Testing POST Endpoints

def test_create_stream(client):
    """Test creating a new stream."""
    response = client.post(
        "/api/v1/hashtags",
        json={
            "hashtag": "python",
            "instance": "mastodon.social",
            "active": True
        }
    )

    assert response.status_code == 201
    data = response.json()
    assert data["hashtag"] == "python"

Testing Error Cases

def test_create_stream_validation_error(client):
    """Test validation on stream creation."""
    response = client.post(
        "/api/v1/hashtags",
        json={
            "hashtag": "",  # Empty hashtag
            "instance": "mastodon.social"
        }
    )

    assert response.status_code == 422
    assert "detail" in response.json()

Database Testing

Using Fresh Database

Each test gets a fresh in-memory database:

@pytest.fixture
def db_session():
    """Create fresh database for each test."""
    # Uses test database URL from conftest.py
    # Automatically rolled back after test

Testing Relationships

def test_stream_post_relationship(db_session: Session):
    """Test that posts belong to streams."""
    stream = HashtagStream(hashtag="python", instance="mastodon.social")
    post = Post(
        post_id="https://example.com/1",
        content="Python tips",
        author_username="user",
        instance="mastodon.social",
        stream=stream
    )

    db_session.add(stream)
    db_session.add(post)
    db_session.flush()

    retrieved = db_session.get(Post, post.id)
    assert retrieved.stream.hashtag == "python"

Integration Tests

Full Workflow Test

@pytest.mark.asyncio
async def test_fetch_and_review_workflow(client, db_session):
    """Test complete fetch → review → queue flow."""
    # 1. Create a stream
    stream_response = client.post(
        "/api/v1/hashtags",
        json={"hashtag": "python", "instance": "mastodon.social"}
    )
    stream_id = stream_response.json()["id"]

    # 2. Mock fetch posts
    with patch('fenliu.services.fediverse.fetch_posts') as mock_fetch:
        mock_fetch.return_value = [
            Post(
                post_id="https://example.com/1",
                content="Great Python article",
                author_username="user",
                instance="mastodon.social"
            )
        ]

        # 3. Fetch posts
        fetch_response = client.post(f"/api/v1/hashtags/{stream_id}/fetch")
        assert fetch_response.status_code == 200

    # 4. List posts
    posts_response = client.get("/api/v1/posts")
    posts = posts_response.json()
    assert len(posts) > 0

    # 5. Approve a post
    post_id = posts[0]["id"]
    approve_response = client.patch(
        f"/api/v1/posts/{post_id}",
        json={"approved": True}
    )
    assert approve_response.status_code == 200

    # 6. Check queue
    queue_response = client.get("/api/v1/curated/next")
    assert queue_response.status_code == 200

Testing Best Practices

Use Descriptive Test Names

# Good
def test_spam_score_increases_with_repetitive_content():
    pass

# Bad
def test_spam():
    pass

Arrange-Act-Assert Pattern

def test_example():
    # Arrange: Set up test data
    post = create_test_post("content")

    # Act: Perform the action
    score = calculate_spam_score(post)

    # Assert: Check the result
    assert score > 50

Test One Thing

Each test should verify one behavior. Don't test multiple scenarios in one test.

Use Fixtures

Avoid repeating test setup with fixtures:

@pytest.fixture
def test_post(db_session):
    """Reusable test post."""
    post = Post(...)
    db_session.add(post)
    db_session.flush()
    return post

Mock External Calls

Don't make real HTTP requests to Mastodon. Mock them:

@patch('fenliu.services.fediverse.httpx.AsyncClient')
async def test_fetch(mock_client):
    mock_response = AsyncMock()
    mock_response.json.return_value = {...}

Test Coverage

Current coverage: 305+ tests across all major components.

Check coverage: 

pytest --cov=src/fenliu tests/ --cov-report=html
open htmlcov/index.html

Continuous Integration

Tests run automatically on every commit via .woodpecker.yml.

Before pushing: 

# Run all checks
nox

This runs: - Linting (ruff check) - Formatting (ruff format) - Type checking (ty) - Tests (pytest)

Debugging Tests

def test_example():
    result = some_function()
    print(f"Debug: {result}")  # Shows with -s flag
    assert result == expected

Run with output: 

pytest -s tests/test_example.py::test_example

Use pdb

def test_example():
    result = some_function()
    breakpoint()  # Stops here in debugger
    assert result == expected

Run and debug: 

pytest -s tests/test_example.py::test_example

Check Database State

def test_example(db_session):
    post = create_post()
    db_session.flush()

    # Inspect database
    all_posts = db_session.query(Post).all()
    print(f"Posts in DB: {len(all_posts)}")

Next Steps