Mocks & Fixtures — Aquilia Documentation
Comprehensive guide and documentation for Mocks & Fixtures in the Aquilia framework. View API reference, examples, and implementation patterns.
Testing / Mocks & Mixins Mocks & Test Mixins Aquilia provides specialized mock objects, context overrides, and testing mixins. Swap dependencies, assert on background flows, and inspect side-effects cleanly from your tests. MockFaultEngine MockFaultEngine replaces the framework default FaultEngine during test phases, capturing all raised faults into an internal list instead of dispatching them to production handlers. from aquilia.testing import MockFaultEngine from aquilia.faults import Fault # Setup engine engine = MockFaultEngine() # Simulate code emitting a fault my_fault = Fault(code="DATABASE_TIMEOUT", message="DB took too long") engine.emit(my_fault, app_name="billing") # Check captured faults assert engine.has_fault("DATABASE_TIMEOUT") assert engine.fault_count == 1 assert engine.last_fault_code == "DATABASE_TIMEOUT" # Reset history between assertions engine.reset() assert engine.fault_count == 0 Testing Assertions If you subclass AquiliaTestCase , you gain access to the following fault assertion methods: self.assert_fault_raised(engine, code=None, domain=None): Asserts that at least one fault matching the code or domain was captured. self.assert_no_faults(engine): Asserts that no faults were captured. self.assert_fault_count(engine, expected): Asserts that exactly the expected number of faults were captured. MockEffectRegistry & MockFlowContext The effect system isolates side effects like databases, cache backends, and task queues. With MockEffectRegistry and MockFlowContext, you can stub these operations and inject mock values into your pipeline handlers: from aquilia.testing import MockEffectRegistry, MockFlowContext # Create registry and register mock effects registry = MockEffectRegistry() mock_db = registry.register_mock("DBTx", return_value="fake_connection") # Acquire effect resources provider = registry.get_provider("DBTx") conn = await provider.acquire(mode="write") assert conn == "fake_connection" assert provider.acquire_count == 1 # Setup sequential returns (for multiple calls) mock_cache = registry.register_mock("Cache", return_sequence=["val1", "val2"]) provider_cache = registry.get_provider("Cache") assert await provider_cache.acquire() == "val1" assert await provider_cache.acquire() == "val2" assert await provider_cache.acquire() == "val2" # Repeats last item # Inject into MockFlowContext for pipeline node testing ctx = MockFlowContext.from_registry(registry) db_resource = ctx.get_effect("DBTx") assert db_resource == "fake_connection" Dependency Injection Overrides Aquilia's DI system includes testing helpers to override services and monitor calls within the container: mock_provider(token, value): Creates a mock provider that resolves to a fixed stub object. override_provider(container, token, mock_value): Async context manager that temporarily swaps a token provider in the container, restoring the original on exit. spy_provider(container, token): Async context manager that wraps a real provider. It monitors and logs instantiation counts and values, but still delegates calls to the real service. from aquilia.testing import override_provider, spy_provider class TestServiceDI(AquiliaTestCase): async def test_repository_override(self): # Override the real database repo with a mock repo fake_repo = MockUserRepository() async with override_provider(self.di_container, UserRepository, fake_repo): resolved = await self.di_container.resolve_async(UserRepository) assert resolved is fake_repo async def test_email_spy(self): # Spy on the real email service async with spy_provider(self.di_container, EmailService) as spy: # Trigger logic that invokes EmailService await self.client.post("/api/register", json= ) # Assert details on spy assert spy.resolve_count == 1 assert len(spy.resolved_values) == 1 Subsystem Mixins Aquilia includes three mixins to easily interact with and assert on cache, authentication, and mail payloads: 1. CacheTestMixin Integrates cache assertions. Automatically hooks into self.cache_service: self.populate_cache(data, ttl=None): Populate keys into the cache. self.assert_cached(key): Assert key is present. self.assert_not_cached(key): Assert key is missing. self.assert_cache_value(key, expected): Assert key value matches expected. self.assert_cache_count(expected, pattern="*"): Assert number of matching keys. self.flush_cache(): Flush all keys. 2. AuthTestMixin Enables quick authentication mocking, letting you bypass credential verification: self.force_login(identity): Injects an identity into the TestClient session headers, making subsequent requests appear authenticated. self.authenticate_as(identity): thorough than force_login, registers the identity directly into the server's identity store. self.login_as_admin(id=None, **kw): Helper that builds an admin identity and authenticates it. self.login_as_user(id=None, **kw): Helper that builds a regular user identity and authenticates it. from aquilia.testing import AquiliaTestCase, AuthTestMixin class TestBilling(AuthTestMixin, AquiliaTestCase): enable_auth = True async def test_access_billing(self): # Build and authenticate user identity self.login_as_admin(id="user-123", email="admin@test.com") # Injected identity header is sent automatically resp = await self.client.get("/api/billing") self.assert_status(resp, 200) 3. MailTestMixin Captures outgoing emails sent via the mail subsystem, placing them in an outbox array instead of triggering SMTP: self.mail_outbox: Read-only list containing all sent CapturedMail objects. self.latest_mail: Returns the most recently sent CapturedMail object. self.assert_mail_count(outbox, expected): Asserts exact number of sent messages. self.assert_mail_to(outbox, address): Asserts email was sent to address. )
Go to Homepage