본 콘텐츠의 이미지 및 내용은 AI로 생성되었습니다.
본 콘텐츠의 이미지 및 내용을 무단으로 복제, 배포, 수정하여 사용할 경우 저작권법에 의해 법적 제재를 받을 수 있습니다.
이미지 로딩 중...
AI Generated
2025. 11. 30. · 19 Views
자동화된 모델 테스트 및 검증 완벽 가이드
머신러닝 모델을 프로덕션에 배포하기 전 반드시 거쳐야 하는 자동화 테스트와 검증 기법을 다룹니다. 테스트 주도 개발 관점에서 모델의 품질을 보장하는 실무 노하우를 배워봅니다.
목차
- 단위_테스트로_모델_함수_검증하기
- 모델_성능_메트릭_자동_검증
- 데이터_검증으로_품질_보장하기
- 모델_입출력_스키마_테스트
- 예측_일관성_테스트
- 모델_직렬화_역직렬화_테스트
- 회귀_테스트로_성능_저하_감지
- 엣지_케이스_테스트
- 통합_테스트로_전체_파이프라인_검증
- CI_CD_파이프라인에_테스트_통합하기
1. 단위 테스트로 모델 함수 검증하기
김개발 씨는 첫 번째 머신러닝 프로젝트를 마무리하고 뿌듯한 마음으로 퇴근 준비를 하고 있었습니다. 그때 박시니어 씨가 다가와 물었습니다.
"테스트 코드는 작성했어요?" 김개발 씨는 멈칫했습니다. 모델 정확도는 확인했지만, 테스트 코드라니?
단위 테스트는 코드의 가장 작은 단위인 함수나 메서드가 올바르게 동작하는지 검증하는 것입니다. 마치 자동차를 조립하기 전에 각 부품이 정상인지 하나씩 점검하는 것과 같습니다.
머신러닝에서는 데이터 전처리 함수, 피처 엔지니어링 로직, 예측 함수 등을 개별적으로 테스트합니다.
다음 코드를 살펴봅시다.
import pytest
import numpy as np
from sklearn.preprocessing import StandardScaler
# 전처리 함수 정의
def preprocess_features(data):
"""입력 데이터를 정규화하고 결측치를 처리합니다"""
data = np.nan_to_num(data, nan=0.0)
scaler = StandardScaler()
return scaler.fit_transform(data.reshape(-1, 1)).flatten()
# 단위 테스트 작성
def test_preprocess_handles_nan():
"""결측치가 0으로 대체되는지 확인합니다"""
input_data = np.array([1.0, np.nan, 3.0])
result = preprocess_features(input_data)
assert not np.isnan(result).any()
def test_preprocess_output_shape():
"""출력 형태가 입력과 동일한지 확인합니다"""
input_data = np.array([1.0, 2.0, 3.0])
result = preprocess_features(input_data)
assert result.shape == input_data.shape
김개발 씨는 입사 6개월 차 주니어 데이터 사이언티스트입니다. 오늘 처음으로 혼자서 분류 모델을 만들어 선배에게 리뷰를 요청했습니다.
모델 정확도도 92%나 나왔고, 나름 깔끔하게 코드를 정리했다고 생각했습니다. 하지만 박시니어 씨의 첫 마디는 예상과 달랐습니다.
"모델 성능은 좋네요. 그런데 테스트 코드가 없으면 이 코드를 어떻게 믿고 프로덕션에 올려요?" 김개발 씨는 당황했습니다.
머신러닝에도 테스트 코드가 필요한 걸까요? 그렇다면 단위 테스트란 정확히 무엇일까요?
쉽게 비유하자면, 단위 테스트는 마치 자동차 공장에서 조립 전에 각 부품을 검사하는 것과 같습니다. 엔진 따로, 바퀴 따로, 핸들 따로 검사한 뒤에야 조립을 시작합니다.
불량 부품이 하나라도 섞이면 완성된 차에 문제가 생기기 때문입니다. 코드도 마찬가지입니다.
작은 함수 하나가 잘못되면 전체 파이프라인이 무너질 수 있습니다. 단위 테스트가 없던 시절에는 어땠을까요?
개발자들은 코드를 수정할 때마다 전체 프로그램을 실행해서 눈으로 결과를 확인해야 했습니다. 작은 수정이 어디에 영향을 미치는지 파악하기 어려웠습니다.
더 큰 문제는 버그가 발생했을 때 원인을 찾기가 매우 힘들었다는 점입니다. 수십 개의 함수 중 어디서 문제가 생겼는지 하나씩 디버깅해야 했습니다.
바로 이런 문제를 해결하기 위해 단위 테스트가 등장했습니다. 단위 테스트를 작성하면 코드를 수정할 때마다 자동으로 모든 함수가 정상 동작하는지 확인할 수 있습니다.
또한 버그가 발생했을 때 어느 부분에서 문제가 생겼는지 즉시 알 수 있습니다. 무엇보다 새로운 팀원이 코드를 이해하는 데 테스트 코드가 훌륭한 문서 역할을 합니다.
위의 코드를 한 줄씩 살펴보겠습니다. 먼저 preprocess_features 함수를 보면 결측치를 0으로 대체하고 데이터를 정규화합니다.
이 함수가 제대로 동작하는지 어떻게 확신할 수 있을까요? 바로 테스트 함수를 통해서입니다.
test_preprocess_handles_nan 함수는 결측치가 포함된 데이터를 넣었을 때 결과에 NaN이 없는지 확인합니다. test_preprocess_output_shape 함수는 출력 형태가 입력과 동일한지 검증합니다.
실제 현업에서는 어떻게 활용할까요? 예를 들어 추천 시스템을 개발한다고 가정해봅시다.
사용자 행동 데이터를 전처리하는 함수가 있다면, 이 함수에 대한 단위 테스트를 작성합니다. 나중에 전처리 로직을 수정하더라도 테스트가 자동으로 기존 기능이 깨지지 않았는지 확인해줍니다.
이것을 회귀 테스트라고 부릅니다. 하지만 주의할 점도 있습니다.
초보 개발자들이 흔히 하는 실수 중 하나는 테스트를 너무 세세하게 작성하는 것입니다. 모든 줄에 테스트를 작성하면 코드 수정 시 테스트도 함께 수정해야 하는 부담이 커집니다.
따라서 핵심 로직과 경계 조건에 집중해서 테스트를 작성해야 합니다. 다시 김개발 씨의 이야기로 돌아가 봅시다.
박시니어 씨의 설명을 들은 김개발 씨는 고개를 끄덕였습니다. "아, 테스트가 있으면 안심하고 코드를 수정할 수 있겠네요!" 단위 테스트를 제대로 작성하면 코드의 신뢰성이 높아지고, 리팩토링도 두렵지 않게 됩니다.
여러분도 오늘부터 pytest로 첫 테스트를 작성해 보세요.
실전 팁
💡 - pytest 명령어 하나로 모든 테스트를 실행할 수 있습니다: pytest -v
- 테스트 함수 이름은 test_로 시작해야 pytest가 인식합니다
- 경계 조건(빈 배열, 큰 숫자, 음수 등)을 반드시 테스트하세요
2. 모델 성능 메트릭 자동 검증
김개발 씨가 만든 모델의 정확도가 어제는 92%였는데 오늘 다시 훈련하니 87%가 나왔습니다. 데이터는 똑같은데 왜 이런 차이가 생기는 걸까요?
박시니어 씨는 웃으며 말했습니다. "그래서 성능 메트릭 검증을 자동화해야 해요."
성능 메트릭 자동 검증은 모델이 일정 수준 이상의 성능을 유지하는지 자동으로 확인하는 것입니다. 마치 품질 검사관이 공장에서 생산된 제품이 기준을 충족하는지 매번 검사하는 것과 같습니다.
정확도, F1 점수, AUC 등 다양한 지표를 기준값과 비교하여 모델 품질을 보장합니다.
다음 코드를 살펴봅시다.
import pytest
from sklearn.datasets import make_classification
from sklearn.model_selection import train_test_split
from sklearn.ensemble import RandomForestClassifier
from sklearn.metrics import accuracy_score, f1_score
# 모델 훈련 함수
def train_model(X_train, y_train):
model = RandomForestClassifier(n_estimators=100, random_state=42)
model.fit(X_train, y_train)
return model
# 성능 메트릭 검증 테스트
def test_model_accuracy_threshold():
"""모델 정확도가 최소 기준(85%)을 충족하는지 검증합니다"""
X, y = make_classification(n_samples=1000, random_state=42)
X_train, X_test, y_train, y_test = train_test_split(X, y, random_state=42)
model = train_model(X_train, y_train)
predictions = model.predict(X_test)
accuracy = accuracy_score(y_test, predictions)
assert accuracy >= 0.85, f"정확도 {accuracy:.2f}가 기준 0.85 미만입니다"
def test_model_f1_score():
"""F1 점수가 최소 기준(0.80)을 충족하는지 검증합니다"""
X, y = make_classification(n_samples=1000, random_state=42)
X_train, X_test, y_train, y_test = train_test_split(X, y, random_state=42)
model = train_model(X_train, y_train)
predictions = model.predict(X_test)
f1 = f1_score(y_test, predictions)
assert f1 >= 0.80, f"F1 점수 {f1:.2f}가 기준 0.80 미만입니다"
김개발 씨는 며칠 전 만든 모델을 다시 훈련시켜 보았습니다. 분명 똑같은 코드인데 정확도가 5%나 떨어졌습니다.
무엇이 잘못된 걸까요? 박시니어 씨가 원인을 설명해주었습니다.
"랜덤 시드를 고정하지 않으면 매번 다른 결과가 나올 수 있어요. 그리고 더 중요한 건, 어떤 결과가 나오든 기준 이하면 자동으로 알려주는 시스템이 필요해요." 그렇다면 성능 메트릭 자동 검증이란 무엇일까요?
쉽게 비유하자면, 이것은 마치 공장의 품질 검사 라인과 같습니다. 제품이 컨베이어 벨트를 타고 지나갈 때 센서가 자동으로 불량품을 걸러냅니다.
크기가 기준보다 작거나, 무게가 기준보다 가벼우면 자동으로 탈락시킵니다. 머신러닝 모델도 마찬가지입니다.
정확도나 F1 점수가 기준 이하면 배포되어서는 안 됩니다. 성능 검증이 자동화되지 않으면 어떤 문제가 생길까요?
개발자가 매번 직접 성능 지표를 확인해야 합니다. 바쁜 일정 속에서 실수로 낮은 성능의 모델이 프로덕션에 배포될 수 있습니다.
더 심각한 문제는 성능 저하를 뒤늦게 발견하는 경우입니다. 이미 서비스에 영향을 준 후에야 문제를 알게 됩니다.
바로 이런 문제를 해결하기 위해 자동화된 성능 검증이 필요합니다. CI/CD 파이프라인에 성능 테스트를 추가하면 모델이 기준 성능을 충족하지 못할 때 자동으로 배포가 중단됩니다.
또한 시간이 지나면서 모델 성능이 저하되는 모델 드리프트도 조기에 발견할 수 있습니다. 위의 코드를 살펴보겠습니다.
test_model_accuracy_threshold 함수는 모델의 정확도가 85% 이상인지 확인합니다. 만약 기준에 미달하면 테스트가 실패하고 에러 메시지를 출력합니다.
test_model_f1_score 함수는 불균형 데이터에서 더 의미 있는 지표인 F1 점수를 검증합니다. 두 테스트 모두 random_state를 고정하여 재현 가능한 결과를 보장합니다.
실제 현업에서는 더 다양한 지표를 검증합니다. 금융 사기 탐지 모델이라면 **재현율(Recall)**이 매우 중요합니다.
사기를 놓치는 것보다 정상 거래를 사기로 잘못 판단하는 것이 차라리 낫기 때문입니다. 반면 스팸 필터라면 **정밀도(Precision)**가 중요할 수 있습니다.
중요한 이메일을 스팸으로 분류하면 큰 문제가 되기 때문입니다. 주의해야 할 점이 있습니다.
기준값을 너무 높게 설정하면 사소한 변동에도 테스트가 실패합니다. 반대로 너무 낮게 설정하면 검증의 의미가 없어집니다.
따라서 과거 모델들의 성능을 참고하여 합리적인 기준을 설정해야 합니다. 또한 여러 번의 교차 검증 결과의 평균을 사용하면 더 안정적인 검증이 가능합니다.
박시니어 씨가 덧붙였습니다. "성능 기준은 비즈니스 요구사항과 맞춰야 해요.
기술적으로 가능한 최고 성능이 아니라, 서비스에 필요한 최소 성능을 기준으로 삼으세요." 김개발 씨는 이제 모든 모델에 성능 검증 테스트를 추가하기로 마음먹었습니다.
실전 팁
💡 - random_state를 고정하여 재현 가능한 테스트를 작성하세요
- 비즈니스 요구사항에 맞는 적절한 임계값을 설정하세요
- 정확도 외에도 Precision, Recall, AUC 등 다양한 지표를 검증하세요
3. 데이터 검증으로 품질 보장하기
어느 날 김개발 씨의 모델이 갑자기 이상한 예측을 내놓기 시작했습니다. 원인을 찾아보니 입력 데이터에 문제가 있었습니다.
나이 컬럼에 -5라는 값이 들어있었던 것입니다. 박시니어 씨가 말했습니다.
"데이터 검증을 했다면 미리 잡을 수 있었을 텐데요."
데이터 검증은 모델에 입력되는 데이터가 예상한 형식과 범위를 따르는지 확인하는 것입니다. 마치 식당에서 재료를 받을 때 신선도와 품질을 확인하는 것과 같습니다.
아무리 훌륭한 요리사라도 상한 재료로는 좋은 음식을 만들 수 없듯이, 아무리 좋은 모델이라도 잘못된 데이터가 들어오면 엉뚱한 결과를 내놓습니다.
다음 코드를 살펴봅시다.
import pytest
import pandas as pd
import numpy as np
from typing import Dict, Any
# 데이터 검증 함수
def validate_input_data(df: pd.DataFrame) -> Dict[str, Any]:
"""입력 데이터의 품질을 검증합니다"""
errors = []
# 필수 컬럼 확인
required_columns = ['age', 'income', 'credit_score']
missing = set(required_columns) - set(df.columns)
if missing:
errors.append(f"필수 컬럼 누락: {missing}")
# 값 범위 검증
if (df['age'] < 0).any() or (df['age'] > 120).any():
errors.append("나이가 유효 범위(0-120)를 벗어났습니다")
# 결측치 비율 확인
null_ratio = df.isnull().sum() / len(df)
high_null_cols = null_ratio[null_ratio > 0.3].index.tolist()
if high_null_cols:
errors.append(f"결측치 30% 초과 컬럼: {high_null_cols}")
return {'valid': len(errors) == 0, 'errors': errors}
# 데이터 검증 테스트
def test_valid_data_passes():
"""정상 데이터가 검증을 통과하는지 확인합니다"""
df = pd.DataFrame({
'age': [25, 30, 35],
'income': [50000, 60000, 70000],
'credit_score': [700, 750, 800]
})
result = validate_input_data(df)
assert result['valid'] is True
def test_invalid_age_detected():
"""비정상 나이 값이 감지되는지 확인합니다"""
df = pd.DataFrame({
'age': [-5, 30, 150],
'income': [50000, 60000, 70000],
'credit_score': [700, 750, 800]
})
result = validate_input_data(df)
assert result['valid'] is False
assert any('나이' in err for err in result['errors'])
김개발 씨는 고객 이탈 예측 모델을 운영하고 있었습니다. 어느 날 모니터링 대시보드를 보니 모델의 예측 결과가 이상했습니다.
대부분의 고객이 이탈할 것이라고 예측하고 있었습니다. 원인을 추적해보니 입력 데이터에 문제가 있었습니다.
데이터 파이프라인에서 오류가 발생해 나이 컬럼에 음수 값이 들어갔고, 소득 컬럼에는 문자열이 섞여 있었습니다. 박시니어 씨가 말했습니다.
"데이터는 모델의 연료예요. 연료가 오염되면 엔진이 고장 나듯이, 데이터가 잘못되면 모델도 엉뚱한 결과를 내놓아요." 그렇다면 데이터 검증이란 무엇일까요?
쉽게 비유하자면, 데이터 검증은 마치 식당의 식재료 검수와 같습니다. 좋은 식당은 재료가 도착하면 먼저 신선도를 확인합니다.
상한 재료는 즉시 반품하고, 기준에 맞는 재료만 주방으로 보냅니다. 데이터도 마찬가지입니다.
모델에 입력되기 전에 품질을 확인해야 합니다. 데이터 검증이 없으면 어떤 일이 벌어질까요?
잘못된 데이터가 모델에 입력되어 엉뚱한 예측을 내놓습니다. 문제는 이런 오류가 조용히 발생한다는 점입니다.
모델은 에러를 내지 않고 그냥 잘못된 결과를 출력합니다. 나중에 비즈니스 지표가 이상해지고 나서야 문제를 발견하게 됩니다.
위의 코드를 살펴보겠습니다. validate_input_data 함수는 세 가지를 검증합니다.
첫째, 필수 컬럼이 모두 존재하는지 확인합니다. 둘째, 나이 값이 0에서 120 사이인지 확인합니다.
셋째, 결측치가 30%를 넘는 컬럼이 있는지 확인합니다. 검증 결과는 딕셔너리로 반환되어 어떤 문제가 있는지 명확하게 알려줍니다.
테스트 코드도 중요합니다. test_valid_data_passes는 정상 데이터가 검증을 통과하는지 확인합니다.
test_invalid_age_detected는 비정상적인 나이 값이 제대로 감지되는지 확인합니다. 이렇게 검증 함수 자체를 테스트하면 검증 로직의 신뢰성도 보장됩니다.
실제 현업에서는 더 다양한 검증을 수행합니다. 데이터 타입 검증, 고유값 개수 확인, 분포 변화 감지 등이 있습니다.
Great Expectations나 Pandera 같은 라이브러리를 사용하면 더 체계적인 데이터 검증이 가능합니다. 주의할 점도 있습니다.
검증 규칙을 너무 엄격하게 설정하면 정상 데이터도 거부될 수 있습니다. 반대로 너무 느슨하면 검증의 의미가 없어집니다.
실제 데이터의 분포를 충분히 분석한 후 합리적인 기준을 설정해야 합니다. 김개발 씨는 이제 모든 데이터 파이프라인 앞에 검증 단계를 추가하기로 했습니다.
"데이터 품질이 모델 품질의 시작이군요."
실전 팁
💡 - 검증 실패 시 명확한 에러 메시지를 제공하세요
- 검증 규칙은 실제 데이터 분포를 분석한 후 설정하세요
- Great Expectations 라이브러리로 더 체계적인 검증을 구현할 수 있습니다
4. 모델 입출력 스키마 테스트
김개발 씨가 만든 모델을 다른 팀에서 사용하려고 했습니다. 그런데 계속 에러가 발생했습니다.
알고 보니 입력 데이터의 컬럼 순서가 달랐던 것입니다. 박시니어 씨가 말했습니다.
"스키마 테스트를 해뒀다면 이런 문제를 바로 잡을 수 있었을 거예요."
모델 입출력 스키마 테스트는 모델이 예상하는 입력 형식과 출력 형식이 정확히 지켜지는지 검증하는 것입니다. 마치 전자제품의 플러그와 콘센트가 맞아야 작동하는 것처럼, 모델도 정해진 형식의 데이터를 받아야 올바르게 동작합니다.
이 테스트는 특히 여러 시스템이 연동될 때 중요합니다.
다음 코드를 살펴봅시다.
import pytest
import numpy as np
from pydantic import BaseModel, field_validator
from typing import List
# 입력 스키마 정의
class ModelInput(BaseModel):
features: List[float]
@field_validator('features')
@classmethod
def check_feature_count(cls, v):
if len(v) != 10:
raise ValueError('features는 정확히 10개여야 합니다')
return v
# 출력 스키마 정의
class ModelOutput(BaseModel):
prediction: int
probability: float
@field_validator('prediction')
@classmethod
def check_prediction_range(cls, v):
if v not in [0, 1]:
raise ValueError('prediction은 0 또는 1이어야 합니다')
return v
@field_validator('probability')
@classmethod
def check_probability_range(cls, v):
if not 0 <= v <= 1:
raise ValueError('probability는 0과 1 사이여야 합니다')
return v
# 스키마 테스트
def test_valid_input_schema():
"""올바른 입력이 스키마를 통과하는지 확인합니다"""
valid_input = ModelInput(features=[0.1] * 10)
assert len(valid_input.features) == 10
def test_invalid_feature_count_rejected():
"""잘못된 피처 개수가 거부되는지 확인합니다"""
with pytest.raises(ValueError):
ModelInput(features=[0.1] * 5)
def test_valid_output_schema():
"""올바른 출력이 스키마를 통과하는지 확인합니다"""
valid_output = ModelOutput(prediction=1, probability=0.85)
assert valid_output.prediction in [0, 1]
김개발 씨의 모델이 드디어 다른 팀과 연동될 날이 왔습니다. 백엔드 팀에서 API를 통해 모델을 호출하기로 했습니다.
하지만 연동 첫날부터 문제가 터졌습니다. 백엔드 개발자가 전화를 걸어왔습니다.
"김개발 씨, 모델이 자꾸 에러를 내요. 데이터는 제대로 보내고 있는데요." 확인해보니 입력 데이터의 형식이 달랐습니다.
모델은 10개의 피처를 리스트로 받아야 하는데, 백엔드에서는 딕셔너리로 보내고 있었습니다. 또한 피처 순서도 달랐습니다.
박시니어 씨가 조언했습니다. "입출력 스키마를 명확하게 정의하고 테스트해두면 이런 문제를 방지할 수 있어요." 모델 입출력 스키마란 무엇일까요?
쉽게 비유하자면, 스키마는 마치 계약서와 같습니다. "이 모델을 사용하려면 이런 형식의 데이터를 보내야 하고, 이런 형식의 결과를 받게 됩니다"라고 명확하게 약속하는 것입니다.
계약서가 없으면 서로 다른 이해로 인해 분쟁이 생기듯이, 스키마가 없으면 시스템 간 연동에서 문제가 발생합니다. 위의 코드를 살펴보겠습니다.
ModelInput 클래스는 입력 스키마를 정의합니다. features는 반드시 float 타입의 리스트여야 하며, 정확히 10개의 값이 있어야 합니다.
ModelOutput 클래스는 출력 스키마를 정의합니다. prediction은 0 또는 1이어야 하고, probability는 0과 1 사이의 값이어야 합니다.
Pydantic이라는 라이브러리를 사용하면 이런 검증을 자동화할 수 있습니다. 스키마에 맞지 않는 데이터가 들어오면 즉시 에러를 발생시킵니다.
테스트 코드를 보겠습니다. test_valid_input_schema는 올바른 입력이 통과하는지 확인합니다.
test_invalid_feature_count_rejected는 피처 개수가 틀렸을 때 에러가 발생하는지 확인합니다. pytest.raises를 사용하면 특정 에러가 발생하는지 테스트할 수 있습니다.
실제 현업에서는 어떻게 활용할까요? REST API로 모델을 서빙할 때 FastAPI와 Pydantic을 함께 사용하면 요청과 응답의 스키마를 자동으로 검증하고 문서화까지 해줍니다.
또한 데이터 파이프라인에서 모델 앞단에 스키마 검증을 추가하면 잘못된 데이터가 모델에 도달하는 것을 방지할 수 있습니다. 주의할 점도 있습니다.
스키마는 너무 엄격해도, 너무 느슨해도 문제입니다. 예를 들어 숫자 필드에 정수만 허용하면 소수점이 있는 데이터가 거부됩니다.
실제 사용 패턴을 고려하여 적절한 유연성을 두어야 합니다. 김개발 씨는 이제 모든 모델에 스키마를 정의하고 문서화하기로 했습니다.
"스키마가 있으면 다른 팀과 소통도 훨씬 쉬워지겠네요."
실전 팁
💡 - Pydantic을 사용하면 타입 검증과 데이터 변환을 자동화할 수 있습니다
- API 문서에 스키마를 명확하게 포함하세요
- 스키마 변경 시에는 버전 관리를 통해 하위 호환성을 유지하세요
5. 예측 일관성 테스트
김개발 씨가 모델을 새로 배포한 후 이상한 리포트를 받았습니다. 같은 입력을 넣었는데 어제와 오늘의 예측 결과가 다르다는 것입니다.
버그일까요? 박시니어 씨가 차분하게 말했습니다.
"예측 일관성 테스트를 해봐야겠네요."
예측 일관성 테스트는 동일한 입력에 대해 모델이 항상 같은 결과를 내놓는지 확인하는 것입니다. 마치 자판기에 같은 금액을 넣으면 같은 음료가 나와야 하는 것처럼, 모델도 같은 입력에는 같은 출력을 내야 합니다.
이 테스트는 모델의 결정론적 동작을 보장합니다.
다음 코드를 살펴봅시다.
import pytest
import numpy as np
import joblib
from sklearn.ensemble import RandomForestClassifier
# 모델 저장 및 로드 함수
def save_model(model, path):
joblib.dump(model, path)
def load_model(path):
return joblib.load(path)
# 예측 일관성 테스트
class TestPredictionConsistency:
@pytest.fixture
def trained_model(self, tmp_path):
"""테스트용 모델을 생성하고 저장합니다"""
np.random.seed(42)
X = np.random.randn(100, 5)
y = np.random.randint(0, 2, 100)
model = RandomForestClassifier(n_estimators=10, random_state=42)
model.fit(X, y)
model_path = tmp_path / "model.pkl"
save_model(model, model_path)
return model_path
def test_same_input_same_output(self, trained_model):
"""같은 입력에 대해 같은 출력이 나오는지 확인합니다"""
model = load_model(trained_model)
test_input = np.array([[0.5, -0.3, 0.8, -0.1, 0.2]])
predictions = [model.predict(test_input)[0] for _ in range(10)]
assert len(set(predictions)) == 1, "예측이 일관되지 않습니다"
def test_probability_consistency(self, trained_model):
"""확률 예측도 일관되는지 확인합니다"""
model = load_model(trained_model)
test_input = np.array([[0.5, -0.3, 0.8, -0.1, 0.2]])
probabilities = [model.predict_proba(test_input)[0] for _ in range(10)]
for prob in probabilities[1:]:
np.testing.assert_array_almost_equal(probabilities[0], prob)
김개발 씨는 고객 서비스팀으로부터 황당한 리포트를 받았습니다. "같은 고객 정보를 넣었는데 어제는 '우량 고객'이라고 했다가 오늘은 '주의 고객'이라고 해요.
어떤 게 맞는 거예요?" 김개발 씨는 당황했습니다. 분명히 코드를 바꾼 적이 없는데 왜 결과가 다른 걸까요?
박시니어 씨가 함께 원인을 분석해주었습니다. "모델 자체에 랜덤 요소가 있거나, 모델 로딩 과정에서 문제가 생겼을 수 있어요.
예측 일관성 테스트를 해봐야겠네요." 예측 일관성이란 무엇일까요? 쉽게 비유하자면, 이것은 마치 계산기의 신뢰성과 같습니다.
계산기에 2+2를 입력하면 항상 4가 나와야 합니다. 어떤 날은 4가 나오고 어떤 날은 5가 나오는 계산기는 아무도 믿지 않을 것입니다.
머신러닝 모델도 마찬가지입니다. 같은 입력에는 같은 출력이 나와야 사용자가 신뢰할 수 있습니다.
예측이 일관되지 않으면 어떤 문제가 생길까요? 사용자가 모델을 신뢰하지 못하게 됩니다.
또한 디버깅이 매우 어려워집니다. 버그가 있어도 재현이 안 되기 때문입니다.
더 심각한 문제는 A/B 테스트나 모델 성능 비교가 무의미해진다는 것입니다. 위의 코드를 살펴보겠습니다.
test_same_input_same_output은 같은 입력으로 10번 예측을 수행한 후 모든 결과가 동일한지 확인합니다. set으로 변환했을 때 길이가 1이면 모든 예측이 같다는 의미입니다.
test_probability_consistency는 확률 예측도 동일한지 확인합니다. np.testing.assert_array_almost_equal을 사용하면 부동소수점 오차를 고려하여 비교할 수 있습니다.
pytest의 fixture도 주목할 만합니다. @pytest.fixture는 테스트에 필요한 자원을 미리 준비하는 기능입니다.
여기서는 테스트용 모델을 훈련하고 저장하는 과정을 fixture로 분리했습니다. tmp_path는 pytest가 제공하는 임시 디렉토리입니다.
예측이 일관되지 않는 원인은 무엇일까요? 여러 가지 원인이 있습니다.
모델 내부에 랜덤 요소가 있거나, GPU 연산의 비결정성, 또는 멀티스레딩 환경에서의 경쟁 조건 등이 있습니다. 대부분의 경우 random_state를 고정하거나 numpy.random.seed를 설정하면 해결됩니다.
김개발 씨는 문제의 원인을 찾았습니다. 모델에서 dropout 레이어가 추론 시에도 활성화되어 있었던 것입니다.
추론 모드로 전환하니 문제가 해결되었습니다.
실전 팁
💡 - 모델 훈련과 추론 시 모든 랜덤 시드를 고정하세요
- 딥러닝 모델은 추론 시 반드시 eval() 모드로 전환하세요
- GPU 연산의 결정론적 동작을 위해 프레임워크별 설정을 확인하세요
6. 모델 직렬화 역직렬화 테스트
김개발 씨가 열심히 훈련한 모델을 저장하고 다음 날 로드했더니 에러가 발생했습니다. 어제까지 잘 되던 모델인데 왜 갑자기 로드가 안 되는 걸까요?
박시니어 씨가 웃으며 말했습니다. "직렬화 테스트를 안 했구나."
직렬화는 모델을 파일로 저장하는 것이고, 역직렬화는 저장된 파일에서 모델을 복원하는 것입니다. 마치 레고 작품을 분해해서 박스에 넣고(직렬화), 나중에 박스에서 꺼내 다시 조립하는 것(역직렬화)과 같습니다.
이 과정에서 문제가 생기면 모델을 재사용할 수 없게 됩니다.
다음 코드를 살펴봅시다.
import pytest
import numpy as np
import pickle
import joblib
from sklearn.ensemble import GradientBoostingClassifier
from sklearn.preprocessing import StandardScaler
from sklearn.pipeline import Pipeline
# 모델 파이프라인 생성
def create_model_pipeline():
return Pipeline([
('scaler', StandardScaler()),
('classifier', GradientBoostingClassifier(n_estimators=10, random_state=42))
])
# 직렬화/역직렬화 테스트
class TestModelSerialization:
@pytest.fixture
def trained_pipeline(self):
"""훈련된 파이프라인을 생성합니다"""
np.random.seed(42)
X = np.random.randn(100, 5)
y = np.random.randint(0, 2, 100)
pipeline = create_model_pipeline()
pipeline.fit(X, y)
return pipeline, X[:5]
def test_joblib_serialization(self, trained_pipeline, tmp_path):
"""joblib으로 저장/로드가 정상 동작하는지 확인합니다"""
pipeline, test_data = trained_pipeline
original_predictions = pipeline.predict(test_data)
# 저장
model_path = tmp_path / "model.joblib"
joblib.dump(pipeline, model_path)
# 로드
loaded_pipeline = joblib.load(model_path)
loaded_predictions = loaded_pipeline.predict(test_data)
np.testing.assert_array_equal(original_predictions, loaded_predictions)
def test_model_attributes_preserved(self, trained_pipeline, tmp_path):
"""모델 속성이 보존되는지 확인합니다"""
pipeline, _ = trained_pipeline
model_path = tmp_path / "model.joblib"
joblib.dump(pipeline, model_path)
loaded = joblib.load(model_path)
original_clf = pipeline.named_steps['classifier']
loaded_clf = loaded.named_steps['classifier']
assert original_clf.n_estimators == loaded_clf.n_estimators
assert original_clf.random_state == loaded_clf.random_state
김개발 씨는 일주일 동안 공들여 모델을 훈련했습니다. 성능도 좋고, 코드도 깔끔하게 정리했습니다.
마지막으로 모델을 저장하고 퇴근했습니다. 다음 날 출근해서 모델을 로드하려는데 에러가 발생했습니다.
"ModuleNotFoundError: No module named 'custom_transformer'" 어제까지 잘 되던 모델인데 왜 갑자기 안 되는 걸까요? 박시니어 씨가 원인을 설명해주었습니다.
"커스텀 전처리 클래스를 사용했는데, 그 클래스가 정의된 모듈을 import하지 않아서 생긴 문제예요. 직렬화 테스트를 했다면 미리 발견할 수 있었을 거예요." 직렬화와 역직렬화란 무엇일까요?
쉽게 비유하자면, 직렬화는 마치 가구를 이사할 때 분해해서 박스에 넣는 것과 같습니다. 그리고 역직렬화는 새 집에서 박스를 열어 가구를 다시 조립하는 것입니다.
분해할 때 나사를 잃어버리면 조립이 안 되듯이, 직렬화할 때 정보가 누락되면 역직렬화에 실패합니다. 직렬화가 실패하면 어떤 문제가 생길까요?
훈련에 들인 시간이 모두 물거품이 됩니다. 또한 모델 버전 관리가 불가능해집니다.
프로덕션 환경에서 모델을 배포할 수도 없습니다. 위의 코드를 살펴보겠습니다.
test_joblib_serialization은 가장 기본적인 테스트입니다. 모델을 저장하고 로드한 후 예측 결과가 동일한지 확인합니다.
np.testing.assert_array_equal로 원본 예측과 로드된 모델의 예측을 비교합니다. test_model_attributes_preserved는 모델의 하이퍼파라미터가 보존되는지 확인합니다.
n_estimators와 random_state 같은 설정 값이 저장 후에도 동일해야 합니다. 실제 현업에서는 어떤 문제가 자주 발생할까요?
첫째, 커스텀 클래스나 함수를 사용할 때 해당 모듈을 import하지 않으면 역직렬화에 실패합니다. 둘째, 라이브러리 버전이 다르면 호환성 문제가 생깁니다.
sklearn 1.0에서 저장한 모델을 sklearn 0.24에서 로드하면 에러가 날 수 있습니다. 이런 문제를 방지하려면 어떻게 해야 할까요?
모델과 함께 의존성 정보를 저장하는 것이 좋습니다. MLflow나 BentoML 같은 도구를 사용하면 모델, 환경, 코드를 함께 패키징할 수 있습니다.
또한 모델 저장 직후 바로 로드 테스트를 수행하는 습관을 들이세요. 김개발 씨는 이제 모든 모델을 저장한 직후 바로 로드 테스트를 수행하기로 했습니다.
"저장이 끝이 아니라 시작이군요."
실전 팁
💡 - joblib은 numpy 배열을 효율적으로 저장하므로 sklearn 모델에 권장됩니다
- 커스텀 클래스를 사용할 때는 해당 모듈의 import 경로를 주의하세요
- 모델과 함께 라이브러리 버전 정보를 기록해두세요
7. 회귀 테스트로 성능 저하 감지
김개발 씨가 데이터 전처리 코드를 조금 수정했습니다. 더 효율적으로 바꾼 것뿐인데 모델 성능이 3% 떨어졌습니다.
코드는 정상적으로 돌아가는데 왜 성능이 떨어진 걸까요? 박시니어 씨가 말했습니다.
"회귀 테스트가 있었다면 바로 잡을 수 있었을 텐데요."
회귀 테스트는 코드를 수정한 후에도 기존 기능이 정상 동작하는지 확인하는 테스트입니다. 마치 자동차를 수리한 후 다른 부분이 고장 나지 않았는지 전체를 점검하는 것과 같습니다.
머신러닝에서는 코드 변경 후 모델 성능이 기존보다 떨어지지 않았는지 자동으로 검증합니다.
다음 코드를 살펴봅시다.
import pytest
import json
import numpy as np
from pathlib import Path
from sklearn.datasets import make_classification
from sklearn.model_selection import cross_val_score
from sklearn.ensemble import RandomForestClassifier
# 기준 성능 저장 및 로드
def save_baseline_metrics(metrics: dict, path: str):
with open(path, 'w') as f:
json.dump(metrics, f)
def load_baseline_metrics(path: str) -> dict:
with open(path, 'r') as f:
return json.load(f)
# 회귀 테스트
class TestModelRegression:
@pytest.fixture
def baseline_path(self, tmp_path):
"""기준 성능 파일 경로를 생성합니다"""
path = tmp_path / "baseline.json"
baseline = {'accuracy': 0.85, 'f1': 0.83}
save_baseline_metrics(baseline, str(path))
return str(path)
def test_no_performance_regression(self, baseline_path):
"""성능이 기준 대비 저하되지 않았는지 확인합니다"""
X, y = make_classification(n_samples=500, random_state=42)
model = RandomForestClassifier(n_estimators=50, random_state=42)
current_accuracy = cross_val_score(model, X, y, cv=5).mean()
baseline = load_baseline_metrics(baseline_path)
# 5% 이상 성능 저하 시 실패
threshold = baseline['accuracy'] * 0.95
assert current_accuracy >= threshold, \
f"성능 저하 감지: {current_accuracy:.3f} < {threshold:.3f}"
def test_update_baseline_if_improved(self, baseline_path):
"""성능이 개선되면 기준을 업데이트합니다"""
X, y = make_classification(n_samples=500, random_state=42)
model = RandomForestClassifier(n_estimators=100, random_state=42)
current_accuracy = cross_val_score(model, X, y, cv=5).mean()
baseline = load_baseline_metrics(baseline_path)
if current_accuracy > baseline['accuracy']:
baseline['accuracy'] = current_accuracy
save_baseline_metrics(baseline, baseline_path)
print(f"기준 업데이트: {current_accuracy:.3f}")
김개발 씨는 데이터 전처리 코드를 리팩토링했습니다. 중복된 코드를 함수로 분리하고, 비효율적인 반복문을 벡터 연산으로 바꿨습니다.
코드가 훨씬 깔끔해졌고, 실행 속도도 빨라졌습니다. 그런데 다음 날 모니터링 대시보드를 보니 모델 정확도가 3% 떨어져 있었습니다.
분명히 로직은 그대로인데 왜 성능이 떨어진 걸까요? 원인을 추적해보니 미묘한 차이가 있었습니다.
기존 코드는 결측치를 평균값으로 채웠는데, 리팩토링하면서 실수로 중앙값으로 바꿔버린 것입니다. 박시니어 씨가 말했습니다.
"이런 실수를 잡으려면 회귀 테스트가 필요해요. 코드를 바꿀 때마다 자동으로 성능을 체크하는 거예요." 회귀 테스트란 무엇일까요?
쉽게 비유하자면, 회귀 테스트는 마치 건강검진과 같습니다. 우리 몸에 이상이 없는지 정기적으로 체크하듯이, 코드를 변경할 때마다 기존 기능이 여전히 정상인지 확인하는 것입니다.
한 부분을 고쳤는데 엉뚱한 곳에서 문제가 생기는 것을 회귀 버그라고 하며, 이를 잡아내는 것이 회귀 테스트의 목적입니다. 회귀 테스트가 없으면 어떤 문제가 생길까요?
코드를 수정할 때마다 두려움이 생깁니다. "이거 고치면 다른 데서 문제 생기는 거 아닐까?" 결국 코드 수정을 꺼리게 되고, 기술 부채가 쌓입니다.
또한 문제가 발생해도 언제부터 생긴 건지 추적하기 어렵습니다. 위의 코드를 살펴보겠습니다.
baseline.json 파일에 기준 성능을 저장합니다. 이것은 "현재 코드가 달성해야 하는 최소 성능"입니다.
test_no_performance_regression은 현재 모델의 성능이 기준의 95% 이상인지 확인합니다. 5% 이상 떨어지면 테스트가 실패합니다.
test_update_baseline_if_improved는 성능이 개선되면 기준을 업데이트합니다. 이렇게 하면 성능이 점점 좋아지는 방향으로 발전할 수 있습니다.
실제 CI/CD 파이프라인에서는 어떻게 사용할까요? 코드를 푸시할 때마다 회귀 테스트가 자동으로 실행됩니다.
성능이 기준 이하로 떨어지면 머지가 차단됩니다. 이렇게 하면 성능이 저하된 코드가 프로덕션에 배포되는 것을 방지할 수 있습니다.
주의할 점도 있습니다. 허용 오차를 너무 작게 설정하면 랜덤 변동에도 테스트가 실패합니다.
반대로 너무 크게 설정하면 실제 성능 저하를 놓칠 수 있습니다. 과거 데이터를 분석하여 적절한 임계값을 설정해야 합니다.
김개발 씨는 이제 모든 코드 변경에 대해 회귀 테스트를 실행하기로 했습니다. "이제 안심하고 리팩토링할 수 있겠네요."
실전 팁
💡 - 기준 성능은 버전 관리 시스템에 함께 커밋하세요
- 허용 오차는 과거 성능 변동폭을 참고하여 설정하세요
- CI/CD 파이프라인에 회귀 테스트를 필수 단계로 추가하세요
8. 엣지 케이스 테스트
김개발 씨의 모델이 프로덕션에서 갑자기 에러를 냈습니다. 로그를 확인해보니 입력 데이터가 모두 0이었습니다.
훈련 데이터에는 없던 케이스였습니다. 박시니어 씨가 말했습니다.
"엣지 케이스 테스트를 했어야 해요."
엣지 케이스 테스트는 극단적이거나 예외적인 입력에 대해 모델이 어떻게 동작하는지 확인하는 것입니다. 마치 자동차의 충돌 테스트처럼, 일상적이지 않은 상황에서도 안전하게 동작하는지 검증합니다.
빈 입력, 모두 같은 값, 극단적으로 크거나 작은 값 등을 테스트합니다.
다음 코드를 살펴봅시다.
import pytest
import numpy as np
from sklearn.ensemble import RandomForestClassifier
# 모델 예측 래퍼
class ModelWrapper:
def __init__(self, model):
self.model = model
def predict_safe(self, X):
"""안전하게 예측을 수행합니다"""
X = np.array(X)
# 빈 입력 처리
if X.size == 0:
return np.array([])
# NaN 처리
if np.isnan(X).any():
X = np.nan_to_num(X, nan=0.0)
# 무한대 처리
if np.isinf(X).any():
X = np.clip(X, -1e10, 1e10)
return self.model.predict(X)
# 엣지 케이스 테스트
class TestEdgeCases:
@pytest.fixture
def model_wrapper(self):
np.random.seed(42)
X = np.random.randn(100, 5)
y = np.random.randint(0, 2, 100)
model = RandomForestClassifier(random_state=42)
model.fit(X, y)
return ModelWrapper(model)
def test_empty_input(self, model_wrapper):
"""빈 입력을 처리할 수 있는지 확인합니다"""
result = model_wrapper.predict_safe(np.array([]))
assert len(result) == 0
def test_all_zeros(self, model_wrapper):
"""모든 값이 0인 입력을 처리할 수 있는지 확인합니다"""
result = model_wrapper.predict_safe([[0, 0, 0, 0, 0]])
assert result[0] in [0, 1]
def test_extreme_values(self, model_wrapper):
"""극단적인 값을 처리할 수 있는지 확인합니다"""
result = model_wrapper.predict_safe([[1e10, -1e10, 1e10, -1e10, 1e10]])
assert result[0] in [0, 1]
def test_nan_handling(self, model_wrapper):
"""NaN 값을 처리할 수 있는지 확인합니다"""
result = model_wrapper.predict_safe([[np.nan, 1, 2, 3, 4]])
assert not np.isnan(result).any()
김개발 씨의 모델은 고객의 구매 패턴을 분석하는 것이었습니다. 수개월간 문제없이 잘 동작했습니다.
그런데 어느 날 갑자기 에러가 발생했습니다. 원인을 조사해보니 신규 가입 직후 아무 활동도 하지 않은 고객의 데이터가 들어왔습니다.
모든 피처가 0이었습니다. 훈련 데이터에는 이런 케이스가 없었기 때문에 모델이 어떻게 반응할지 아무도 몰랐던 것입니다.
박시니어 씨가 말했습니다. "엣지 케이스는 항상 발생해요.
미리 테스트해두지 않으면 프로덕션에서 터지게 되어 있어요." 엣지 케이스란 무엇일까요? 쉽게 비유하자면, 엣지 케이스는 마치 자동차의 충돌 테스트와 같습니다.
일상적인 운전에서는 충돌이 거의 일어나지 않지만, 만약의 상황에 대비해서 테스트합니다. 프로그래밍에서도 마찬가지입니다.
흔하지 않지만 발생할 수 있는 상황에 대비하는 것입니다. 어떤 엣지 케이스를 테스트해야 할까요?
머신러닝에서 흔한 엣지 케이스들이 있습니다. 빈 입력, 모든 값이 동일한 입력, 극단적으로 크거나 작은 값, NaN이나 무한대, 데이터 타입이 다른 경우 등입니다.
위의 코드를 살펴보겠습니다. ModelWrapper 클래스는 원본 모델을 감싸서 안전한 예측을 제공합니다.
빈 입력이 들어오면 빈 배열을 반환합니다. NaN이 있으면 0으로 대체합니다.
무한대 값은 적절한 범위로 클리핑합니다. 테스트 코드를 보겠습니다.
test_empty_input은 빈 배열이 들어왔을 때 에러 없이 빈 결과를 반환하는지 확인합니다. test_all_zeros는 모든 값이 0일 때도 정상적인 예측을 하는지 확인합니다.
test_extreme_values는 매우 큰 값이나 작은 값에 대해서도 동작하는지 확인합니다. 실제 현업에서는 더 다양한 엣지 케이스가 있습니다.
문자열 타입이 들어오는 경우, 피처 개수가 다른 경우, 인코딩이 깨진 텍스트 등이 있습니다. 서비스의 특성에 따라 발생 가능한 엣지 케이스를 미리 나열하고 테스트해야 합니다.
주의할 점도 있습니다. 모든 엣지 케이스에 대해 "정상적인 예측"을 기대하면 안 됩니다.
때로는 명확한 에러를 내는 것이 더 좋은 선택일 수 있습니다. 잘못된 예측보다 에러가 나은 경우도 있기 때문입니다.
각 엣지 케이스에 대해 어떻게 동작해야 하는지 미리 정의해두세요. 김개발 씨는 엣지 케이스 목록을 만들고 하나씩 테스트를 추가하기로 했습니다.
"예방이 치료보다 낫군요."
실전 팁
💡 - 서비스 로그를 분석하여 실제로 발생한 엣지 케이스를 파악하세요
- 엣지 케이스마다 예상 동작을 명확히 정의하세요
- 방어적 코딩으로 예외 상황을 우아하게 처리하세요
9. 통합 테스트로 전체 파이프라인 검증
김개발 씨의 모델은 단위 테스트를 모두 통과했습니다. 그런데 막상 전체 파이프라인을 돌리니 에러가 발생했습니다.
개별 부품은 모두 정상인데 왜 전체가 안 돌아가는 걸까요? 박시니어 씨가 말했습니다.
"통합 테스트가 필요한 이유예요."
통합 테스트는 개별 컴포넌트들이 함께 올바르게 동작하는지 확인하는 것입니다. 마치 오케스트라에서 각 악기가 개별적으로는 완벽해도 함께 연주할 때 조화를 이뤄야 하는 것처럼, 데이터 로딩부터 전처리, 모델 예측, 후처리까지 전체 흐름을 테스트합니다.
다음 코드를 살펴봅시다.
import pytest
import numpy as np
import pandas as pd
from sklearn.preprocessing import StandardScaler
from sklearn.ensemble import RandomForestClassifier
# 파이프라인 컴포넌트들
def load_data(data_source):
"""데이터를 로드합니다"""
return pd.DataFrame(data_source)
def preprocess(df):
"""데이터를 전처리합니다"""
scaler = StandardScaler()
return scaler.fit_transform(df), scaler
def train_model(X, y):
"""모델을 훈련합니다"""
model = RandomForestClassifier(random_state=42)
model.fit(X, y)
return model
def predict_and_postprocess(model, X):
"""예측하고 후처리합니다"""
predictions = model.predict(X)
probabilities = model.predict_proba(X)
return {'predictions': predictions, 'confidence': probabilities.max(axis=1)}
# 통합 테스트
class TestEndToEndPipeline:
@pytest.fixture
def sample_data(self):
np.random.seed(42)
return {
'features': np.random.randn(100, 5).tolist(),
'labels': np.random.randint(0, 2, 100).tolist()
}
def test_full_pipeline_execution(self, sample_data):
"""전체 파이프라인이 정상 실행되는지 확인합니다"""
# 1. 데이터 로드
df = load_data(sample_data['features'])
assert len(df) == 100
# 2. 전처리
X, scaler = preprocess(df)
assert X.shape == (100, 5)
# 3. 모델 훈련
y = np.array(sample_data['labels'])
model = train_model(X, y)
assert hasattr(model, 'predict')
# 4. 예측 및 후처리
result = predict_and_postprocess(model, X[:10])
assert 'predictions' in result
assert 'confidence' in result
assert len(result['predictions']) == 10
def test_pipeline_with_new_data(self, sample_data):
"""새로운 데이터로 예측이 가능한지 확인합니다"""
df = load_data(sample_data['features'])
X, scaler = preprocess(df)
y = np.array(sample_data['labels'])
model = train_model(X, y)
# 새로운 데이터
new_data = np.random.randn(5, 5)
new_scaled = scaler.transform(new_data)
result = predict_and_postprocess(model, new_scaled)
assert len(result['predictions']) == 5
김개발 씨는 머신러닝 파이프라인을 모듈화해서 잘 정리했습니다. 데이터 로딩 모듈, 전처리 모듈, 모델 모듈, 후처리 모듈.
각 모듈에 대한 단위 테스트도 모두 통과했습니다. 그런데 이상했습니다.
전체 파이프라인을 실행하면 에러가 발생했습니다. "shape mismatch"라는 에러 메시지가 떴습니다.
원인을 찾아보니 전처리 모듈의 출력 형태와 모델 모듈이 기대하는 입력 형태가 달랐습니다. 전처리 모듈은 2D 배열을 반환했지만, 모델 모듈은 DataFrame을 기대하고 있었던 것입니다.
박시니어 씨가 설명했습니다. "단위 테스트는 각 부품이 정상인지 확인하는 거예요.
하지만 부품들이 서로 잘 맞물리는지는 통합 테스트로 확인해야 해요." 통합 테스트란 무엇일까요? 쉽게 비유하자면, 통합 테스트는 마치 오케스트라 리허설과 같습니다.
바이올리니스트는 혼자서 완벽하게 연주할 수 있고, 피아니스트도 마찬가지입니다. 하지만 함께 연주할 때는 서로의 타이밍을 맞추고 음량을 조절해야 합니다.
시스템도 마찬가지입니다. 개별 컴포넌트는 정상이어도 함께 동작할 때 문제가 생길 수 있습니다.
통합 테스트가 없으면 어떤 문제가 생길까요? 인터페이스 불일치 문제를 늦게 발견하게 됩니다.
프로덕션 직전에야 문제를 발견하면 수정 비용이 매우 커집니다. 또한 어디서 문제가 생겼는지 추적하기도 어렵습니다.
위의 코드를 살펴보겠습니다. test_full_pipeline_execution은 전체 파이프라인을 처음부터 끝까지 실행합니다.
각 단계의 출력이 다음 단계의 입력으로 올바르게 전달되는지 확인합니다. test_pipeline_with_new_data는 훈련 후 새로운 데이터로 예측이 가능한지 테스트합니다.
scaler가 새 데이터에도 적용될 수 있는지 확인합니다. 실제 현업에서는 더 복잡한 통합 테스트가 필요합니다.
데이터베이스 연결, API 호출, 파일 시스템 접근 등 외부 의존성도 테스트해야 합니다. 이때 목(Mock) 객체를 사용하면 외부 시스템 없이도 테스트할 수 있습니다.
주의할 점도 있습니다. 통합 테스트는 단위 테스트보다 실행 시간이 오래 걸립니다.
따라서 모든 코드 변경마다 전체 통합 테스트를 실행하기는 어렵습니다. CI/CD 파이프라인에서 단계별로 테스트를 구성하는 것이 좋습니다.
김개발 씨는 이제 단위 테스트와 통합 테스트를 모두 작성하기로 했습니다. "개별 부품도 중요하지만 전체 조립도 중요하군요."
실전 팁
💡 - 통합 테스트는 실제 환경과 유사하게 구성하세요
- 외부 의존성은 Mock 객체로 대체하면 테스트가 안정적입니다
- CI/CD에서 단위 테스트는 자주, 통합 테스트는 주기적으로 실행하세요
10. CI CD 파이프라인에 테스트 통합하기
김개발 씨는 테스트 코드를 열심히 작성했습니다. 하지만 바쁠 때는 테스트 실행을 잊어버리곤 했습니다.
어느 날 테스트를 건너뛰고 배포했다가 버그가 프로덕션에 올라갔습니다. 박시니어 씨가 말했습니다.
"테스트를 CI/CD에 통합해야 해요."
CI/CD 파이프라인에 테스트를 통합하면 코드를 푸시할 때마다 자동으로 테스트가 실행됩니다. 마치 공항의 보안 검색대처럼, 모든 코드가 배포되기 전에 반드시 테스트를 통과해야 합니다.
사람이 깜빡해도 시스템이 자동으로 품질을 검증합니다.
다음 코드를 살펴봅시다.
# .github/workflows/ml-tests.yml
name: ML Model Tests
on:
push:
branches: [main, develop]
pull_request:
branches: [main]
jobs:
test:
runs-on: ubuntu-latest
steps:
- uses: actions/checkout@v3
- name: Set up Python
uses: actions/setup-python@v4
with:
python-version: '3.9'
- name: Install dependencies
run: |
pip install -r requirements.txt
pip install pytest pytest-cov
- name: Run unit tests
run: pytest tests/unit/ -v --cov=src
- name: Run model validation tests
run: pytest tests/model/ -v
- name: Run integration tests
run: pytest tests/integration/ -v
- name: Check test coverage
run: |
coverage report --fail-under=80
김개발 씨는 부지런히 테스트 코드를 작성해왔습니다. 하지만 마감에 쫓기다 보면 테스트 실행을 건너뛰는 일이 생겼습니다.
"일단 배포하고 나중에 테스트하자"는 생각이었습니다. 그러던 어느 날 대형 사고가 터졌습니다.
테스트를 건너뛰고 배포한 코드에 버그가 있었고, 고객 데이터가 잘못 처리되었습니다. 롤백하느라 야근을 했고, 보고서도 써야 했습니다.
박시니어 씨가 말했습니다. "사람은 실수하게 되어 있어요.
그래서 시스템으로 강제해야 해요. CI/CD에 테스트를 통합하면 테스트를 통과하지 못한 코드는 아예 배포가 안 돼요." CI/CD란 무엇일까요?
CI는 Continuous Integration(지속적 통합), CD는 Continuous Deployment(지속적 배포)입니다. 쉽게 말해, 코드를 자주 합치고 자동으로 배포하는 방식입니다.
이 과정에 테스트를 추가하면 품질을 자동으로 보장할 수 있습니다. 위의 코드는 GitHub Actions 설정 파일입니다.
on 섹션은 언제 이 워크플로우가 실행되는지 정의합니다. main이나 develop 브랜치에 푸시하거나, main으로 PR을 올리면 자동으로 실행됩니다.
steps는 실행할 작업들입니다. 코드를 체크아웃하고, Python을 설정하고, 의존성을 설치하고, 테스트를 실행합니다.
테스트는 세 단계로 나뉘어 있습니다. unit tests는 개별 함수를 테스트합니다.
model validation tests는 모델 성능을 검증합니다. integration tests는 전체 파이프라인을 테스트합니다.
마지막으로 테스트 커버리지가 80% 이상인지 확인합니다. 실제 현업에서는 어떻게 운영할까요?
PR을 올리면 자동으로 테스트가 실행되고, 모든 테스트가 통과해야 머지가 허용됩니다. 이것을 머지 게이트라고 합니다.
테스트가 실패하면 PR이 차단되어 버그가 있는 코드가 메인 브랜치에 들어가는 것을 방지합니다. 주의할 점도 있습니다.
테스트 실행 시간이 너무 길면 개발 속도가 느려집니다. 빠른 테스트(단위 테스트)는 모든 커밋에, 느린 테스트(통합 테스트)는 PR이나 야간 빌드에 실행하는 식으로 분리하는 것이 좋습니다.
김개발 씨는 CI/CD 파이프라인을 설정한 후 한결 마음이 편해졌습니다. "이제 테스트를 까먹을 일이 없겠네요.
시스템이 알아서 확인해주니까요." 박시니어 씨가 덧붙였습니다. "좋은 개발 문화는 개인의 의지가 아니라 시스템이 만드는 거예요."
실전 팁
💡 - pytest-cov로 테스트 커버리지를 측정하고 최소 기준을 설정하세요
- 느린 테스트는 별도 워크플로우로 분리하여 야간에 실행하세요
- 테스트 실패 시 슬랙이나 이메일로 알림을 받도록 설정하세요
이상으로 학습을 마칩니다. 위 내용을 직접 코드로 작성해보면서 익혀보세요!
댓글 (0)
함께 보면 좋은 카드 뉴스
Helm 마이크로서비스 패키징 완벽 가이드
Kubernetes 환경에서 마이크로서비스를 효율적으로 패키징하고 배포하는 Helm의 핵심 기능을 실무 중심으로 학습합니다. Chart 생성부터 릴리스 관리까지 체계적으로 다룹니다.
보안 아키텍처 구성 완벽 가이드
프로젝트의 보안을 처음부터 설계하는 방법을 배웁니다. AWS 환경에서 VPC부터 WAF, 암호화, 접근 제어까지 실무에서 바로 적용할 수 있는 보안 아키텍처를 단계별로 구성해봅니다.
AWS Organizations 완벽 가이드
여러 AWS 계정을 체계적으로 관리하고 통합 결제와 보안 정책을 적용하는 방법을 실무 스토리로 쉽게 배워봅니다. 초보 개발자도 바로 이해할 수 있는 친절한 설명과 실전 예제를 제공합니다.
AWS KMS 암호화 완벽 가이드
AWS KMS(Key Management Service)를 활용한 클라우드 데이터 암호화 방법을 초급 개발자를 위해 쉽게 설명합니다. CMK 생성부터 S3, EBS 암호화, 봉투 암호화까지 실무에 필요한 모든 내용을 담았습니다.
AWS Secrets Manager 완벽 가이드
AWS에서 데이터베이스 비밀번호, API 키 등 민감한 정보를 안전하게 관리하는 Secrets Manager의 핵심 개념과 실무 활용법을 배워봅니다. 초급 개발자도 쉽게 따라할 수 있도록 실전 예제와 함께 설명합니다.