"""중복 unit 조각 복제(수정사항 6번)와 병렬 실행(10번), 요약→색인 반영(1번)·텍스트심볼 색인(3번). _BAK / _COPY 관행 때문에 같은 코드가 여러 프로그램에 그대로 들어 있다. code_hash 가 같으면 대표 1건만 LLM 에 보내고 나머지는 조각을 평행이동해 복제한다. """ from __future__ import annotations import threading import pytest from fastapi.testclient import TestClient from config.settings import settings from index.db import connect, loads import summarize.runner as runner BODY = ( "FORM select_t001.\n" " SELECT * FROM t001 INTO TABLE gt_t001.\n" " SORT gt_t001 BY bukrs.\n" "ENDFORM.\n" ) # 같은 FORM 을 서로 다른 줄 위치에 둔 두 프로그램 — 복제 시 줄 평행이동이 필요하다 ORIG = { "MAIN_PROGRAM": "ZDUP_A", "DESCRIPTION": "원본", "TEXT_SYMBOL": [{"SYMBOL": "001", "TEXT": "회사코드 목록 조회"}], "INCLUDE_PROGRAM": [{"INCLUDE": "ZDUP_A", "SOURCE_CODE": "REPORT zdup_a.\n" + BODY}], } COPY = { "MAIN_PROGRAM": "ZDUP_A_COPY", "DESCRIPTION": "복사본", "INCLUDE_PROGRAM": [{ "INCLUDE": "ZDUP_A_COPY", # 앞에 주석 3줄을 더 넣어 FORM 시작 줄을 밀어낸다 "SOURCE_CODE": "REPORT zdup_a_copy.\n* c1\n* c2\n* c3\n" + BODY, }], } class CountingLLM: """호출 수와 호출 스레드를 기록하는 스텁.""" def __init__(self) -> None: self.usage = {"calls": 0, "prompt_tokens": 0, "completion_tokens": 0, "cost_usd": 0.0} self.unit_calls: list[str] = [] self.threads: set[str] = set() self._lock = threading.Lock() def complete_json(self, system: str, user: str) -> dict: with self._lock: self.usage["calls"] += 1 self.threads.add(threading.current_thread().name) if "[작업] program_summary" in user: return {"program": "x", "business_purpose_ko": "회사코드 마스터를 조회하는 테스트 리포트", "main_flow": ["회사코드 조회"], "business_tags": ["마스터관리"], "keywords_ko": ["회사코드", "마스터"], "keywords_en": ["company code"], "sap_module": "FI", "confidence": 0.8} with self._lock: self.unit_calls.append(user) if "name: SELECT_T001" in user: return {"unit_purpose_ko": "회사코드 마스터를 읽는다", "chunks": [{"line_start": 2, "line_end": 3, "first_line": "SELECT * FROM t001 INTO TABLE gt_t001.", "kind": "sql_select", "purpose_ko": "회사코드 마스터(T001) 전체 조회", "keywords_ko": ["회사코드"], "keywords_en": ["company code"], "sap_objects": ["T001"], "confidence": 0.9}]} return {"unit_purpose_ko": "요약", "chunks": []} @pytest.fixture() def two_programs(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", "") c = TestClient(api.app) for payload in (ORIG, COPY): assert c.post("/ingest", json=payload).json()["status"] == "loaded" return c def _chunks(con, program: str): return con.execute( "SELECT unit_id, line_start, line_end, purpose_ko, code_hash FROM logic_chunk " "WHERE program=? ORDER BY seq", (program,) ).fetchall() def test_duplicate_units_share_one_llm_call(two_programs, monkeypatch): llm = CountingLLM() monkeypatch.setattr(runner, "create_llm", lambda fake=False, **kw: llm) stats = runner.summarize_units(None, None, trigger="test") # SELECT_T001 은 두 프로그램에 동일 코드로 있다 → unit 추출 호출은 1회뿐 select_prompts = [u for u in llm.unit_calls if "name: SELECT_T001" in u] assert len(select_prompts) == 1, "같은 code_hash 는 대표 1건만 호출해야 한다" assert stats["cloned"] == 1 assert stats["llm_calls_saved"] == 1 con = connect() try: a, b = _chunks(con, "ZDUP_A"), _chunks(con, "ZDUP_A_COPY") assert len(a) == 1 and len(b) == 1 # 코드가 같으므로 조각 해시와 설명은 같고, 줄 번호는 주석 3줄만큼 밀려 있어야 한다 assert a[0]["code_hash"] == b[0]["code_hash"] assert a[0]["purpose_ko"] == b[0]["purpose_ko"] assert b[0]["line_start"] - a[0]["line_start"] == 3 assert b[0]["line_end"] - a[0]["line_end"] == 3 # 복제 사실을 unit 요약에 남긴다 tgt = con.execute("SELECT summary_json, chunk_count, summary_status FROM unit " "WHERE program='ZDUP_A_COPY' AND name='SELECT_T001'").fetchone() assert tgt["summary_status"] == "done" and tgt["chunk_count"] == 1 assert loads(tgt["summary_json"])["cloned_from"].startswith("ZDUP_A#") # 복제된 조각도 검색 가능해야 한다 n = con.execute("SELECT COUNT(*) c FROM chunk_fts WHERE program='ZDUP_A_COPY'").fetchone()["c"] assert n == 1 finally: con.close() def test_cloned_chunk_lines_point_at_same_code(two_programs, monkeypatch): """복제된 줄 범위가 실제로 같은 코드를 가리키는지 원문으로 확인한다.""" monkeypatch.setattr(runner, "create_llm", lambda fake=False, **kw: CountingLLM()) runner.summarize_units(None, None, trigger="test") con = connect() try: out = {} for prog in ("ZDUP_A", "ZDUP_A_COPY"): c = _chunks(con, prog)[0] code = con.execute("SELECT code FROM include WHERE program=? AND include=?", (prog, prog)).fetchone()["code"].split("\n") out[prog] = "\n".join(code[c["line_start"] - 1 : c["line_end"]]) assert out["ZDUP_A"] == out["ZDUP_A_COPY"] assert "SELECT * FROM t001" in out["ZDUP_A"] finally: con.close() def test_no_dedupe_flag_calls_each_unit(two_programs, monkeypatch): llm = CountingLLM() monkeypatch.setattr(runner, "create_llm", lambda fake=False, **kw: llm) stats = runner.summarize_units(None, None, trigger="test", dedupe=False) assert len([u for u in llm.unit_calls if "name: SELECT_T001" in u]) == 2 assert stats["cloned"] == 0 def test_parallel_and_serial_give_same_result(two_programs, monkeypatch): llm = CountingLLM() monkeypatch.setattr(runner, "create_llm", lambda fake=False, **kw: llm) stats = runner.summarize_units(None, None, trigger="test", concurrency=4) assert stats["failed"] == 0 and stats["chunks"] >= 1 # 동시성 4로 돌렸으면 워커 스레드에서 호출돼야 한다 (unit 이 2개 이상일 때) assert llm.threads, "호출 스레드가 기록돼야 한다" def test_summary_and_text_symbol_reach_program_index(two_programs, monkeypatch): """수정사항 1·3 — 요약 문장과 텍스트 심볼 한국어로 프로그램이 검색돼야 한다.""" monkeypatch.setattr(runner, "create_llm", lambda fake=False, **kw: CountingLLM()) runner.summarize_units(None, None, trigger="test") from query import tools by_purpose = [r["program"] for r in tools.search_programs("마스터를 조회하는 테스트", top_k=5)] assert "ZDUP_A" in by_purpose, "요약 본문이 program_fts 에 반영돼야 한다" by_tag = [r["program"] for r in tools.search_programs("마스터관리", top_k=5)] assert "ZDUP_A" in by_tag, "business_tags 가 색인돼야 한다" by_text_symbol = [r["program"] for r in tools.search_programs("회사코드 목록 조회", top_k=5)] assert "ZDUP_A" in by_text_symbol, "텍스트 심볼 한국어가 색인돼야 한다" def test_long_summary_does_not_outrank_title_match(tmp_path, monkeypatch): """회귀 방지 — 요약을 붙인 프로그램이 타이틀만 있는 프로그램에 밀리면 안 된다. bm25 는 행 전체 길이로 정규화한다. 짧은 타이틀과 긴 요약을 한 FTS 행에 넣었더니 타이틀이 정확히 일치하는 프로그램이 1위 → 33위로 밀렸다. 그래서 색인을 program_fts(이름·타이틀) / program_desc_fts(서술) 로 분리했다. """ import query.api as api from index.loader import refresh_program_fts from query import tools monkeypatch.setattr(settings, "database_url", f"sqlite:///{tmp_path / 'i.db'}") monkeypatch.setattr(settings, "data_normalized", tmp_path / "n") monkeypatch.setattr(settings, "data_parsed", tmp_path / "p") monkeypatch.setattr(settings, "llm_base_url", "") c = TestClient(api.app) src = "REPORT z.\nSTART-OF-SELECTION.\n WRITE 1." # 타이틀이 정확히 일치하는 프로그램 (요약 있음, 매우 김) c.post("/ingest", json={"MAIN_PROGRAM": "ZTITLE_HIT", "DESCRIPTION": "총계정원장 조회", "INCLUDE_PROGRAM": [{"INCLUDE": "ZTITLE_HIT", "SOURCE_CODE": src}]}) # 타이틀만 있는 경쟁 프로그램들 (요약 없음, 매우 짧음) for i in range(4): c.post("/ingest", json={"MAIN_PROGRAM": f"ZSHORT{i}", "DESCRIPTION": "총계정원장(월별)", "INCLUDE_PROGRAM": [{"INCLUDE": f"ZSHORT{i}", "SOURCE_CODE": src}]}) con = connect() try: long_summary = { "program": "ZTITLE_HIT", "business_purpose_ko": "회사코드와 회계기간 조건으로 " * 30, "main_flow": ["단계 " + "설명 " * 20] * 6, "keywords_ko": [f"키워드{i}" for i in range(40)], "business_tags": ["총계정원장", "계정잔액"], "sap_module": "FI", "prompt_version": 2, } con.execute("UPDATE program SET summary_json=?, summary_status='done' WHERE name='ZTITLE_HIT'", (__import__("json").dumps(long_summary, ensure_ascii=False),)) refresh_program_fts(con) con.commit() finally: con.close() found = [r["program"] for r in tools.search_programs("총계정원장 조회하는 프로그램", top_k=5)] assert "ZTITLE_HIT" in found, f"긴 요약 때문에 타이틀 일치가 밀렸다: {found}" assert found[0] == "ZTITLE_HIT", f"타이틀 정확 일치가 1위여야 한다: {found}" def test_unit_fts_purpose_has_no_decoration(tmp_path, monkeypatch): """수정사항 1 — 적재 시점 unit_fts.purpose 에 '----' 장식이 들어가면 안 된다.""" import query.api as api monkeypatch.setattr(settings, "database_url", f"sqlite:///{tmp_path / 'i.db'}") monkeypatch.setattr(settings, "data_normalized", tmp_path / "n") monkeypatch.setattr(settings, "data_parsed", tmp_path / "p") monkeypatch.setattr(settings, "llm_base_url", "") c = TestClient(api.app) c.post("/ingest", json={ "MAIN_PROGRAM": "ZDECO", "INCLUDE_PROGRAM": [{"INCLUDE": "ZDECO", "SOURCE_CODE": "REPORT zdeco.\n" "*&---------------------------------------------------------------------*\n" "*& Form BUILD_LIST\n" "*&---------------------------------------------------------------------*\n" "FORM build_list.\n WRITE 1.\nENDFORM."}], }) con = connect() try: rows = [r["purpose"] for r in con.execute("SELECT purpose FROM unit_fts WHERE program='ZDECO'")] assert not any("----" in (p or "") for p in rows), rows assert any("BUILD_LIST" in (p or "") for p in rows), rows finally: con.close()