"""OpenAI 키 2슬롯 failover — 결정론 계약 (2026-07-30).

전환 규칙은 수학적으로 판정 가능한 영역이라 유닛이 유효하다. 여기서 고정
하는 것은 셋이다.

    1. **전환 트리거가 키 수준 실패로 좁혀져 있는가** — billing/quota/인증만.
       타임아웃·5xx·단순 429 에서 키를 바꾸면 멀쩡한 보조 키를 태우고 진짜
       원인을 가린다.
    2. **조용히 전환하지 않는가** — 슬롯이 없으면 False 를 돌려 호출자가
       원래 예외를 올리게 한다(빈 결과로 흘리지 않는다).
    3. **기본값이 기존 동작인가** — 보조 슬롯이 비어 있으면 슬롯 1개.
"""
from __future__ import annotations

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()


def _with_keys(primary: str, secondary: str):
    return patch.multiple(
        "app.core.config.settings",
        openai_api_key=primary,
        openai_api_key_secondary=secondary,
    )


class _Err(Exception):
    """OpenAI SDK 예외 흉내 — 구조화 필드를 갖는다."""

    def __init__(self, msg="", *, code=None, err_type=None, status_code=None,
                 llm_provider="openai", body=None):
        super().__init__(msg)
        if code is not None:
            self.code = code
        if err_type is not None:
            # OpenAI 응답의 `type` — `code` 와 다른 이름이 실릴 수 있다.
            self.type = err_type
        if status_code is not None:
            self.status_code = status_code
        if llm_provider is not None:
            self.llm_provider = llm_provider
        if body is not None:
            self.body = body


# ─────────────────────────────────────────────────────────────────────
# 1. 슬롯 구성
# ─────────────────────────────────────────────────────────────────────
def test_secondary_blank_means_single_slot_default():
    """보조 키를 넣지 않으면 슬롯 1개 — 기존 동작 그대로."""
    with _with_keys("KEY-A", ""):
        assert openai_keys.slot_count() == 1
        assert openai_keys.active_slot() == "primary"
        assert openai_keys.active_key() == "KEY-A"


def test_primary_is_used_first_when_both_present():
    """사용자 지시: 기본은 기존 키, 보조는 다음 차례."""
    with _with_keys("KEY-A", "KEY-B"):
        assert openai_keys.slot_count() == 2
        assert openai_keys.active_slot() == "primary"
        assert openai_keys.active_key() == "KEY-A"


def test_no_keys_at_all_is_empty_not_crash():
    with _with_keys("", ""):
        assert openai_keys.slot_count() == 0
        assert openai_keys.active_key() == ""
        assert openai_keys.active_slot() is None


def test_only_secondary_configured_is_usable():
    """1차를 비우고 보조만 채워도 그 키를 쓴다(슬롯이 비면 건너뛴다)."""
    with _with_keys("", "KEY-B"):
        assert openai_keys.slot_count() == 1
        assert openai_keys.active_key() == "KEY-B"


# ─────────────────────────────────────────────────────────────────────
# 2. 키 수준 실패 판정 — 트리거 경계
# ─────────────────────────────────────────────────────────────────────
@pytest.mark.parametrize("exc", [
    _Err("Billing hard limit has been reached.",
         code="billing_hard_limit_reached", status_code=400),
    _Err("quota", code="insufficient_quota", status_code=429),
    _Err("bad key", code="invalid_api_key", status_code=401),
    _Err("forbidden", status_code=403),
    _Err("unauthorized", status_code=401),
    _Err("You exceeded your current quota, please check your plan."),
    _Err("body 경유", body={"error": {"code": "billing_hard_limit_reached"}}),
    # ★2026-07-31 실측 — 크레딧 소진. 아는 이름이 `type` 에만 있고 `code` 는
    # 새 이름이라, `code` 만 읽던 판정이 이 실패를 통과시켰다(보조 키가
    # 멀쩡한데 전환하지 않고 실행이 죽었다).
    _Err("You have no credits remaining. Add credits to continue using "
         "the API at https://platform.openai.com/settings/…",
         code="credit_balance_exhausted", err_type="insufficient_quota",
         status_code=429),
    # 같은 실패가 LiteLLM 을 거치면 구조화 필드가 전부 지워진다 — 실측값
    # 그대로(code='429' / type='throttling_error' / body 없음). 남는 단서는
    # 메시지 문구뿐이다.
    _Err("litellm.RateLimitError: RateLimitError: OpenAIException - You "
         "have no credits remaining. Add credits to continue using the API "
         "at https://platform.openai.com/settings/organization/billing/.",
         code="429", err_type="throttling_error", status_code=429),
    # 본문이 `{"error": {...}}` 로 감싸이지 않고 평면으로 오는 모양.
    _Err("평면 body", body={"message": "You have no credits remaining.",
                          "type": "insufficient_quota",
                          "code": "credit_balance_exhausted"}),
])
def test_key_level_failures_are_recognized(exc):
    assert openai_keys.is_key_level_failure(exc) is True


