Testing Infrastructure (ZTestClient)
Explore how ZCore orchestrates IoC sandboxes, savepoint database rollbacks, context mocking, and application lifespans.
Testing multi-layered enterprise applications is notoriously difficult because of shared mutable state (database connections, IoC container singletons, context variables). ZCore solves this with ZTestClient and a suite of composable ZTestFixture orchestrators.
1. The Fixture Composition Pipeline
When you open an async with ZTestClient(...) context, ZCore orchestrates a composite pipeline of isolated fixtures:
2. IoC Container Sandboxing (ContainerSandbox)
During integration testing, you may override dependencies or register mock services in the container. If test A mutates the container, test B might fail unexpectedly.
ContainerSandbox captures a complete snapshot of container._singletons, _scoped_definitions, and _factories during setUp(). During tearDown(), it restores the exact snapshot, guaranteeing 100% test isolation.
# Internal ContainerSandbox mechanics
async def setUp(self):
self._singletons = dict(container._singletons)
self._scoped = dict(container._scoped_definitions)
self._factories = dict(container._factories)
async def tearDown(self):
container._singletons = self._singletons
container._scoped_definitions = self._scoped
container._factories = self._factories3. Event Bus Isolation (EventDispatcherSandbox)
Domain-driven systems frequently attach ephemeral or test-specific event listeners. Without rigorous sandboxing, event subscriptions registered in one test case bleed into subsequent executions.
EventDispatcherSandbox captures an isolated dictionary snapshot of all registered event callbacks in EventDispatcher._subscribers. Upon test completion, it clears active subscribers and restores the pristine pre-test subscriber registry.
4. Lightning-Fast Database Rollbacks (DatabaseRollback)
Dropping tables, re-running database migrations, or truncating rows between test runs introduces severe latency.
The Savepoint Strategy:
DatabaseRollbackopens a dedicated connection on the async engine and begins a root transaction.- It initializes an
AsyncSessionusingjoin_transaction_mode="create_savepoint". - It overrides
db_manager.sessionand registers this session as the active request-scoped database dependency in the IoC container. - All test inserts, updates, and repository operations execute inside this savepoint.
- Upon test exit, it executes
transaction.rollback(). The database is instantly reverted to its pre-test state in milliseconds.
Schema Provisioning with setup_test_database
Before running tests that interact with transactional rollbacks, the database tables themselves must exist. ZCore provides setup_test_database() to synchronously create or refresh your schema:
# conftest.py
import pytest
from zcore.testing import setup_test_database
@pytest.fixture(scope="session", autouse=True)
def initialize_test_database():
# Safely creates all tables across existing event loops without async boilerplate
setup_test_database(drop_first=True)setup_test_database() automatically detects whether an event loop is already active in the current thread (e.g. under pytest-asyncio) and safely dispatches execution via a background worker thread, eliminating loop conflicts.
5. Context & User Dependency Overrides
Testing role-based endpoints in standard FastAPI requires manually manipulating app.dependency_overrides and generating dummy JWT tokens.
ZTestClient coordinates this automatically when user_id is passed:
UserContext: Injectsuser_id,scopes,is_superuser, andextra_contextdirectly into theZContextstore for the active async coroutine tree.DependencyOverride: Replacesget_current_user_stubandget_optional_user_stub(along with any targets inuser_dependency) with an authentic Pydantic user instance (whenuser_modelis specified) or an internalGenericMockUserimplementingUserProtocol.
# Seamless user mocking without JWT logic
async with ZTestClient(
app=app,
user_id=uuid.uuid4(),
user_model=AppUser,
scopes=["tasks:delete", "tasks:create"],
extra_context={"restricted_fields": ["tasks.view.salary"]}
) as client:
response = await client.delete(f"/tasks/{task_id}")
assert response.status_code == 2006. Application Lifespan Execution (AppLifespan)
ZTestClient automatically invokes FastAPI's router.lifespan_context. This guarantees that:
- Kernel plugins execute their
before_startupandon_startuphooks. - Global singletons (like
EventDispatcherand cache workers) are active during tests. - Upon test completion,
on_shutdownhooks execute in reverse order to release connection pools cleanly.
Caching & Real-Time Streaming
Deep dive into ZCore's distributed Redis cache with resilient in-memory LRU fallback and the cluster-wide PubSub streaming engine.
File Storage Security Architecture
Deep dive into how ZCore prevents path traversal, DoS through large files, arbitrary file deletions, and executable MIME spoofing.