Skip to content

TESTING_GUIDE.md

Latest commit

 

History

History
612 lines (449 loc) · 16.4 KB

TESTING_GUIDE.md

File metadata and controls

612 lines (449 loc) · 16.4 KB

Testing Guide

📋 Document Summary

What This Document Covers:

  • Testing philosophy: real services, no mocks, verbose output for debugging
  • Unit testing with pytest (fixtures, async tests, parameterized tests)
  • Integration testing patterns (run_*.py executables, real Firebase/Solana)
  • Test organization and naming conventions
  • Running tests and interpreting results
  • Common testing patterns and best practices

Sections in This Document:

Related Documentation:

Context Tags: #testing #pytest #integration #unit-tests #quality-assurance

Testing Philosophy

Core Principles

LaunchAgencyBot follows a "real services" testing philosophy:

  1. No Mock Services - Tests run against real services and data
  2. Verbose Tests - Designed for debugging with detailed output
  3. Devnet First - Blockchain tests on devnet before mainnet
  4. Complete Current Test - Finish one test before moving to next
  5. Fail Fast - Tests fail immediately on first error

Why Real Services?

# ❌ BAD: Mocked service doesn't catch real issues
@patch('src.services.simple_firebase_service.SimpleFirebaseService')
def test_wallet_creation(mock_firebase):
    mock_firebase.create_wallet.return_value = {"success": True}
    # Test passes but doesn't validate actual Firebase behavior

# ✅ GOOD: Real service catches real issues
async def test_wallet_creation_integration():
    service = SimpleFirebaseService()
    await service.initialize()
    wallet = await service.create_wallet("test_wallet_001")
    # Test validates actual Firebase behavior and data persistence

Benefits:

  • Catches integration issues early
  • Validates real API behavior
  • Tests actual error scenarios
  • Ensures production readiness

Test Organization

Directory Structure

tests/
├── unit/                           # pytest unit tests
│   ├── domain/
│   │   ├── processor/
│   │   │   ├── stages/            # Stage tests (test_*.py)
│   │   │   └── test_*_processor.py
│   │   ├── model/
│   │   │   ├── actions/           # Action model tests
│   │   │   ├── events/            # Event model tests
│   │   │   └── test_*.py
│   │   └── executor/              # Executor tests
│   ├── services/                  # Service tests (test_*.py)
│   └── models/                    # Data model tests
└── integration/                    # Integration tests
    ├── run_*_integration.py       # Runnable integration tests
    └── orchestrator/              # Orchestrator integration tests
        └── run_*_integration.py

Naming Conventions

Unit Tests (pytest):

  • File: test_<component_name>.py
  • Class: Test<ComponentName> (optional grouping)
  • Function: test_<specific_behavior>

Integration Tests (executable scripts):

  • File: run_<feature>_integration.py
  • Class: <Feature>IntegrationTest
  • Function: Descriptive method names (no test_ prefix)

Examples

# Unit test file: tests/unit/services/test_simple_firebase_service.py
class TestSimpleFirebaseService:
    def test_wallet_creation_with_valid_data(self):
        """Test wallet creation with valid parameters"""
        pass

    def test_wallet_creation_with_invalid_address(self):
        """Test wallet creation fails with invalid address"""
        pass

# Integration test file: tests/integration/run_simple_firebase_service_integration.py
class SimpleFirebaseServiceIntegrationTest:
    async def test_create_wallet_end_to_end(self):
        """Create wallet, verify persistence, clean up"""
        pass

Unit Testing with pytest

Basic Test Structure

"""
Comprehensive tests for TagClassificationStage

Tests cover tag classification with MemecoinEntry objects, LLM integration,
multimodal vs text-only processing, error handling, and stage graph integration.
"""

import pytest
import asyncio
from unittest.mock import Mock, AsyncMock

from src.domain.processor.stages.tag_classification_stage import TagClassificationStage
from src.ai_generation.models import ProcessingContext, MemecoinEntry


class TestTagClassificationStage:
    """Test suite for TagClassificationStage functionality"""

    @pytest.fixture
    def mock_llm_service(self):
        """Mock LLM service for testing"""
        service = Mock(spec=LiteLLMService)
        service.query_llm = AsyncMock()
        service._initialized = True
        return service

    @pytest.mark.asyncio
    async def test_tag_classification_with_valid_input(self, mock_llm_service):
        """Test successful tag classification"""
        # Setup
        stage = TagClassificationStage(llm_service=mock_llm_service)
        context = ProcessingContext()

        # Execute
        result = await stage.execute(context)

        # Assert
        assert result is not None
        assert len(result.tags) > 0

Async Test Patterns

Pytest Mark Pattern (recommended): Use @pytest.mark.asyncio decorator for async test functions. Automatically manages event loop. Cleaner syntax and integrates with pytest fixtures.

Asyncio.run Pattern (alternative): For tests that need explicit event loop control, wrap async logic in nested function and call asyncio.run() at test level. Less common but useful for complex event loop scenarios.