@pytest.mark.parametrize("exc", [
    _Err("Rate limit reached for gpt", code="rate_limit_exceeded",
         status_code=429),
    _Err("server error", status_code=500),
    _Err("Read timed out", status_code=None),
    _Err("context_length_exceeded", code="context_length_exceeded",
         status_code=400),
    _Err("content filtered", code="content_policy_violation",
         status_code=400),
    # ★경계 — LiteLLM 을 거친 **진짜** rate limit 은 크레딧 소진과 code·
    # type 이 똑같다(429 / throttling_error). 문구만이 둘을 가른다. 여기서
    # 전환하면 멀쩡한 보조 키를 태우고 진짜 원인을 가린다.
    _Err("litellm.RateLimitError: RateLimitError: OpenAIException - Rate "
         "limit reached for gpt-5.6 in organization org-x. Please try "
         "again in 6ms.", code="429", err_type="throttling_error",
         status_code=429),
])
def test_transient_and_request_failures_are_not_key_level(exc):
    """★단순 429·5xx·타임아웃은 기존 재시도 경로 몫 — 키를 바꾸지 않는다."""
    assert openai_keys.is_key_level_failure(exc) is False


def test_non_openai_provider_error_never_switches_openai_key():
    """Gemini 쪽 401 로 OpenAI 키를 갈아 끼우는 헛수고를 막는다."""
    gem = _Err("API key not valid", status_code=401, llm_provider="gemini")
    assert openai_keys.is_key_level_failure(gem) is False
    with _with_keys("KEY-A", "KEY-B"):
        assert openai_keys.failover_on(gem, where="test") is False
        assert openai_keys.active_key() == "KEY-A"


# ─────────────────────────────────────────────────────────────────────
# 3. 전환 동작
# ─────────────────────────────────────────────────────────────────────
def test_failover_moves_to_secondary_and_stays():
    """전환은 프로세스 전역이고 되돌아가지 않는다(죽은 키 재시도 금지)."""
    with _with_keys("KEY-A", "KEY-B"):
        exc = _Err("billing", code="billing_hard_limit_reached")
        assert openai_keys.failover_on(exc, where="test") is True
        assert openai_keys.active_slot() == "secondary"
        assert openai_keys.active_key() == "KEY-B"
        # 두 번째 호출은 더 갈 곳이 없다 → False (호출자가 원래 예외를 올림)
        assert openai_keys.failover_on(exc, where="test") is False
        assert openai_keys.active_key() == "KEY-B"


def test_failover_returns_false_when_only_one_slot():
    """슬롯이 하나면 키 수준 실패라도 전환하지 않는다 — 조용히 삼키지 않게."""
    with _with_keys("KEY-A", ""):
        exc = _Err("billing", code="billing_hard_limit_reached")
        assert openai_keys.failover_on(exc, where="test") is False


def test_switch_hook_fires_once_per_switch():
    """Router 무효화 훅이 전환 시 호출된다(옛 키가 박힌 Router 재사용 차단)."""
    calls = []
    openai_keys.register_switch_hook(lambda: calls.append(1))
    try:
        with _with_keys("KEY-A", "KEY-B"):
            openai_keys.failover_on(
                _Err("q", code="insufficient_quota"), where="test")
        assert len(calls) == 1
    finally:
        openai_keys._switch_hooks.clear()


def test_hook_failure_does_not_block_switch():
    """훅이 터져도 전환 자체는 성립한다."""
    def _boom():
        raise RuntimeError("hook 고장")

    openai_keys.register_switch_hook(_boom)
    try:
        with _with_keys("KEY-A", "KEY-B"):
            assert openai_keys.failover_on(
                _Err("q", code="insufficient_quota"), where="t") is True
            assert openai_keys.active_key() == "KEY-B"
    finally:
        openai_keys._switch_hooks.clear()


