what is it?
core functionality for spinning up docker containers in test environments
examples
testing fastapi and postgresql with pytest
following this guide
pytest==7.0.1
testcontainers==3.6.1
testcontainers==4.7.2conftest.py
launch the postgres container
import pytest
from fastapi.testclient import TestClient
from sqlalchemy import create_engine
from sqlalchemy.orm.session import Session
from testcontainers.core.waiting_utils import wait_for_logs
from testcontainers.postgres import PostgresContainer
from database.database import Base
from main import app
from services.v1_0_0.services import get_db
POSTGRES_IMAGE = "postgres:14"
POSTGRES_USER = "postgres"
POSTGRES_PASSWORD = "test_password"
POSTGRES_DATABASE = "test_database"
POSTGRES_CONTAINER_PORT = 5432
@pytest.fixture(scope="session")
def postgres_container() -> PostgresContainer:
"""
Setup postgres container
"""
postgres = PostgresContainer(
image=POSTGRES_IMAGE,
username=POSTGRES_USER,
password=POSTGRES_PASSWORD,
dbname=POSTGRES_DATABASE,
port=POSTGRES_CONTAINER_PORT,
)
with postgres:
wait_for_logs(
postgres,
r"UTC \[1\] LOG: database system is ready to accept connections",
10,
)
yield postgrescreate the engine that connects to our test database and initialize the db with the project’s tables
@pytest.fixture(scope="session")
def db(postgres_container: PostgresContainer):
url = postgres_container.get_connection_url()
engine = create_engine(url, echo=False)
Base.metadata.create_all(engine)
with Session(engine) as session:
yield sessionoverride the database dependency as explained in the fastapi documentation, using the injected db fixture that contains our test database
@pytest.fixture(scope="session")
def test_client(db):
app.dependency_overrides[get_db] = lambda: db
with TestClient(app) as c:
yield c
app.dependency_overrides.clear()test_<FOO>.py implementation
# using the fixtures configured above
def create_model_factory(db_session, field):
model_obj = Model(
model_field=field,
)
db_session.add(model_obj)
db_session.commit()
return model_obj
def test_foo_something(test_client, db)
field = "foo"
model_obj = create_model_factory(db, field)
headers = {"X-API-Version": "latest"}
resp = test_client.get("endpoint/whatever", headers=headers)
assert resp.status_code == 200
assert resp.json()["model_field"] == "foo"