"""A-1(#58) 기록 분류 — **나가는 것**에서 잰다. 2026-08-28 완주 주행 Opik 실측이 뿌리다: ⑤ `step:step:` 이중 접두사 span **6건** / 세 자리 (`shot_conti_light` 3 · `background_render` 2 · `floor_plan_render` 1) ⑥ `text_steps.py:33` 이 계보를 넘기는데 `extract_text_from_pdf_llm` 이 받아만 놓고 안 썼다 ★조립부를 손으로 부르고 「끝점에서 쟀다」고 하지 않는다. ⑥은 `router_completion` 이 **실제로 받은 metadata** 를 잡아서 본다. """ from __future__ import annotations import sys import types import pytest from app.modules.llm.image_tracer import resolve_step_name from app.modules.llm.opik_trace import STEP_TAG_PREFIX, build_axis_tags # ── ⑤ 태그에서 스텝 이름을 되찾을 때 접두사를 벗긴다 ────────────── def test_step_name_from_axis_tag_has_no_prefix(monkeypatch): """`step:shot_conti_light` 태그에서 `shot_conti_light` 를 얻는다.""" monkeypatch.setattr( "app.modules.llm.llm_client._get_thread_opik_meta", lambda: {"tags": build_axis_tags(step="shot_conti_light")}, raising=False) assert resolve_step_name(None, {}) == "shot_conti_light" def test_round_trip_never_doubles_the_prefix(monkeypatch): """되찾은 이름을 다시 태그로 만들어도 `step:step:` 이 안 생긴다. ★이게 실제로 깨졌던 자리다 — `image_tracer.py:209` 가 되찾은 이름을 그대로 `build_axis_tags(step=...)` 에 넣는다. """ for step in ("shot_conti_light", "background_render", "floor_plan_render"): monkeypatch.setattr( "app.modules.llm.llm_client._get_thread_opik_meta", lambda step=step: {"tags": build_axis_tags(step=step)}, raising=False) again = build_axis_tags(step=resolve_step_name(None, {})) assert again == [f"{STEP_TAG_PREFIX}{step}"], ( f"{step}: 이중 접두사가 다시 생겼다 → {again}") def test_step_axis_is_picked_by_name_not_by_position(monkeypatch): """축 순서에 기대지 않는다 — 앞에 다른 축이 와도 step 을 고른다. 종전 코드는 `tags[0]` 이라 `_AXIS_ORDER` 가 step 을 맨 앞에 두는 덕에 **우연히** 맞고 있었다. 순서가 바뀌면 모델명이 스텝 이름이 됐다. """ monkeypatch.setattr( "app.modules.llm.llm_client._get_thread_opik_meta", lambda: {"tags": ["model:gpt-image-2", "provider:openai", f"{STEP_TAG_PREFIX}background_render"]}, raising=False) assert resolve_step_name(None, {}) == "background_render" def test_capture_scope_stage_still_wins(monkeypatch): """capture scope 의 stage 가 있으면 그것이 먼저다 (기존 계약 보존).""" monkeypatch.setattr( "app.modules.llm.llm_client._get_thread_opik_meta", lambda: {"tags": [f"{STEP_TAG_PREFIX}background_render"]}, raising=False) assert resolve_step_name(None, {"stage": "scene_image_pipeline"}) \ == "scene_image_pipeline" def test_no_step_tag_falls_back(monkeypatch): """step 축이 없으면 fallback — 다른 축을 스텝 이름으로 쓰지 않는다.""" monkeypatch.setattr( "app.modules.llm.llm_client._get_thread_opik_meta", lambda: {"tags": ["model:gpt-image-2", "provider:openai"]}, raising=False) assert resolve_step_name("ref_image_gen", {}) == "ref_image_gen" assert resolve_step_name(None, {}) == "image_gen" # ── ⑥ PDF 추출이 호출자 계보를 버리지 않는다 ───────────────────── @pytest.fixture def captured_metadata(monkeypatch): """`router_completion` 이 **실제로 받은** metadata 를 잡는다.""" seen: list[dict] = [] def fake_router_completion(**kw): seen.append(dict(kw.get("metadata") or {})) msg = types.SimpleNamespace(content="추출된 본문") choice = types.SimpleNamespace(message=msg, finish_reason="stop") return types.SimpleNamespace(choices=[choice]) import app.modules.llm.llm_client as lc monkeypatch.setattr(lc, "router_completion", fake_router_completion) return seen def _fake_fitz(monkeypatch, pages: int): """`fitz.open(path)` 이 pages 쪽짜리 문서를 주는 것처럼.""" import app.modules.pipeline.text_cleaner as tc doc = types.SimpleNamespace(page_count=pages, close=lambda: None) monkeypatch.setattr(tc, "fitz", types.SimpleNamespace( open=lambda *a, **k: doc), raising=False) return doc def _run_extract(monkeypatch, tmp_path): import app.modules.pipeline.text_cleaner as tc pdf = tmp_path / "대본.pdf" pdf.write_bytes(b"%PDF-1.4 fake") _fake_fitz(monkeypatch, pages=1) monkeypatch.setattr(tc, "_resolve_model", lambda *a, **k: "gemini-lite") return tc.extract_text_from_pdf_llm(str(pdf), None) def test_pdf_extract_lineage_rides_the_thread_local(monkeypatch, captured_metadata, tmp_path): """★계보는 **thread-local 로** 실린다 — 인자로 넘기지 않는다. `StepRunner._execute_and_finalize`(`step_runner.py:1599`)가 스텝 실행 직전에 `set_opik_context(self.build_opik_metadata())` 를 부른다. 이 시험은 그 상태를 그대로 만들고, 나가는 호출에 계보가 실리는지 본다. """ from app.modules.llm.llm_client import set_opik_context # ★태그는 **프로덕션 모양**으로 준다 — `build_opik_metadata` # (`step_runner.py:2144`)가 `build_axis_tags(step=...)` 로 축 접두사를 # 붙여 보낸다. 맨 이름은 `_build_opik_metadata` 가 일부러 버린다 # (`llm_client.py:181-184`, 카디널리티 방어). step_tag = f"{STEP_TAG_PREFIX}text_cleanup" lineage = {"project_id": "P1", "episode_id": "E1", "thread_id": "episode:E1", "tags": [step_tag]} set_opik_context(lineage) try: assert _run_extract(monkeypatch, tmp_path) == "추출된 본문" finally: set_opik_context(None) assert captured_metadata, "router_completion 이 한 번도 안 불렸다" opik = captured_metadata[0].get("opik") assert isinstance(opik, dict), ( f"★litellm 은 metadata['opik'] 만 읽는다 — 최상위에 두면 안 닿는다. " f"받은 것: {sorted(captured_metadata[0])}") for key in ("project_id", "episode_id", "thread_id"): assert opik.get(key) == lineage[key], ( f"계보 {key} 가 metadata['opik'] 에 없다 — 있는 것: {sorted(opik)}") assert step_tag in (opik.get("tags") or []), ( f"스텝 축 태그가 안 실렸다 — {opik.get('tags')}") def test_pdf_extract_names_the_op_even_without_a_step(monkeypatch, captured_metadata, tmp_path): """thread-local 이 비어도 op 이름은 남는다 (호출이 이름 없이 안 나간다).""" from app.modules.llm.llm_client import set_opik_context set_opik_context(None) _run_extract(monkeypatch, tmp_path) opik = (captured_metadata[0] or {}).get("opik") or {} assert "op:pdf_text_extract" in (opik.get("tags") or []), ( f"op 축 이름이 없다 — {opik}") def test_pdf_extract_takes_no_lineage_argument(): """★받는데 안 쓰는 인자를 되살리지 못하게 한다. 2026-08-28: 이 함수에 `opik_metadata` 인자가 **있었는데 본문에서 한 번도 안 썼다.** 그래서 「호출자가 넘기는데 함수가 버린다」는 결함으로 읽혔고 나는 그것을 고친다며 계보를 **최상위**에 붙였다 — litellm 이 안 읽는 층이라 아무 데도 안 닿았고, 내 시험은 최상위를 봐서 초록이었다(Codex BLOCK). 되살리려면 `metadata["opik"]` 중첩 층에 넣어야 한다. 그 전까지는 인자를 두지 않는다. """ import inspect import app.modules.pipeline.text_cleaner as tc params = inspect.signature(tc.extract_text_from_pdf_llm).parameters assert "opik_metadata" not in params, ( "인자를 되살렸다면 metadata['opik'] 중첩 층에 넣었는지 확인하고 " "이 시험을 그 계약으로 바꿔라 — 최상위에 붙이면 안 닿는다") if __name__ == "__main__": # pragma: no cover sys.exit(pytest.main([__file__, "-q"]))