# ─────────────────────────────────────────────────────────────────────
# 4. 클라이언트 프록시 — 호출 형태 보존 + 1회 전환 재시도
# ─────────────────────────────────────────────────────────────────────
class _FakeOpenAI:
    """`client.images.edit(...)` 형태를 흉내내는 가짜 SDK 클라이언트."""

    made: list = []

    def __init__(self, api_key=None, **kw):
        self.api_key = api_key
        _FakeOpenAI.made.append(api_key)
        self.images = _Images(self)
        self.responses = _Responses(self)


class _Images:
    def __init__(self, owner):
        self._owner = owner

    def edit(self, **kw):
        if self._owner.api_key == "KEY-A":
            raise _Err("billing", code="billing_hard_limit_reached")
        return {"ok": True, "key": self._owner.api_key, "kw": kw}


class _Responses:
    def __init__(self, owner):
        self._owner = owner

    def create(self, **kw):
        raise _Err("server error", status_code=500)


@pytest.fixture
def _fake_sdk(monkeypatch):
    _FakeOpenAI.made = []
    monkeypatch.setattr("openai.OpenAI", _FakeOpenAI)
    return _FakeOpenAI


def test_proxy_retries_on_secondary_key_after_key_level_failure(_fake_sdk):
    """1차 키가 billing 으로 죽으면 보조 키로 같은 호출을 다시 태운다."""
    with _with_keys("KEY-A", "KEY-B"):
        client = openai_keys.openai_client(timeout=1.0)
        out = client.images.edit(model="x", prompt="p")
        assert out["ok"] is True
        assert out["key"] == "KEY-B"
        assert out["kw"] == {"model": "x", "prompt": "p"}
        assert _fake_sdk.made == ["KEY-A", "KEY-B"]
        assert openai_keys.active_slot() == "secondary"


def test_proxy_propagates_non_key_failures_without_switching(_fake_sdk):
    """5xx 는 그대로 올린다 — 키를 바꾸지도, 재시도하지도 않는다."""
    with _with_keys("KEY-A", "KEY-B"):
        client = openai_keys.openai_client()
        with pytest.raises(_Err):
            client.responses.create(model="x")
        assert _fake_sdk.made == ["KEY-A"]
        assert openai_keys.active_slot() == "primary"


def test_proxy_does_not_take_api_key_argument(_fake_sdk):
    """키는 브로커가 정한다 — 호출자가 넘긴 api_key 는 무시한다."""
    with _with_keys("KEY-A", "KEY-B"):
        client = openai_keys.openai_client(api_key="MANUAL", timeout=2.0)
        client.images.edit(model="x")   # KEY-A 실패 → KEY-B 성공
        assert "MANUAL" not in _fake_sdk.made


def test_proxy_reuses_cached_client_per_slot(_fake_sdk):
    """같은 슬롯에서 반복 호출이 클라이언트를 새로 만들지 않는다."""
    with _with_keys("", "KEY-B"):        # 1차 비움 → 처음부터 KEY-B
        client = openai_keys.openai_client()
        client.images.edit(model="x")
        client.images.edit(model="y")
        assert _fake_sdk.made == ["KEY-B"]


# ─────────────────────────────────────────────────────────────────────
# 5. 동시 호출 (A5 — Codex 코드 리뷰 BLOCKING 5, 실측 재현)
# ─────────────────────────────────────────────────────────────────────
# 전환 판단이 **전역 활성 index** 를 기준으로 이뤄졌다. 두 호출이 같은 슬롯을
# 잡고 나란히 실패하면, 먼저 도착한 쪽이 0→1 로 옮기고 뒤이어 도착한 쪽은
# 이미 1이 된 전역을 읽어 1→2 로 넘기려 한다. 슬롯이 2개면 그 자리에서
# "소진"으로 판정되어 **한 번도 보조 키를 못 써 본 요청이 그대로 버려진다.**
#
# 전환은 "지금 전역이 어디인가"가 아니라 **"이 호출이 어느 슬롯에서 실패했는가"**
# 를 기준으로 판단해야 한다.

def _billing():
    return _Err("billing", code="billing_hard_limit_reached")


def test_concurrent_failures_on_the_same_slot_do_not_burn_the_next_slot():
    """같은 슬롯에서 나란히 실패한 두 호출 — 어느 쪽도 버려지지 않는다."""
    with _with_keys("KEY-A", "KEY-B"):
        first = openai_keys.failover_on(
            _billing(), where="t1", attempted_slot="primary")
        second = openai_keys.failover_on(
            _billing(), where="t2", attempted_slot="primary")
        assert (first, second) == (True, True)
        assert openai_keys.active_key() == "KEY-B"


