Testing

This guide covers testing strategies, setup, and best practices for the Quantbot project.

Overview

Quantbot uses a comprehensive testing approach with multiple types of tests:

• Unit Tests: Test individual components and functions
• Integration Tests: Test component interactions and API endpoints
• Evaluation Tests: Test AI agent behavior and accuracy
• End-to-End Tests: Full system testing (frontend + backend)

Testing Framework

Backend Testing Stack

• pytest: Main testing framework with async support
• pytest-asyncio: For testing async functions
• pytest-cov: Code coverage reporting
• factory-boy: Test data factories
• faker: Generate realistic test data
• httpx: HTTP client for API testing

Configuration

Testing is configured in pyproject.toml:

[tool.pytest.ini_options]
minversion = "6.0"
addopts = "-ra -q --strict-markers --strict-config"
testpaths = ["tests", "evals"]
python_files = ["test_*.py", "*_test.py"]
python_classes = ["Test*"]
python_functions = ["test_*"]
asyncio_default_fixture_loop_scope = "function"
markers = [
    "slow: marks tests as slow (deselect with '-m \"not slow\"')",
    "integration: marks tests as integration tests",
    "unit: marks tests as unit tests", 
    "eval: marks tests as evaluation tests",
]

Test Structure

Directory Organization

Quantbot-BE/
├── tests/                    # Unit and integration tests
│   ├── unit/                # Unit tests
│   │   ├── test_services/   # Service layer tests
│   │   ├── test_models/     # Model tests
│   │   └── test_utils/      # Utility function tests
│   ├── integration/         # Integration tests
│   │   ├── test_api/        # API endpoint tests
│   │   ├── test_database/   # Database tests
│   │   └── test_external/   # External API tests
│   ├── factories.py         # Test data factories
│   └── conftest.py          # Shared test configuration
└── evals/                   # AI evaluation tests
    ├── eval_tests/          # Python evaluation tests
    ├── eval_json/           # Test case definitions
    └── __init__.py

Running Tests

Quick Commands

cd Quantbot-BE

# Run all tests with coverage
uv run pytest --cov=app --cov-report=term-missing tests/

# Run specific test types
uv run pytest tests/unit/ -v                # Unit tests only
uv run pytest tests/integration/ -v         # Integration tests only
uv run pytest evals/eval_tests/ -v          # AI evaluation tests

# Run with specific markers
uv run pytest -m "not slow"                 # Skip slow tests
uv run pytest -m "unit"                     # Run only unit tests
uv run pytest -m "integration"              # Run only integration tests
uv run pytest -m "eval"                     # Run only evaluation tests

Detailed Test Commands

# Run tests with verbose output and stop on first failure
uv run pytest -v -x tests/

# Run tests in parallel (if pytest-xdist is installed)
uv run pytest -n auto tests/

# Run specific test file
uv run pytest tests/unit/test_services/test_auth_service.py

# Run specific test function
uv run pytest tests/unit/test_services/test_auth_service.py::test_create_user

# Run with coverage and generate HTML report
uv run pytest --cov=app --cov-report=html tests/
open htmlcov/index.html  # View coverage report

# Run with profiling (to identify slow tests)
uv run pytest --durations=10 tests/

# Run in Docker container
docker compose exec backend uv run pytest --cov=app tests/

Coverage Goals

Target coverage levels: - Overall: 90%+ - Services: 95%+ - Models: 85%+ - API Endpoints: 90%+

Writing Tests

Unit Test Example

# tests/unit/test_services/test_auth_service.py
import pytest
from unittest.mock import AsyncMock, patch
from app.services.auth_service import AuthService
from app.models.user import User
from tests.factories import UserFactory

class TestAuthService:
    @pytest.fixture
    def auth_service(self, mock_db_session):
        return AuthService(db=mock_db_session)

    @pytest.mark.asyncio
    async def test_create_user_success(self, auth_service):
        # Arrange
        user_data = UserFactory.build()

        # Act
        result = await auth_service.create_user(
            email=user_data.email,
            password="test123",
            full_name=user_data.full_name
        )

        # Assert
        assert result.email == user_data.email
        assert result.full_name == user_data.full_name
        assert result.id is not None

    @pytest.mark.asyncio
    async def test_create_user_duplicate_email(self, auth_service):
        # Arrange
        existing_user = UserFactory.create()

        # Act & Assert
        with pytest.raises(ValueError, match="Email already registered"):
            await auth_service.create_user(
                email=existing_user.email,
                password="test123",
                full_name="Test User"
            )

