How to test API endpoints
Write isolated, lightning-fast integration tests using ZTestClient with automatic database rollback and user mocking.
Testing layered architectures often requires mocking database connections, overriding dependencies, and cleaning up test records. ZCore's ZTestClient automates this entire pipeline using transactions with savepoints and container sandboxing.
1. Initializing Test Database Schemas
Before executing your test suite, use setup_test_database in your conftest.py to ensure test tables are created cleanly across synchronous or asynchronous test runners:
# conftest.py
import pytest
from zcore.testing import setup_test_database
@pytest.fixture(scope="session", autouse=True)
def initialize_test_db():
setup_test_database(drop_first=True)2. Basic Endpoint Test with Database Rollback
By default (use_db=True), any database mutations performed during the test are rolled back on exit:
# test_tasks.py
import pytest
from zcore.testing import ZTestClient
from main import app
@pytest.mark.asyncio
async def test_create_and_get_tasks():
async with ZTestClient(app=app) as client:
# 1. Create a task
create_res = await client.post("/tasks", json={"title": "Test Task"})
assert create_res.status_code == 201
task_id = create_res.json()["data"]["id"]
# 2. Fetch the created task
get_res = await client.get(f"/tasks/{task_id}")
assert get_res.status_code == 200
assert get_res.json()["data"]["title"] == "Test Task"
# The database transaction is automatically rolled back here!3. Testing Authenticated Routes & Permissions
Pass user_id, scopes, and is_superuser to simulate authenticated requests without generating real JWT tokens:
import uuid
@pytest.mark.asyncio
async def test_admin_delete_operation():
admin_id = uuid.uuid4()
async with ZTestClient(
app=app,
user_id=admin_id,
scopes=["tasks:delete", "admin:all"],
is_superuser=True,
extra_context={"restricted_fields": []}
) as client:
response = await client.delete(f"/tasks/{uuid.uuid4()}")
# Validates that HasScopes dependency allows execution
assert response.status_code in [200, 404]4. Class-Based Testing with BaseZTest
For test suites sharing identical user attributes and configuration, inherit from BaseZTest:
from zcore.testing import BaseZTest
from main import app
class TestTaskSuite(BaseZTest):
app = app
scopes = ["tasks:view", "tasks:create"]
extra_user_attrs = {"department": "Engineering"}
@pytest.mark.asyncio
async def test_suite_operation(self):
async with self.run() as client:
res = await client.get("/tasks")
assert res.status_code == 200Behind the Scenes Orchestration:
- Database Savepoints: Connects to the engine, opens a transaction with
create_savepoint, and binds it todb_manager.session. It rolls back instantly on exit without dropping or truncating tables. - IoC Container Sandboxing: Backs up singletons, scoped instances, and factories before each test, completely preventing test pollution.
- Event Dispatcher Sandboxing: Automatically snapshots and restores registered event subscribers so test-specific event listeners never leak across test cases.
- Application Lifespan: Automatically executes plugin
before_startup,on_startup, andon_shutdownhooks.