Skip to content

Database and broker test contexts

Install the contexts you need:

pip install 'fastsprout[testing-data,testing-events]'

testing.core provides typed pytest phases. testing.data.TestSQLite(*entities) creates an isolated in-memory database and exposes a typed TestSQLDatabase with engine, session, and signpost.

The context creates the selected tables, binds the entities, and removes the tables on exit. Previous entity bindings are restored, including inherited bindings. Isolation does not depend on rolling back application transactions: a test can commit through DataR and read the result in another transaction. Use dedicated test databases; existing selected tables are rejected.

Pass signpost=CRM for an entity family declared with that source type. The resulting database.signpost retains the concrete CRM type.

TestSQLite accepts a caller-owned engine and a sessions factory. Caller-owned engines remain open after exit; selected tables are still removed. Default session factories use expire_on_commit=False.

See the SQLite example.

PostgreSQL

TestPostgreSQL(*entities) starts a disposable PostgreSQL container and reuses the SQL context's table, session, and binding lifecycle. Pass a configured PostgresContainer, an engine factory, or a session factory for native settings. The default image is the public postgres:18-alpine image.

Container lifecycle tests require a local Docker API. Select them with the docker marker:

uv run --extra testing-data \
  pytest src/tests/fastsprout/testing/data -m docker --no-cov

Use pytest -m "not integration" to run without external services. Unavailable Docker infrastructure fails the selected container lifecycle tests.

CI runs database and broker integration tests against ordinary GitLab service containers. It does not start Docker-in-Docker, mount a Docker socket, or require privileged jobs. PostgreSQL, MySQL, MariaDB, Oracle, Kafka, NATS, and RabbitMQ each run in their own job on Python 3.12–3.14. Service jobs are serialized to limit cluster resource use. The integration jobs, changelog, and public API type coverage run in merge request pipelines as well as branch pipelines.

For example, with an existing PostgreSQL service:

TEST_SERVICE_URL=postgresql+asyncpg://tester:testing@localhost:5432/heroes \
  uv run --extra testing-data pytest \
  src/tests/fastsprout/testing/data/test_services.py -m integration --no-cov

Broker service tests use TEST_SERVICE_URL and the corresponding class in src/tests/fastsprout/events/backend/test_services.py: TestKafkaService, TestNATSService, or TestRabbitMQService. Each test uses its own route to isolate messages on a shared service. Container factory configuration, startup failure, cancellation, and cleanup contracts remain covered by unit tests; the docker-marked tests additionally exercise a real local Docker runtime.

See the PostgreSQL example.

MySQL and MariaDB

TestMySQL(*entities) and TestMariaDB(*entities) reuse the SQL context's table, session, and binding lifecycle with the asyncmy driver. Pass a configured MySqlContainer, an engine factory, or a session factory for native settings. The default images are public mysql:8.4 and mariadb:11.4; override image= to select another version or registry.

See the MySQL and MariaDB example.

Oracle

TestOracle(*entities) runs Oracle Free with asynchronous oracledb connections. Pass a configured OracleDbContainer to select credentials and service, or engine/session factories for SQLAlchemy settings. The public default image is gvenzl/oracle-free:23-slim-faststart; override image= for another version or registry.

Oracle string columns need an explicit length, for example name: Field[str] = SQLField(max_length=100). The context preserves the model's column definitions.

See the Oracle example.

RabbitMQ

Install fastsprout[testing-events] for TestRabbitMQ(). It starts a disposable RabbitMQ container and yields a RabbitMQBroker implementing Brokerable. Pass it to EventBus and register subscribers before publishing. Dispatch connects the adapter and starts its consumers. A receiver which does not publish calls await adapter.start() after registering its handlers.

Pass a configured RabbitMqContainer, a native broker factory, or an adapter factory for native configuration. The public default image is rabbitmq:4.1.0-alpine; image= selects another version or registry. The context stops the adapter and removes its container on exit.

See the RabbitMQ example.

Local event brokers

Local brokers use the same typed pytest phases without a Docker server. The @given fixture creates the broker and closes it after @then. The @when step publishes the event and waits for its subscribers.

NATS

TestNATS() starts a disposable NATS container and yields a NATSBroker. Pass a configured NatsContainer, native broker factory, or adapter factory for native settings. Its public default image is nats:2-alpine. The default Core NATS server provides live delivery without durable replay.

See the NATS example.

Kafka

TestKafka() starts a disposable Kafka container and yields a KafkaBroker. Pass a configured KafkaContainer, native broker factory, or adapter factory for native settings. Its public default image is confluentinc/cp-kafka:7.6.0. Partitions, consumer groups, offsets, and commits remain available through the native Kafka API.

See the Kafka example.