Integration Test Example

# tests/integration/test_api/test_auth.py
import pytest
from httpx import AsyncClient
from app.main import app
from tests.factories import UserFactory

class TestAuthAPI:
    @pytest.mark.asyncio
    async def test_register_user_success(self, async_client: AsyncClient):
        # Arrange
        user_data = {
            "email": "test@example.com",
            "password": "test123",
            "full_name": "Test User"
        }

        # Act
        response = await async_client.post("/api/v1/auth/register", json=user_data)

        # Assert
        assert response.status_code == 201
        data = response.json()
        assert data["email"] == user_data["email"]
        assert data["full_name"] == user_data["full_name"]
        assert "id" in data

    @pytest.mark.asyncio
    async def test_login_success(self, async_client: AsyncClient, test_user):
        # Arrange
        login_data = {
            "username": test_user.email,  # FastAPI OAuth2 uses 'username' field
            "password": "test123"
        }

        # Act
        response = await async_client.post("/api/v1/auth/login", data=login_data)

        # Assert
        assert response.status_code == 200
        data = response.json()
        assert "access_token" in data
        assert "refresh_token" in data
        assert data["token_type"] == "bearer"

Test Fixtures

# tests/conftest.py
import pytest
import asyncio
from httpx import AsyncClient
from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine
from sqlalchemy.pool import StaticPool
from app.main import app
from app.core.database import get_db
from app.models.user import User
from tests.factories import UserFactory

# Database fixtures
@pytest.fixture(scope="session")
def event_loop():
    """Create an instance of the default event loop for the test session."""
    loop = asyncio.get_event_loop_policy().new_event_loop()
    yield loop
    loop.close()

@pytest.fixture
async def mock_db_session():
    """Mock database session for unit tests."""
    # Use in-memory SQLite for testing
    engine = create_async_engine(
        "sqlite+aiosqlite:///:memory:",
        poolclass=StaticPool,
        connect_args={"check_same_thread": False}
    )

    async with engine.begin() as conn:
        # Create all tables
        await conn.run_sync(Base.metadata.create_all)

    async with AsyncSession(engine) as session:
        yield session

@pytest.fixture
async def async_client(mock_db_session):
    """HTTP client for API testing."""
    app.dependency_overrides[get_db] = lambda: mock_db_session

    async with AsyncClient(app=app, base_url="http://test") as client:
        yield client

    app.dependency_overrides.clear()

@pytest.fixture
async def test_user(mock_db_session):
    """Create a test user."""
    user = UserFactory.create(session=mock_db_session)
    await mock_db_session.commit()
    await mock_db_session.refresh(user)
    return user

@pytest.fixture
async def authenticated_client(async_client, test_user):
    """HTTP client with authentication headers."""
    # Login to get token
    login_data = {"username": test_user.email, "password": "test123"}
    response = await async_client.post("/api/v1/auth/login", data=login_data)
    token = response.json()["access_token"]

    # Add authorization header
    async_client.headers.update({"Authorization": f"Bearer {token}"})
    return async_client

Test Factories

# tests/factories.py
import factory
from factory.alchemy import SQLAlchemyModelFactory
from app.models.user import User
from app.models.portfolio import Portfolio

class UserFactory(SQLAlchemyModelFactory):
    class Meta:
        model = User
        sqlalchemy_session_persistence = "commit"

    email = factory.Faker("email")
    full_name = factory.Faker("name")
    hashed_password = factory.LazyFunction(
        lambda: "$2b$12$test_hashed_password"  # Pre-hashed test password
    )
    is_active = True

class PortfolioFactory(SQLAlchemyModelFactory):
    class Meta:
        model = Portfolio
        sqlalchemy_session_persistence = "commit"

    name = factory.Faker("company")
    description = factory.Faker("text", max_nb_chars=200)
    initial_balance = factory.Faker("pydecimal", left_digits=5, right_digits=2, positive=True)
    user = factory.SubFactory(UserFactory)

