"""tests/test_engine_v2_phase4.py — Phase 4 TDD 테스트 스위트. G09+G10+G11+G12: ConsensusPipeline, CircuitBreaker, CostTracker 검증. 작성 순서: 테스트 먼저(RED), 구현 후 GREEN 확인. """ from __future__ import annotations import json import os import sys import time sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..")) from pathlib import Path from unittest.mock import AsyncMock, patch import pytest from engine_v2.engine_result import EngineResult # --------------------------------------------------------------------------- # 공통 헬퍼 # --------------------------------------------------------------------------- def _make_result( content: str = "정상 응답", engine: str = "claude", error: bool = False, task_id: str = "task-001", step: int = 1, ) -> EngineResult: """테스트용 EngineResult 생성.""" return EngineResult( engine=engine, # type: ignore[arg-type] content=content, clean=content, task_id=task_id, step=step, error=error, ) # --------------------------------------------------------------------------- # G10: CircuitBreaker 테스트 # --------------------------------------------------------------------------- class TestCircuitBreakerClosed: """CLOSED 상태 기본 동작 테스트.""" def test_circuit_breaker_closed_allows(self) -> None: """CLOSED 상태에서 요청이 허용되어야 한다.""" from engine_v2.circuit_breaker import CBState, CircuitBreaker cb = CircuitBreaker() assert cb.state == CBState.CLOSED assert cb.allow_request() is True class TestCircuitBreakerOpen: """OPEN 상태 전환 테스트.""" def test_circuit_breaker_open_after_3_failures(self) -> None: """3회 실패 후 OPEN 상태로 전환되어야 한다.""" from engine_v2.circuit_breaker import CBState, CircuitBreaker cb = CircuitBreaker(fail_threshold=3) assert cb.state == CBState.CLOSED cb.record_failure() assert cb.state == CBState.CLOSED cb.record_failure() assert cb.state == CBState.CLOSED cb.record_failure() assert cb.state == CBState.OPEN assert cb.allow_request() is False class TestCircuitBreakerHalfOpen: """HALF_OPEN 상태 전환 테스트.""" def test_circuit_breaker_recovery_to_half_open(self) -> None: """recovery_sec 경과 후 HALF_OPEN으로 전환되어야 한다.""" from engine_v2.circuit_breaker import CBState, CircuitBreaker cb = CircuitBreaker(fail_threshold=3, recovery_sec=60) cb.record_failure() cb.record_failure() cb.record_failure() # state 프로퍼티 호출 전에 _state를 직접 확인 (recovery_sec=60이므로 아직 OPEN) assert cb._state == CBState.OPEN # recovery_sec=0인 CB로 다시 테스트 — 즉시 HALF_OPEN으로 전환 cb2 = CircuitBreaker(fail_threshold=3, recovery_sec=0) cb2.record_failure() cb2.record_failure() cb2.record_failure() # 약간의 시간 경과 후 state 조회 → HALF_OPEN time.sleep(0.01) assert cb2.state == CBState.HALF_OPEN def test_circuit_breaker_half_open_success_closes(self) -> None: """HALF_OPEN에서 성공 기록 시 CLOSED로 전환되어야 한다.""" from engine_v2.circuit_breaker import CBState, CircuitBreaker cb = CircuitBreaker(fail_threshold=3, recovery_sec=0) cb.record_failure() cb.record_failure() cb.record_failure() time.sleep(0.01) assert cb.state == CBState.HALF_OPEN cb.allow_request() cb.record_success() assert cb.state == CBState.CLOSED def test_circuit_breaker_half_open_max_calls(self) -> None: """HALF_OPEN 상태에서 half_max_calls=1이면 1회만 허용되어야 한다.""" from engine_v2.circuit_breaker import CBState, CircuitBreaker cb = CircuitBreaker(fail_threshold=3, recovery_sec=0, half_max_calls=1) cb.record_failure() cb.record_failure() cb.record_failure() time.sleep(0.01) assert cb.state == CBState.HALF_OPEN # 1회 허용 assert cb.allow_request() is True # 2회째 거부 assert cb.allow_request() is False # --------------------------------------------------------------------------- # G11: CostTracker 테스트 # --------------------------------------------------------------------------- class TestCostTrackerLogUsage: """log_usage() 기록 테스트.""" def test_cost_tracker_log_usage(self, tmp_path: Path) -> None: """JSONL 파일에 사용량이 기록되어야 한다.""" from engine_v2 import cost_tracker result = _make_result(content="테스트 결과", engine="claude") # tmp_path로 로그 경로 오버라이드 with patch.object(cost_tracker, "_get_log_path", return_value=tmp_path / "test_usage.jsonl"): cost_tracker.log_usage(result, prompt_chars=100, duration_sec=1.23) log_file = tmp_path / "test_usage.jsonl" assert log_file.exists() lines = log_file.read_text(encoding="utf-8").strip().splitlines() assert len(lines) == 1 entry = json.loads(lines[0]) assert entry["engine"] == "claude" assert entry["prompt_chars"] == 100 assert entry["duration_sec"] == 1.23 assert entry["task_id"] == "task-001" assert entry["step"] == 1 def test_cost_tracker_read_usage(self, tmp_path: Path) -> None: """기록된 JSONL을 읽어 리스트로 반환해야 한다.""" from engine_v2 import cost_tracker result1 = _make_result(content="응답1", engine="claude") result2 = _make_result(content="응답2", engine="gemini", step=2) log_file = tmp_path / "engine_usage_2026-03.jsonl" with patch.object(cost_tracker, "_get_log_path", return_value=log_file): cost_tracker.log_usage(result1, prompt_chars=50, duration_sec=0.5) cost_tracker.log_usage(result2, prompt_chars=80, duration_sec=0.8) # read_usage에서 파일 경로를 직접 지정하기 위해 함수 내부 경로 우회 # _LOG_DIR을 임시로 패치 with patch.object(cost_tracker, "_LOG_DIR", tmp_path): entries = cost_tracker.read_usage("2026-03") assert len(entries) == 2 assert entries[0]["engine"] == "claude" assert entries[1]["engine"] == "gemini" def test_cost_tracker_monthly_file(self, tmp_path: Path) -> None: """월별로 다른 파일에 기록되어야 한다.""" from datetime import datetime, timezone from unittest.mock import patch as up from engine_v2 import cost_tracker result = _make_result(content="월별 테스트") # 2026-01 jan_file = tmp_path / "engine_usage_2026-01.jsonl" # 2026-02 feb_file = tmp_path / "engine_usage_2026-02.jsonl" with patch.object(cost_tracker, "_get_log_path", return_value=jan_file): cost_tracker.log_usage(result, prompt_chars=10, duration_sec=0.1) with patch.object(cost_tracker, "_get_log_path", return_value=feb_file): cost_tracker.log_usage(result, prompt_chars=20, duration_sec=0.2) assert jan_file.exists() assert feb_file.exists() # 파일이 분리되어 있어야 함 jan_data = json.loads(jan_file.read_text().strip()) feb_data = json.loads(feb_file.read_text().strip()) assert jan_data["prompt_chars"] == 10 assert feb_data["prompt_chars"] == 20 # --------------------------------------------------------------------------- # G09: ConsensusPipeline 테스트 # --------------------------------------------------------------------------- class TestConsensusEvaluateHighScore: """합의 점수 >= threshold → converged 테스트.""" def test_consensus_evaluate_high_score(self) -> None: """긍정 시그널만 있으면 합의 점수 >= 0.75이고 converged=True여야 한다.""" from publishing.consensus_pipeline import ConsensusPipeline pipeline = ConsensusPipeline(threshold=0.75) results = [ _make_result(content="수정 불필요. 충분히 적절하고 우수한 내용입니다.", engine="claude"), _make_result(content="양호합니다. 수정 불필요.", engine="gemini"), ] should_continue = pipeline.should_continue(results) assert should_continue is False assert pipeline.state.converged is True assert pipeline.state.consensus_score >= 0.75 class TestConsensusEvaluateLowScore: """합의 점수 < threshold → continue 테스트.""" def test_consensus_evaluate_low_score(self) -> None: """부정 시그널만 있으면 합의 점수 < 0.75이고 continue=True여야 한다.""" from publishing.consensus_pipeline import ConsensusPipeline pipeline = ConsensusPipeline(threshold=0.75) results = [ _make_result(content="수정 필요. 오류가 있으며 누락된 내용이 많습니다.", engine="claude"), _make_result(content="부족합니다. major issue가 있고 missing 항목이 있습니다.", engine="gemini"), ] should_continue = pipeline.should_continue(results) assert should_continue is True assert pipeline.state.converged is False assert pipeline.state.consensus_score < 0.75 class TestConsensusMaxRounds: """최대 라운드 강제 종료 테스트.""" def test_consensus_max_rounds(self) -> None: """3라운드 후 강제 종료되어야 한다 (MAX_CONSENSUS_ROUNDS=3).""" from publishing.consensus_pipeline import MAX_CONSENSUS_ROUNDS, ConsensusPipeline assert MAX_CONSENSUS_ROUNDS == 3 pipeline = ConsensusPipeline(threshold=0.75) low_score_results = [ _make_result(content="수정 필요. 오류가 있습니다.", engine="claude"), ] # 라운드 1 — continue r1 = pipeline.should_continue(low_score_results) assert r1 is True assert pipeline.state.round_number == 1 # 라운드 2 — continue r2 = pipeline.should_continue(low_score_results) assert r2 is True assert pipeline.state.round_number == 2 # 라운드 3 — 강제 종료 (MAX_CONSENSUS_ROUNDS 도달) r3 = pipeline.should_continue(low_score_results) assert r3 is False assert pipeline.state.round_number == 3 assert pipeline.state.converged is False class TestConsensusMinorityPreservation: """소수 의견 보존 테스트.""" def test_consensus_minority_preservation(self) -> None: """반대/우려 키워드가 있는 의견이 minority_opinions에 보존되어야 한다.""" from publishing.consensus_pipeline import ConsensusPipeline pipeline = ConsensusPipeline(threshold=0.75) results = [ _make_result(content="수정 불필요. 우수합니다.", engine="claude"), _make_result(content="그러나 concern이 있습니다. however 추가 검토 필요.", engine="gemini"), ] pipeline.should_continue(results) assert len(pipeline.state.minority_opinions) >= 1 assert any("gemini" in op for op in pipeline.state.minority_opinions) # --------------------------------------------------------------------------- # G10+G12: EngineOrchestrator + CircuitBreaker 통합 테스트 # --------------------------------------------------------------------------- class TestOrchestratorCircuitBreakerBlocks: """Circuit Breaker OPEN 시 에러 EngineResult 반환 테스트.""" @pytest.mark.asyncio async def test_orchestrator_circuit_breaker_blocks(self) -> None: """CircuitBreaker가 OPEN 상태이면 에러 EngineResult를 반환해야 한다.""" from engine_v2.circuit_breaker import CBState, CircuitBreaker from engine_v2.engine_orchestrator import EngineOrchestrator orchestrator = EngineOrchestrator() # claude 브레이커를 강제로 OPEN orchestrator._breakers["claude"].record_failure() orchestrator._breakers["claude"].record_failure() orchestrator._breakers["claude"].record_failure() assert orchestrator._breakers["claude"].state == CBState.OPEN # SEQUENTIAL 실행 — claude는 CB에 의해 차단되어야 함 results = await orchestrator.run( mode="SEQUENTIAL", prompts=["테스트 프롬프트"], engines=["claude"], task_id="cb-test-001", step=1, ) assert len(results) == 1 assert results[0].error is True assert results[0].engine == "claude"