Testing Exceptions

import pytest

from ragkit.chunking import chunk
from ragkit.errors import InputValidationError


def test_overlap_must_be_smaller_than_size():
    with pytest.raises(ValueError, match=r"overlap \(64\) must be smaller than size \(32\)"):
        chunk("some text", size=32, overlap=64)


def test_domain_error_is_also_a_value_error():
    with pytest.raises(InputValidationError):
        chunk("", size=32, overlap=0)

def test_missing_variable(prompt_registry):
    template = prompt_registry.get("qa")                   # outside: must not raise
    with pytest.raises(InputValidationError, match="missing variables: context"):
        template.render(question="What is RAG?")

import re

model = "gpt-4o (2024-08-06)"
with pytest.raises(ProviderError, match=re.escape(f"unknown model {model!r}")):
    get_client(model)

def test_rate_limit_is_translated_with_retry_after(fake_openai):
    fake_openai.fail_with(status=429, headers={"retry-after": "7"})

    with pytest.raises(RateLimitError) as excinfo:
        client.chat([{"role": "user", "content": "hi"}], model="m")

    err = excinfo.value
    assert err.retryable is True
    assert err.retry_after == 7.0
    assert err.provider == "openai"
    assert isinstance(err.__cause__, openai.RateLimitError)     # chained with `from err`


def test_note_added_per_batch():
    with pytest.raises(ProviderError) as excinfo:
        embed_all(["a", "b"], embedder=AlwaysFailing())
    assert any("batch 0" in note for note in excinfo.value.__notes__)   # 3.11+


def test_lookup_hides_internal_key_error():
    with pytest.raises(ConfigurationError) as excinfo:
        price_for("no-such-model")
    assert excinfo.value.__cause__ is None
    assert excinfo.value.__suppress_context__ is True          # raised `from None`

async def test_taskgroup_reports_every_failure():
    with pytest.raises(ExceptionGroup) as excinfo:
        await fan_out([fail_with(RateLimitError("429", provider="a")), fail_with(ValueError("bad"))])

    group = excinfo.value
    assert group.subgroup(RateLimitError) is not None
    assert group.subgroup(ValueError) is not None
    assert len(group.exceptions) == 2

def test_group_contains():
    with pytest.raises(ExceptionGroup) as excinfo:
        validate_tools([{"name": "search"}, {"name": 3}])
    assert excinfo.group_contains(ValueError, match="tool=3")
    assert excinfo.group_contains(ValueError, depth=1)            # only direct children


def test_exact_group_shape():                                     # pytest >= 8.4
    with pytest.RaisesGroup(ValueError, ValueError, match="2 invalid tool schemas"):
        validate_tools([{"name": 1}, {"name": 2}])

def test_engine_argument_is_deprecated():
    with pytest.warns(DeprecationWarning, match="`engine` is deprecated"):
        chat("hi", engine="old-model")


def test_no_warnings_on_happy_path(recwarn):
    chat("hi", model="small")
    assert not recwarn.list

# pyproject.toml
[tool.pytest.ini_options]
filterwarnings = [
    "error",
    "ignore::DeprecationWarning:some_third_party.*",
]

import logging


def test_fallback_is_logged_with_traceback(caplog, failing_store):
    with caplog.at_level(logging.WARNING, logger="ragkit.retrieval"):
        docs = retrieve("what is hybrid search?", store=failing_store)

    assert docs                                                    # BM25 fallback returned results
    record = next(r for r in caplog.records if "falling back to BM25" in r.getMessage())
    assert record.levelno == logging.WARNING
    assert record.exc_info is not None                             # traceback attached
    assert isinstance(record.exc_info[1], VectorStoreError)

import itertools


class Flaky:
    """Callable that raises the given errors in order, then returns `result`."""

    def __init__(self, *errors: BaseException, result="ok"):
        self.errors = list(errors)
        self.result = result
        self.calls = 0

    async def __call__(self, *args, **kwargs):
        self.calls += 1
        if self.errors:
            raise self.errors.pop(0)
        return self.result


@pytest.fixture
def no_sleep(monkeypatch):
    slept = []

    async def fake_sleep(delay):
        slept.append(delay)

    monkeypatch.setattr("ragkit.retry.asyncio.sleep", fake_sleep)
    return slept


async def test_retries_transient_then_succeeds(no_sleep):
    call = Flaky(ProviderOverloadedError("503", provider="p"), RateLimitError("429", provider="p", retry_after=2))
    assert await retry_async(call, attempts=5) == "ok"
    assert call.calls == 3
    assert no_sleep[1] == 2                                        # honoured Retry-After


async def test_does_not_retry_permanent_errors(no_sleep):
    call = Flaky(ContextLengthError("too long", provider="p", tokens=9000, limit=8192))
    with pytest.raises(ContextLengthError):
        await retry_async(call, attempts=5)
    assert call.calls == 1
    assert no_sleep == []


async def test_gives_up_with_original_error(no_sleep):
    errors = [ProviderOverloadedError("503", provider="p") for _ in range(5)]
    with pytest.raises(ProviderOverloadedError):
        await retry_async(Flaky(*errors), attempts=3)


