paulpas Domain Testing Strategies
Tests DDD domain models to verify invariants hold at construction time, aggregates enforce consistency boundaries, specifications correctly classify candidates, and domain events fire predictably after state transitions. Produces unit tests for value objects, entities, aggregate roots, and domain services with concrete assertion strategies for each pattern.
TL;DR Checklist
- Test every value object constructor — valid inputs pass, invalid inputs raise
- Test aggregate root as a single unit — never test internal entities in isolation
- For each specification: write one passing candidate test and one failing candidate test
- Verify domain events are published after mutation, not before, with correct payload data
- Use stubs for read-only dependencies (repositories returning fixed aggregates), mocks only for side-effect producers (publishers, notification senders)
- Test invariants at construction time — never test objects that violate invariants because they cannot exist
When to Use
- Writing unit tests for a DDD domain layer with value objects, entities, and aggregate roots
- Verifying that an aggregate enforces its consistency boundary (e.g., order total recalculated after every item change)
- Testing specification implementations against known passing and failing candidates
- Building test suites that verify domain events are published with correct payloads after state mutations
- Setting up test doubles (stubs, mocks) for domain-layer dependencies in a clean architecture
When NOT to Use
- Testing infrastructure layers (database persistence, HTTP clients, message brokers) — use integration tests or separate infrastructure testing skills
- Testing application services that orchestrate multiple aggregates — this is cross-cutting concern testing, not pure domain testing
- Testing presentation/UI layers where domain logic has no presence
- When the codebase uses anemic domain models (plain data carriers with no behavior) — fix the domain model first before writing meaningful tests
Core Workflow
Map Domain Elements to Test Categories — Identify which elements need which testing strategy: value objects → invariant tests at construction, entities → lifecycle + identity tests, aggregates → whole-unit mutation tests, specifications → passing/failing candidate pairs, domain events → publish-after-mutation verification. Checkpoint: Every test must map to exactly one domain element; if a test touches three different aggregates or services, split it.
Write Value Object Tests First — These are the simplest: construct valid instances, assert equality by value, then construct invalid inputs and assert exceptions. Test edge cases like zero amounts, maximum precision, empty strings for required fields, and currency mismatches. Checkpoint: Every public factory method or
__post_init__validation rule must have at least one test — if a validation branch has no test, the invariant is unverified.Test Aggregates as Whole Units — Test every public command (method) on the aggregate root. Assert invariants after each mutation by examining the aggregate's state or published events. Never mock away the aggregate root itself — you are testing its behavior, not isolating it from its children. For methods that span child entities, verify cross-entity consistency (e.g., order total matches sum of item prices). Checkpoint: After every method call under test, the aggregate must be in a logically consistent state — either by inspecting internal fields or by verifying published events capture the full state transition.
Write Specification Tests in Pairs — For each specification class, write two tests: one with an object that must pass and one that must fail. The failing test is equally important — it verifies the rejection logic matches business rules exactly. Use edge cases: boundary values, empty collections, single-element collections, null-or-absent fields. Checkpoint: A specification is fully tested when both the positive and negative tests cover every branch in its
is_satisfied_bymethod.Verify Domain Event Publishing — After a state-changing aggregate method, assert that the correct events were published with the expected data payload. Do not use spies for event publishing unless you need to verify the exact number of publications — stubs or simple list assertions are sufficient. Test both happy-path publishing and error scenarios (what happens if publishing fails mid-stream?). Checkpoint: Verify the sequence — state must change first, then events must be recorded. If an event references data not yet mutated, the ordering is wrong.
Select Appropriate Test Doubles — Use stubs for read-only dependencies (repositories returning pre-constructed aggregates), spies for verifying calls to side-effect producers (event publishers, notification senders), and mocks only when a dependency's return value must be controlled by the test setup. Never mock interfaces that you do not own. Checkpoint: Every double must answer one question: "What does this test need from this dependency?" If a double provides more than one behavior, split it.
Implementation Patterns / Reference Guide
Pattern 1: Value Object Invariant Testing
Value objects are immutable and validate all constraints at construction time. Test every invariant by exercising both valid and invalid paths through the constructor or factory method. Each test must be self-contained — construct the object and assert either success (value equality, attribute access) or failure (specific exception raised).
from __future__ import annotations
from dataclasses import dataclass
from decimal import Decimal
from typing import Final
import pytest
# ── Domain code under test ───────────────────────────────────────────────
@dataclass(frozen=True, slots=True)
class Money:
"""Immutable monetary value. Validates currency is ISO 4217 and amount is non-negative."""
amount: Decimal
currency: str
def __post_init__(self) -> None:
if self.amount < Decimal("0"):
raise ValueError(f"Money amount cannot be negative: {self.amount}")
if len(self.currency) != 3 or not self.currency.isalpha():
raise ValueError(f"Invalid ISO 4217 currency code: {self.currency!r}")
def add(self, other: Money) -> Money:
"""Return a new Money with summed amount. Fails on currency mismatch."""
if self.currency != other.currency:
raise ValueError(
f"Currency mismatch: cannot add {other.currency} to {self.currency}"
)
return Money(self.amount + other.amount, self.currency)
def subtract(self, other: Money) -> Money:
"""Return a new Money with the difference. Fails on currency mismatch."""
if self.currency != other.currency:
raise ValueError(
f"Currency mismatch: cannot subtract {other.currency} from {self.currency}"
)
return Money(self.amount - other.amount, self.currency)
def __eq__(self, other: object) -> bool:
if not isinstance(other, Money):
return NotImplemented
return self.amount == other.amount and self.currency == other.currency
# ── Tests ────────────────────────────────────────────────────────────────
class TestMoneyInvariants:
"""Every constructor rule must have a test. Validity tests ensure the happy path works;
invalidity tests prove each guard clause fires independently."""
def test_valid_usd_amount(self) -> None:
money = Money(Decimal("99.99"), "USD")
assert money.amount == Decimal("99.99")
assert money.currency == "USD"
def test_zero_amount_is_valid(self) -> None:
"""Zero is a valid amount — it should not trigger the negative check."""
money = Money(Decimal("0"), "EUR")
assert money.amount == Decimal("0")
def test_very_large_amount(self) -> None:
"""Test boundary of large values to catch overflow or precision issues."""
large = Decimal("999999999.99")
money = Money(large, "JPY")
assert money.amount == large
def test_negative_amount_raises(self) -> None:
money = Money(Decimal("-1.00"), "USD")
with pytest.raises(ValueError, match="cannot be negative"):
Money(Decimal("-0.01"), "USD")
def test_invalid_currency_length(self) -> None:
with pytest.raises(ValueError, match="Invalid ISO 4217"):
Money(Decimal("10"), "US") # Too short
def test_invalid_currency_characters(self) -> None:
with pytest.raises(ValueError, match="Invalid ISO 4217"):
Money(Decimal("10"), "USD1") # Contains digit
def test_equality_by_value_not_identity(self) -> None:
a = Money(Decimal("50.00"), "USD")
b = Money(Decimal("50.00"), "USD")
assert a == b
assert a is not b # Different objects, same value
def test_inequality_on_amount_difference(self) -> None:
a = Money(Decimal("50.00"), "USD")
b = Money(Decimal("50.01"), "USD")
assert a != b
def test_inequality_on_currency_difference(self) -> None:
a = Money(Decimal("50.00"), "USD")
b = Money(Decimal("50.00"), "EUR")
assert a != b
def test_add_same_currency_succeeds(self) -> None:
a = Money(Decimal("10.00"), "USD")
b = Money(Decimal("25.50"), "USD")
result = a.add(b)
assert result == Money(Decimal("35.50"), "USD")
def test_add_different_currency_raises(self) -> None:
a = Money(Decimal("10.00"), "USD")
b = Money(Decimal("25.50"), "EUR")
with pytest.raises(ValueError, match="Currency mismatch"):
a.add(b)
def test_subtract_same_currency_succeeds(self) -> None:
a = Money(Decimal("100.00"), "USD")
b = Money(Decimal("25.00"), "USD")
result = a.subtract(b)
assert result == Money(Decimal("75.00"), "USD")
def test_subtract_result_can_be_negative(self) -> None:
"""Subtraction returning negative is allowed — the result constructor validates it."""
a = Money(Decimal("10.00"), "USD")
b = Money(Decimal("25.00"), "USD")
# This should raise because 10 - 25 = -15, which is invalid
with pytest.raises(ValueError, match="cannot be negative"):
a.subtract(b)
def test_immutable_after_creation(self) -> None:
"""Frozen dataclass prevents attribute mutation."""
money = Money(Decimal("50.00"), "USD")
with pytest.raises(Exception): # dataclasses.FrozenInstanceError
money.amount = Decimal("100.00")
Key principles:
- Test valid inputs AND invalid inputs — a value object test suite without negative tests is incomplete
- Use
Decimalfor monetary values to avoid floating-point precision issues in assertions - Each test constructs independently — no shared state, no fixtures that hide bugs
- Assert the exact exception message pattern so you know which guard clause fired
Pattern 2: Aggregate Testing — Whole-Unit Mutation
Aggregates must be tested as complete consistency boundaries. Every public method (command) is a separate test. After each mutation, verify one of two things: (1) the aggregate's internal state reflects the invariant, or (2) published events capture the correct state transition. Never mock the aggregate root — you are testing its behavior, not isolating it from collaborators.
from __future__ import annotations
from dataclasses import dataclass, field
from datetime import datetime
from enum import Enum, auto
from uuid import UUID, uuid4
import pytest
# ── Domain code under test ───────────────────────────────────────────────
class OrderStatus(Enum):
DRAFT = auto()
CONFIRMED = auto()
CANCELLED = auto()
SHIPPED = auto()
@dataclass(frozen=True, slots=True)
class OrderItem:
product_id: str
quantity: int
unit_price: float
@property
def line_total(self) -> float:
return self.quantity * self.unit_price
def __post_init__(self) -> None:
if self.quantity <= 0:
raise ValueError("Quantity must be positive")
if self.unit_price < 0:
raise ValueError("Unit price cannot be negative")
class OrderTotalMismatchError(Exception):
"""Raised when recalculated total does not match stored total."""
class Order:
"""Aggregate root for the Order bounded context.
Invariants:
- At least one item required before confirmation
- Total must equal sum of all item line totals at all times
- Once confirmed, no items may be added or removed
- A cancelled order cannot be re-confirmed
"""
def __init__(
self,
order_id: UUID,
customer_email: str,
publisher: OrderEventPublisher | None = None,
) -> None:
self.order_id = order_id
self.customer_email = customer_email
self._items: list[OrderItem] = []
self._total: float = 0.0
self.status: OrderStatus = OrderStatus.DRAFT
self._publisher = publisher or _NoOpPublisher()
@property
def items(self) -> tuple[OrderItem, ...]:
return tuple(self._items)
@property
def total(self) -> float:
return self._total
def add_item(
self, product_id: str, quantity: int, unit_price: float
) -> None:
"""Add an item to the order. Only allowed in DRAFT status."""
if self.status != OrderStatus.DRAFT:
raise RuntimeError(
f"Cannot add items to {self.status.name} order {self.order_id}"
)
item = OrderItem(product_id, quantity, unit_price)
self._items.append(item)
self._recalculate_total()
def remove_item(self, product_id: str) -> None:
"""Remove an item from the order by product ID. Only allowed in DRAFT status."""
if self.status != OrderStatus.DRAFT:
raise RuntimeError(
f"Cannot remove items from {self.status.name} order {self.order_id}"
)
original_count = len(self._items)
self._items = [i for i in self._items if i.product_id != product_id]
if len(self._items) == original_count:
raise ValueError(f"No item with product_id {product_id} found")
if not self._items:
raise RuntimeError("Order must have at least one item")
self._recalculate_total()
def confirm(self) -> None:
"""Confirm the order. Requires at least one item."""
if self.status != OrderStatus.DRAFT:
raise RuntimeError(
f"Cannot confirm order in {self.status.name} state"
)
if not self._items:
raise RuntimeError("Cannot confirm order with no items")
previous_status = self.status
self.status = OrderStatus.CONFIRMED
self._publisher.publish_order_confirmed(
order_id=self.order_id,
customer_email=self.customer_email,
item_count=len(self._items),
total_amount=self._total,
)
def cancel(self) -> None:
"""Cancel the order. Cannot cancel an already cancelled order."""
if self.status == OrderStatus.CANCELLED:
raise RuntimeError("Order is already cancelled")
previous_status = self.status
self.status = OrderStatus.CANCELLED
self._publisher.publish_order_cancelled(
order_id=self.order_id,
reason=f"Cancelled from {previous_status.name}",
)
def ship(self) -> None:
"""Mark as shipped. Only confirmed orders can be shipped."""
if self.status != OrderStatus.CONFIRMED:
raise RuntimeError(f"Cannot ship order in {self.status.name} state")
self.status = OrderStatus.SHIPPED
self._publisher.publish_order_shipped(order_id=self.order_id)
def _recalculate_total(self) -> None:
"""Recalculate total from items. Internal method — always kept in sync."""
self._total = sum(item.line_total for item in self._items)
class OrderEventPublisher:
"""Interface for publishing order events."""
def publish_order_confirmed(
self, order_id: UUID, customer_email: str, item_count: int, total_amount: float
) -> None: ...
def publish_order_cancelled(self, order_id: UUID, reason: str) -> None: ...
def publish_order_shipped(self, order_id: UUID) -> None: ...
class _NoOpPublisher(OrderEventPublisher):
"""Do-nothing publisher used when no event handling is needed."""
def publish_order_confirmed(
self, order_id: UUID, customer_email: str, item_count: int, total_amount: float
) -> None:
pass
def publish_order_cancelled(self, order_id: UUID, reason: str) -> None:
pass
def publish_order_shipped(self, order_id: UUID) -> None:
pass
# ── Tests using a recording publisher to verify events ───────────────────
@dataclass
class _RecordingPublisher(OrderEventPublisher):
"""Captures all published events for test assertions."""
events: list[dict] = field(default_factory=list)
def publish_order_confirmed(
self, order_id: UUID, customer_email: str, item_count: int, total_amount: float
) -> None:
self.events.append({
"type": "confirmed",
"order_id": order_id,
"customer_email": customer_email,
"item_count": item_count,
"total_amount": total_amount,
})
def publish_order_cancelled(self, order_id: UUID, reason: str) -> None:
self.events.append({
"type": "cancelled",
"order_id": order_id,
"reason": reason,
})
def publish_order_shipped(self, order_id: UUID) -> None:
self.events.append({"type": "shipped", "order_id": order_id})
class TestOrderAggregate:
"""Tests for the Order aggregate root.
Each test exercises one public command and verifies either:
- The aggregate's state invariant holds after mutation
- The correct events were published with correct payloads
"""
def _make_order(
self,
) -> tuple[Order, _RecordingPublisher]:
"""Factory helper — returns order with recording publisher."""
pub = _RecordingPublisher()
order = Order(uuid4(), "customer@example.com", pub)
return order, pub
# ── add_item tests ──────────────────────────────────────────────────
def test_add_item_sets_total(self) -> None:
order, pub = self._make_order()
order.add_item("PROD-001", 2, 25.00)
assert len(order.items) == 1
assert order.total == 50.00
def test_add_multiple_items_accumulates_total(self) -> None:
order, pub = self._make_order()
order.add_item("PROD-001", 2, 25.00)
order.add_item("PROD-002", 3, 10.00)
assert order.total == 80.00 # (2*25) + (3*10)
def test_add_item_on_confirmed_order_raises(self) -> None:
order, pub = self._make_order()
order.add_item("PROD-001", 1, 10.00)
order.confirm()
with pytest.raises(RuntimeError, match="Cannot add items"):
order.add_item("PROD-002", 1, 20.00)
def test_add_item_with_invalid_quantity_raises(self) -> None:
order, _ = self._make_order()
with pytest.raises(ValueError, match="Quantity must be positive"):
order.add_item("PROD-001", 0, 10.00)
# ── remove_item tests ───────────────────────────────────────────────
def test_remove_item_updates_total(self) -> None:
order, pub = self._make_order()
order.add_item("PROD-001", 2, 25.00)
order.add_item("PROD-002", 3, 10.00)
order.remove_item("PROD-001")
assert len(order.items) == 1
assert order.total == 30.00
def test_remove_item_when_none_exists_raises(self) -> None:
order, _ = self._make_order()
order.add_item("PROD-001", 1, 10.00)
with pytest.raises(ValueError, match="No item with product_id"):
order.remove_item("NONEXISTENT")
def test_remove_last_item_raises(self) -> None:
order, _ = self._make_order()
order.add_item("PROD-001", 1, 10.00)
with pytest.raises(RuntimeError, match="must have at least one item"):
order.remove_item("PROD-001")
# ── confirm tests ───────────────────────────────────────────────────
def test_confirm_publishes_event_with_correct_payload(self) -> None:
order, pub = self._make_order()
order.add_item("PROD-001", 2, 25.00)
order.confirm()
assert order.status == OrderStatus.CONFIRMED
assert len(pub.events) == 1
event = pub.events[0]
assert event["type"] == "confirmed"
assert event["item_count"] == 1
assert event["total_amount"] == 50.00
assert event["customer_email"] == "customer@example.com"
def test_confirm_empty_order_raises(self) -> None:
order, pub = self._make_order()
with pytest.raises(RuntimeError, match="no items"):
order.confirm()
assert len(pub.events) == 0 # No events should publish on failure
def test_confirm_already_confirmed_order_raises(self) -> None:
order, _ = self._make_order()
order.add_item("PROD-001", 1, 10.00)
order.confirm()
with pytest.raises(RuntimeError, match="Cannot confirm"):
order.confirm()
def test_cancel_publishes_event(self) -> None:
order, pub = self._make_order()
order.add_item("PROD-001", 1, 10.00)
order.cancel()
assert order.status == OrderStatus.CANCELLED
assert len(pub.events) == 1
assert pub.events[0]["type"] == "cancelled"
def test_cancel_already_cancelled_raises(self) -> None:
order, _ = self._make_order()
order.add_item("PROD-001", 1, 10.00)
order.cancel()
with pytest.raises(RuntimeError, match="already cancelled"):
order.cancel()
def test_ship_confirmed_order_publishes_event(self) -> None:
order, pub = self._make_order()
order.add_item("PROD-001", 1, 10.00)
order.confirm()
order.ship()
assert order.status == OrderStatus.SHIPPED
assert len(pub.events) == 2
assert pub.events[0]["type"] == "confirmed"
assert pub.events[1]["type"] == "shipped"
def test_cancelled_order_cannot_be_shipped(self) -> None:
order, _ = self._make_order()
order.add_item("PROD-001", 1, 10.00)
order.cancel()
with pytest.raises(RuntimeError, match="Cannot ship"):
order.ship()
def test_items_are_immutable_tuple(self) -> None:
"""Items property returns a tuple, preventing external mutation."""
order, _ = self._make_order()
order.add_item("PROD-001", 2, 25.00)
with pytest.raises(TypeError):
order.items[0].quantity = 999
Key principles:
- Test each public command in isolation — one mutation per test, one assertion group per test
- Use a recording publisher to verify events — it's a simple stub, not a complex mock framework
- Always test failure paths (wrong status, empty collections, invalid inputs)
- Verify event sequence when multiple commands are chained (confirm → ship should produce two events in order)
Pattern 3: Specification Testing — Passing and Failing Candidate Pairs
Specifications define business rules as reusable predicates. Each specification must have at least two tests: one candidate that satisfies the rule, and one that does not. Edge cases go in the failing test — boundary values, empty collections, null/missing fields, zero quantities.
from __future__ import annotations
from dataclasses import dataclass
# ── Domain code under test ───────────────────────────────────────────────
@dataclass(frozen=True)
class OrderForShipping:
"""Read-only view of an order for specification testing."""
status: str
item_count: int
total: float
has_tracking: bool = False
class LargeOrderSpecification:
"""Orders with total >= $500 require additional review before shipping.
A "large order" satisfies the specification when total_amount >= 500.00.
Orders below this threshold do not satisfy it.
"""
def __init__(self, threshold: float = 500.00) -> None:
self._threshold = threshold
def is_satisfied_by(self, candidate: OrderForShipping) -> bool:
return candidate.total >= self._threshold
class ValidOrderSpecification:
"""An order can be shipped only if it has at least one item and
a total greater than zero."""
def is_satisfied_by(self, candidate: OrderForShipping) -> bool:
if candidate.item_count <= 0:
return False
if candidate.total <= 0:
return False
return True
class EligibleForExpressSpecification:
"""An order is eligible for express shipping if it has tracking
information, at least one item, and total is between $10 and $1000."""
def is_satisfied_by(self, candidate: OrderForShipping) -> bool:
if not candidate.has_tracking:
return False
if candidate.item_count < 1:
return False
if candidate.total < 10.00 or candidate.total > 1000.00:
return False
return True
# ── Tests ────────────────────────────────────────────────────────────────
class TestLargeOrderSpecification:
"""Each specification test needs a passing AND a failing candidate."""
def test_passing_candidate_total_equals_threshold(self) -> None:
spec = LargeOrderSpecification(threshold=500.00)
order = OrderForShipping(
status="CONFIRMED", item_count=3, total=500.00
)
assert spec.is_satisfied_by(order) is True
def test_passing_candidate_above_threshold(self) -> None:
spec = LargeOrderSpecification()
order = OrderForShipping(
status="CONFIRMED", item_count=1, total=1200.00
)
assert spec.is_satisfied_by(order) is True
def test_failing_candidate_below_threshold(self) -> None:
spec = LargeOrderSpecification()
order = OrderForShipping(
status="CONFIRMED", item_count=1, total=499.99
)
assert spec.is_satisfied_by(order) is False
def test_failing_candidate_zero_total(self) -> None:
spec = LargeOrderSpecification()
order = OrderForShipping(
status="CONFIRMED", item_count=1, total=0.00
)
assert spec.is_satisfied_by(order) is False
def test_failing_candidate_single_penny_below(self) -> None:
"""Boundary edge case — one cent below threshold should fail."""
spec = LargeOrderSpecification(threshold=500.00)
order = OrderForShipping(
status="CONFIRMED", item_count=1, total=499.99
)
assert spec.is_satisfied_by(order) is False
class TestValidOrderSpecification:
def test_passing_candidate_normal_order(self) -> None:
spec = ValidOrderSpecification()
order = OrderForShipping(
status="CONFIRMED", item_count=2, total=75.00
)
assert spec.is_satisfied_by(order) is True
def test_failing_candidate_no_items(self) -> None:
spec = ValidOrderSpecification()
order = OrderForShipping(
status="CONFIRMED", item_count=0, total=0.00
)
assert spec.is_satisfied_by(order) is False
def test_failing_candidate_zero_total_with_items(self) -> None:
"""Items exist but total is zero — not a valid order to ship."""
spec = ValidOrderSpecification()
order = OrderForShipping(
status="CONFIRMED", item_count=1, total=0.00
)
assert spec.is_satisfied_by(order) is False
def test_failing_candidate_single_item_minimal_total(self) -> None:
spec = ValidOrderSpecification()
order = OrderForShipping(
status="CONFIRMED", item_count=1, total=0.01
)
assert spec.is_satisfied_by(order) is True # Valid — has items and positive total
class TestEligibleForExpressSpecification:
def test_passing_candidate_within_range(self) -> None:
spec = EligibleForExpressSpecification()
order = OrderForShipping(
status="CONFIRMED", item_count=1, total=500.00, has_tracking=True
)
assert spec.is_satisfied_by(order) is True
def test_passing_candidate_exact_lower_boundary(self) -> None:
spec = EligibleForExpressSpecification()
order = OrderForShipping(
status="CONFIRMED", item_count=1, total=10.00, has_tracking=True
)
assert spec.is_satisfied_by(order) is True
def test_failing_candidate_no_tracking(self) -> None:
spec = EligibleForExpressSpecification()
order = OrderForShipping(
status="CONFIRMED", item_count=1, total=500.00, has_tracking=False
)
assert spec.is_satisfied_by(order) is False
def test_failing_candidate_over_upper_limit(self) -> None:
spec = EligibleForExpressSpecification()
order = OrderForShipping(
status="CONFIRMED", item_count=1, total=1000.01, has_tracking=True
)
assert spec.is_satisfied_by(order) is False
def test_failing_candidate_below_lower_limit(self) -> None:
spec = EligibleForExpressSpecification()
order = OrderForShipping(
status="CONFIRMED", item_count=1, total=9.99, has_tracking=True
)
assert spec.is_satisfied_by(order) is False
def test_failing_candidate_multiple_conditions_fail(self) -> None:
"""When multiple conditions are violated, the specification still returns False."""
spec = EligibleForExpressSpecification()
order = OrderForShipping(
status="CONFIRMED", item_count=0, total=5.00, has_tracking=False
)
assert spec.is_satisfied_by(order) is False
Key principles:
- Always write both a passing and a failing test per specification — the failing test catches logic errors in rejection paths
- Test boundary values explicitly: threshold equals, one cent below, zero, single item
- For multi-condition specifications, test each condition independently with failing candidates before testing combined failures
Pattern 4: Domain Event Testing — Publish After Mutation
Domain events must be verified to fire after state changes, not before. Use a recording publisher (stub) to capture events and assert both timing (order of mutations vs. events) and payload accuracy. Test the happy path and the failure path (what happens if an event fails to publish?).
from __future__ import annotations
from dataclasses import dataclass, field
from datetime import datetime
from uuid import UUID
import pytest
# ── Domain code under test ───────────────────────────────────────────────
@dataclass(frozen=True)
class OrderCreatedEvent:
order_id: UUID
customer_email: str
item_count: int
total_amount: float
created_at: datetime = field(default_factory=datetime.now)
@dataclass(frozen=True)
class OrderCancelledEvent:
order_id: UUID
reason: str
cancelled_at: datetime = field(default_factory=datetime.now)
@dataclass
class InMemoryEventStore:
"""In-memory store that records all published events for test assertions."""
events: list[object] = field(default_factory=list)
def record(self, event: object) -> None:
self.events.append(event)
def clear(self) -> None:
self.events.clear()
@property
def created_events(self) -> list[OrderCreatedEvent]:
return [e for e in self.events if isinstance(e, OrderCreatedEvent)]
@property
def cancelled_events(self) -> list[OrderCancelledEvent]:
return [e for e in self.events if isinstance(e, OrderCancelledEvent)]
class SimpleOrder:
"""Minimal order that publishes events through a store."""
def __init__(self, order_id: UUID, customer_email: str, store: InMemoryEventStore) -> None:
self.order_id = order_id
self.customer_email = customer_email
self.status: str = "DRAFT"
self._store = store
def confirm(self) -> None:
if self.status != "DRAFT":
raise RuntimeError("Order not in draft state")
# State change FIRST
self.status = "CONFIRMED"
# THEN publish event
self._store.record(OrderCreatedEvent(
order_id=self.order_id,
customer_email=self.customer_email,
item_count=1,
total_amount=99.99,
))
def cancel(self, reason: str) -> None:
if self.status == "CANCELLED":
raise RuntimeError("Already cancelled")
# State change FIRST
self.status = "CANCELLED"
# THEN publish event
self._store.record(OrderCancelledEvent(
order_id=self.order_id,
reason=reason,
))
# ── Tests ────────────────────────────────────────────────────────────────
class TestOrderEventPublishing:
"""Verifies events are published AFTER state mutations and contain correct data."""
def test_event_published_after_status_change(self) -> None:
store = InMemoryEventStore()
order = SimpleOrder(UUID("11111111-0000-0000-0000-000000000001"), "a@b.com", store)
order.confirm()
# State changed first, then event recorded
assert order.status == "CONFIRMED"
assert len(store.created_events) == 1
def test_event_payload_contains_all_required_data(self) -> None:
store = InMemoryEventStore()
order_id = UUID("22222222-0000-0000-0000-000000000002")
email = "buyer@example.com"
order = SimpleOrder(order_id, email, store)
order.confirm()
event = store.created_events[0]
assert event.order_id == order_id
assert event.customer_email == email
assert event.item_count == 1
assert event.total_amount == 99.99
def test_cancel_publishes_correct_event(self) -> None:
store = InMemoryEventStore()
order_id = UUID("33333333-0000-0000-0000-000000000003")
order = SimpleOrder(order_id, "buyer@test.com", store)
order.confirm()
order.cancel(reason="Customer requested cancellation")
assert order.status == "CANCELLED"
assert len(store.cancelled_events) == 1
assert store.cancelled_events[0].reason == "Customer requested cancellation"
def test_no_events_published_on_failed_mutation(self) -> None:
"""If a mutation raises an exception, no events should be recorded."""
store = InMemoryEventStore()
order = SimpleOrder(UUID("44444444-0000-0000-0000-000000000004"), "a@b.com", store)
with pytest.raises(RuntimeError, match="not in draft"):
order.cancel(reason="early cancel")
assert len(store.created_events) == 0
assert len(store.cancelled_events) == 0
def test_duplicate_cancel_raises_without_publishing(self) -> None:
store = InMemoryEventStore()
order = SimpleOrder(UUID("55555555-0000-0000-0000-000000000005"), "a@b.com", store)
order.confirm()
order.cancel("first reason")
initial_count = len(store.cancelled_events)
with pytest.raises(RuntimeError, match="Already cancelled"):
order.cancel("second reason")
# Event count should not increase on the failed call
assert len(store.cancelled_events) == initial_count
Key principles:
- Always test the ordering: state change must happen before event recording
- Test that failed mutations do NOT produce events (the guard clause fires before either mutation or publishing)
- Use an in-memory store as a stub — it's simpler than mocking frameworks and fully transparent for inspection
- Assert all fields of published events, not just their count
Pattern 5: Test Double Selection Strategy for DDD
Choosing the right test double type is critical in DDD testing. The wrong choice leads to tests that pass but don't verify behavior. Use this decision guide:
| Dependency Type | Use | Why | Example |
|---|---|---|---|
| Read-only (returns data) | Stub | Test needs fixed return values, no verification needed | Repository returning a pre-constructed aggregate |
| Side-effect only (no return value) | Spy | Test needs to verify the call happened with correct arguments | Event publisher, notification sender |
| Behavior-controllable | Mock | Test setup must determine what the dependency returns | External pricing service returning different rates |
| Owned interface (your code) | Real implementation | Never mock interfaces you own — test the real behavior | Specification implementations, value objects |
from __future__ import annotations
from dataclasses import dataclass, field
from uuid import UUID
# ── Domain code under test ───────────────────────────────────────────────
class ProductRepository:
"""Interface for loading products. In production, this talks to a database."""
def get(self, product_id: UUID) -> Product | None: ...
@dataclass(frozen=True)
class Product:
product_id: UUID
name: str
price: float
in_stock: bool = True
class OrderPlacingService:
"""Application service that uses a repository and publishes events.
In production, this depends on a real ProductRepository and a real EventPublisher.
In tests, we inject controlled doubles.
"""
def __init__(
self,
product_repo: ProductRepository,
event_publisher: object,
) -> None:
self._product_repo = product_repo
self._event_publisher = event_publisher
def place_order(self, customer_id: UUID, product_id: UUID) -> dict:
"""Place an order for a single product.
Returns a result dict or raises if the product is not found or out of stock.
Publishes OrderPlacedEvent on success.
"""
product = self._product_repo.get(product_id)
if product is None:
raise ValueError(f"Product {product_id} not found")
if not product.in_stock:
raise RuntimeError(f"Product {product.name} is out of stock")
# Publish event — we only verify this was called, not the internal implementation
if hasattr(self._event_publisher, "publish"):
self._event_publisher.publish("OrderPlaced", {
"customer_id": customer_id,
"product_id": product_id,
"product_name": product.name,
"price": product.price,
})
return {
"status": "placed",
"product": product.name,
"price": product.price,
}
# ── Test doubles ─────────────────────────────────────────────────────────
class ProductNotFound:
"""Stub repository that returns None for any product_id — simulates DB miss."""
def get(self, product_id: UUID) -> None:
return None # type: ignore
class InStockProduct:
"""Stub repository that returns a known in-stock product."""
def __init__(self, product: Product = None) -> None:
self._product = product or Product(
UUID("aaaaaaaa-0000-0000-0000-000000000001"),
"Widget", 29.99,
in_stock=True,
)
def get(self, product_id: UUID) -> Product:
return self._product
class OutOfStockProduct:
"""Stub repository that returns a product marked out of stock."""
def __init__(self, product: Product = None) -> None:
self._product = product or Product(
UUID("bbbbbbbb-0000-0000-0000-000000000002"),
"Gadget", 49.99,
in_stock=False,
)
def get(self, product_id: UUID) -> Product:
return self._product
cl
…(truncated)