"""키 failover 가 **실제 소비 경로 전부에 도달하는가** (2026-08-01, A5 마무리). ## 무엇이 잘못됐었나 A5 로 `_completion` 을 binding-aware 로 만들었지만, production 소비자 4곳은 `_get_router().completion(...)` 을 **직접** 불러 그 helper 를 통째로 지나쳤다. 그러면 1차 키가 billing 으로 죽어도 **보조 키를 한 번도 안 써 보고** 예외가 그대로 올라간다. 직접 재현: 예외 그대로 전파: Err Billing hard limit has been reached. active_slot 최종: primary ← 전환이 아예 없었다 Router 빌드 이력: [('primary', 'KEY-A')] 우회하던 곳(전부 확인): app/services/fal_angle_helpers.py:89 select_and_recommend_angle app/services/fal_angle_helpers.py:152 select_final_best app/modules/pipeline/ref_image_pipeline.py:127 _call_gpt_lvm app/modules/pipeline/text_cleaner.py:122 _extract_pdf_bytes_llm `select_final_best` 는 휴면이 아니다 — `scene_variation_service.py:165` 가 후보가 2장 이상이면 무조건 부른다(fal.ai 플래그와 무관). ## 계약 Router 호출은 **`router_completion` 하나**만 쓴다. 규칙을 문서로 두면 또 새로 생기므로, 원시 Router 를 꺼내는 경로 자체를 없애고 잠근다. """ from __future__ import annotations import json from types import SimpleNamespace from unittest.mock import patch import pytest from app.core import openai_keys @pytest.fixture(autouse=True) def _reset_slot(): openai_keys.reset() yield openai_keys.reset() class _Billing(Exception): """1차 키 소진 — 키 수준 실패로 판정되는 모양.""" def __init__(self) -> None: super().__init__("Billing hard limit has been reached.") self.code = "billing_hard_limit_reached" self.status_code = 400 self.llm_provider = "openai" def _response(payload: str) -> SimpleNamespace: return SimpleNamespace(choices=[SimpleNamespace( message=SimpleNamespace(content=payload), finish_reason="stop")]) class _SlotRouter: """지어질 때의 키를 기억하는 가짜 Router — 1차 키로는 실패한다.""" def __init__(self, api_key: str, payload: str): self.api_key = api_key self._payload = payload self.calls = 0 def completion(self, **_kw): self.calls += 1 if self.api_key == "KEY-A": raise _Billing() return _response(self._payload) @pytest.fixture def failing_primary(monkeypatch): """1차 키는 billing 으로 죽고 보조 키는 살아 있는 상태를 만든다. ★fake 를 소비자와 같은 함수로 만들지 않는다 — 실제 `_build_router` 를 갈아 끼워 **production 코드가 스스로 Router 를 받아 가게** 두고, 그 Router 만 슬롯 키를 기억하는 가짜로 바꾼다. """ from app.modules.llm import llm_client as lc state = {"payload": "{}", "built": []} def _fake_build(): slot, key = openai_keys.active_slot_and_key() r = _SlotRouter(key, state["payload"]) state["built"].append((slot, key)) lc._router = r lc._binding = lc._RouterBinding(slot=slot, router=r) return r monkeypatch.setattr(lc, "_build_router", _fake_build) monkeypatch.setattr(lc, "_init_opik", lambda: None) monkeypatch.setattr(lc, "_router", None, raising=False) monkeypatch.setattr(lc, "_binding", None, raising=False) with patch.multiple("app.core.config.settings", openai_api_key="KEY-A", openai_api_key_secondary="KEY-B"): yield state def _assert_failed_over(state) -> None: """1차에서 죽고 보조로 넘어갔음을 **양쪽으로** 확인한다.""" assert openai_keys.active_slot() == "secondary" assert state["built"] == [("primary", "KEY-A"), ("secondary", "KEY-B")] # ── 실제 production 소비자 ──────────────────────────────────────────── def test_select_final_best_reaches_the_secondary_key(failing_primary): """★live 경로 — scene_variation_service 가 후보 2장 이상이면 무조건 탄다.""" from app.services.fal_angle_helpers import select_final_best failing_primary["payload"] = json.dumps( {"selected_index": 2, "reason": "SAMPLE"}) got = select_final_best([b"img-a", b"img-b"], "SAMPLE beat") assert got == 1 # 1-based 2 → 0-based 1 _assert_failed_over(failing_primary) def test_select_and_recommend_angle_reaches_the_secondary_key(failing_primary): from app.services.fal_angle_helpers import select_and_recommend_angle failing_primary["payload"] = json.dumps({ "best_for_angle": 0, "horizontal_angle": 40, "vertical_angle": 0, "zoom": 5, "reason": "SAMPLE"}) got = select_and_recommend_angle( [b"img-a", b"img-b"], "SAMPLE beat", "SAMPLE prompt") assert got is not None, "실패를 None 으로 삼키면 호출자가 결함을 못 본다" assert got["horizontal_angle"] == 40 _assert_failed_over(failing_primary) def test_call_gpt_lvm_reaches_the_secondary_key(failing_primary): from app.modules.pipeline.ref_image_pipeline import _call_gpt_lvm failing_primary["payload"] = json.dumps({"verdict": "SAMPLE"}) got = _call_gpt_lvm(b"img", "SAMPLE prompt", {"type": "object"}, schema_name="sample_fixture_schema") assert got == {"verdict": "SAMPLE"} _assert_failed_over(failing_primary) def test_pdf_extraction_reaches_the_secondary_key(failing_primary): from app.modules.pipeline.text_cleaner import _extract_pdf_bytes_llm failing_primary["payload"] = "SAMPLE extracted text" got = _extract_pdf_bytes_llm(b"%PDF-sample", "gemini-lite", "SAMPLE") assert got == "SAMPLE extracted text" _assert_failed_over(failing_primary) # ── 재발 방지 잠금 ──────────────────────────────────────────────────── # 규칙을 문서로만 두면 또 생긴다 — 실제로 네 번 생겼다. 원시 Router 를 꺼내는 # 경로 자체를 없애고, 없어졌다는 것을 테스트가 지킨다. def test_raw_router_accessor_is_gone(): """`_get_router` 가 남아 있으면 다음 사람이 또 그것을 쓴다.""" from app.modules.llm import llm_client as lc assert not hasattr(lc, "_get_router") def test_no_production_module_bypasses_the_failover_helper(): """`llm_client` 밖에서 Router 를 직접 꺼내 호출하는 곳이 없어야 한다.""" from pathlib import Path root = Path(__file__).resolve().parents[2] / "app" owner = root / "modules" / "llm" / "llm_client.py" offenders = [] for path in sorted(root.rglob("*.py")): if path == owner: continue text = path.read_text(encoding="utf-8") for lineno, line in enumerate(text.splitlines(), 1): if "_get_router" in line or "router.completion(" in line: offenders.append(f"{path.relative_to(root.parent)}:{lineno}") assert offenders == [], ( "Router 를 직접 꺼내 호출하면 키 슬롯 전환이 일어나지 않는다 — " f"`router_completion` 을 쓸 것: {offenders}") def test_router_completion_is_the_public_entry(): from app.modules.llm import llm_client as lc assert callable(lc.router_completion)