"""canary 주행의 **글 호출 문** — 모든 스레드에서 세어지나. ★유료 0.

Codex BLOCK 1 (2026-08-31) — 새 예산을 만들지 말고 **기존 `ResearchCallBudget`**
한 벌을 canary 의 전체 text 범위에 쓸 것. 다만 thread-local 이라 팬아웃
worker 에서 `reserve` 가 no-op 이 되는 구멍을 닫을 것. 요구 끝점 둘 —

    ①main + worker **둘 다**에 budget 과 armed 가 간다
    ②cap N 에서 **N+1 번째가 provider 앞에서** 선다
"""
from __future__ import annotations

import threading

import pytest

from app.core.research_call_budget import (ResearchCallBudgetExceeded,
                                           get_current_budget, is_armed,
                                           reserve_current_research_call)
from tools.grounding_audit import canary_text_budget as tb


class TestTheDoorReachesEveryThread:
    def test_a_worker_thread_is_counted(self):
        """★★앞 판은 워커에서 **그냥 지나갔다** — 상한이 통째로 no-op."""
        seen = []
        with tb.canary_text_scope(cap=10) as b:
            def w():
                seen.append((is_armed(), get_current_budget() is b))
                reserve_current_research_call(source="worker")

            t = threading.Thread(target=w)
            t.start()
            t.join()
        assert seen == [(True, True)], f"★워커가 문을 못 봤다: {seen}"
        assert b.snapshot()["used"] == 1

    def test_many_workers_share_one_count(self):
        with tb.canary_text_scope(cap=100) as b:
            def w():
                reserve_current_research_call(source="w")

            ts = [threading.Thread(target=w) for _ in range(12)]
            for t in ts:
                t.start()
            for t in ts:
                t.join()
        assert b.snapshot()["used"] == 12, "★따로 세었다"

    def test_the_scope_does_not_leak(self):
        with tb.canary_text_scope(cap=1):
            assert is_armed() is True
        assert is_armed() is False
        assert get_current_budget() is None

    def test_it_restores_even_on_error(self):
        with pytest.raises(RuntimeError):
            with tb.canary_text_scope(cap=1):
                raise RuntimeError("일부러")
        assert is_armed() is False


class TestTheNextOneStopsBeforeTheProvider:
    def _client(self, sent):
        """진짜 `FailoverOpenAIClient` 에 **가짜 운반층**만 끼운다."""
        from app.core.openai_keys import FailoverOpenAIClient

        c = FailoverOpenAIClient.__new__(FailoverOpenAIClient)
        c._cache = {}
        c._cache_lock = threading.Lock()
        c._client_kwargs = {}

        class _Leaf:
            def create(self, **kw):
                sent.append(kw)
                return {"ok": True}

        class _Chat:
            completions = _Leaf()

        class _Raw:
            chat = _Chat()

        c._raw = lambda: ("slot-1", _Raw())
        return c

    def test_the_cap_stops_the_real_central_door(self):
        """★★`openai_keys._invoke` 는 **물리 전송 자리**다 (production)."""
        sent = []
        c = self._client(sent)
        with tb.canary_text_scope(cap=2) as b:
            for _ in range(2):
                c._invoke(("chat", "completions", "create"), (), {})
            assert len(sent) == 2
            with pytest.raises(ResearchCallBudgetExceeded):
                c._invoke(("chat", "completions", "create"), (), {})
        assert len(sent) == 2, f"★상한을 넘겨 {len(sent)}번 나갔다"
        assert b.snapshot() == {"cap": 2, "used": 2, "denied": 1,
                                "remaining": 0}

    def test_a_worker_cannot_go_over_either(self):
        """★워커에서도 넘으면 선다 — 거기가 새면 상한이 거짓말이다."""
        sent, errs = [], []
        c = self._client(sent)
        with tb.canary_text_scope(cap=3) as b:
            def w():
                try:
                    c._invoke(("chat", "completions", "create"), (), {})
                except ResearchCallBudgetExceeded:
                    errs.append(1)

            ts = [threading.Thread(target=w) for _ in range(6)]
            for t in ts:
                t.start()
            for t in ts:
                t.join()
        assert len(sent) == 3 and len(errs) == 3
        assert b.snapshot()["denied"] == 3

    def test_outside_the_scope_nothing_changes(self):
        """★팔을 안 들면 **한 글자도 안 바뀐다** — 다른 호출부는 그대로."""
        sent = []
        c = self._client(sent)
        for _ in range(4):
            c._invoke(("chat", "completions", "create"), (), {})
        assert len(sent) == 4


