Python + pytest + Testcontainers 통합 테스트
import pytest
from testcontainers.postgres import PostgresContainer
from testcontainers.redis import RedisContainer
from sqlalchemy import create_engine, text
from sqlalchemy.orm import sessionmaker
import redis
# 실제 PostgreSQL 컨테이너를 사용한 통합 테스트
class TestUserRepository:
"""사용자 저장소 통합 테스트 - 실제 DB 연동"""
@pytest.fixture(scope="class")
def postgres_container(self):
"""테스트용 PostgreSQL 컨테이너 시작"""
with PostgresContainer("postgres:15-alpine") as postgres:
yield postgres
@pytest.fixture(scope="class")
def db_session(self, postgres_container):
"""데이터베이스 세션 생성"""
engine = create_engine(postgres_container.get_connection_url())
# 스키마 초기화
with engine.connect() as conn:
conn.execute(text("""
CREATE TABLE IF NOT EXISTS users (
id SERIAL PRIMARY KEY,
email VARCHAR(255) UNIQUE NOT NULL,
name VARCHAR(100) NOT NULL,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
)
"""))
conn.commit()
Session = sessionmaker(bind=engine)
return Session()
def test_create_and_retrieve_user(self, db_session):
"""사용자 생성 및 조회 통합 테스트"""
# Given: 새로운 사용자 정보
user_email = "test@example.com"
user_name = "테스트 사용자"
# When: 사용자 생성
db_session.execute(
text("INSERT INTO users (email, name) VALUES (:email, :name)"),
{"email": user_email, "name": user_name}
)
db_session.commit()
# Then: 생성된 사용자 조회 확인
result = db_session.execute(
text("SELECT * FROM users WHERE email = :email"),
{"email": user_email}
).fetchone()
assert result is not None
assert result.email == user_email
assert result.name == user_name
def test_duplicate_email_fails(self, db_session):
"""중복 이메일 등록 실패 테스트"""
from sqlalchemy.exc import IntegrityError
email = "duplicate@example.com"
# 첫 번째 사용자 등록
db_session.execute(
text("INSERT INTO users (email, name) VALUES (:email, :name)"),
{"email": email, "name": "First User"}
)
db_session.commit()
# 같은 이메일로 두 번째 등록 시도
with pytest.raises(IntegrityError):
db_session.execute(
text("INSERT INTO users (email, name) VALUES (:email, :name)"),
{"email": email, "name": "Second User"}
)
db_session.commit()
# Redis 캐시 통합 테스트
class TestCacheService:
"""캐시 서비스 통합 테스트 - 실제 Redis 연동"""
@pytest.fixture(scope="class")
def redis_container(self):
"""테스트용 Redis 컨테이너 시작"""
with RedisContainer("redis:7-alpine") as redis_cont:
yield redis_cont
@pytest.fixture
def redis_client(self, redis_container):
"""Redis 클라이언트 생성"""
client = redis.Redis(
host=redis_container.get_container_host_ip(),
port=redis_container.get_exposed_port(6379),
decode_responses=True
)
yield client
client.flushall() # 테스트 후 정리
def test_cache_set_and_get(self, redis_client):
"""캐시 저장 및 조회 테스트"""
# Given
cache_key = "user:123:profile"
cache_value = '{"id": 123, "name": "Test User"}'
# When
redis_client.setex(cache_key, 3600, cache_value)
# Then
result = redis_client.get(cache_key)
assert result == cache_value
def test_cache_expiration(self, redis_client):
"""캐시 만료 테스트"""
import time
# Given: 1초 TTL 캐시
cache_key = "temp:data"
redis_client.setex(cache_key, 1, "temporary")
# When: 1.5초 대기
time.sleep(1.5)
# Then: 캐시 만료됨
assert redis_client.get(cache_key) is None
# API 통합 테스트 예제 (FastAPI + httpx)
import httpx
from fastapi import FastAPI
from fastapi.testclient import TestClient
app = FastAPI()
@app.get("/api/health")
def health_check():
return {"status": "healthy"}
@app.post("/api/users")
def create_user(email: str, name: str):
return {"id": 1, "email": email, "name": name}
class TestAPIIntegration:
"""API 엔드포인트 통합 테스트"""
@pytest.fixture
def client(self):
return TestClient(app)
def test_health_endpoint(self, client):
"""헬스체크 API 테스트"""
response = client.get("/api/health")
assert response.status_code == 200
assert response.json()["status"] == "healthy"
def test_create_user_api(self, client):
"""사용자 생성 API 통합 테스트"""
response = client.post(
"/api/users",
params={"email": "new@example.com", "name": "New User"}
)
assert response.status_code == 200
data = response.json()
assert data["email"] == "new@example.com"
assert "id" in data