Fixtures

@pytest.fixture
def sample_memecoin_entry():
    """Sample memecoin entry for testing"""
    return MemecoinEntry(
        token_name="Test Coin",
        ticker="TEST",
        description="A test memecoin",
        tags=["meme", "test"]
    )


@pytest.fixture
async def initialized_service():
    """Initialized service fixture with cleanup"""
    service = SomeService()
    await service.initialize()
    yield service  # Provide to test
    await service.cleanup()  # Cleanup after test

Parameterized Tests

Use @pytest.mark.parametrize to test multiple input combinations without code duplication. Pass parameter names and a list of tuples with test values. Pytest generates separate test runs for each tuple.

Example: @pytest.mark.parametrize("status,expected", [(FUNDED, True), (PENDING, False)]) generates 2 test runs with different inputs.

Integration Testing

Integration Test Structure

#!/usr/bin/env python3
"""
Integration tests for SimpleFirebaseService

Tests against real Firebase/Firestore instance to validate
end-to-end functionality and data persistence.

Prerequisites:
- Firebase project configured
- FIREBASE_PROJECT_ID environment variable set
- Service account with proper permissions

These tests create, modify, and clean up real data in Firestore.
Use a test project to avoid affecting production data.
"""

import asyncio
import os
import sys

# Add src to path
sys.path.append(os.path.join(os.path.dirname(__file__), '..', '..'))

from src.services.simple_firebase_service import SimpleFirebaseService


class SimpleFirebaseServiceIntegrationTest:
    """Integration test suite for SimpleFirebaseService"""

    def __init__(self):
        self.service = SimpleFirebaseService()
        self.test_data = []

    async def setup(self):
        """Initialize service and prepare test environment"""
        print("🔧 Setting up integration test...")

        # Validate environment
        if not os.getenv('FIREBASE_PROJECT_ID'):
            raise Exception("FIREBASE_PROJECT_ID not set")

        # Initialize service
        await self.service.initialize()
        print(f"✅ Connected to Firebase: {self.service._project_id}")

        # Clean up previous test data
        await self.cleanup()

    async def cleanup(self):
        """Clean up test data"""
        print("🧹 Cleaning up test data...")

        for item_id in self.test_data:
            await self.service.delete(item_id)

        self.test_data.clear()

    async def run_all_tests(self):
        """Run all integration tests"""
        print("\\n" + "="*80)
        print("🧪 SimpleFirebaseService Integration Tests")
        print("="*80 + "\\n")

        await self.setup()

        try:
            await self.test_create_and_retrieve()
            await self.test_update_operation()
            await self.test_delete_operation()

            print("\\n✅ All tests passed!")

        except Exception as e:
            print(f"\\n❌ Test failed: {e}")
            raise
        finally:
            await self.cleanup()

    async def test_create_and_retrieve(self):
        """Test creating and retrieving data"""
        print("Test: Create and retrieve data...")

        # Create
        wallet_id = "test_wallet_001"
        wallet_data = {"address": wallet_id, "balance": 10.0}

        result = await self.service.create_wallet(wallet_data)
        self.test_data.append(wallet_id)

        # Retrieve
        retrieved = await self.service.get_wallet(wallet_id)

        assert retrieved["address"] == wallet_id
        assert retrieved["balance"] == 10.0

        print("  ✓ Create and retrieve successful")


async def main():
    """Run integration tests"""
    test = SimpleFirebaseServiceIntegrationTest()
    await test.run_all_tests()


if __name__ == "__main__":
    asyncio.run(main())

Running Integration Tests

# Run single integration test
PYTHONPATH=. python tests/integration/run_simple_firebase_service_integration.py

# Run with specific network
NETWORK=devnet PYTHONPATH=. python tests/integration/run_launch_workflow_integration.py

# Run with verbose output
PYTHONPATH=. python -v tests/integration/run_rag_v2_workflow_integration.py

Running Tests

Running pytest Unit Tests

# Run all unit tests
pytest tests/unit/

# Run specific test file
pytest tests/unit/services/test_simple_firebase_service.py

# Run specific test class
pytest tests/unit/services/test_simple_firebase_service.py::TestSimpleFirebaseService

# Run specific test function
pytest tests/unit/services/test_simple_firebase_service.py::TestSimpleFirebaseService::test_wallet_creation

# Run with verbose output
pytest -v tests/unit/

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

# Run with coverage
pytest --cov=src tests/unit/

Running Integration Tests

# Run single integration test (executable script)
PYTHONPATH=. python tests/integration/run_simple_firebase_service_integration.py

# Run orchestrator integration suite
PYTHONPATH=. python tests/integration/orchestrator/run_orchestrator_integration_suite.py

# Run with environment variables
NETWORK=devnet PYTHONPATH=. python tests/integration/run_launch_workflow_integration.py

Common Patterns

Testing Async Workflows