AI Agent Evaluation

Evaluation Framework

Quantbot uses Google ADK Evaluator for testing AI agent behavior:

from google.adk.evaluation import AgentEvaluator

# Run evaluation with test cases
eval_results = AgentEvaluator.evaluate(
    agent_module='app.agents.quantbot',
    eval_dataset_file_path_or_dir='evals/eval_json/stock_price_analyst_test.json',
)

Evaluation Test Cases

Test cases are defined in JSON format:

{
  "name": "Stock Price Query Test",
  "test_cases": [
    {
      "eval_id": "session_tesla_price_01",
      "session": {
        "turns": [
          {
            "user_message": "What is the current price of Tesla stock (TSLA)?",
            "expected_tools": ["get_stock_data"],
            "expected_response": "Tesla (TSLA) is currently trading at $XXX.XX"
          }
        ]
      }
    }
  ]
}

Custom Evaluation Metrics

@pytest.mark.asyncio
async def test_factual_accuracy(quantbot_agent):
    """Test factual accuracy of agent responses."""
    eval_results = AgentEvaluator.evaluate(
        agent_module='app.agents.quantbot',
        eval_dataset_file_path_or_dir='evals/eval_json/stock_price_test.json'
    )

    for result in eval_results:
        # Extract price from agent response
        agent_response = result.final_response.text_content
        price_match = re.search(r'\$?(\d+\.?\d*)', agent_response)
        agent_price = float(price_match.group(1)) if price_match else None

        # Get actual price for comparison
        actual_data = get_stock_data('TSLA', '1d')
        actual_price = actual_data.get('TSLA', {}).get('latest_price')

        # Assert accuracy within tolerance
        tolerance = 0.5
        assert abs(agent_price - actual_price) <= tolerance

Evaluation Metrics

• Tool Trajectory Average Score: Accuracy of tool usage
• Response Match Score: ROUGE similarity to expected response
• Custom Factual Accuracy: Domain-specific accuracy checks
• Response Time: Performance benchmarks

Mocking External Services

Market Data APIs

# tests/unit/test_services/test_market_service.py
import pytest
from unittest.mock import patch, AsyncMock
from app.services.market_service import MarketService

class TestMarketService:
    @pytest.mark.asyncio
    @patch('app.services.market_service.yfinance.download')
    async def test_get_stock_data_success(self, mock_yf_download, mock_db_session):
        # Arrange
        mock_yf_download.return_value = MockDataFrame(...)
        service = MarketService(db=mock_db_session)

        # Act
        result = await service.get_stock_data("AAPL", "1d")

        # Assert
        assert result["AAPL"]["latest_price"] > 0
        mock_yf_download.assert_called_once_with("AAPL", period="1d")

News APIs

@patch('app.services.news_service.NewsApiClient')
async def test_get_news_success(self, mock_news_client, mock_db_session):
    # Arrange
    mock_news_client.return_value.get_everything.return_value = {
        'articles': [
            {
                'title': 'Test Article',
                'description': 'Test Description',
                'url': 'https://example.com',
                'publishedAt': '2025-01-01T00:00:00Z'
            }
        ]
    }

    service = NewsService(db=mock_db_session)

    # Act
    result = await service.get_news(query="AAPL")

    # Assert
    assert len(result) == 1
    assert result[0]['title'] == 'Test Article'

Memory System Testing

Testing Neo4j Integration

@pytest.mark.integration
async def test_memory_storage_and_retrieval(test_user):
    """Test memory system integration."""
    from app.agents.quantbot.utils.memory_manager import MemoryManager

    memory_manager = MemoryManager()

    # Store episode
    await memory_manager.add_episode(
        group_id=f"user_{test_user.id}",
        name="quantbot",
        episode_body="User asked about Tesla stock price",
        reference_time=datetime.utcnow()
    )

    # Search memories
    results = await memory_manager.search(
        group_ids=[f"user_{test_user.id}"],
        query="Tesla stock",
        num_results=10
    )

    assert len(results) > 0
    assert "Tesla" in results[0].content

Continuous Integration

GitHub Actions

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

on: [push, pull_request]