class TestItSaysWhatItDoesNotCover:
    """★★이 시험은 **뒤집힌 것**이다 (2026-09-01).

    앞에는 `openai_keys.llm_completion` 을 「문 없음」으로 적어 두고 시험이
    그것을 **잠그고** 있었다. 그런데 그 자리에는 이미 문을 달았다 — 즉 시험이
    낡은 주장을 지키느라 「그 스텝은 미측정이니 안 돌린다」는 **틀린 판단**을
    떠받치고 있었다. 지우지 않고 **반대 방향으로** 잠근다.
    """

    def test_every_named_door_is_really_in_the_code(self):
        """★적어 둔 자리마다 그 줄에 문이 **있어야** 한다."""
        from pathlib import Path as _P

        backend = _P(tb.__file__).resolve().parents[2]
        named = tb.uncovered_paths()["text_doors"]
        assert named, "★문 자리를 안 적었다"
        for ref in named:
            rel, ln = ref.rsplit(":", 1)
            line = (backend / rel).read_text(
                encoding="utf-8").splitlines()[int(ln) - 1]
            assert "reserve_current_research_call" in line, (
                f"{ref} 에 문이 없다 — 글과 코드가 갈라졌다")

    def test_it_admits_that_images_do_not_pass_here(self):
        """★★글 예산이 **이미지를 못 센다**고 스스로 말해야 한다.

        이것을 안 적어 두면 「글 장부 0」을 「아무것도 안 샀다」로 읽는다.
        """
        got = tb.uncovered_paths()
        assert "이미지" in got["★covers"]
        assert "image_call_budget" in got["image_gate"]

    def test_the_snapshot_says_it_is_counted_not_raw(self):
        with tb.canary_text_scope(cap=1) as b:
            pass
        got = tb.snapshot_of(b)
        assert "counted" in got["★means"] and "SDK" in got["★means"]


class TestItRefusesToOverlap:
    """★모듈 전역을 바꾸는 문이다 — 겹쳐 쓰면 남의 수를 먹는다."""

    def test_nesting_stops(self):
        with tb.canary_text_scope(cap=1):
            with pytest.raises(RuntimeError):
                with tb.canary_text_scope(cap=1):
                    pass

    def test_an_existing_budget_stops(self):
        from app.core.research_call_budget import (ResearchCallBudget,
                                                   install_budget,
                                                   uninstall_budget)

        install_budget(ResearchCallBudget(cap=5))
        try:
            with pytest.raises(RuntimeError):
                with tb.canary_text_scope(cap=1):
                    pass
        finally:
            uninstall_budget()

    def test_it_still_restores_after_a_refusal(self):
        import threading

        from app.core import research_call_budget as rb

        with tb.canary_text_scope(cap=1):
            try:
                with tb.canary_text_scope(cap=1):
                    pass
            except RuntimeError:
                pass
        assert isinstance(rb._local, threading.local), "★원래대로 안 왔다"


# ─────────────────────────────────────────────────────────────────────
# ★★★**네 경계 모두** 같은 문을 갖는다 (Codex BLOCK 2, 2026-08-31)
#
# > 정적 추적을 더 쌓지 말고 그 **세 실제 dispatch 직전**에도
# > `reserve_current_research_call` 을 거십시오. armed 밖에서는 no-op 이라
# > 기존 경로 영향 0 이고, 네 text 경계가 모두 같은 문을 갖게 됩니다.
# ─────────────────────────────────────────────────────────────────────


