byeongwook.choiandClaude Fable 5.1
11ae3629b2
Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
247 lines
12 KiB
Python
247 lines
12 KiB
Python
"""중복 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()
|