@pytest.mark.asyncio
async def test_workflow_execution():
    """Test complete workflow execution"""
    # Initialize
    workflow = RAGMemecoinGenerationWorkflow()
    await workflow.initialize()

    # Execute
    request = {"prompt": "Create a funny cat meme", "generation_count": 1}
    session_id = await workflow.create_session(request)

    # Poll for completion
    max_polls = 30
    for _ in range(max_polls):
        status = await workflow.get_status(session_id)
        if status["status"] == "completed":
            break
        await asyncio.sleep(1)

    # Verify results
    assert status["status"] == "completed"
    results = await workflow.get_results(session_id)
    assert len(results) == 1

    # Cleanup
    await workflow.cleanup()

Testing Error Scenarios

@pytest.mark.asyncio
async def test_service_handles_missing_data():
    """Test service handles missing data gracefully"""
    service = SomeService()
    await service.initialize()

    # Test with invalid ID
    with pytest.raises(ValueError, match="not found"):
        await service.get("invalid_id")

    await service.cleanup()


def test_validation_fails_on_invalid_input():
    """Test validation rejects invalid input"""
    with pytest.raises(ValidationError):
        MemecoinEntry(
            token_name="",  # Invalid: empty name
            ticker="INVALID"
        )

Testing with Real Firebase

async def test_firebase_data_persistence():
    """Test data persists correctly in Firebase"""
    service = SimpleFirebaseService()
    await service.initialize()

    # Create test wallet
    wallet_id = "test_wallet_persistence"
    wallet_data = {
        "address": wallet_id,
        "sol_balance": 5.0,
        "status": "FUNDED"
    }

    try:
        # Create
        await service.create_wallet(wallet_data)

        # Retrieve and verify
        retrieved = await service.get_wallet(wallet_id)
        assert retrieved["sol_balance"] == 5.0
        assert retrieved["status"] == "FUNDED"

        # Update
        await service.update_wallet(wallet_id, {"sol_balance": 10.0})

        # Verify update
        updated = await service.get_wallet(wallet_id)
        assert updated["sol_balance"] == 10.0

    finally:
        # Cleanup
        await service.delete_wallet(wallet_id)

Debugging Tests

Verbose Output

# Enable verbose logging in tests
import logging

def test_with_verbose_logging():
    """Test with detailed logging"""
    logging.basicConfig(level=logging.DEBUG)
    logger = logging.getLogger(__name__)

    logger.debug("Starting test...")
    result = some_function()
    logger.debug(f"Result: {result}")

    assert result.success

Print Debugging

# Run pytest with output capture disabled
pytest -s tests/unit/test_something.py

# Run with very verbose output
pytest -vv tests/unit/test_something.py

Debugging Async Tests

@pytest.mark.asyncio
async def test_async_with_debugging():
    """Debug async test execution"""
    print("\\n=== Starting async test ===")

    service = SomeService()
    await service.initialize()
    print(f"Service initialized: {service._initialized}")

    result = await service.execute()
    print(f"Execution result: {result}")

    assert result.success
    print("=== Test completed ===\\n")

Best Practices

1. Test Isolation

# ✅ GOOD: Each test is independent
@pytest.mark.asyncio
async def test_wallet_creation():
    service = SimpleFirebaseService()
    await service.initialize()

    wallet_id = f"test_wallet_{uuid.uuid4()}"  # Unique ID

    try:
        wallet = await service.create_wallet(wallet_id)
        assert wallet.address == wallet_id
    finally:
        await service.delete_wallet(wallet_id)


# ❌ BAD: Tests share state
shared_wallet_id = "test_wallet_001"  # Reused across tests

async def test_create():
    await service.create_wallet(shared_wallet_id)

async def test_update():
    # Depends on test_create running first
    await service.update_wallet(shared_wallet_id, {...})

2. Cleanup Always Runs

@pytest.fixture
async def test_service():
    """Service fixture with guaranteed cleanup"""
    service = SomeService()
    await service.initialize()

    yield service

    # Cleanup always runs, even if test fails
    await service.cleanup()

3. Clear Test Names

# ✅ GOOD: Descriptive test names
def test_wallet_creation_fails_with_insufficient_balance()
def test_tag_classification_returns_4_tags_for_memecoin_entry()
def test_image_generation_retries_on_service_unavailable()

# ❌ BAD: Vague test names
def test_wallet()
def test_tags()
def test_image()

4. One Assert Per Logical Concept

# ✅ GOOD: Related assertions grouped
def test_wallet_creation_result():
    wallet = create_wallet("test_001")

    # Assert wallet properties as a group
    assert wallet.address == "test_001"
    assert wallet.sol_balance == 0.0
    assert wallet.status == WalletStatus.PENDING


# ❌ BAD: Unrelated assertions
def test_everything():
    wallet = create_wallet("test_001")
    assert wallet.address == "test_001"

    memecoin = create_memecoin("test")
    assert memecoin.ticker == "TEST"  # Different concern

Last Updated: January 2025 - LaunchAgencyBot v2.0