"""`litellm.completion` 직접 호출 경로의 키 슬롯 도달 (2026-08-01).

## 무엇이 잘못됐었나

Router 경로를 다 고친 뒤에도 **더 큰 구멍**이 남아 있었다. production 10개
파일이 `litellm.completion(...)` 을 직접 부르는데(13 호출) 전부 기본 모델이
`openai/gpt-5.6-sol` 이고, **`api_key=` 를 하나도 넘기지 않는다.** 그러면
litellm 이 `os.environ["OPENAI_API_KEY"]` 로 키를 정한다.

그런데 슬롯 브로커(`app/core/openai_keys.py`)는 그 환경변수를 **한 번도
건드리지 않는다**(`environ` 참조 0건). 게다가
`dwelling_zone_map_provider._require_api_key` 는 그 변수를
`settings.openai_api_key` = **1차 키로 고정**한다. 즉 보조 슬롯으로 전환돼도
이 13곳은 계속 죽은 1차 키를 쓴다.

환경변수를 동기화하는 것만으로는 부족하다 — **이 경로들은 `failover_on` 을
부르지 않아 전환이 일어날 계기 자체가 없다.** 그래서 호출을 감싸 ①활성 슬롯의
키를 명시 전달하고 ②키 수준 실패면 다음 슬롯으로 재시도해야 한다.

## 계약

`litellm.completion` 직접 호출은 **`llm_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 _SleepProbe:
    """그 모듈이 부르는 `time.sleep` 만 잰다 — 배경 스레드는 진짜로 잔다.

    ★`monkeypatch.setattr("time.sleep", …)` 도 `setattr(mod.time, "sleep", …)`
     도 **`time` 모듈 자체**를 바꾼다. 전체 시험을 돌리면 litellm/Opik 의
     배경 스레드가 주기적으로 자면서 그 sleep 까지 이 목록에 들어와,
     `sleeps == []` 같은 단언이 간헐적으로 깨졌다(2026-08-27 실측:
     `[0.1, 0.1]`). 모듈의 `time` **이름**만 대리물로 바꾸면 그 모듈의
     소비만 잡힌다.
    """

    def __init__(self, sink):
        self._sink = sink

    def sleep(self, seconds):
        self._sink.append(seconds)

    def __getattr__(self, name):
        import time as _real_time
        return getattr(_real_time, name)


class _Billing(Exception):
    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")])


@pytest.fixture
def failing_primary(monkeypatch):
    """1차 키는 billing 으로 죽고 보조 키는 살아 있는 상태.

    ★가짜는 `litellm.completion` **하나만** 갈아 끼운다 — 키를 어떻게 정하는지는
    production 코드가 스스로 하게 두어야 그 계약을 실제로 시험한다.
    """
    import litellm

    state = {"payload": "{}", "keys": []}

    def _fake_completion(**kw):
        key = kw.get("api_key")
        state["keys"].append(key)
        if key == "KEY-A":
            raise _Billing()
        return _response(state["payload"])

    monkeypatch.setattr(litellm, "completion", _fake_completion)
    monkeypatch.setenv("OPENAI_API_KEY", "ENV-STALE-KEY")
    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:
    assert state["keys"] == ["KEY-A", "KEY-B"], (
        "활성 슬롯 키를 명시 전달하고 실패 시 다음 슬롯으로 넘어가야 한다 "
        f"— 실제 사용 키: {state['keys']}")
    assert openai_keys.active_slot() == "secondary"


# ── 공용 진입점 ──────────────────────────────────────────────────────

def test_llm_completion_retries_on_the_next_slot(failing_primary):
    from app.core.openai_keys import llm_completion

    failing_primary["payload"] = json.dumps({"ok": True})
    resp = llm_completion(model="openai/gpt-5.6-sol", messages=[], num_retries=0)
    assert json.loads(resp.choices[0].message.content) == {"ok": True}
    _assert_failed_over(failing_primary)


def test_llm_completion_does_not_touch_non_openai_models(failing_primary):
    """Gemini 호출에 OpenAI 키를 끼워 넣지 않는다 — provider 는 litellm 이 판정."""
    from app.core.openai_keys import llm_completion

    failing_primary["payload"] = "SAMPLE"
    llm_completion(model="gemini/gemini-3.1-pro-preview", messages=[])
    assert failing_primary["keys"] == [None]
    assert openai_keys.active_slot() == "primary"


def test_llm_completion_ignores_the_stale_environment_key(failing_primary):
    """환경변수의 옛 키가 아니라 **활성 슬롯**의 키를 쓴다."""
    from app.core.openai_keys import llm_completion

    llm_completion(model="openai/gpt-5.6-sol", messages=[])
    assert "ENV-STALE-KEY" not in failing_primary["keys"]


def test_llm_completion_raises_when_all_slots_are_exhausted(monkeypatch):
    """보조 키가 없으면 조용히 삼키지 않고 원래 예외를 올린다."""
    import litellm

    from app.core.openai_keys import llm_completion

    def _always_billing(**_kw):
        raise _Billing()

    monkeypatch.setattr(litellm, "completion", _always_billing)
    with patch.multiple("app.core.config.settings",
                        openai_api_key="KEY-A", openai_api_key_secondary=""):
        with pytest.raises(_Billing):
            llm_completion(model="openai/gpt-5.6-sol", messages=[])


# ── 실제 production 소비자 ────────────────────────────────────────────

def test_edge_judge_provider_reaches_the_secondary_key(failing_primary):
    """★실제 소비자 — 1차 키가 죽어도 판정이 완주한다."""
    from app.modules.pipeline.bg_space_partition_provider import (
        litellm_edge_judge_provider,
    )

    failing_primary["payload"] = json.dumps({"verdict": "SAMPLE"})
    got = litellm_edge_judge_provider(prompt_bundle={
        "system": "SAMPLE system", "user": "SAMPLE user",
        "schema": {"type": "object"}})
    assert got == {"verdict": "SAMPLE"}
    _assert_failed_over(failing_primary)


# ── 재발 방지 잠금 ────────────────────────────────────────────────────

def test_no_production_module_calls_litellm_completion_directly():
    """직접 부르면 키 슬롯이 닿지 않는다 — `llm_completion` 을 쓸 것."""
    from pathlib import Path

    root = Path(__file__).resolve().parents[2] / "app"
    owner = root / "core" / "openai_keys.py"
    offenders = []
    for path in sorted(root.rglob("*.py")):
        if path == owner:
            continue
        for lineno, line in enumerate(
                path.read_text(encoding="utf-8").splitlines(), 1):
            if "litellm.completion(" in line and not line.lstrip().startswith(
                    ("#", "*", "``")):
                offenders.append(f"{path.relative_to(root.parent)}:{lineno}")
    assert offenders == [], (
        "litellm.completion 직접 호출은 활성 키 슬롯에 닿지 않는다: "
        f"{offenders}")


def test_broker_is_the_only_key_authority_for_preflights():
    """preflight 가 환경변수만 보면 보조 슬롯만 있을 때 없다고 세운다.

    `space_set_bg_provider` 가 2026-07-30 에 이미 세운 패턴인데 나머지가
    따라오지 않았다. 특히 `dwelling_zone_map_provider` 는 환경변수를 **1차
    키로 고정**해 전환을 무효로 만들었다.
    """
    from pathlib import Path

    root = Path(__file__).resolve().parents[2] / "app"
    offenders = []
    for path in sorted(root.rglob("*.py")):
        text = path.read_text(encoding="utf-8")
        if 'os.environ["OPENAI_API_KEY"] =' in text:
            offenders.append(str(path.relative_to(root.parent)))
    assert offenders == [], (
        "환경변수에 키를 직접 써 넣으면 슬롯 전환이 무효가 된다: "
        f"{offenders}")


# ── 보조 슬롯 단독 구성이 소비 경로에 도달하는가 (Codex 재리뷰 BLOCKING) ──
# `llm_completion` 이 맞아도 소비자가 **브로커 진입 전에** 막으면 소용이 없다.
# 19곳이 "OpenAI 키가 있는가"를 브로커가 아니라 각자 판단하고 있었다 —
# 환경변수만 읽는 preflight 8곳, `settings.openai_api_key`(=**1차 슬롯 필드**)로
# 게이트하는 곳 11곳(스텝 2·서비스 9). 직접 재현:
#
#   브로커가 보는 것 → active_slot: secondary | active_key 설정됨: True
#   → 소비자에서 차단: EdgeJudgeProviderError OPENAI_API_KEY missing or empty
#
# 즉 이번 wave 가 지원하는 **보조 키 단독 상태**가 실제 실행에 도달하지 않는다.
# "키가 있는가"의 답은 브로커 하나가 갖는다.

def _secondary_only(monkeypatch):
    monkeypatch.delenv("OPENAI_API_KEY", raising=False)
    return patch.multiple("app.core.config.settings",
                          openai_api_key="", openai_api_key_secondary="KEY-B")


def test_has_openai_key_accepts_a_secondary_only_configuration(monkeypatch):
    from app.core.openai_keys import has_openai_key

    with _secondary_only(monkeypatch):
        assert openai_keys.active_key() == "KEY-B"
        assert has_openai_key() is True


def test_has_openai_key_falls_back_to_the_environment(monkeypatch):
    """슬롯이 하나도 없어도 환경에 키가 있으면 있다고 답한다(기존 동작 보존)."""
    from app.core.openai_keys import has_openai_key

    monkeypatch.setenv("OPENAI_API_KEY", "ENV-KEY")
    with patch.multiple("app.core.config.settings",
                        openai_api_key="", openai_api_key_secondary=""):
        assert has_openai_key() is True


def test_has_openai_key_is_false_when_nothing_is_configured(monkeypatch):
    from app.core.openai_keys import has_openai_key

    monkeypatch.delenv("OPENAI_API_KEY", raising=False)
    with patch.multiple("app.core.config.settings",
                        openai_api_key="", openai_api_key_secondary=""):
        assert has_openai_key() is False


def test_consumer_completes_on_a_secondary_only_configuration(monkeypatch):
    """★실제 소비자 — 1차 슬롯이 비어 있어도 보조 키로 완주한다."""
    import litellm

    from app.modules.pipeline.bg_space_partition_provider import (
        litellm_edge_judge_provider,
    )

    used = []

    def _fake_completion(**kw):
        used.append(kw.get("api_key"))
        return _response(json.dumps({"verdict": "SAMPLE"}))

    monkeypatch.setattr(litellm, "completion", _fake_completion)
    with _secondary_only(monkeypatch):
        got = litellm_edge_judge_provider(prompt_bundle={
            "system": "SAMPLE system", "user": "SAMPLE user",
            "schema": {"type": "object"}})
    assert got == {"verdict": "SAMPLE"}
    assert used == ["KEY-B"], f"보조 키로 나가야 한다 — 실제: {used}"


def test_only_the_broker_decides_whether_an_openai_key_exists():
    """존재 판단이 흩어지면 새 슬롯 구성이 그중 하나에서 막힌다.

    ★잠금을 두 번 틀렸다. ①문자열 휴리스틱으로 주석·docstring 을 걸렀더니
    홑따옴표 backtick 이 담긴 docstring 한 줄이 통과했다. ②`tokenize` 로 바꿨는데
    **토큰 타입만 건너뛰고 `tok.line`(물리적 줄 전체)을 검사**해서, docstring
    전용 줄이 같은 줄의 INDENT/NEWLINE 토큰으로 다시 잡히고(오탐) 여러 줄에
    걸친 attribute 는 놓쳤다(미탐). 둘 다 직접 재현했다.

    소스를 글자로 훑는 방식 자체가 틀렸다. **AST 로 실제 참조를 본다** —
    docstring 은 문자열 상수라 참조가 아니고, 줄바꿈은 구조에 영향이 없다.
    """
    import ast
    from pathlib import Path

    root = Path(__file__).resolve().parents[2] / "app"
    owner = root / "core" / "openai_keys.py"
    offenders = []
    for path in sorted(root.rglob("*.py")):
        if path == owner:
            continue
        tree = ast.parse(path.read_text(encoding="utf-8"))
        for node in ast.walk(tree):
            # settings.openai_api_key — 1차 슬롯 필드 직접 참조
            if (isinstance(node, ast.Attribute)
                    and node.attr == "openai_api_key"):
                offenders.append(
                    f"{path.relative_to(root.parent)}:{node.lineno} field")
            # os.environ 에서 그 키를 읽는 모든 형태(구독·get)
            if isinstance(node, ast.Constant) and node.value == "OPENAI_API_KEY":
                offenders.append(
                    f"{path.relative_to(root.parent)}:{node.lineno} env")
    offenders = sorted(set(offenders))

    assert offenders == [], (
        "OpenAI 키의 유무·값은 브로커(active_key/has_openai_key)가 답한다 — "
        f"1차 슬롯 필드나 환경변수를 직접 보면 보조 키 단독 구성이 막힌다: "
        f"{offenders}")


# ── 실효 키의 단일 SOT (Codex 재리뷰 BLOCKING-1) ──────────────────────
# `has_openai_key()` 는 슬롯 0개여도 환경변수가 있으면 True 라고 약속했는데
# `active_key()` 는 "" 를 돌려줬다. preflight 만 통과하고 실제 호출은 빈 키로
# 나간다. 직접 재현: has_openai_key=True 인데 OpenAIClient._api_key=''.
# 약속과 값이 갈라지면 안 된다 — 환경변수를 **진짜 슬롯**으로 편입한다.

def test_environment_key_is_a_real_slot(monkeypatch):
    monkeypatch.setenv("OPENAI_API_KEY", "ENV-ONLY")
    with patch.multiple("app.core.config.settings",
                        openai_api_key="", openai_api_key_secondary=""):
        assert openai_keys.has_openai_key() is True
        assert openai_keys.active_key() == "ENV-ONLY"
        assert openai_keys.active_slot_and_key()[1] == "ENV-ONLY"
        assert openai_keys.slot_count() == 1


def test_configured_slots_take_precedence_over_the_environment(monkeypatch):
    """설정 슬롯이 있으면 그것이 먼저다 — 환경변수는 최후 수단이다."""
    monkeypatch.setenv("OPENAI_API_KEY", "ENV-ONLY")
    with patch.multiple("app.core.config.settings",
                        openai_api_key="KEY-A", openai_api_key_secondary="KEY-B"):
        assert openai_keys.active_key() == "KEY-A"
        assert [n for n, _ in openai_keys.available_slots()][:2] == [
            "primary", "secondary"]


def test_environment_duplicate_does_not_add_a_slot(monkeypatch):
    """환경변수가 설정 슬롯과 같은 키면 슬롯을 늘리지 않는다(헛 전환 방지)."""
    monkeypatch.setenv("OPENAI_API_KEY", "KEY-A")
    with patch.multiple("app.core.config.settings",
                        openai_api_key="KEY-A", openai_api_key_secondary=""):
        assert openai_keys.slot_count() == 1


# ── raw urllib 소비자도 전환한다 (Codex 재리뷰 BLOCKING-2) ────────────
# 5곳이 활성 키를 **한 번 복사**했을 뿐 키 수준 실패를 브로커에 알리지 않는다.
# 게다가 `openai_client` 는 **생성 시점**에 복사해 이후 전환이 닿지도 않는다.

def test_call_with_key_failover_moves_to_the_next_slot():
    from app.core.openai_keys import call_with_key_failover

    seen = []

    def _fn(api_key):
        seen.append(api_key)
        if api_key == "KEY-A":
            exc = RuntimeError("OpenAI API error 429: insufficient_quota")
            exc.llm_provider = "openai"
            raise exc
        return "ok"

    with patch.multiple("app.core.config.settings",
                        openai_api_key="KEY-A", openai_api_key_secondary="KEY-B"):
        assert call_with_key_failover(_fn, where="t") == "ok"
    assert seen == ["KEY-A", "KEY-B"]


def test_call_with_key_failover_honours_an_explicit_key():
    """호출자가 키를 명시했으면 그건 고정 계약이다 — 슬롯을 옮기지 않는다."""
    from app.core.openai_keys import call_with_key_failover

    seen = []

    def _fn(api_key):
        seen.append(api_key)
        return "ok"

    with patch.multiple("app.core.config.settings",
                        openai_api_key="KEY-A", openai_api_key_secondary="KEY-B"):
        assert call_with_key_failover(_fn, where="t", fixed_key="MANUAL") == "ok"
    assert seen == ["MANUAL"]


def test_raw_urllib_consumer_reaches_the_secondary_key(monkeypatch):
    """★실제 raw-urllib 소비자 — 1차 quota 실패 후 보조 키로 완주한다."""
    import io
    import json as _json
    import urllib.error
    import urllib.request

    from app.modules.llm.openai_client import OpenAIClient

    auth = []

    def _fake_urlopen(request, timeout=None):
        auth.append(request.headers.get("Authorization"))
        if request.headers.get("Authorization") == "Bearer KEY-A":
            # Codex 재현과 같은 형태 — quota 소진이 400 으로 온다.
            raise urllib.error.HTTPError(
                "u", 400, "quota", {},
                io.BytesIO(_json.dumps(
                    {"error": {"code": "insufficient_quota"}}).encode()))

        class _R:
            def read(self_inner):
                return _json.dumps({"output": [{"content": [
                    {"type": "output_text",
                     "text": _json.dumps({"ok": True})}]}]}).encode()

            def __enter__(self_inner):
                return self_inner

            def __exit__(self_inner, *a):
                return False
        return _R()

    monkeypatch.setattr(urllib.request, "urlopen", _fake_urlopen)
    with patch.multiple("app.core.config.settings",
                        openai_api_key="KEY-A", openai_api_key_secondary="KEY-B"):
        OpenAIClient().generate_structured(
            "SAMPLE system", "SAMPLE user", {"type": "object"})
    assert auth == ["Bearer KEY-A", "Bearer KEY-B"], f"실제 순서: {auth}"
    assert openai_keys.active_slot() == "secondary"


# ── 나머지 raw urllib 소비자 4곳 (Codex 재리뷰 BLOCKING-2 잔여) ────────
# 넷 다 LiteLLM 을 안 거치고 urllib 로 OpenAI 를 직접 두드린다. 키를 재시도
# 루프 **밖**에서 한 번 정하므로 키 수준 실패에도 같은 키로만 재시도한다.
# ①② 는 생성 시점에 복사해 인스턴스를 들고 있으면 전환이 닿지도 않는다.

def _quota_then_ok(monkeypatch, payload_text):
    """1차 키는 quota 400, 보조 키는 성공 — 쓰인 Authorization 을 기록한다."""
    import io
    import json as _json
    import urllib.error
    import urllib.request

    auth = []

    def _fake_urlopen(request, timeout=None):
        auth.append(request.headers.get("Authorization"))
        if request.headers.get("Authorization") == "Bearer KEY-A":
            raise urllib.error.HTTPError(
                "u", 400, "quota", {},
                io.BytesIO(_json.dumps(
                    {"error": {"code": "insufficient_quota"}}).encode()))

        class _R:
            def read(self_inner):
                return payload_text.encode()

            def __enter__(self_inner):
                return self_inner

            def __exit__(self_inner, *a):
                return False
        return _R()

    monkeypatch.setattr(urllib.request, "urlopen", _fake_urlopen)
    return auth


_VISION_OK = json.dumps({"output": [{"content": [
    {"type": "output_text", "text": json.dumps(
        {"passed": True, "issues": [], "severity": "ok",
         "matches_prompt": True, "reason": "SAMPLE"})}]}]})


def test_image_validator_reaches_the_secondary_key(monkeypatch):
    from app.modules.image_validator import ImageValidator

    auth = _quota_then_ok(monkeypatch, _VISION_OK)
    with patch.multiple("app.core.config.settings",
                        openai_api_key="KEY-A", openai_api_key_secondary="KEY-B"):
        # ★생성 후에 전환이 일어나도 닿아야 한다 — 생성 시점 복사 금지.
        v = ImageValidator()
        v.validate_scene_image(
            b"png", {"prompt": "SAMPLE"}, ["SAMPLE entity"])
    assert auth == ["Bearer KEY-A", "Bearer KEY-B"], f"실제 순서: {auth}"
    assert openai_keys.active_slot() == "secondary"


def test_variation_recommender_reaches_the_secondary_key(monkeypatch):
    from app.modules.variation_recommender_v2 import VariationRecommenderV2

    auth = _quota_then_ok(monkeypatch, json.dumps({"output": [{"content": [
        {"type": "output_text", "text": json.dumps(
            {"variations": [], "reason": "SAMPLE"})}]}]}))
    with patch.multiple("app.core.config.settings",
                        openai_api_key="KEY-A", openai_api_key_secondary="KEY-B"):
        VariationRecommenderV2().recommend_from_image(b"png", "SAMPLE prompt")
    assert auth == ["Bearer KEY-A", "Bearer KEY-B"], f"실제 순서: {auth}"


def test_pdf_validator_reaches_the_secondary_key(monkeypatch):
    from app.modules.pdf_validator import _call_vision_api

    auth = _quota_then_ok(monkeypatch, json.dumps({"output": [{"content": [
        {"type": "output_text", "text": json.dumps({"ok": True})}]}]}))
    with patch.multiple("app.core.config.settings",
                        openai_api_key="KEY-A", openai_api_key_secondary="KEY-B"):
        _call_vision_api(b"png", "SAMPLE system")
    assert auth == ["Bearer KEY-A", "Bearer KEY-B"], f"실제 순서: {auth}"


def test_no_raw_openai_consumer_holds_a_key_across_calls():
    """키를 생성 시점에 복사하면 이후 전환이 그 인스턴스에 닿지 않는다.

    ★`openai_client` 이 정확히 그래서 죽었다 — 첫 OpenAI 호출이 그 경로면
    보조 키가 있어도 전환 계기가 없다.
    """
    import ast
    from pathlib import Path

    root = Path(__file__).resolve().parents[2] / "app"
    offenders = []
    for path in sorted(root.rglob("*.py")):
        tree = ast.parse(path.read_text(encoding="utf-8"))
        for node in ast.walk(tree):
            if not isinstance(node, ast.FunctionDef) or node.name != "__init__":
                continue
            for sub in ast.walk(node):
                if (isinstance(sub, ast.Call)
                        and isinstance(sub.func, ast.Name)
                        and sub.func.id in ("_active_openai_key", "active_key")):
                    offenders.append(
                        f"{path.relative_to(root.parent)}:{sub.lineno}")
    assert offenders == [], (
        "생성 시점에 활성 키를 복사하면 이후 슬롯 전환이 닿지 않는다: "
        f"{offenders}")


def test_key_level_429_does_not_retry_the_dead_key(monkeypatch):
    """★quota 가 429 로 오면 죽은 키를 반복 두드린 뒤에야 전환했다.

    소비자의 내부 재시도는 `exc.code in {429, ...}` 만 보고 backoff 재시도를
    건다. 그런데 quota 소진은 **재시도해도 풀리지 않는다** — 브로커가 이미
    그 구분(단순 rate limit vs 크레딧 소진)을 갖고 있는데 소비자가 그 앞에서
    같은 키를 max_retries 번 더 태운다. 실측: 이 경로가 13.8초 걸렸다.
    """
    import io
    import json as _json
    import urllib.error
    import urllib.request

    from app.modules.llm.openai_client import OpenAIClient

    auth = []
    sleeps = []
    monkeypatch.setattr(
        "app.modules.llm.openai_client.time", _SleepProbe(sleeps))

    def _fake_urlopen(request, timeout=None):
        auth.append(request.headers.get("Authorization"))
        if request.headers.get("Authorization") == "Bearer KEY-A":
            raise urllib.error.HTTPError(
                "u", 429, "quota", {},
                io.BytesIO(_json.dumps(
                    {"error": {"code": "insufficient_quota"}}).encode()))

        class _R:
            def read(self_inner):
                return _json.dumps({"output": [{"content": [
                    {"type": "output_text",
                     "text": _json.dumps({"ok": True})}]}]}).encode()

            def __enter__(self_inner):
                return self_inner

            def __exit__(self_inner, *a):
                return False
        return _R()

    monkeypatch.setattr(urllib.request, "urlopen", _fake_urlopen)
    with patch.multiple("app.core.config.settings",
                        openai_api_key="KEY-A", openai_api_key_secondary="KEY-B"):
        OpenAIClient().generate_structured(
            "SAMPLE system", "SAMPLE user", {"type": "object"})
    assert auth == ["Bearer KEY-A", "Bearer KEY-B"], (
        f"죽은 키를 반복하지 않고 바로 전환해야 한다 — 실제: {auth}")
    assert sleeps == [], f"키 수준 실패에 backoff 를 걸면 안 된다 — {sleeps}"


def test_ordinary_429_still_retries_on_the_same_key(monkeypatch):
    """경계 — 단순 rate limit 은 기존 backoff 재시도 그대로다."""
    import io
    import json as _json
    import urllib.error
    import urllib.request

    from app.modules.llm.openai_client import OpenAIClient

    auth = []
    monkeypatch.setattr(
        "app.modules.llm.openai_client.time", _SleepProbe([]))
    state = {"n": 0}

    def _fake_urlopen(request, timeout=None):
        auth.append(request.headers.get("Authorization"))
        state["n"] += 1
        if state["n"] == 1:
            raise urllib.error.HTTPError(
                "u", 429, "rate", {},
                io.BytesIO(_json.dumps({"error": {
                    "code": "rate_limit_exceeded",
                    "message": "Rate limit reached. try again in 6ms"}}).encode()))

        class _R:
            def read(self_inner):
                return _json.dumps({"output": [{"content": [
                    {"type": "output_text",
                     "text": _json.dumps({"ok": True})}]}]}).encode()

            def __enter__(self_inner):
                return self_inner

            def __exit__(self_inner, *a):
                return False
        return _R()

    monkeypatch.setattr(urllib.request, "urlopen", _fake_urlopen)
    # 이 시험의 전제는 「재시도가 한 번은 일어난다」이다 — 그 값을 여기서
    # 못박는다. 2026-08-20 부터 시험 기본 재시도는 0이다(돈 가드가 막은 호출을
    # 다시 보내면 느려지기만 한다, tests/conftest.py 참조). 기계 기본값에
    # 기대면 이 시험이 무엇을 재는지가 환경에 따라 달라진다.
    with patch.multiple("app.core.config.settings",
                        openai_api_key="KEY-A", openai_api_key_secondary="KEY-B",
                        llm_max_retries=3):
        OpenAIClient().generate_structured(
            "SAMPLE system", "SAMPLE user", {"type": "object"})
    assert auth == ["Bearer KEY-A", "Bearer KEY-A"], f"실제: {auth}"
    assert openai_keys.active_slot() == "primary"


def test_swallowing_fallback_does_not_hide_a_key_level_failure(monkeypatch):
    """★방어적 fallback 이 키 수준 실패를 삼키면 전환이 원천 봉쇄된다.

    `_gpt_review_entity_list` 는 함수 **안**에 `except Exception` 이 있어
    실패를 "원본 목록 사용"으로 조용히 흘렸다. 감싸기를 붙여도 바깥
    wrapper 가 볼 예외 자체가 없으니 슬롯이 옮겨질 수 없다.
    (Codex 재현: auth=['Bearer KEY-A'], active_slot=primary)

    표시 없는 urllib HTTPError 를 브로커가 못 알아보는 것도 함께 확인한다 —
    둘 중 하나만 고치면 여전히 안 넘어간다.
    """
    import io
    import json as _json
    import urllib.error
    import urllib.request

    from app.modules.pipeline import entity_extractor_v2_legacy as legacy

    auth = []

    def _fake_urlopen(request, timeout=None):
        auth.append(request.headers.get("Authorization"))
        if request.headers.get("Authorization") == "Bearer KEY-A":
            raise urllib.error.HTTPError(
                "u", 400, "quota", {},
                io.BytesIO(_json.dumps(
                    {"error": {"code": "insufficient_quota"}}).encode()))

        class _R:
            def read(self_inner):
                return _json.dumps({"output_text": _json.dumps(
                    {"entities": [{"name": "SAMPLE", "entity_type": "prop",
                                   "importance": "high", "appearances": 3}]}
                )}).encode()

            def __enter__(self_inner):
                return self_inner

            def __exit__(self_inner, *a):
                return False
        return _R()

    monkeypatch.setattr(urllib.request, "urlopen", _fake_urlopen)
    monkeypatch.setattr(legacy, "_load_system_prompt", lambda: "SAMPLE")
    with patch.multiple("app.core.config.settings",
                        openai_api_key="KEY-A", openai_api_key_secondary="KEY-B"):
        out = legacy._gpt_review_entity_list(
            {"characters": [{"name": "SAMPLE", "appearances": 3}],
             "locations": [], "props": []}, "SAMPLE text")
    assert auth == ["Bearer KEY-A", "Bearer KEY-B"], (
        f"삼키지 말고 브로커에 올려야 전환된다 — 실제: {auth}")
    assert openai_keys.active_slot() == "secondary"
    # ★보조 키의 응답을 실제로 썼는지 — 원본 목록 fallback 이 아니어야 한다.
    assert out["review"], "보조 응답이 아니라 fallback 을 돌려주면 안 된다"
    assert out["importance_map"] == {"SAMPLE": "high"}


def test_plain_text_body_is_judged_by_the_same_markers():
    """★비JSON 본문도 마커 판정에 넣는다.

    나는 "파싱 실패면 문자열로 두어 메시지 마커 경로가 받는다"고 썼는데
    **틀렸다** — 마커 판정은 `str(exc)` 만 본다. 재현: body='insufficient_quota'
    인데 is_key_level_failure=False. proxy·plain-text 응답에서는 그대로 삼킨다.
    """
    import io
    import urllib.error

    from app.core.openai_keys import is_key_level_failure, mark_openai_failure

    exc = urllib.error.HTTPError(
        "u", 400, "quota", {}, io.BytesIO(b"insufficient_quota"))
    mark_openai_failure(exc, status=400, body="insufficient_quota")
    assert is_key_level_failure(exc) is True


def test_plain_text_body_without_a_marker_is_not_key_level():
    """경계 — 아무 문자열이나 키 수준으로 승격시키지 않는다."""
    import io
    import urllib.error

    from app.core.openai_keys import is_key_level_failure, mark_openai_failure

    exc = urllib.error.HTTPError(
        "u", 500, "oops", {}, io.BytesIO(b"upstream connect error"))
    mark_openai_failure(exc, status=500, body="upstream connect error")
    assert is_key_level_failure(exc) is False