"""Phase T 전체 파이프라인 시뮬레이션 (Stage 0 ~ Stage 5). API 호출을 mock으로 대체하여 코드 경로 전체를 검증. 실제 Kei 응답(기존 run) + mock Sonnet/Selenium으로 전 Stage 통과 여부 확인. """ import asyncio import json import sys import logging from pathlib import Path from unittest.mock import patch, AsyncMock, MagicMock sys.path.insert(0, ".") logging.basicConfig(level=logging.WARNING) # ── 실제 데이터 로드 ── RUN_DIR = Path("data/runs/1774922951020") STAGE_0_DIR = Path("data/runs/20260401_151426") stage0_ctx = json.loads((STAGE_0_DIR / "stage_0_context.json").read_text(encoding="utf-8")) raw_content = stage0_ctx["raw_content"] analysis_1a = json.loads((RUN_DIR / "step1_analysis.json").read_text(encoding="utf-8")) concepts_1b = json.loads((RUN_DIR / "step1b_concepts.json").read_text(encoding="utf-8")) # ── Mock 응답 정의 ── async def mock_classify_content(content): """Stage 1A mock: 실제 Kei 응답 반환""" return analysis_1a async def mock_refine_concepts(content, analysis): """Stage 1B mock: 실제 Kei 1B 응답을 analysis에 병합하여 반환""" result = dict(analysis) concepts = concepts_1b.get("concepts", []) for t in result.get("topics", []): match = next((c for c in concepts if c.get("id") == t.get("id")), None) if match: t["relation_type"] = match.get("relation_type", "") t["expression_hint"] = match.get("expression_hint", "") t["source_data"] = match.get("source_data", "") return result # Stage 2 mock: generate_with_retry → mock HTML 반환 MOCK_BODY_HTML = """

용어 혼용

DX와 BIM이 혼용되어 사용되고 있다

DX와 핵심기술의 올바른 관계

DX는 BIM, GIS, 디지털트윈을 포함하는 상위개념이다

BIM ≠ DX

""" MOCK_SIDEBAR_HTML = """

용어 정의

건설산업: 종합산업

BIM: 정보관리도구

DX: 디지털 전환

""" MOCK_FOOTER_HTML = """

BIM은 DX의 기초가 되는 일부분이다

""" MOCK_GENERATED = { "body_html": MOCK_BODY_HTML, "sidebar_html": MOCK_SIDEBAR_HTML, "footer_html": MOCK_FOOTER_HTML, "reasoning": "mock", } MOCK_VERIFICATION = {} # verify_all_areas 결과 async def mock_generate_with_retry(content, analysis, container_specs, preset, images=None): """Stage 2 mock""" from src.content_verifier import VerificationResult verification = { "body_bg": VerificationResult(passed=True, area_name="body_bg", score=1.0), "body_core": VerificationResult(passed=True, area_name="body_core", score=1.0), "sidebar": VerificationResult(passed=True, area_name="sidebar", score=1.0), "footer": VerificationResult(passed=True, area_name="footer", score=1.0), } return MOCK_GENERATED, verification def mock_render_slide_from_html(generated, analysis, preset): """Stage 3 mock""" return f"""

{analysis.get('title','')}

{generated.get('body_html','')}