jobs:
  test:
    runs-on: ubuntu-latest

    services:
      postgres:
        image: pgvector/pgvector:pg16
        env:
          POSTGRES_PASSWORD: test_password
        options: >-
          --health-cmd pg_isready
          --health-interval 10s
          --health-timeout 5s
          --health-retries 5

      neo4j:
        image: neo4j:latest
        env:
          NEO4J_AUTH: neo4j/test_password

    steps:
    - uses: actions/checkout@v4

    - name: Set up Python
      uses: actions/setup-python@v4
      with:
        python-version: '3.12'

    - name: Install UV
      run: curl -LsSf https://astral.sh/uv/install.sh | sh

    - name: Install dependencies
      run: |
        cd Quantbot-BE
        uv sync

    - name: Run tests
      run: |
        cd Quantbot-BE
        uv run pytest --cov=app --cov-report=xml tests/

    - name: Upload coverage
      uses: codecov/codecov-action@v3

Testing Best Practices

General Guidelines

1. Test Naming: Use descriptive names that explain what is being tested
2. AAA Pattern: Arrange, Act, Assert structure
3. Isolation: Tests should not depend on each other
4. Fast Tests: Unit tests should run quickly (< 1 second each)
5. Reliable: Tests should pass consistently

Database Testing

# Good: Use transactions that rollback
@pytest.fixture
async def db_session():
    async with async_session() as session:
        transaction = await session.begin()
        yield session
        await transaction.rollback()

# Good: Use factories for test data
user = UserFactory.build()  # Don't save to DB
user = UserFactory.create()  # Save to DB

Async Testing

# Good: Proper async test setup
@pytest.mark.asyncio
async def test_async_function():
    result = await some_async_function()
    assert result is not None

# Good: Mock async functions
@patch('app.services.some_service.async_method')
async def test_with_async_mock(mock_async_method):
    mock_async_method.return_value = AsyncMock(return_value="test")
    result = await service.method_that_calls_async_method()
    assert result == "test"

Error Testing

# Test error conditions
@pytest.mark.asyncio
async def test_invalid_input_raises_error():
    with pytest.raises(ValueError, match="Invalid email format"):
        await auth_service.create_user("invalid-email", "password", "Name")

# Test API error responses
@pytest.mark.asyncio
async def test_unauthorized_request(async_client):
    response = await async_client.get("/api/v1/protected-endpoint")
    assert response.status_code == 401
    assert "Unauthorized" in response.json()["detail"]

Performance Testing

Load Testing

@pytest.mark.slow
@pytest.mark.asyncio
async def test_api_performance(async_client):
    """Test API endpoint performance under load."""
    import time

    start_time = time.time()

    # Make multiple concurrent requests
    tasks = []
    for _ in range(100):
        task = async_client.get("/api/v1/market/data?symbol=AAPL")
        tasks.append(task)

    responses = await asyncio.gather(*tasks)
    end_time = time.time()

    # Assert performance requirements
    assert end_time - start_time < 5.0  # Should complete in < 5 seconds
    assert all(r.status_code == 200 for r in responses)

Debugging Tests

Common Issues

# Test discovery issues
uv run pytest --collect-only  # List all discovered tests

# Import errors
uv run pytest -v --tb=short   # Show shorter tracebacks

# Async/await issues
uv run pytest -s             # Show print statements

# Database issues
uv run pytest --pdb          # Drop into debugger on failure

# Memory leaks in tests
uv run pytest --memray       # Profile memory usage (if memray installed)

Test Debugging Tips

1. Use print statements in tests (run with -s to see output)
2. Use pytest.set_trace() for interactive debugging
3. Check test isolation by running tests in different orders
4. Use factories instead of hardcoded test data
5. Mock external dependencies to avoid network calls

Frontend Testing

JavaScript Testing Stack

cd Quantbot-FE

# Install testing dependencies
npm install --save-dev vitest @testing-library/react @testing-library/jest-dom

# Run tests
npm run test        # Run all tests
npm run test:watch  # Watch mode
npm run test:ui     # Visual test UI
npm run coverage    # Coverage report

This comprehensive testing approach ensures Quantbot maintains high quality, reliability, and performance across all components.