자막 파이프라인을 만들면서 테스트를 어떻게 짜야 할지 한참 고민했다. 화자분리, ASR, 교정, 자막 분할까지 파이프라인 곳곳에 GPU 모델이 끼어 있으니 테스트를 돌리려면 매번 GPU가 필요했다. CI에서는 GPU를 쓸 수 없고, 로컬에서도 모델 로딩만 수백 초가 걸린다. 개발 중 피드백 루프가 너무 길었다.
결국 답은 의존성 경계를 명확히 긋는 거였다. GPU가 정말 필요한 부분과 순수 로직을 분리하고 3단계 테스트 피라미드를 세웠다.
테스트를 세 층으로 나눴다.
피라미드 아래로 갈수록 테스트가 많고 빠르다. CI에서는 1번과 2번만 돌리고, GPU가 있는 환경에서만 3번을 실행한다.
파이프라인의 각 모듈에서 GPU 모델 호출 부분만 mock으로 대체하면 나머지 로직은 전부 GPU 없이 테스트할 수 있다. 예를 들어 교정 모듈은 내부적으로 Claude Code CLI를 subprocess로 호출하는데, 이 subprocess.run을 mock하면 응답 파싱 로직을 빠르게 검증할 수 있다.
def test_correct_parses_structured_output(mock_subprocess, segments):
mock_subprocess.return_value = CompletedProcess(
args=[], returncode=0,
stdout=json.dumps({
"structured_output": {
"segments": [{"index": 0, "text": "교정된 텍스트"}]
},
"usage": {"input_tokens": 100, "output_tokens": 50},
"total_cost_usd": 0.001,
}),
)
result = correct(segments)
assert result[0].text == "교정된 텍스트"subprocess mock의 핵심은 실제 CLI 응답 형식을 정확히 재현하는 거다. structured_output 필드로 오는 경우와 result 필드로 오는 경우를 둘 다 테스트해야 한다. 처음엔 한쪽만 테스트했다가 프로덕션에서 터졌다.
PersistentWorker 테스트는 좀 더 까다로웠다. 이전에 만든 HTTP 서버 방식의 GPU 워커는 stdin/stdout/stderr을 모두 사용하기 때문에 이 세 스트림을 전부 mock해야 했고, SimpleQueue와 threading.Event의 상호작용까지 시뮬레이션해야 했다.
단위 테스트만으로는 단계 간 데이터 흐름을 검증하기 어렵다. 화자분리 출력이 ASR 입력으로 제대로 변환되는지, 교정 결과가 자막 분할기에 정상적으로 들어가는지 확인하려면 실제 데이터가 필요하다.
그래서 Golden File Testing 방식을 택했다. 파이프라인을 한번 돌려서 각 단계의 출력을 JSON 파일로 저장해 둔다. tests/fixtures/ 디렉토리에 diarize.json, asr.json, corrected.json, aligned.json을 넣어두고 이걸 fixture로 로딩한다.
def test_diarize_to_asr_data_flow(diarize_data, asr_data):
# 화자분리 출력이 ASR 입력으로 변환 가능한지 검증
for seg in diarize_data:
assert hasattr(seg, "speaker")
assert hasattr(seg, "start")
assert hasattr(seg, "end")
for seg in asr_data:
assert hasattr(seg, "text")
assert hasattr(seg, "language")이 방식의 장점은 GPU 없이도 데이터 모델 변환을 꼼꼼히 검증할 수 있다는 점이다. 단점도 분명하다. 파이프라인 출력 형식이 바뀌면 fixture도 갱신해야 하는데, 이걸 깜빡하면 테스트가 거짓 통과하거나 거짓 실패한다. CLAUDE.md에 fixture 갱신 절차를 문서화해 놓은 것도 이 삽질 때문이다.
피라미드 꼭대기는 실제 GPU 서버에 HTTP 요청을 보내는 통합 테스트다. 여기서는 mock을 쓰지 않는다. 진짜 모델이 돌아가고 진짜 결과가 나온다.
핵심은 환경변수 기반 조건 실행이다. WORKER_URL 환경변수가 설정되어 있을 때만 통합 테스트가 실행되고, 없으면 자동으로 skip된다.
WORKER_URL = os.environ.get("WORKER_URL")
@pytest.mark.skipif(not WORKER_URL, reason="WORKER_URL not set")
def test_diarize_via_http():
resp = httpx.post(f"{WORKER_URL}/run/diarize", json={"wav_path": "..."})
# NDJSON 스트리밍 응답 파싱 ...덕분에 CI에서는 GPU 없이 단위 테스트와 데이터 흐름 테스트만 빠르게 돌리고, 배포 전 GPU 환경에서 통합 테스트를 한 번 돌리는 흐름이 자연스럽게 만들어졌다.
처음에는 모든 테스트가 GPU를 필요로 해서 CI 구축 자체가 불가능했다. "GPU 없으면 테스트를 못 하는 거 아닌가?"라고 생각했는데, 막상 뜯어보니 GPU가 진짜 필요한 건 모델 추론 한 줄뿐이었다. 나머지는 전부 데이터 변환과 파싱 로직이다.
비싼 리소스에 의존하는 코드를 테스트할 때 기억할 것은 하나다. 의존성 경계를 명확히 긋고 그 경계 바깥은 mock과 fixture로 대체하면 된다. GPU든, 외부 API든, 유료 서비스든 마찬가지다.