def test_failover_from_a_stale_slot_does_not_advance_again():
    """이미 다른 호출이 옮겨 놓았으면 또 옮기지 않는다 — 재시도만 허용."""
    with _with_keys("KEY-A", "KEY-B"):
        openai_keys.failover_on(_billing(), where="t", attempted_slot="primary")
        assert openai_keys.active_slot() == "secondary"
        for _ in range(3):
            assert openai_keys.failover_on(
                _billing(), where="t", attempted_slot="primary") is True
        assert openai_keys.active_slot() == "secondary"


def test_failure_on_the_last_slot_is_still_exhaustion():
    """마지막 슬롯에서 실패했으면 갈 곳이 없다 — 조용히 삼키지 않는다."""
    with _with_keys("KEY-A", "KEY-B"):
        openai_keys.failover_on(_billing(), where="t", attempted_slot="primary")
        assert openai_keys.failover_on(
            _billing(), where="t", attempted_slot="secondary") is False


def test_switch_hook_fires_once_even_when_two_calls_fail_together():
    """전환은 한 번만 일어났으므로 Router 무효화도 한 번이다."""
    calls = []
    openai_keys.register_switch_hook(lambda: calls.append(1))
    try:
        with _with_keys("KEY-A", "KEY-B"):
            openai_keys.failover_on(
                _billing(), where="t", attempted_slot="primary")
            openai_keys.failover_on(
                _billing(), where="t", attempted_slot="primary")
        assert len(calls) == 1
    finally:
        openai_keys._switch_hooks.clear()


def test_concurrent_proxy_calls_all_reach_the_secondary_key(monkeypatch):
    """★실제 동시 호출 — 1차 키가 죽었을 때 어느 요청도 실패로 끝나지 않는다.

    ★단순히 스레드를 동시에 띄우는 것으로는 부족하다 — 먼저 시작한 스레드가
    실패·전환·재시도까지 끝내 버리면 경합이 아예 일어나지 않아 **결함이 있어도
    통과한다**(처음 쓴 판본이 그렇게 통과했다). 그래서 1차 키 실패 지점에
    장벽을 두어 **모든 호출이 같은 슬롯에서 실패한 상태**를 강제한다.
    """
    import threading

    parties = 4
    at_failure = threading.Barrier(parties, timeout=10)

    class _BlockingImages:
        def __init__(self, owner):
            self._owner = owner

        def edit(self, **kw):
            if self._owner.api_key == "KEY-A":
                at_failure.wait()      # 전원이 1차 키에서 실패할 때까지 대기
                raise _Err("billing", code="billing_hard_limit_reached")
            return {"ok": True, "key": self._owner.api_key, "kw": kw}

    class _BlockingOpenAI:
        def __init__(self, api_key=None, **kw):
            self.api_key = api_key
            self.images = _BlockingImages(self)

    monkeypatch.setattr("openai.OpenAI", _BlockingOpenAI)

    with _with_keys("KEY-A", "KEY-B"):
        client = openai_keys.openai_client()
        results: dict = {}
        errors: dict = {}

        def _call(i: int) -> None:
            try:
                results[i] = client.images.edit(model=f"m{i}")
            except BaseException as exc:  # noqa: BLE001
                errors[i] = exc

        threads = [threading.Thread(target=_call, args=(i,))
                   for i in range(parties)]
        for t in threads:
            t.start()
        for t in threads:
            t.join(timeout=15)

        assert errors == {}
        assert len(results) == parties
        assert {r["key"] for r in results.values()} == {"KEY-B"}


def test_slot_and_key_are_read_as_one_value(_fake_sdk):
    """캐시 키(슬롯)와 그 클라이언트의 실제 키가 어긋나지 않는다.

    슬롯과 키를 따로 읽으면 그 사이의 전환이 "primary 슬롯에 보조 키가 박힌
    클라이언트"를 캐시에 남긴다 — 그 뒤로는 영구히 어긋난다.
    """
    with _with_keys("KEY-A", "KEY-B"):
        client = openai_keys.openai_client()
        slot, raw = client._raw()
        assert (slot, raw.api_key) == ("primary", "KEY-A")
        openai_keys.failover_on(_billing(), where="t", attempted_slot=slot)
        slot2, raw2 = client._raw()
        assert (slot2, raw2.api_key) == ("secondary", "KEY-B")