async def test_fallback_chain_uses_second_model(monkeypatch):
    monkeypatch.setitem(CLIENTS, "openai", FakeClient(fail=ProviderOverloadedError("529", provider="openai")))
    monkeypatch.setitem(CLIENTS, "anthropic", FakeClient(reply="answer"))
    assert await complete_with_fallback([{"role": "user", "content": "q"}]) == "answer"

@pytest.mark.parametrize(
    ("status", "expected", "retryable"),
    [(429, RateLimitError, True), (500, ProviderOverloadedError, True), (503, ProviderOverloadedError, True),
     (400, ProviderError, False), (401, ProviderAuthError, False)],
)
def test_status_translation(fake_openai, status, expected, retryable):
    fake_openai.fail_with(status=status)
    with pytest.raises(expected) as excinfo:
        client.chat([], model="m")
    assert excinfo.type is expected
    assert is_retryable(excinfo.value) is retryable

async def test_repair_loop_feeds_errors_back(fake_llm):
    fake_llm.replies = ['{"company": "Acme", "amount": "lots"}', '{"company": "Acme", "amount": 12, "currency": "USD"}']
    result = await extract("invoice text", model="m", max_repairs=2)
    assert result.amount == 12
    repair_prompt = fake_llm.requests[1][-1]["content"]
    assert "amount" in repair_prompt                    # validation error was sent back


async def test_repair_loop_gives_up(fake_llm):
    fake_llm.replies = ["not json"] * 3
    with pytest.raises(StructuredOutputError, match="schema validation"):
        await extract("invoice text", model="m", max_repairs=2)
    assert len(fake_llm.requests) == 3

def test_encode_halves_batch_on_oom(monkeypatch):
    seen = []

    class FakeModel:
        def encode(self, batch, convert_to_tensor=True):
            seen.append(len(batch))
            if len(batch) > 2:
                raise torch.cuda.OutOfMemoryError("CUDA out of memory")
            return torch.zeros(len(batch), 4)

    monkeypatch.setattr(torch.cuda, "empty_cache", lambda: None)
    out = encode_adaptive(FakeModel(), ["t"] * 5, batch_size=8)
    assert out.shape == (5, 4)
    assert seen[:3] == [5, 4, 2]

import asyncio


async def test_stream_reraises_cancellation_and_closes(fake_stream):
    task = asyncio.create_task(consume(stream_tokens(fake_stream, "hi")))
    await asyncio.sleep(0)                          # let it start
    task.cancel()
    with pytest.raises(asyncio.CancelledError):
        await task
    assert fake_stream.closed                       # cleanup ran


async def test_deadline_raises_domain_timeout(slow_llm):
    with pytest.raises(ProviderTimeoutError) as excinfo:
        await chat_with_deadline([{"role": "user", "content": "hi"}], seconds=0.01)
    assert isinstance(excinfo.value.__cause__, TimeoutError)

import pickle

import pytest

from ragkit import errors

ALL_ERRORS = [cls for cls in vars(errors).values()
              if isinstance(cls, type) and issubclass(cls, errors.RagkitError)]


@pytest.mark.parametrize("cls", ALL_ERRORS, ids=lambda c: c.__name__)
def test_every_error_derives_from_base_and_pickles(cls):
    err = cls("boom")
    assert isinstance(err, errors.RagkitError)
    assert str(err).startswith("boom")
    clone = pickle.loads(pickle.dumps(err))
    assert type(clone) is cls and clone.message == err.message


@pytest.mark.parametrize("cls", ALL_ERRORS, ids=lambda c: c.__name__)
def test_every_error_has_http_status(cls):
    assert 400 <= errors.http_status_for(cls("x")) <= 599


def test_retryable_flags():
    assert errors.RateLimitError("x").retryable
    assert not errors.ContextLengthError("x").retryable
    assert errors.is_retryable(TimeoutError())
    assert not errors.is_retryable(KeyError("k"))

from fastapi import FastAPI
from fastapi.testclient import TestClient

from ragkit.errors import RateLimitError, register_exception_handlers

app = FastAPI()
register_exception_handlers(app)


@app.get("/boom")
async def boom():
    raise RateLimitError("slow down", retry_after=2.5, provider="openai")


def test_rate_limit_maps_to_429_with_retry_after():
    client = TestClient(app)
    resp = client.get("/boom")
    assert resp.status_code == 429
    assert resp.headers["retry-after"] == "3"
    assert resp.json()["error"] == "rate_limit"


def test_unexpected_errors_do_not_leak_internals():
    client = TestClient(app, raise_server_exceptions=False)
    resp = client.get("/crash")                     # route raising KeyError("api_key=sk-...")
    assert resp.status_code == 500
    assert "sk-" not in resp.text

def test_cli_reports_domain_error(capsys, monkeypatch):
    def fail(args):
        raise ConfigurationError("no API key")

    monkeypatch.setattr("ragkit.cli.run", fail)
    assert main(["ask", "hi"]) == 1
    assert "error: no API key" in capsys.readouterr().err


def test_argparse_errors_exit_2():
    with pytest.raises(SystemExit) as excinfo:
        main(["--no-such-flag"])
    assert excinfo.value.code == 2