"""i2i 직접 호출도 DB(llm_call_log)에 남아야 한다. 실측(2026-08-23): Opik 에는 i2i_edit trace 가 있는데 llm_call_log 에는 step_name/operation_type 에 i2i 가 든 행이 0건이었다. 한쪽에만 남으면 '이 호출이 무엇을 만들었나'를 DB 로 못 묻는다. ★이 시험이 **측정 도구이기도 하다.** record_provider_call 은 to_db=True 가 기본이라 일반 환경에서 그냥 부르면 프로덕션 llm_call_log·Opik 에 진짜 행이 남는다 — 이 문서가 고치려는 바로 그 오염을 진단이 반복하는 꼴이 된다. pytest 는 conftest 가 시험 DB·시험 Opik 프로젝트로 갈아입혀 주므로 여기서 재는 것이 가장 안전하다. """ def test_record_provider_call_passes_step_and_operation(monkeypatch): """헬퍼가 DB 에 무엇을 넘기는지 — 이것이 멀쩡하면 원인은 호출 자리다.""" seen = {} def _fake_log(**kw): seen.update(kw) return "call-1" import app.modules.llm.llm_logger as llm_logger monkeypatch.setattr(llm_logger, "log_llm_call", _fake_log) from app.modules.llm.image_tracer import record_provider_call cid = record_provider_call( step="i2i_edit", model="gemini-3.1-flash-image-preview", prompt="p", status="success", duration_ms=12, meta={"project_id": "p1", "episode_id": "e1"}, operation="i2i_edit", output_text="[image generated]") assert cid == "call-1" assert seen["step_name"] == "i2i_edit" assert seen["operation_type"] == "i2i_edit" assert seen["project_id"] == "p1" def test_db_failure_does_not_lose_opik_record(monkeypatch, caplog): """DB 가 터져도 Opik 기록은 남아야 한다 — 둘이 함께 죽으면 안 된다. ★**실제 실패 경계를 태운다** (2026-08-24 Codex BLOCK). 앞 판은 `log_llm_call` 통째로를 예외 던지는 대역으로 바꿔, 그 함수가 **내부에서 예외를 잡고도 id 를 돌려주던** 진짜 경계를 안 봤다. 여기서는 DB 세션이 터지게 해서 실제 경로를 지나게 한다. None 반환·Opik 1회 기록·warning **셋 다** 본다. """ from app.core import database from app.modules.llm import image_tracer class _DeadSession: def add(self, *a, **kw): raise RuntimeError("DB 죽음") def commit(self): raise RuntimeError("DB 죽음") def close(self): pass monkeypatch.setattr(database, "SessionLocal", lambda: _DeadSession()) logged = [] class _T: def log(self, **kw): logged.append(kw) monkeypatch.setattr(image_tracer, "get_image_tracer", lambda: _T()) cid = image_tracer.record_provider_call( step="i2i_edit", model="m", prompt="p", status="success", duration_ms=1, meta={}, operation="i2i_edit") assert cid is None, ( "DB 저장이 실패했는데 id 를 돌려줬다 — 그 값이 자산의 " "generation_call_id 로 붙으면 없는 행을 가리킨다") assert len(logged) == 1, "DB 가 죽었다고 Opik 기록까지 잃으면 안 된다" # ★실제로 warning 이 나오는지까지 본다 (2026-08-24 Codex 비차단 메모). # 「warning 을 본다」고 적어 놓고 레벨을 안 재면, 조용히 debug 로 # 되돌아가도 이 시험은 초록이다 — 그것이 이 결함을 숨겼던 방식이다. assert any(r.levelname == "WARNING" and "LLM call log failed" in r.message for r in caplog.records), \ f"DB 기록 실패가 warning 으로 안 보인다: {[r.levelname for r in caplog.records]}" def test_log_llm_call_returns_none_when_the_row_is_not_saved(monkeypatch): """★계약 자체를 못박는다 — 저장 실패면 None.""" from app.core import database from app.modules.llm.llm_logger import log_llm_call class _DeadSession: def add(self, *a, **kw): raise RuntimeError("DB 죽음") def commit(self): raise RuntimeError("DB 죽음") def close(self): pass monkeypatch.setattr(database, "SessionLocal", lambda: _DeadSession()) assert log_llm_call(model_name="m", user_prompt="p", status="success") is None def test_log_llm_call_really_inserts_the_row(): """★성공 경로는 **시험 DB 에 실제로 넣고 되읽어** 확인한다. `_fake_log(**kw)` 대역은 무엇이든 받아 주므로 인자 이름이 어긋나도 모른다. 실제 insert 는 그 사각을 함께 닫는다. (conftest 가 시험 DB 로 갈아입힌다 — 프로덕션에 안 쓴다.) """ from app.core.database import SessionLocal, engine from app.models.project import LLMCallLog from app.modules.llm.llm_logger import log_llm_call # ★테이블을 스스로 보장한다 — 같은 걷기에서 safe_drop_all 을 쓰는 시험 # 뒤에 돌면 이 표가 없어 순서에 따라 결과가 달라진다(실측). LLMCallLog.__table__.create(bind=engine, checkfirst=True) cid = log_llm_call( model_name="probe-model", user_prompt="확인", status="success", duration_ms=7, project_id="p-probe", episode_id="e-probe", operation_type="i2i_edit", step_name="i2i_edit", output_text="[image generated]", metadata={"still_id": "s-probe"}) assert cid, "성공 경로가 id 를 안 준다" db = SessionLocal() try: row = db.query(LLMCallLog).filter(LLMCallLog.id == cid).first() assert row is not None, "id 는 줬는데 행이 없다" assert row.step_name == "i2i_edit" assert row.operation_type == "i2i_edit" assert row.project_id == "p-probe" assert row.model_name == "probe-model" finally: db.close() def test_db_failure_is_not_silent(): """★DB 기록 실패는 **보여야** 한다 — debug 로 삼키면 아무도 모른다. 조용히 삼키는 것이 「i2i 가 DB 에 없다」는 조사를 넉 달 뒤에 부르게 했다. (그때 본 것은 실은 시험 기록이었다 — 아래 시험 참조.) """ import inspect from app.modules.llm import image_tracer src = inspect.getsource(image_tracer.record_provider_call) db_block = src.split("llm_call_log 기록 실패")[0][-400:] assert "logger.warning" in src.split("Opik 기록 실패")[0], \ "DB 기록 실패를 아직 debug 로 삼킨다" assert "logger.debug" in src.split("llm_call_log 기록 실패")[1], \ "Opik 기록 실패까지 warning 이면 자체 호스팅이 죽을 때 로그가 덮인다" assert db_block # 자리 확인용 def test_the_i2i_gap_was_test_records_not_a_production_defect(): """★★계획의 전제가 틀렸다 — 이 사실을 여기 못박는다 (2026-08-24 실측). 계획 Task 16 은 「Opik 에는 i2i_edit trace 가 있는데 llm_call_log 에는 0건」을 프로덕션 결함으로 읽었다. 실제로 재 보니 **Opik 의 i2i trace 84건이 전부 시험 기록**이었다: · project_id="p-i2i" 21건 — test_i2i_capture 의 가짜 신원 · project_id=None · duration_ms=0 · model="m" 63건 — scope 밖 호출 즉 프로덕션에서 그 경로가 안 돌았고, DB 0건이 **정상**이다. 고칠 결함이 아니었다. ★이것은 단계 1 이 고친 바로 그 문제의 결과다 — 시험이 프로덕션 Opik 에 쓰고 있었고, 그 기록을 보고 프로덕션 결함이라고 판단했다. **오염된 자료로는 진단도 오염된다.** 이 시험은 그 두 갈래를 감사 필터가 실제로 거르는지 본다. 나중에 진짜 프로덕션 i2i 기록이 생기면(신원 있고 시간 > 0) 안 걸러진다. """ import sys from pathlib import Path sys.path.insert(0, str(Path(__file__).resolve().parents[3])) from tools.opik_prompt_audit.audit.fetch import is_test_origin fake_identity = {"metadata": {"step": "i2i_edit", "duration_ms": 0, "project_id": "p-i2i"}} no_identity = {"metadata": {"step": "i2i_edit", "duration_ms": 0}} real_call = {"metadata": {"step": "i2i_edit", "duration_ms": 1840, "project_id": "5bddbdfc-2681-42a6-9837-43f35f60049d"}} assert is_test_origin(fake_identity) is True assert is_test_origin(no_identity) is True assert is_test_origin(real_call) is False, \ "진짜 프로덕션 i2i 기록까지 거르면 결함을 영영 못 본다"