class TestEveryTextBoundaryHasTheSameDoor:
    def test_the_direct_path_is_counted_for_a_non_openai_model(self,
                                                               monkeypatch):
        """★경계 ③ — Router 를 안 거치는 직접 호출(비 OpenAI 모델)."""
        import litellm

        from app.core import openai_keys as ok

        sent = []
        monkeypatch.setattr(litellm, "completion",
                            lambda **kw: sent.append(kw) or {"ok": True})
        monkeypatch.setattr(ok, "_init_opik", lambda *a, **k: None,
                            raising=False)
        with tb.canary_text_scope(cap=2) as b:
            for _ in range(2):
                ok.llm_completion(model="gemini/무엇", messages=[])
            with pytest.raises(ResearchCallBudgetExceeded):
                ok.llm_completion(model="gemini/무엇", messages=[])
        assert len(sent) == 2, f"★상한을 넘겨 {len(sent)}번 나갔다"
        assert b.snapshot()["denied"] == 1

    def test_the_direct_path_is_counted_inside_the_slot_loop(self,
                                                             monkeypatch):
        """★★경계 ③ — 슬롯 loop **안**이라 논리 하나가 물리 둘이 될 수 있다."""
        import litellm

        from app.core import openai_keys as ok

        sent = []
        monkeypatch.setattr(litellm, "completion",
                            lambda **kw: sent.append(kw) or {"ok": True})
        monkeypatch.setattr(ok, "_is_openai_model", lambda m: True)
        monkeypatch.setattr(ok, "slot_count", lambda: 2)
        monkeypatch.setattr(ok, "active_slot_and_key",
                            lambda: ("slot-1", "키"))
        with tb.canary_text_scope(cap=1) as b:
            ok.llm_completion(model="gpt-무엇", messages=[])
            with pytest.raises(ResearchCallBudgetExceeded):
                ok.llm_completion(model="gpt-무엇", messages=[])
        assert len(sent) == 1
        assert b.snapshot()["used"] == 1 and b.snapshot()["denied"] == 1

    def test_a_missing_key_is_counted_too(self, monkeypatch):
        """★키가 없어 그냥 보내는 갈래도 **나가는 것은 나가는 것**이다."""
        import litellm

        from app.core import openai_keys as ok

        sent = []
        monkeypatch.setattr(litellm, "completion",
                            lambda **kw: sent.append(kw) or {"ok": True})
        monkeypatch.setattr(ok, "_is_openai_model", lambda m: True)
        monkeypatch.setattr(ok, "slot_count", lambda: 1)
        monkeypatch.setattr(ok, "active_slot_and_key", lambda: ("slot-1", ""))
        with tb.canary_text_scope(cap=1):
            ok.llm_completion(model="gpt-무엇", messages=[])
            with pytest.raises(ResearchCallBudgetExceeded):
                ok.llm_completion(model="gpt-무엇", messages=[])
        assert len(sent) == 1

    def test_outside_the_scope_the_direct_path_is_untouched(self,
                                                            monkeypatch):
        """★★팔을 안 들면 **한 글자도 안 바뀐다** — 기존 경로 영향 0."""
        import litellm

        from app.core import openai_keys as ok

        sent = []
        monkeypatch.setattr(litellm, "completion",
                            lambda **kw: sent.append(kw) or {"ok": True})
        monkeypatch.setattr(ok, "_init_opik", lambda *a, **k: None,
                            raising=False)
        for _ in range(5):
            ok.llm_completion(model="gemini/무엇", messages=[])
        assert len(sent) == 5

    def test_the_router_path_is_counted(self, monkeypatch):
        """★경계 ① — `router_completion` → `_completion` → 문."""
        from app.modules.llm import llm_client as lc

        sent = []

        class _B:
            slot = "slot-1"

            class router:
                @staticmethod
                def completion(**kw):
                    sent.append(kw)
                    return {"ok": True}

        from app.core import openai_keys as ok

        monkeypatch.setattr(lc, "_get_router_binding", lambda: _B())
        monkeypatch.setattr(ok, "slot_count", lambda: 1)
        with tb.canary_text_scope(cap=2):
            for _ in range(2):
                lc.router_completion(model="무엇", messages=[])
            with pytest.raises(ResearchCallBudgetExceeded):
                lc.router_completion(model="무엇", messages=[])
        assert len(sent) == 2