Testing Framework

The Corgi Recommender Service employs a comprehensive, multi-layered testing strategy to ensure reliability, performance, and security. Our commitment to quality is reflected in our 100% test pass requirement—every test must succeed before code can be merged.

Testing Philosophy

Zero-Tolerance Quality Gate

Our CI/CD pipeline enforces a strict quality gate:

• 100% Test Success Required: All 396+ tests must pass
• No Performance Regressions: Tests must complete within established thresholds
• Security First: Automated vulnerability scanning on every commit
• Continuous Validation: Automated health checks and browser testing

This approach ensures that the main branch is always in a deployable state.

Test Categories

We organize tests into distinct categories for targeted execution:

# pytest markers from pytest.ini
markers =
    unit: Core functionality tests
    integration: Cross-component tests
    performance: Speed and resource tests
    security: Vulnerability tests
    api: REST API endpoint tests
    db: Database operation tests
    auth: Authentication flow tests
    privacy: Data protection tests

Unit Testing

Running Unit Tests

Unit tests verify individual components in isolation:

# Run all unit tests
pytest -m unit

# Run specific test file
pytest tests/test_recommendations.py

# Run with coverage report
pytest --cov=. --cov-report=term-missing

# Run tests in parallel for speed
pytest -n auto

Writing Unit Tests

Follow our conventions for consistency:

# tests/test_example.py
import pytest
from unittest.mock import Mock, patch

class TestRecommendationEngine:
    """Test suite for recommendation engine."""

    @pytest.fixture
    def mock_user(self):
        """Fixture for test user data."""
        return {
            'id': 'test_user_123',
            'preferences': ['technology', 'science']
        }

    def test_generate_recommendations(self, mock_user):
        """Test recommendation generation for a user."""
        # Arrange
        from core.ranking_algorithm import generate_recommendations
        expected_count = 20

        # Act
        recommendations = generate_recommendations(
            user_id=mock_user['id'],
            limit=expected_count
        )

        # Assert
        assert len(recommendations) == expected_count
        assert all(0 <= r.score <= 1 for r in recommendations)
        assert recommendations == sorted(
            recommendations, 
            key=lambda x: x.score, 
            reverse=True
        )

Test Naming Convention

Use descriptive test names that explain what is being tested: - test_<function>_<scenario>_<expected_result> - Example: test_authentication_with_expired_token_returns_401

Integration Testing

API Integration Tests

Test complete API flows end-to-end:

# Run all API integration tests
pytest -m api

# Run with real database (requires setup)
POSTGRES_DB=corgi_test pytest tests/test_api_flow.py

# Test specific endpoint group
pytest tests/test_mastodon_api.py -v

Database Integration Tests

Verify database operations and migrations:

# Run database tests
pytest -m db

# Test with specific database
pytest tests/test_db_connection.py --db-url postgresql://localhost/corgi_test

Example Integration Test

# tests/test_api_integration.py
@pytest.mark.integration
@pytest.mark.api
class TestRecommendationAPI:
    """Integration tests for recommendation API."""

    @pytest.fixture
    def client(self):
        """Create test client."""
        from app import app
        app.config['TESTING'] = True
        with app.test_client() as client:
            yield client

    def test_get_recommendations_flow(self, client):
        """Test complete recommendation flow."""
        # Authenticate
        auth_response = client.post('/api/v1/oauth/token', json={
            'grant_type': 'authorization_code',
            'code': 'test_code'
        })
        assert auth_response.status_code == 200
        token = auth_response.json['access_token']

        # Get recommendations
        rec_response = client.get(
            '/api/v1/recommendations',
            headers={'Authorization': f'Bearer {token}'}
        )
        assert rec_response.status_code == 200
        assert 'recommendations' in rec_response.json

Performance Testing

Benchmark Tests

Monitor performance across different load scenarios:

# Run performance benchmarks
pytest -m performance

# Run specific benchmark suite
pytest tests/test_performance_benchmarks.py -v

# Generate performance report
pytest tests/test_performance_monitoring.py --benchmark-json=perf.json

Performance Thresholds

Our performance tests enforce strict thresholds:

Metric	Light Load	Standard Load	Heavy Load
P95 Latency	< 50ms	< 100ms	< 200ms
Throughput	> 50 RPS	> 25 RPS	> 15 RPS
Success Rate	> 99%	> 98%	> 95%

Load Testing

Use Locust for realistic load testing:

# Run load test
locust -f tests/locustfile_recommendations.py --host=http://localhost:5002

# Headless load test with specific parameters
locust -f tests/locustfile_recommendations.py \
    --headless \
    --users 100 \
    --spawn-rate 10 \
    --run-time 5m

Security Testing

Automated Security Scans

Security testing runs automatically in CI:

# Run all security tests
make security-scan

# Individual security tools
pip-audit                    # Check for vulnerable dependencies
bandit -r . -ll             # Static security analysis
safety check                # Additional vulnerability database
semgrep --config=auto .     # Advanced pattern matching

Security Test Example

# tests/test_security_interactions.py
@pytest.mark.security
class TestSecurityFeatures:
    """Test security controls and protections."""

    def test_sql_injection_prevention(self, client):
        """Verify SQL injection attempts are blocked."""
        malicious_input = "'; DROP TABLE users; --"

        response = client.get(
            f'/api/v1/search?q={malicious_input}',
            headers={'Authorization': 'Bearer valid_token'}
        )

        # Should sanitize input, not execute SQL
        assert response.status_code in [200, 400]
        assert 'error' not in response.json

        # Verify database is intact
        with get_db_connection() as conn:
            cursor = conn.cursor()
            cursor.execute("SELECT 1 FROM users LIMIT 1")
            assert cursor.fetchone() is not None

