Fake Helpers: Design Base Classes for Common Testing Patterns #352
Description
Summary
Design reusable base classes and utilities for common fake implementations, reducing boilerplate when creating test adapters.
Problem Statement
Every project using dioxide needs to create fakes for testing:
- In-memory repositories
- Fake clocks (controllable time)
- Fake event buses (captured events)
- Fake HTTP clients (recorded requests)
This is repetitive. We can provide well-tested base implementations.
Acceptance Criteria
-
InMemoryRepository[T]generic base class designed -
FakeClockcontrollable time implementation designed -
FakeEventBusevent capture implementation designed -
FakeHttpClientrequest recording implementation designed - All base classes have clear extension points
- Type-safe generics where applicable
- API documented with examples
Design
InMemoryRepository[T] - Generic CRUD Repository
from typing import TypeVar, Generic, Protocol, Callable
from dioxide.testing import InMemoryRepository
T = TypeVar("T")
ID = TypeVar("ID")
class Repository(Protocol[T, ID]):
"""Standard repository protocol."""
async def get(self, id: ID) -> T | None: ...
async def save(self, entity: T) -> T: ...
async def delete(self, id: ID) -> bool: ...
async def find_all(self) -> list[T]: ...
class InMemoryRepository(Generic[T, ID]):
"""In-memory repository for testing.
Usage:
@adapter.for_(UserRepository, profile=Profile.TEST)
class FakeUserRepository(InMemoryRepository[User, int]):
def get_id(self, entity: User) -> int:
return entity.id
def set_id(self, entity: User, id: int) -> User:
entity.id = id
return entity
Features:
- Auto-incrementing IDs (optional)
- Seed data helpers
- Query by predicate
- Clear/reset for test isolation
"""
def __init__(self):
self._store: dict[ID, T] = {}
self._next_id: int = 1
# Abstract: subclass must implement
def get_id(self, entity: T) -> ID:
"""Extract ID from entity."""
raise NotImplementedError
def set_id(self, entity: T, id: ID) -> T:
"""Set ID on entity (for auto-increment)."""
raise NotImplementedError
# Implemented CRUD operations
async def get(self, id: ID) -> T | None:
return self._store.get(id)
async def save(self, entity: T) -> T:
id = self.get_id(entity)
if id is None:
id = self._next_id
self._next_id += 1
entity = self.set_id(entity, id)
self._store[id] = entity
return entity
async def delete(self, id: ID) -> bool:
if id in self._store:
del self._store[id]
return True
return False
async def find_all(self) -> list[T]:
return list(self._store.values())
# Test helpers (not in Protocol)
def seed(self, *entities: T) -> None:
"""Seed repository with test data."""
for entity in entities:
id = self.get_id(entity)
self._store[id] = entity
def find_by(self, predicate: Callable[[T], bool]) -> list[T]:
"""Find entities matching predicate."""
return [e for e in self._store.values() if predicate(e)]
def clear(self) -> None:
"""Clear all data (for test isolation)."""
self._store.clear()
self._next_id = 1FakeClock - Controllable Time
from datetime import datetime, timedelta
from zoneinfo import ZoneInfo
class FakeClock:
"""Controllable clock for testing time-dependent logic.
Usage:
@adapter.for_(Clock, profile=Profile.TEST)
class TestClock(FakeClock):
pass
# In tests:
clock = container.resolve(Clock)
clock.set_time(datetime(2024, 1, 1))
clock.advance(days=30)
"""
def __init__(self, initial: datetime | None = None):
self._now = initial or datetime(2024, 1, 1, tzinfo=ZoneInfo("UTC"))
def now(self) -> datetime:
"""Get current (fake) time."""
return self._now
def set_time(self, dt: datetime) -> None:
"""Set the current time."""
self._now = dt
def advance(self, **kwargs) -> datetime:
"""Advance time by a delta.
Args:
**kwargs: Arguments to timedelta (days, hours, minutes, seconds, etc.)
Returns:
The new current time
"""
self._now += timedelta(**kwargs)
return self._nowFakeEventBus - Event Capture
from typing import TypeVar, Generic
from dataclasses import dataclass
E = TypeVar("E")
@dataclass
class RecordedEvent(Generic[E]):
"""Recorded event with metadata."""
event: E
timestamp: datetime
class FakeEventBus(Generic[E]):
"""Event bus that captures published events for testing.
Usage:
@adapter.for_(EventBus, profile=Profile.TEST)
class TestEventBus(FakeEventBus[DomainEvent]):
pass
# In tests:
bus = container.resolve(EventBus)
await service.do_something()
assert bus.has_event(UserCreated)
assert bus.last_event.user_id == 123
"""
def __init__(self):
self._events: list[RecordedEvent[E]] = []
async def publish(self, event: E) -> None:
"""Publish (record) an event."""
self._events.append(RecordedEvent(
event=event,
timestamp=datetime.now(ZoneInfo("UTC")),
))
@property
def events(self) -> list[E]:
"""All recorded events."""
return [r.event for r in self._events]
@property
def last_event(self) -> E | None:
"""Most recently published event."""
return self._events[-1].event if self._events else None
def has_event(self, event_type: type[E]) -> bool:
"""Check if an event of given type was published."""
return any(isinstance(r.event, event_type) for r in self._events)
def events_of_type(self, event_type: type[E]) -> list[E]:
"""Get all events of a specific type."""
return [r.event for r in self._events if isinstance(r.event, event_type)]
def clear(self) -> None:
"""Clear recorded events."""
self._events.clear()FakeHttpClient - Request Recording
from dataclasses import dataclass
from typing import Any
@dataclass
class RecordedRequest:
"""Recorded HTTP request."""
method: str
url: str
headers: dict[str, str]
body: Any
response: Any # What to return
class FakeHttpClient:
"""HTTP client that records requests for testing.
Usage:
@adapter.for_(HttpClient, profile=Profile.TEST)
class TestHttpClient(FakeHttpClient):
pass
# In tests:
client = container.resolve(HttpClient)
client.stub_response("GET", "https://api.example.com/users", [{"id": 1}])
result = await service.fetch_users()
assert client.was_called("GET", "https://api.example.com/users")
"""
def __init__(self):
self._requests: list[RecordedRequest] = []
self._stubs: dict[tuple[str, str], Any] = {}
def stub_response(self, method: str, url: str, response: Any) -> None:
"""Stub a response for a specific request."""
self._stubs[(method.upper(), url)] = response
async def request(
self,
method: str,
url: str,
headers: dict[str, str] | None = None,
body: Any = None,
) -> Any:
"""Make a (fake) HTTP request."""
key = (method.upper(), url)
response = self._stubs.get(key)
self._requests.append(RecordedRequest(
method=method.upper(),
url=url,
headers=headers or {},
body=body,
response=response,
))
if response is None:
raise ValueError(f"No stub for {method} {url}")
return response
def was_called(self, method: str, url: str) -> bool:
"""Check if a specific request was made."""
return any(
r.method == method.upper() and r.url == url
for r in self._requests
)
def clear(self) -> None:
"""Clear recorded requests and stubs."""
self._requests.clear()
self._stubs.clear()Module Structure
dioxide/
├── testing.py # Existing: fresh_container
├── testing/
│ ├── __init__.py # Export all fakes
│ ├── repository.py # InMemoryRepository
│ ├── clock.py # FakeClock
│ ├── events.py # FakeEventBus
│ └── http.py # FakeHttpClient
Dependencies
- Blocked by: None
- Blocks: Fake Helpers: Implement Testing Utilities Package #353 (Fake Helpers Implementation)
Parent Epic
Part of #336 (Production-Ready Developer Experience Improvements)
Definition of Ready
- Problem clearly stated
- Acceptance criteria are testable
- All base classes designed with examples