import asyncio
import threading
from types import SimpleNamespace

import pydantic
import pytest

from pageindex.config import (
    IndexConfig,
    _env_max_concurrency_default,
    get_max_concurrency,
    max_concurrency_scope,
    set_max_concurrency,
)
from pageindex.index.utils import _llm_semaphore, llm_acompletion


@pytest.fixture(autouse=True)
def _restore_max_concurrency():
    """Keep tests isolated — the concurrency setting is a module global."""
    prev = get_max_concurrency()
    yield
    set_max_concurrency(prev)


async def _nested_llm_load(state, *, branches=5, leaves=5):
    """Drive branches*leaves leaf calls, nested two levels deep, each holding
    the shared per-loop LLM semaphore — the exact shape of the indexing pipeline
    (tree_parser gather -> per-node gather -> leaf LLM call)."""

    async def leaf():
        async with _llm_semaphore():
            state["in_flight"] += 1
            state["peak"] = max(state["peak"], state["in_flight"])
            await asyncio.sleep(0.01)
            state["in_flight"] -= 1

    async def branch():
        await asyncio.gather(*(leaf() for _ in range(leaves)))

    await asyncio.gather(*(branch() for _ in range(branches)))


def test_llm_semaphore_bounds_concurrency_even_when_nested():
    # The core fix: the cap is a TRUE global bound even when acquired from
    # deeply nested gathers. 25 leaf calls nested two levels, cap 3 -> peak 3.
    # A per-gather-call semaphore (the previous design) would let this reach
    # branches*leaves and blow past the cap.
    set_max_concurrency(3)
    state = {"in_flight": 0, "peak": 0}
    asyncio.run(_nested_llm_load(state, branches=5, leaves=5))
    assert state["peak"] == 3


def test_llm_semaphore_uses_scoped_override():
    # A per-index max_concurrency_scope active when the loop's semaphore is first
    # created must set its size, and must not mutate the process default.
    set_max_concurrency(10)
    state = {"in_flight": 0, "peak": 0}

    async def run():
        with max_concurrency_scope(2):
            await _nested_llm_load(state, branches=4, leaves=4)

    asyncio.run(run())
    assert state["peak"] == 2
    assert get_max_concurrency() == 10


def test_llm_acompletion_holds_the_shared_semaphore(monkeypatch):
    # Prove llm_acompletion (the single chokepoint every LLM call funnels
    # through) actually acquires the shared cap around the network call.
    set_max_concurrency(3)
    state = {"in_flight": 0, "peak": 0}

    async def fake_acompletion(**kwargs):
        state["in_flight"] += 1
        state["peak"] = max(state["peak"], state["in_flight"])
        await asyncio.sleep(0.01)
        state["in_flight"] -= 1
        return SimpleNamespace(
            choices=[SimpleNamespace(message=SimpleNamespace(content="ok"))]
        )

    monkeypatch.setattr("litellm.acompletion", fake_acompletion)

    async def run():
        await asyncio.gather(*(llm_acompletion("gpt-x", f"p{i}") for i in range(20)))

    asyncio.run(run())
    assert state["peak"] == 3


def test_run_async_propagates_scope_into_worker_thread():
    # When build_index runs inside an already-running loop, _run_async hops to a
    # worker thread. The max_concurrency_scope override must ride along (copied
    # context) and still bound the (nested) LLM load in that worker loop.
    from pageindex.index.pipeline import _run_async

    set_max_concurrency(10)
    state = {"in_flight": 0, "peak": 0}

    async def outer():
        # We're inside a running loop -> _run_async uses the worker thread.
        with max_concurrency_scope(3):
            _run_async(_nested_llm_load(state, branches=4, leaves=4))

    asyncio.run(outer())
    assert state["peak"] == 3


def test_set_get_max_concurrency_round_trip():
    set_max_concurrency(3)
    assert get_max_concurrency() == 3


def test_set_max_concurrency_rejects_invalid():
    # bool is an int subclass -> must be rejected, not silently -> Semaphore(1).
    for bad in (0, -1, True, False, 2.5, "3", None):
        with pytest.raises(ValueError):
            set_max_concurrency(bad)


def test_env_default_parsing(monkeypatch):
    monkeypatch.delenv("PAGEINDEX_MAX_CONCURRENCY", raising=False)
    assert _env_max_concurrency_default() == 5
    monkeypatch.setenv("PAGEINDEX_MAX_CONCURRENCY", "20")
    assert _env_max_concurrency_default() == 20
    monkeypatch.setenv("PAGEINDEX_MAX_CONCURRENCY", "garbage")
    assert _env_max_concurrency_default() == 5
    monkeypatch.setenv("PAGEINDEX_MAX_CONCURRENCY", "0")
    assert _env_max_concurrency_default() == 5


def test_index_config_max_concurrency_field():
    # Default is None → "use the global/env default"; explicit value overrides.
    assert IndexConfig().max_concurrency is None
    assert IndexConfig(max_concurrency=7).max_concurrency == 7


def test_index_config_rejects_bool_and_non_positive_max_concurrency():
    # bool would otherwise be coerced by pydantic to 1/0; both must be rejected.
    for bad in (True, False, 0, -1):
        with pytest.raises(pydantic.ValidationError):
            IndexConfig(max_concurrency=bad)


def test_max_concurrency_scope_overrides_then_restores():
    # A per-index override applies inside the scope and, crucially, does NOT
    # stick as the new process default afterwards (no stickiness).
    set_max_concurrency(10)
    with max_concurrency_scope(3):
        assert get_max_concurrency() == 3
    assert get_max_concurrency() == 10


def test_max_concurrency_scope_none_is_a_no_op():
    set_max_concurrency(8)
    with max_concurrency_scope(None):
        assert get_max_concurrency() == 8
    assert get_max_concurrency() == 8


def test_max_concurrency_scope_rejects_invalid():
    for bad in (0, -1, True, False):
        with pytest.raises(ValueError):
            with max_concurrency_scope(bad):
                pass


def test_max_concurrency_scope_is_isolated_across_threads():
    # A per-index override in one indexing thread must not leak into another
    # thread indexing a different document concurrently. The override is a
    # ContextVar, so it's invisible outside its own context.
    set_max_concurrency(10)
    seen = {}
    barrier = threading.Barrier(2)

    def worker():
        with max_concurrency_scope(2):
            barrier.wait()            # let main read while we're inside the scope
            seen["worker"] = get_max_concurrency()
            barrier.wait()

    t = threading.Thread(target=worker)
    t.start()
    barrier.wait()
    seen["main"] = get_max_concurrency()
    barrier.wait()
    t.join()

    assert seen["worker"] == 2   # worker sees its own scoped override
    assert seen["main"] == 10    # main is unaffected by the worker's scope


def test_utils_star_import_does_not_leak_config_name():
    # `from .utils import *` (used by the page_index modules) must not export a
    # name `config` that would shadow the real pageindex.config submodule for
    # those modules. The SimpleNamespace alias is now `_config`.
    ns = {}
    exec("from pageindex.index.utils import *", ns)
    assert "config" not in ns