Test Doubles

PSP provides in-memory implementations of all platform primitives:

FixedClock

Returns a time you control. Essential for testing time-dependent logic.

InMemoryEventBus

Captures published events for inspection. No async delivery.

InMemoryIdempotencyStore

Tracks idempotency keys in memory.

AllowAllPolicy

Permits all actions. Skips authorization in tests.

Testing Use Cases

Use cases are tested with real domain objects and test doubles:

tests/test_complete_task.py
import pytest
from datetime import datetime
from uuid import uuid4

from psp.platform.clock import FixedClock
from psp.platform.eventbus import InMemoryEventBus
from psp.components.todos.application.use_cases import CompleteTask
from psp.components.todos.adapters.persistence import InMemoryTaskRepo
from psp.components.todos.domain.models import Task, TaskStatus
from psp.components.todos.domain.events import TaskCompleted

@pytest.fixture
def clock():
    return FixedClock(datetime(2024, 1, 15, 10, 30))

@pytest.fixture
def event_bus():
    return InMemoryEventBus()

@pytest.fixture
def repo():
    return InMemoryTaskRepo()

def test_complete_task_changes_status(repo, clock, event_bus):
    # Arrange
    task = Task(
        id=uuid4(),
        owner_id=uuid4(),
        title="Test task",
        status=TaskStatus.PENDING,
        ...
    )
    repo.save(task)

    use_case = CompleteTask(repo, clock, event_bus)

    # Act
    use_case.execute(task.id, actor_id=task.owner_id)

    # Assert
    updated = repo.get(task.id)
    assert updated.status == TaskStatus.COMPLETED

def test_complete_task_emits_event(repo, clock, event_bus):
    task = Task(id=uuid4(), owner_id=uuid4(), ...)
    repo.save(task)

    use_case = CompleteTask(repo, clock, event_bus)
    use_case.execute(task.id, actor_id=task.owner_id)

    assert len(event_bus.published) == 1
    event = event_bus.published[0]
    assert isinstance(event, TaskCompleted)
    assert event.task_id == task.id

Testing Time-Dependent Logic

FixedClock makes time deterministic:

tests/test_due_date.py
def test_task_is_overdue():
    clock = FixedClock(datetime(2024, 1, 20))  # "today"

    task = Task(
        due_date=date(2024, 1, 15),  # 5 days ago
        ...
    )

    assert task.is_overdue(clock.now()) is True

def test_task_is_not_overdue():
    clock = FixedClock(datetime(2024, 1, 10))  # "today"

    task = Task(
        due_date=date(2024, 1, 15),  # 5 days from now
        ...
    )

    assert task.is_overdue(clock.now()) is False

Testing Hooks

Test hook handlers directly:

tests/test_budget_hook.py
from psp.platform.hooks import Allow, Veto

def test_allows_when_within_budget(budget_module, ledger, cost_repo):
    # Setup: 10 tokens available, task costs 5
    ledger.credit(owner_id, token_type_id, 10)
    cost_repo.save(TaskCost(task_id=task_id, amount=5, ...))

    result = budget_module._check_budget({
        "task_id": task_id,
        "owner_id": owner_id
    })

    assert isinstance(result, Allow)

def test_vetoes_when_over_budget(budget_module, ledger, cost_repo):
    # Setup: 3 tokens available, task costs 5
    ledger.credit(owner_id, token_type_id, 3)
    cost_repo.save(TaskCost(task_id=task_id, amount=5, ...))

    result = budget_module._check_budget({
        "task_id": task_id,
        "owner_id": owner_id
    })

    assert isinstance(result, Veto)
    assert "Insufficient" in result.reason

Testing Event Subscribers

Test idempotency explicitly:

tests/test_budget_subscriber.py
def test_processes_event_once(budget_module, ledger):
    ledger.credit(owner_id, token_type_id, 10)

    envelope = EventEnvelope(
        id=uuid4(),
        payload={"task_id": task_id, "owner_id": owner_id}
    )

    # Process twice
    budget_module._on_task_completed(envelope)
    budget_module._on_task_completed(envelope)

    # Should only debit once
    assert ledger.get_balance(owner_id, token_type_id) == 7  # not 4

Boundary Enforcement Tests

Verify components don't import each other:

tests/test_boundaries.py
from pathlib import Path

COMPONENTS = ["todos", "budget", "notes"]

def test_no_cross_component_imports():
    for component in COMPONENTS:
        component_path = Path(f"src/psp/components/{component}")
        other_components = [c for c in COMPONENTS if c != component]

        for py_file in component_path.rglob("*.py"):
            content = py_file.read_text()
            for other in other_components:
                assert f"from psp.components.{other}" not in content, \
                    f"{py_file} imports from {other}"
                assert f"import psp.components.{other}" not in content, \
                    f"{py_file} imports {other}"

Testing Domain Invariants

Test that domain rules are enforced:

tests/test_task.py
def test_cannot_complete_already_completed_task():
    task = Task(status=TaskStatus.COMPLETED, ...)

    with pytest.raises(AlreadyCompletedError):
        task.complete()

def test_task_requires_owner_id():
    with pytest.raises(TypeError):
        Task(id=uuid4(), title="No owner")  # missing owner_id

def test_note_cannot_have_empty_content():
    with pytest.raises(EmptyContentError):
        Note(content="   ", ...)  # whitespace only

Running Tests

terminal
# Run all tests
uv run pytest

# Run with coverage
uv run pytest --cov=src/psp

# Run specific component tests
uv run pytest tests/components/todos/

# Run boundary tests only
uv run pytest tests/test_boundaries.py -v
Testing Checklist
  • Use FixedClock, never real time
  • Use InMemoryEventBus to capture events
  • Test idempotency for event subscribers
  • Test hook responses (Allow, Veto, Patch)
  • Enforce component boundaries with import tests
  • Test domain invariants directly on entities