Skip to main content
Unlisted page
This page is unlisted. Search engines will not index it, and only users having a direct link can access it.

Unit Tests

info

Learn how to write unit tests for Fluvius Framework applications.

Testing Aggregates​

Basic Aggregate Test​

import pytest
from myapp.domains.user import UserAggregate, UserDomain
from fluvius.domain.context import SanicContext

@pytest.mark.asyncio
async def test_create_user():
    ctx = SanicContext.create(namespace='test-user')
    domain = UserDomain(ctx)
    aggregate = UserAggregate(domain)
    
    result = await aggregate.create_user('John', 'john@example.com')
    
    assert result['name'] == 'John'
    assert result['email'] == 'john@example.com'
    assert result['active'] is True

Testing Validation​

@pytest.mark.asyncio
async def test_create_user_invalid_email():
    ctx = SanicContext.create(namespace='test-user')
    domain = UserDomain(ctx)
    aggregate = UserAggregate(domain)
    
    with pytest.raises(ValueError, match="Invalid email"):
        await aggregate.create_user('John', 'invalid-email')

Testing State Updates​

@pytest.mark.asyncio
async def test_update_user():
    ctx = SanicContext.create(namespace='test-user')
    domain = UserDomain(ctx)
    aggregate = UserAggregate(domain)
    
    # Create user
    await aggregate.create_user('John', 'john@example.com')
    
    # Update user
    result = await aggregate.update_user(name='Jane')
    
    assert result['name'] == 'Jane'
    assert result['email'] == 'john@example.com'

Testing Commands​

Command Processing Test​

@pytest.mark.asyncio
async def test_process_command():
    ctx = SanicContext.create(namespace='test-user')
    domain = UserDomain(ctx)
    domain.set_aggroot('user', 'user-123')
    
    command = domain.create_command('create-user', {
        'name': 'John',
        'email': 'john@example.com'
    })
    
    response = await domain.process_command(command)
    
    assert response['name'] == 'John'

Testing Events​

Event Generation Test​

@pytest.mark.asyncio
async def test_event_generation():
    ctx = SanicContext.create(namespace='test-user')
    domain = UserDomain(ctx)
    domain.set_aggroot('user', 'user-123')
    
    command = domain.create_command('create-user', {...})
    await domain.process_command(command)
    
    # Check events
    events = await domain.logstore.get_events('user', 'user-123')
    assert len(events) == 1
    assert events[0].key == 'user-created'

Test Fixtures​

Domain Fixture​

import pytest
from myapp.domains.user import UserDomain
from fluvius.domain.context import SanicContext

@pytest.fixture
async def user_domain():
    ctx = SanicContext.create(namespace='test-user')
    domain = UserDomain(ctx)
    yield domain
    # Cleanup if needed

@pytest.mark.asyncio
async def test_with_fixture(user_domain):
    user_domain.set_aggroot('user', 'user-123')
    command = user_domain.create_command('create-user', {...})
    response = await user_domain.process_command(command)
    assert response['name'] == 'John'

Next Steps​