Testing
Test doubles, contract tests, and boundary enforcement
Test Doubles
PSP provides in-memory implementations of all platform primitives:
Returns a time you control. Essential for testing time-dependent logic.
Captures published events for inspection. No async delivery.
Tracks idempotency keys in memory.
Permits all actions. Skips authorization in tests.
Testing Use Cases
Use cases are tested with real domain objects and test doubles:
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:
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:
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:
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:
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:
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
# 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
- Use
FixedClock, never real time - Use
InMemoryEventBusto 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