"""summarize.runner — 프로그램 요약 + unit 조각 추출의 상태·실패 사유·실행로그 라이프사이클. - 프로그램 요약이 먼저 생성되고 program.summary_json 에 저장된다 - 실패 unit → summary_error 에 예외명+메시지, 재실행 시 실패분만 재시도 - 조각은 logic_chunk / chunk_fts 에 저장되고 unit.chunk_count 에 반영된다 - --fake 는 FakeLLM 으로 파이프라인 전체(조각 1개/unit)를 통과한다 """ from __future__ import annotations import pytest from fastapi.testclient import TestClient from config.settings import settings from index.db import connect, loads import summarize.runner as runner PAYLOAD = { "MAIN_PROGRAM": "ZRUNNER_T1", "DESCRIPTION": "러너 테스트", "TEXT_SYMBOL": [{"SYMBOL": "001", "TEXT": "회사코드 조회"}], "INCLUDE_PROGRAM": [ { "INCLUDE": "ZRUNNER_T1", "SOURCE_CODE": ( "REPORT zrunner_t1.\n" "START-OF-SELECTION.\n" " PERFORM ok_one.\n" " PERFORM fail_me.\n" "FORM ok_one.\n" " SELECT * FROM t001 INTO TABLE gt_t001.\n" " SORT gt_t001 BY bukrs.\n" "ENDFORM.\n" "FORM fail_me.\n" " WRITE 2.\n" "ENDFORM." ), } ], } class StubLLM: """FAIL_ME unit 에서만 예외. ok_one 에는 SELECT 조각 하나를 (줄 번호를 틀리게) 돌려준다.""" def __init__(self, fail_names: set[str]): self.fail_names = fail_names self.usage = {"calls": 0, "prompt_tokens": 0, "completion_tokens": 0, "cost_usd": 0.0} self.prompts: list[str] = [] def complete_json(self, system: str, user: str) -> dict: self.prompts.append(user) for name in self.fail_names: if f"name: {name}" in user: raise RuntimeError("HTTP Error 429: rate limited") self.usage["calls"] += 1 self.usage["prompt_tokens"] += 10 self.usage["completion_tokens"] += 20 if "[작업] program_summary" in user: return {"program": "x", "business_purpose_ko": "회사코드 마스터 조회 테스트", "main_flow": ["회사코드 조회", "출력"], "business_tags": ["마스터관리"], "confidence": 0.7} if "name: OK_ONE" in user: return {"unit_purpose_ko": "회사코드 마스터를 읽는다", "chunks": [{"line_start": 7, "line_end": 8, # 실제는 6~7 — 앵커로 보정되어야 함 "first_line": "SELECT * FROM t001 INTO TABLE gt_t001.", "kind": "sql_select", "purpose_ko": "회사코드 마스터(T001) 전체 조회", "purpose_en": "Read company code master", "keywords_ko": ["회사코드"], "keywords_en": ["company code"], "sap_objects": ["T001"], "confidence": 0.9}]} return {"unit_purpose_ko": "요약", "chunks": []} @pytest.fixture() def ingested(tmp_path, monkeypatch): import query.api as api monkeypatch.setattr(settings, "database_url", f"sqlite:///{tmp_path / 'index.db'}") monkeypatch.setattr(settings, "data_normalized", tmp_path / "normalized") monkeypatch.setattr(settings, "data_parsed", tmp_path / "parsed") monkeypatch.setattr(settings, "llm_base_url", "") # ingest 의 자동 요약은 끔 c = TestClient(api.app) assert c.post("/ingest", json=PAYLOAD).json()["status"] == "loaded" return c def _unit(con, name: str): return con.execute( "SELECT summary_status, summary_error, chunk_count, summary_json FROM unit " "WHERE program='ZRUNNER_T1' AND name=?", (name,), ).fetchone() def test_program_summary_chunks_failure_and_retry(ingested, monkeypatch): stub = StubLLM({"FAIL_ME"}) monkeypatch.setattr(runner, "create_llm", lambda fake=False, **kw: stub) stats = runner.summarize_units("ZRUNNER_T1", None, fake=False, dry_run=False, trigger="test") assert stats["program_summaries"] == 1 assert stats["failed"] == 1 assert stats["done"] == 2 # OK_ONE + START-OF-SELECTION assert stats["chunks"] == 1 # 프롬프트에 프로그램 요약 문맥과 텍스트 심볼·파서 힌트가 들어갔는가 unit_prompt = next(p for p in stub.prompts if "name: OK_ONE" in p) assert "회사코드 마스터 조회 테스트" in unit_prompt # 프로그램 요약 문맥 assert "L6 SELECT" in unit_prompt # 파서 힌트 assert " 6| " in unit_prompt # 줄번호 붙은 코드 con = connect() try: p = con.execute("SELECT summary_status, summary_json FROM program WHERE name='ZRUNNER_T1'").fetchone() assert p["summary_status"] == "done" assert loads(p["summary_json"])["prompt_version"] == runner.PROMPT_VERSION fail = _unit(con, "FAIL_ME") assert fail["summary_status"] == "failed" assert "RuntimeError: HTTP Error 429" in fail["summary_error"] ok = _unit(con, "OK_ONE") assert ok["summary_status"] == "done" and ok["summary_error"] is None assert ok["chunk_count"] == 1 thin = loads(ok["summary_json"]) assert thin["purpose_ko"] == "회사코드 마스터를 읽는다" and thin["chunk_count"] == 1 ch = con.execute("SELECT * FROM logic_chunk WHERE program='ZRUNNER_T1'").fetchone() assert ch["chunk_id"].endswith("#FORM#OK_ONE#C1") assert (ch["line_start"], ch["line_end"]) == (6, 7) # 앵커로 보정됨 assert loads(ch["tables_read"]) == ["T001"] # 파서가 채움 assert ch["kind"] == "sql_select" # 검색 색인에 들어갔는가 hit = con.execute("SELECT chunk_id FROM chunk_fts WHERE chunk_fts MATCH '\"회사코드\"'").fetchone() assert hit and hit["chunk_id"] == ch["chunk_id"] # 얇은 unit 색인도 한 줄 요약으로 갱신 u = con.execute("SELECT purpose FROM unit_fts WHERE unit_id=?", (ch["unit_id"],)).fetchone() assert u["purpose"] == "회사코드 마스터를 읽는다" log = con.execute("SELECT * FROM llm_usage_log ORDER BY id DESC").fetchall() assert len(log) == 1 assert log[0]["status"] == "done" assert log[0]["total"] == stats["done"] + stats["failed"] assert log[0]["failed"] == 1 and log[0]["chunks"] == 1 assert log[0]["trigger_by"] == "test" finally: con.close() # 재실행(전부 성공) → 실패분만 재시도, 프로그램 요약은 스킵 stub2 = StubLLM(set()) monkeypatch.setattr(runner, "create_llm", lambda fake=False, **kw: stub2) stats2 = runner.summarize_units("ZRUNNER_T1", None, fake=False, dry_run=False, trigger="test") assert stats2["failed"] == 0 and stats2["done"] == 1 and stats2["program_summaries"] == 0 assert stats2["skipped"] == stats["done"] assert not any("[작업] program_summary" in p for p in stub2.prompts) con = connect() try: fail = _unit(con, "FAIL_ME") assert fail["summary_status"] == "done" and fail["summary_error"] is None and fail["chunk_count"] == 0 assert con.execute("SELECT COUNT(*) FROM llm_usage_log").fetchone()[0] == 2 finally: con.close() def test_all_skipped_writes_no_log(ingested, monkeypatch): monkeypatch.setattr(runner, "create_llm", lambda fake=False, **kw: StubLLM(set())) runner.summarize_units("ZRUNNER_T1", None, fake=False, dry_run=False, trigger="test") stats = runner.summarize_units("ZRUNNER_T1", None, fake=False, dry_run=False, trigger="test") assert stats["done"] == 0 and stats["failed"] == 0 and stats["program_summaries"] == 0 con = connect() try: # 두 번째 실행은 전부 스킵 — 로그 행이 추가되지 않는다 assert con.execute("SELECT COUNT(*) FROM llm_usage_log").fetchone()[0] == 1 finally: con.close() def test_fake_llm_pipeline(ingested): """--fake: FakeLLM 이 unit 마다 첫 실행문을 조각으로 만들어 파이프라인 전체를 통과한다.""" stats = runner.summarize_units("ZRUNNER_T1", None, fake=True, dry_run=False, trigger="test") assert stats["failed"] == 0 and stats["program_summaries"] == 1 assert stats["chunks"] >= 1 con = connect() try: n = con.execute("SELECT COUNT(*) FROM logic_chunk WHERE program='ZRUNNER_T1'").fetchone()[0] assert n == stats["chunks"] for r in con.execute("SELECT c.*, u.line_start us, u.line_end ue FROM logic_chunk c " "JOIN unit u ON u.unit_id=c.unit_id"): assert r["us"] <= r["line_start"] <= r["line_end"] <= r["ue"] finally: con.close() def test_reload_keeps_chunks_for_unchanged_units(ingested, monkeypatch): """소스가 바뀐 unit 의 조각만 버리고, 같은 unit 의 조각은 재적재 후에도 남는다.""" monkeypatch.setattr(runner, "create_llm", lambda fake=False, **kw: StubLLM(set())) runner.summarize_units("ZRUNNER_T1", None, fake=False, dry_run=False, trigger="test") changed = dict(PAYLOAD) changed["INCLUDE_PROGRAM"] = [dict(PAYLOAD["INCLUDE_PROGRAM"][0])] changed["INCLUDE_PROGRAM"][0]["SOURCE_CODE"] = PAYLOAD["INCLUDE_PROGRAM"][0]["SOURCE_CODE"].replace( "WRITE 2.", "WRITE 3.") # FAIL_ME 만 바뀜 assert ingested.post("/ingest", json=changed).json()["status"] == "loaded" con = connect() try: assert con.execute("SELECT COUNT(*) FROM logic_chunk WHERE unit_id LIKE '%#OK_ONE'").fetchone()[0] == 1 assert _unit(con, "OK_ONE")["summary_status"] == "done" assert _unit(con, "FAIL_ME")["summary_status"] == "none" # 재추출 대상 assert con.execute("SELECT summary_status FROM program WHERE name='ZRUNNER_T1'").fetchone()[0] == "stale" finally: con.close()