Testing Guide

# Activate virtual environment FIRST
source venv/bin/activate

# Run all tests
pytest

# Run with verbose output
pytest -v

# Run with output capture disabled (see print statements)
pytest -s

# Run with coverage report
pytest --cov=pipeline --cov=api --cov-report=html

# Run only unit tests (fast)
pytest -m unit

# Run only integration tests
pytest -m integration

# Skip slow tests
pytest -m "not slow"

# Run unit tests that are not slow
pytest -m "unit and not slow"

pipeline/
├── steps/
│   ├── template_parser/
│   │   ├── main.py
│   │   └── tests/
│   │       ├── __init__.py
│   │       ├── test_template_parser.py
│   │       └── test_template_parser_logging.py
│   ├── web_scraper/
│   │   ├── main.py
│   │   └── tests/
│   │       ├── __init__.py
│   │       └── test_full_pipeline.py
│   ├── arxiv_helper/
│   │   ├── main.py
│   │   └── tests/
│   │       └── __init__.py
│   └── email_composer/
│       ├── main.py
│       └── tests/
│           ├── __init__.py
│           ├── test_email_composer.py
│           ├── test_db_insert.py
│           └── test_db_performance.py
├── tests/
│   └── test_full_pipeline_request.py
api/
tests/
├── integration/
│   ├── test_email_api.py
│   ├── test_templates_api.py
│   └── test_infrastructure.py

# pipeline/steps/template_parser/tests/test_template_parser.py

import pytest
import logfire
from uuid import uuid4

from pipeline.steps.template_parser.main import TemplateParserStep
from pipeline.models.core import PipelineData, TemplateType


@pytest.mark.unit
@pytest.mark.asyncio
async def test_research_template_parsing():
    """
    Test parsing of RESEARCH type template with real Anthropic API.

    Expected behavior:
    - Returns success=True
    - Correctly identifies template_type as RESEARCH
    - Generates 1-2 relevant search terms
    """
    logfire.info("Starting test: test_research_template_parsing")

    # Arrange
    step = TemplateParserStep()
    data = PipelineData(
        task_id=str(uuid4()),
        user_id=str(uuid4()),
        email_template="Dear {{name}}, I love your work on {{research}}!",
        recipient_name="Dr. Jane Smith",
        recipient_interest="machine learning"
    )

    # Act
    with logfire.span("test_research_template"):
        result = await step.execute(data)

    # Assert
    assert result.success is True
    assert data.template_type == TemplateType.RESEARCH
    assert len(data.search_terms) >= 1
    assert len(data.search_terms) <= 2

    logfire.info(
        "Test passed",
        template_type=data.template_type.value,
        search_terms=data.search_terms
    )

# tests/integration/test_email_api.py

import pytest
from httpx import AsyncClient


@pytest.mark.integration
@pytest.mark.asyncio
async def test_email_generation_api():
    """
    Test complete email generation through API endpoint.

    Requires:
    - Database connection
    - Valid JWT token
    - Anthropic API access
    """
    async with AsyncClient(base_url="http://localhost:8000") as client:
        response = await client.post(
            "/api/email/generate",
            headers={"Authorization": f"Bearer {jwt_token}"},
            json={
                "email_template": "Hi {{name}}, interested in {{research}}!",
                "recipient_name": "Dr. Test",
                "recipient_interest": "AI"
            }
        )

    assert response.status_code == 200
    assert "task_id" in response.json()

import pytest
from pipeline.models.core import PipelineData
from uuid import uuid4


@pytest.fixture
def research_template():
    """Email template requiring research papers (RESEARCH type)"""
    return """
    Dear {{name}},
    I came across your groundbreaking work on {{research_area}}.
    Would love to discuss {{specific_application}}.
    Best regards,
    {{my_name}}
    """.strip()


@pytest.fixture
def pipeline_data(research_template):
    """Create PipelineData instance for testing"""
    return PipelineData(
        task_id=str(uuid4()),
        user_id=str(uuid4()),
        email_template=research_template,
        recipient_name="Dr. Jane Smith",
        recipient_interest="neural networks"
    )


@pytest.mark.unit
@pytest.mark.asyncio
async def test_with_fixture(pipeline_data):
    """Test using fixtures"""
    assert pipeline_data.recipient_name == "Dr. Jane Smith"
    assert len(pipeline_data.email_template) > 0

# All tests
make test

# Unit tests only
make test-unit

# Integration tests only
make test-integration

# With coverage report (htmlcov/index.html)
make test-coverage

# Show test names and status
pytest -v

# Show print statements and logs
pytest -s

# Show local variables on failure
pytest -v --showlocals

# Combine flags
pytest -v -s --showlocals

# Generate HTML coverage report
pytest --cov=pipeline --cov=api --cov-report=html

# Open report
open htmlcov/index.html  # macOS
xdg-open htmlcov/index.html  # Linux

# Terminal report only
pytest --cov=pipeline --cov=api --cov-report=term

[pytest]
# Test discovery paths
testpaths =
    pipeline
    api
    scripts

# Python files/directories to search for tests
python_files = test_*.py *_test.py
python_classes = Test*
python_functions = test_*

# Async configuration
asyncio_mode = auto
asyncio_default_fixture_loop_scope = function

# Output and reporting
addopts =
    -ra                    # Show summary of all test outcomes
    -v                     # Verbose output
    --showlocals           # Show local variables in tracebacks
    --strict-markers       # Fail on unknown markers
    --capture=no           # Better async support
    --disable-warnings     # Reduce noise

# Markers
markers =
    slow: marks tests as slow (deselect with '-m "not slow"')
    integration: marks tests requiring external services
    unit: marks tests as unit tests (fast, no external dependencies)
    asyncio: marks async tests

import pytest


@pytest.mark.asyncio
async def test_async_function():
    result = await some_async_function()
    assert result.success

import pytest
from pipeline.core.exceptions import ValidationError, StepExecutionError


@pytest.mark.asyncio
async def test_empty_template_validation():
    """Test that empty templates are rejected"""
    data = PipelineData(
        task_id="test",
        user_id="user",
        email_template="",  # Empty
        recipient_name="Dr. Test",
        recipient_interest="testing"
    )

    with pytest.raises(StepExecutionError) as exc_info:
        await step.execute(data)

    assert isinstance(exc_info.value.original_error, ValidationError)
    assert "empty" in str(exc_info.value.original_error).lower()

import logfire


@pytest.mark.asyncio
async def test_with_logging():
    """Test with structured logging"""
    logfire.info("Starting test: test_with_logging")

    with logfire.span("test_operation"):
        result = await operation()
        logfire.info("Operation completed", result=result)

    assert result.success
    logfire.info("Test passed")

# Generate coverage report
pytest --cov=pipeline --cov=api --cov-report=term --cov-report=html

# View HTML report
open htmlcov/index.html

# View terminal report with missing lines
pytest --cov=pipeline --cov-report=term-missing

# .github/workflows/test.yml
name: Tests

on: [push, pull_request]

jobs:
  test:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v3
      - uses: actions/setup-python@v4
        with:
          python-version: '3.13'
      - run: pip install -r requirements.txt
      - run: pytest -v --cov=pipeline --cov=api

# Show local variables on failure
pytest -v --showlocals

# Drop into debugger on failure
pytest --pdb

# Show full diff for assertion failures
pytest -vv

# Only run failed tests from last run
pytest --lf

# Run failed tests first, then others
pytest --ff

# See what tests would run (useful for debugging imports)
pytest --collect-only

# See what tests match a marker
pytest -m unit --collect-only