""" def mock_measure_rendered_heights(html): """Stage 4 L4 mock: overflow 없음""" return { "zones": { "body": {"scrollHeight": 400, "clientHeight": 490, "overflowed": False}, "sidebar": {"scrollHeight": 300, "clientHeight": 490, "overflowed": False}, "footer": {"scrollHeight": 55, "clientHeight": 60, "overflowed": False}, } } def mock_capture_slide_screenshot(html): """Stage 4 L5 mock: 빈 스크린샷""" return "" # 빈 문자열 → 비전 품질 게이트 스킵 async def run_full_simulation(): """전체 파이프라인 시뮬레이션""" passed = 0 failed = 0 def check(name, condition, detail=""): nonlocal passed, failed if condition: print(f" ✅ {name}") passed += 1 else: print(f" ❌ {name}") if detail: print(f" → {detail}") failed += 1 # Mock 패치 적용 # _retry_kei는 async fn을 await하는 래퍼이므로, mock도 await 해야 함 async def mock_retry_kei(fn, *a, **kw): return await fn(*a, **kw) with patch("src.pipeline._retry_kei", side_effect=mock_retry_kei), \ patch("src.kei_client.classify_content", side_effect=mock_classify_content), \ patch("src.kei_client.refine_concepts", side_effect=mock_refine_concepts), \ patch("src.content_verifier.generate_with_retry", side_effect=mock_generate_with_retry), \ patch("src.renderer.render_slide_from_html", side_effect=mock_render_slide_from_html), \ patch("src.slide_measurer.measure_rendered_heights", side_effect=mock_measure_rendered_heights), \ patch("src.slide_measurer.capture_slide_screenshot", side_effect=mock_capture_slide_screenshot), \ patch("src.image_utils.get_image_sizes", return_value={}), \ patch("src.image_utils.embed_images", side_effect=lambda html, bp: html): from src.pipeline import generate_slide events = [] print("── 전체 파이프라인 실행 ──") try: async for event in generate_slide(raw_content): events.append(event) evt_type = event.get("event", "") evt_data = event.get("data", "") if evt_type == "progress": print(f" 📌 {evt_data}") elif evt_type == "error": print(f" ❌ ERROR: {evt_data}") elif evt_type == "result": print(f" 📄 result: {len(evt_data)}자 HTML") except Exception as e: print(f" 💥 EXCEPTION: {type(e).__name__}: {e}") import traceback traceback.print_exc() check("파이프라인 예외 없음", False, str(e)) print(f"\n{'═' * 55}") print(f" 전체 시뮬레이션: {passed} passed, {failed} failed") print(f"{'═' * 55}") return False print() # 이벤트 검증 event_types = [e["event"] for e in events] check("progress 이벤트 존재", "progress" in event_types) check("error 이벤트 없음", "error" not in event_types, f"errors: {[e['data'] for e in events if e['event']=='error']}") check("result 이벤트 존재", "result" in event_types) # result HTML 검증 result_events = [e for e in events if e["event"] == "result"] if result_events: html = result_events[0]["data"] check("HTML 비어있지 않음", len(html) > 100, f"길이: {len(html)}") check("HTML에 slide 클래스", "slide" in html) check("HTML에 body 영역", "area-body" in html or "body_html" in html or "bg" in html) check("HTML에 sidebar 영역", "area-sidebar" in html or "sidebar" in html) check("HTML에 footer 영역", "area-footer" in html or "footer" in html) else: check("result HTML", False, "result 이벤트 없음") # 스냅샷 파일 확인 import glob latest_runs = sorted(glob.glob("data/runs/2026*"), reverse=True) if latest_runs: run_dir = latest_runs[0] files = [Path(f).name for f in glob.glob(f"{run_dir}/*.json")] print(f"\n 스냅샷 폴더: {Path(run_dir).name}") print(f" 저장된 파일: {files}") check("stage_0 스냅샷", "stage_0_context.json" in files) check("stage_1a 스냅샷", "stage_1a_context.json" in files) check("stage_1b 스냅샷", "stage_1b_context.json" in files) check("stage_1_5a 스냅샷", "stage_1_5a_context.json" in files) check("stage_1_7 스냅샷", "stage_1_7_context.json" in files) check("stage_1_5b 스냅샷", "stage_1_5b_context.json" in files) check("stage_2 스냅샷", "stage_2_context.json" in files) check("stage_3 스냅샷", "stage_3_context.json" in files) check("stage_4 스냅샷", "stage_4_context.json" in files) check("final 스냅샷", "final_context.json" in files) check("final.html 저장", "final.html" in [Path(f).name for f in glob.glob(f"{run_dir}/*")]) else: check("스냅샷 폴더", False, "run 폴더 없음") print(f"\n{'═' * 55}") print(f" 전체 파이프라인 시뮬레이션: {passed} passed, {failed} failed") if failed == 0: print(" 전체 통과 ✅") else: print(f" ❌ {failed}개 실패") print(f"{'═' * 55}") return failed == 0 if __name__ == "__main__": success = asyncio.run(run_full_simulation()) sys.exit(0 if success else 1)