# ─────────────────────────────────────────────────────────────────────
# 6. LiteLLM Router 경로 (A5 후속 — Codex 재확인 BLOCKING)
# ─────────────────────────────────────────────────────────────────────
# 직접 프록시는 (슬롯, 클라이언트) 를 짝으로 묶어 고쳤지만 Router 경로에는
# 같은 창이 남아 있었다. Router 는 **어느 슬롯 키로 지어졌는지 기록하지 않고**,
# `_completion` 은 호출 직전 전역 활성 슬롯을 다시 읽어 그것을 자기 슬롯으로
# 삼았다. 그래서 이런 일이 난다:
#
#   ① 한 요청이 primary 키로 지은 Router 를 이미 들고 있다
#   ② 다른 요청이 전역을 secondary 로 옮긴다
#   ③ ①이 그 stale Router 로 호출한다 → 실제로는 primary 로 나간다
#   ④ primary billing 실패인데 "secondary 에서 실패"로 보고된다
#   ⑤ 마지막 슬롯 소진으로 판정 — **보조 키를 한 번도 안 써 보고 죽는다**
#
# Router 와 그 생성 슬롯을 하나의 불변 짝으로 다뤄야 한다.

class _SlotRouter:
    """지어질 때의 키를 기억하는 가짜 Router — 그 키로만 성공한다."""

    def __init__(self, api_key: str):
        self.api_key = api_key
        self.calls = 0

    def completion(self, **kwargs):
        self.calls += 1
        if self.api_key == "KEY-A":
            raise _Err("billing", code="billing_hard_limit_reached")
        return {"ok": True, "key": self.api_key}


@pytest.fixture
def _router_env(monkeypatch):
    """`_build_router` 를 슬롯 키를 기억하는 가짜로 바꾼다."""
    from app.core import openai_keys as _keys
    from app.modules.llm import llm_client as lc

    built: list = []

    def _fake_build():
        slot, key = _keys.active_slot_and_key()
        r = _SlotRouter(key)
        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)
    return lc, built


def test_stale_router_failure_is_attributed_to_its_own_slot(_router_env):
    """★재현 — primary Router 를 든 채 전역이 secondary 로 바뀐 뒤 호출."""
    lc, _built = _router_env
    with _with_keys("KEY-A", "KEY-B"):
        stale = lc._get_router_binding()          # ① primary 로 지어진 binding
        assert stale.slot == "primary"
        openai_keys.failover_on(                  # ② 다른 요청이 전역을 옮김
            _billing(), where="other", attempted_slot="primary")
        assert openai_keys.active_slot() == "secondary"

        out = lc._completion(stale, "gpt", {})    # ③ stale binding 으로 호출
        assert out == {"ok": True, "key": "KEY-B"}


def test_router_binding_is_not_handed_out_when_its_slot_is_stale(_router_env):
    """전환 훅이 돌기 전이라도 옛 슬롯의 Router 를 내주지 않는다."""
    lc, _built = _router_env
    with _with_keys("KEY-A", "KEY-B"):
        first = lc._get_router_binding()
        assert first.slot == "primary"
        # 훅을 지우고 전환 — `_router=None` 무효화 없이 전역만 옮긴 상태
        hooks = list(openai_keys._switch_hooks)
        openai_keys._switch_hooks.clear()
        try:
            openai_keys.failover_on(
                _billing(), where="t", attempted_slot="primary")
        finally:
            openai_keys._switch_hooks.extend(hooks)
        again = lc._get_router_binding()
        assert again.slot == "secondary"
        assert again.router.api_key == "KEY-B"


def test_router_binding_slot_and_key_come_from_one_read(_router_env):
    """Router 의 키와 그 binding 의 슬롯이 어긋나지 않는다."""
    lc, built = _router_env
    with _with_keys("KEY-A", "KEY-B"):
        b = lc._get_router_binding()
        assert (b.slot, b.router.api_key) == ("primary", "KEY-A")
        assert built[-1] == ("primary", "KEY-A")


def test_router_binding_is_reused_while_the_slot_holds(_router_env):
    """슬롯이 그대로면 Router 를 다시 짓지 않는다(기존 캐시 계약 유지)."""
    lc, built = _router_env
    with _with_keys("KEY-A", "KEY-B"):
        a = lc._get_router_binding()
        b = lc._get_router_binding()
        assert a is b
        assert len(built) == 1