Testing Guide¶
Testing is essential for reliable content packs. This guide covers testing strategies for systems, commands, events, and persistence using pytest.
Setup¶
Dependencies¶
Add testing dependencies to your pyproject.toml:
Pytest Configuration¶
Configure pytest for async tests:
Or in pytest.ini:
Testing Fixtures¶
World Fixture¶
Create a reusable world fixture:
# tests/conftest.py
import pytest
from uuid import uuid4
from maid_engine.core.world import World
from maid_engine.storage.document_store import InMemoryDocumentStore
@pytest.fixture
def world():
"""Create a fresh World for testing."""
return World()
@pytest.fixture
def document_store():
"""Create an in-memory document store."""
return InMemoryDocumentStore()
@pytest.fixture
def world_with_store(document_store):
"""Create a World with document store."""
world = World()
world.document_store = document_store
return world
Entity Fixtures¶
Create common entity setups:
# tests/conftest.py
from maid_stdlib.components import (
DescriptionComponent,
HealthComponent,
PositionComponent,
)
@pytest.fixture
def room_id():
"""A room UUID for testing."""
return uuid4()
@pytest.fixture
def player_entity(world, room_id):
"""Create a player entity with common components."""
player = world.entities.create()
player.add(PositionComponent(room_id=room_id))
player.add(HealthComponent(current=100, maximum=100))
player.add(DescriptionComponent(name="TestPlayer"))
player.add_tag("player")
return player
@pytest.fixture
def monster_entity(world, room_id):
"""Create a monster entity."""
monster = world.entities.create()
monster.add(PositionComponent(room_id=room_id))
monster.add(HealthComponent(current=50, maximum=50))
monster.add(DescriptionComponent(name="Goblin"))
monster.add_tag("hostile")
return monster
Testing Components¶
Basic Component Tests¶
# tests/test_components.py
from my_pack.components import ReputationComponent
def test_reputation_get_standing_default():
"""Test default standing for unknown faction."""
rep = ReputationComponent()
assert rep.get_standing("unknown_faction") == 0
def test_reputation_modify_standing():
"""Test modifying faction standing."""
rep = ReputationComponent()
new_value = rep.modify_standing("warriors", 50)
assert new_value == 50
assert rep.get_standing("warriors") == 50
def test_reputation_standing_clamped():
"""Test that standing is clamped to valid range."""
rep = ReputationComponent()
# Try to exceed maximum
rep.modify_standing("test", 2000)
assert rep.get_standing("test") == 1000 # Clamped
# Try to go below minimum
rep.modify_standing("test", -3000)
assert rep.get_standing("test") == -1000 # Clamped
Component Integration Tests¶
def test_component_on_entity(world):
"""Test component attached to entity."""
entity = world.entities.create()
entity.add(ReputationComponent())
assert entity.has(ReputationComponent)
rep = entity.get(ReputationComponent)
rep.modify_standing("guild", 100)
# Verify change persists
rep2 = entity.get(ReputationComponent)
assert rep2.get_standing("guild") == 100
Testing Systems¶
Basic System Tests¶
# tests/test_systems.py
import pytest
from maid_stdlib.components import HealthComponent
from my_pack.systems import RegenerationSystem
@pytest.mark.asyncio
async def test_regeneration_system(world, player_entity):
"""Test health regeneration over time."""
# Set up health below maximum
health = player_entity.get(HealthComponent)
health.current = 50 # Half health
# Create and run system
system = RegenerationSystem(world)
world.systems.register(system)
# Simulate time passing (1 second)
await system.update(delta=1.0)
# Check regeneration occurred
assert health.current > 50
@pytest.mark.asyncio
async def test_regeneration_caps_at_maximum(world, player_entity):
"""Test regeneration doesn't exceed maximum."""
health = player_entity.get(HealthComponent)
health.current = 99 # Almost full
system = RegenerationSystem(world)
await system.update(delta=10.0) # Long time
assert health.current == health.maximum
Testing System Lifecycle¶
@pytest.mark.asyncio
async def test_system_startup(world):
"""Test system startup hook."""
system = MySystem(world)
await system.startup()
# Verify startup actions
assert system._initialized
assert world.events.has_handlers(SomeEvent)
@pytest.mark.asyncio
async def test_system_shutdown(world):
"""Test system shutdown hook."""
system = MySystem(world)
await system.startup()
await system.shutdown()
# Verify cleanup
assert not world.events.has_handlers(SomeEvent)
Testing System Priority¶
def test_system_priorities(world):
"""Test systems run in priority order."""
execution_order = []
class FirstSystem(System):
priority = 10
async def update(self, delta):
execution_order.append("first")
class SecondSystem(System):
priority = 20
async def update(self, delta):
execution_order.append("second")
world.systems.register(SecondSystem(world))
world.systems.register(FirstSystem(world))
await world.systems.update(0.1)
assert execution_order == ["first", "second"]
Testing Events¶
Basic Event Tests¶
# tests/test_events.py
import pytest
from my_pack.events import QuestCompletedEvent
def test_event_creation():
"""Test event can be created with required fields."""
event = QuestCompletedEvent(
player_id=uuid4(),
quest_id=uuid4(),
)
assert event.event_type == "QuestCompletedEvent"
assert event.timestamp is not None
assert not event.cancelled
def test_event_cancellation():
"""Test event cancellation."""
event = QuestCompletedEvent(
player_id=uuid4(),
quest_id=uuid4(),
)
event.cancel()
assert event.cancelled
Testing Event Handlers¶
@pytest.mark.asyncio
async def test_event_handler_called(world):
"""Test event handler is called when event is emitted."""
received_events = []
async def handler(event):
received_events.append(event)
world.events.subscribe(QuestCompletedEvent, handler)
event = QuestCompletedEvent(
player_id=uuid4(),
quest_id=uuid4(),
)
await world.events.emit(event)
assert len(received_events) == 1
assert received_events[0] is event
@pytest.mark.asyncio
async def test_event_handler_priority(world):
"""Test handlers run in priority order."""
call_order = []
async def high_priority(event):
call_order.append("high")
async def low_priority(event):
call_order.append("low")
from maid_engine.core.events import EventPriority
world.events.subscribe(
QuestCompletedEvent,
low_priority,
priority=EventPriority.LOW,
)
world.events.subscribe(
QuestCompletedEvent,
high_priority,
priority=EventPriority.HIGH,
)
await world.events.emit(QuestCompletedEvent(
player_id=uuid4(),
quest_id=uuid4(),
))
assert call_order == ["high", "low"]
@pytest.mark.asyncio
async def test_cancelled_event_stops_propagation(world):
"""Test cancelled events don't reach later handlers."""
call_order = []
async def cancelling_handler(event):
call_order.append("canceller")
event.cancel()
async def later_handler(event):
call_order.append("later")
from maid_engine.core.events import EventPriority
world.events.subscribe(
QuestCompletedEvent,
cancelling_handler,
priority=EventPriority.HIGH,
)
world.events.subscribe(
QuestCompletedEvent,
later_handler,
priority=EventPriority.LOW,
)
await world.events.emit(QuestCompletedEvent(
player_id=uuid4(),
quest_id=uuid4(),
))
assert call_order == ["canceller"] # "later" not called
Testing Commands¶
Mock Session¶
Create a mock session for testing commands:
# tests/conftest.py
from dataclasses import dataclass, field
@dataclass
class MockSession:
"""Mock session for testing commands."""
messages: list[str] = field(default_factory=list)
player_id: UUID | None = None
async def send(self, message: str) -> None:
self.messages.append(message)
def clear(self) -> None:
self.messages.clear()
@pytest.fixture
def mock_session(player_entity):
"""Create a mock session."""
session = MockSession(player_id=player_entity.id)
return session
Command Tests¶
# tests/test_commands.py
import pytest
from maid_engine.commands.registry import CommandContext
from my_pack.commands import heal_command
@pytest.fixture
def command_context(world, player_entity, mock_session):
"""Create a command context for testing."""
return CommandContext(
session=mock_session,
player_id=player_entity.id,
command="heal",
args=[],
raw_input="heal",
world=world,
)
@pytest.mark.asyncio
async def test_heal_self(command_context, player_entity):
"""Test healing yourself."""
# Damage the player first
health = player_entity.get(HealthComponent)
health.current = 50
result = await heal_command(command_context)
assert result is True
assert health.current > 50
assert "heal" in command_context.session.messages[0].lower()
@pytest.mark.asyncio
async def test_heal_at_full_health(command_context, player_entity):
"""Test healing when already at full health."""
result = await heal_command(command_context)
assert result is True
assert "full health" in command_context.session.messages[0].lower()
@pytest.mark.asyncio
async def test_heal_target(command_context, monster_entity):
"""Test healing another entity."""
command_context.args = ["goblin"]
health = monster_entity.get(HealthComponent)
health.current = 25
result = await heal_command(command_context)
assert result is True
assert health.current > 25
Testing Persistence¶
Document Store Tests¶
# tests/test_persistence.py
import pytest
from uuid import uuid4
from maid_engine.storage.document_store import (
InMemoryDocumentStore,
QueryOptions,
SortOrder,
)
from my_pack.models import CharacterModel
@pytest.fixture
def store():
"""Create an in-memory store with schemas registered."""
store = InMemoryDocumentStore()
store.register_schema("characters", CharacterModel)
return store
@pytest.mark.asyncio
async def test_create_and_get(store):
"""Test creating and retrieving a document."""
characters = store.get_collection("characters")
char = CharacterModel(
id=uuid4(),
name="Test Character",
level=5,
)
char_id = await characters.create(char)
retrieved = await characters.get(char_id)
assert retrieved is not None
assert retrieved.name == "Test Character"
assert retrieved.level == 5
@pytest.mark.asyncio
async def test_update(store):
"""Test updating a document."""
characters = store.get_collection("characters")
char = CharacterModel(id=uuid4(), name="Original", level=1)
char_id = await characters.create(char)
char.name = "Updated"
char.level = 10
updated = await characters.update(char_id, char)
assert updated is True
retrieved = await characters.get(char_id)
assert retrieved.name == "Updated"
assert retrieved.level == 10
@pytest.mark.asyncio
async def test_delete(store):
"""Test deleting a document."""
characters = store.get_collection("characters")
char = CharacterModel(id=uuid4(), name="ToDelete", level=1)
char_id = await characters.create(char)
deleted = await characters.delete(char_id)
assert deleted is True
assert await characters.get(char_id) is None
@pytest.mark.asyncio
async def test_query_with_filters(store):
"""Test querying with filters."""
characters = store.get_collection("characters")
# Create test data
for i in range(5):
await characters.create(CharacterModel(
id=uuid4(),
name=f"Char{i}",
level=i * 2,
))
# Query for specific level
results = await characters.query(QueryOptions(
filters={"level": 4}
))
assert len(results) == 1
assert results[0].level == 4
@pytest.mark.asyncio
async def test_query_with_sorting(store):
"""Test querying with sorting."""
characters = store.get_collection("characters")
# Create unsorted data
for level in [5, 1, 3, 4, 2]:
await characters.create(CharacterModel(
id=uuid4(),
name=f"Level{level}",
level=level,
))
# Query sorted ascending
results = await characters.query(QueryOptions(
order_by="level",
order=SortOrder.ASC,
))
levels = [r.level for r in results]
assert levels == [1, 2, 3, 4, 5]
@pytest.mark.asyncio
async def test_query_with_pagination(store):
"""Test querying with pagination."""
characters = store.get_collection("characters")
# Create 10 characters
for i in range(10):
await characters.create(CharacterModel(
id=uuid4(),
name=f"Char{i}",
level=i,
))
# Get first page
page1 = await characters.query(QueryOptions(
order_by="level",
limit=3,
offset=0,
))
# Get second page
page2 = await characters.query(QueryOptions(
order_by="level",
limit=3,
offset=3,
))
assert len(page1) == 3
assert len(page2) == 3
assert page1[0].level == 0
assert page2[0].level == 3
Integration Tests¶
Full Content Pack Test¶
# tests/test_integration.py
import pytest
from maid_engine.core.engine import GameEngine
from maid_engine.config.settings import Settings
from maid_stdlib.pack import StdlibContentPack
from my_pack import MyContentPack
@pytest.fixture
async def engine():
"""Create a fully configured engine."""
settings = Settings(debug=True)
engine = GameEngine(settings)
# Load packs
engine.load_content_pack(StdlibContentPack())
engine.load_content_pack(MyContentPack())
yield engine
# Cleanup
await engine.stop()
@pytest.mark.asyncio
async def test_pack_loads_correctly(engine):
"""Test content pack loads without errors."""
await engine.start()
# Verify systems registered
assert len(engine.world.systems) > 0
# Verify commands registered
assert "my_command" in engine.command_registry
@pytest.mark.asyncio
async def test_full_gameplay_scenario(engine):
"""Test a complete gameplay scenario."""
await engine.start()
# Create a player
player = engine.world.entities.create()
# ... set up player ...
# Execute commands
# ... simulate player actions ...
# Verify game state
# ... check expected outcomes ...
Running Tests¶
# Run all tests
uv run pytest
# Run with coverage
uv run pytest --cov=my_pack
# Run specific test file
uv run pytest tests/test_systems.py
# Run specific test
uv run pytest tests/test_systems.py::test_regeneration_system
# Run with verbose output
uv run pytest -v
# Run only marked tests
uv run pytest -m "not slow"
Best Practices¶
Isolate Tests¶
Each test should be independent:
Test Edge Cases¶
def test_health_damage_at_zero():
"""Test damaging entity already at 0 health."""
health = HealthComponent(current=0, maximum=100)
damage = health.damage(50)
assert damage == 0 # Can't deal damage to dead entity
Use Descriptive Names¶
# Good
def test_heal_command_shows_error_when_target_not_found():
...
# Avoid
def test_heal_error():
...
Test Both Success and Failure¶
@pytest.mark.asyncio
async def test_command_succeeds_with_valid_input(context):
result = await my_command(context)
assert result is True
@pytest.mark.asyncio
async def test_command_fails_with_invalid_input(context):
context.args = ["invalid"]
result = await my_command(context)
assert "error" in context.session.messages[0].lower()
Next Steps¶
- Publishing Guide - Share your tested content pack
- Commands Guide - More on command testing
- Persistence Guide - Database testing strategies