End-to-End Browser Testing

Intelligent Browser Agent

Our browser testing uses Playwright for realistic user simulation:

# Run browser tests (headless)
make dev-test

# Run with visible browser for debugging
make dev-test-headed

# Continuous browser monitoring
make dev-test-continuous

Key Browser Tests

The browser agent validates critical user flows:

1. API Connection Test: Ensures frontend connects to backend
2. OAuth Flow Test: Validates authentication process
3. Recommendation Display: Confirms recommendations appear in UI

# Example from browser_agent.py
async def test_elk_corgi_connection(self, page: Page) -> TestResult:
    """Critical test: Check if ELK frontend can connect to Corgi API."""
    # Navigate to frontend
    await page.goto(self.frontend_url)

    # Check for offline mode (FAIL condition)
    offline_messages = [
        msg for msg in self.console_messages 
        if "[Corgi] Running in offline mode" in msg.text
    ]

    if offline_messages:
        return TestResult(
            name="ELK-Corgi API Connection",
            passed=False,
            error="API CONNECTION BROKEN: Frontend in offline mode!"
        )

Validation Framework

Comprehensive System Validation

The validation framework checks system integrity:

# Run full validation suite
make validate

# Dry run (no side effects)
make dry-validate

# Quick health check
make check

# Nightly comprehensive check
make nightly-check

Validation Components

• API Endpoint Validation: All endpoints respond correctly
• Database Integrity: Schema and data consistency
• Configuration Validation: Environment and settings
• Documentation Validation: API docs match implementation

Continuous Testing

Automated Development Workflow

During development, tests run automatically:

# Start development with monitoring
make dev

# This enables:
# - Continuous health checks every 30 seconds
# - Browser testing on file changes
# - Real-time error detection
# - Screenshot capture on failures

Pre-Commit Testing

Git hooks ensure quality before commits:

#!/bin/sh
# .git/hooks/pre-commit
echo "Running health check before commit..."
./dev-monitor health
if [ $? -ne 0 ]; then
    echo "Health check failed! Fix issues before committing."
    exit 1
fi

Test Organization

Directory Structure

Tests mirror the source code structure:

tests/
├── conftest.py              # Shared fixtures
├── test_*.py               # Test files
├── fixtures/               # Test data and helpers
├── config_tests/           # Configuration tests
└── performance_baseline.md # Performance benchmarks

Naming Conventions

• Test Files: test_<module>.py
• Test Classes: Test<Component>
• Test Methods: test_<scenario>_<expected_result>
• Fixtures: mock_<resource> or fake_<data>

Running Tests in CI/CD

Quality Gate Workflow

The CI pipeline runs comprehensive checks:

# From .github/workflows/quality-gate.yml
- name: Run full test suite (ZERO FAILURES REQUIRED)
  run: |
    python -m pytest \
      --tb=short \
      --cov=. \
      --cov-report=xml \
      --timeout=300 \
      --maxfail=1 \
      -v

Test Execution Strategy

1. Parallel Execution: Tests run on Python 3.11 and 3.12
2. Fast Fail: Stops on first failure to save time
3. Timeout Protection: 300-second limit per test
4. Coverage Tracking: Uploaded to Codecov

Best Practices

Writing Effective Tests

1. Isolation: Each test should be independent
2. Clarity: Test one thing per test method
3. Speed: Mock external dependencies
4. Reliability: No flaky tests allowed
5. Coverage: Aim for critical path coverage

Using Fixtures

@pytest.fixture
def authenticated_client(client):
    """Client with valid authentication."""
    client.headers['Authorization'] = 'Bearer test_token'
    client.headers['X-Mastodon-Instance'] = 'mastodon.social'
    return client

def test_protected_endpoint(authenticated_client):
    """Test endpoint requires authentication."""
    response = authenticated_client.get('/api/v1/user/profile')
    assert response.status_code == 200

Mocking Best Practices

@patch('external_service.api_call')
def test_with_mock(mock_api):
    """Test with mocked external service."""
    # Configure mock
    mock_api.return_value = {'status': 'success'}

    # Test your code
    result = your_function()

    # Verify mock was called correctly
    mock_api.assert_called_once_with(expected_params)

Troubleshooting

Common Issues

Tests Failing Locally but Passing in CI - Check environment variables: cat .env.test - Ensure test database is clean: make reset-db - Verify dependencies match: pip freeze | diff requirements.txt -

Slow Test Execution - Run in parallel: pytest -n auto - Skip slow tests temporarily: pytest -m "not slow" - Profile test execution: pytest --durations=10

Flaky Tests - Add retry logic: @pytest.mark.flaky(reruns=3) - Increase timeouts for network operations - Use proper test isolation

Summary

The Corgi testing framework ensures code quality through:

• Comprehensive Coverage: Unit, integration, performance, and security tests
• Automated Execution: CI/CD pipeline with strict quality gates
• Continuous Monitoring: Real-time testing during development
• Clear Standards: Consistent naming and organization
• Zero Tolerance: 100% test success requirement

By following these testing practices, we maintain a robust, reliable recommendation service that users can trust.

---

Quick Test Commands

• pytest - Run all tests
• make validate - Full system validation
• make dev-test - Browser testing
• make check - Quick health check