54 / 75 · 08 Infrastructure as Code Testing · Testcontainers for Infrastructure Testing← prev⊞ allnext →☰ Read as one page
10.4PostgreSQL with Testcontainers
# tests/integration/test_database.py
import pytest
from testcontainers.postgres import PostgresContainer
import psycopg2
@pytest.fixture(scope="module")
def postgres():
"""Start a PostgreSQL container with test schema."""
with PostgresContainer("postgres:16-alpine") as pg:
# Apply migrations
conn = psycopg2.connect(pg.get_connection_url())
cursor = conn.cursor()
cursor.execute("""
CREATE TABLE orders (
id SERIAL PRIMARY KEY,
order_id VARCHAR(50) UNIQUE NOT NULL,
customer_id VARCHAR(50) NOT NULL,
total DECIMAL(10,2) NOT NULL,
status VARCHAR(20) DEFAULT 'pending',
created_at TIMESTAMP DEFAULT NOW()
);
CREATE INDEX idx_orders_customer ON orders(customer_id);
CREATE INDEX idx_orders_status ON orders(status);
""")
conn.commit()
conn.close()
yield pg
@pytest.fixture
def db_conn(postgres):
"""Fresh connection for each test, with rollback for isolation."""
conn = psycopg2.connect(postgres.get_connection_url())
yield conn
conn.rollback()
conn.close()
def test_order_insertion(db_conn):
"""Test that orders can be inserted and retrieved."""
cursor = db_conn.cursor()
cursor.execute(
"INSERT INTO orders (order_id, customer_id, total) VALUES (%s, %s, %s)",
("ORD-001", "CUST-001", 99.99)
)
cursor.execute("SELECT total FROM orders WHERE order_id = %s", ("ORD-001",))
result = cursor.fetchone()
assert float(result[0]) == 99.99
def test_unique_order_id_constraint(db_conn):
"""Test that duplicate order IDs are rejected."""
cursor = db_conn.cursor()
cursor.execute(
"INSERT INTO orders (order_id, customer_id, total) VALUES (%s, %s, %s)",
("ORD-DUP", "CUST-001", 50.00)
)
with pytest.raises(psycopg2.errors.UniqueViolation):
cursor.execute(
"INSERT INTO orders (order_id, customer_id, total) VALUES (%s, %s, %s)",
("ORD-DUP", "CUST-002", 75.00)
)