tests/test_llm.py

from types import SimpleNamespace

import pytest

from agent import llm


def test_retry_call_retries_then_succeeds():
    calls = {"n": 0}

    def fn():
        calls["n"] += 1
        if calls["n"] < 3:
            raise ValueError("transient")
        return "ok"

    out = llm.retry_call(fn, is_retryable=lambda e: isinstance(e, ValueError),
                         max_retries=5, base_delay=0, sleep=lambda s: None)
    assert out == "ok"
    assert calls["n"] == 3


def test_retry_call_non_retryable_raises_immediately():
    calls = {"n": 0}

    def fn():
        calls["n"] += 1
        raise TypeError("nope")

    with pytest.raises(TypeError):
        llm.retry_call(fn, is_retryable=lambda e: isinstance(e, ValueError),
                       max_retries=5, base_delay=0, sleep=lambda s: None)
    assert calls["n"] == 1


def test_retry_call_exhausts_and_raises_last():
    def fn():
        raise ValueError("always")

    with pytest.raises(ValueError):
        llm.retry_call(fn, is_retryable=lambda e: True,
                       max_retries=3, base_delay=0, sleep=lambda s: None)


def test_is_retryable_code():
    # 429 is intentionally NOT retried: retrying wastes scarce free-tier requests and a short
    # backoff cannot clear a per-minute window. Only transient server/timeout errors are retried.
    assert not llm.is_retryable_code(429)
    assert llm.is_retryable_code(408)
    assert llm.is_retryable_code(503)
    assert not llm.is_retryable_code(400)
    assert not llm.is_retryable_code(None)


def _quota_exc(quota_id):
    details = {"error": {"details": [
        {"@type": "type.googleapis.com/google.rpc.QuotaFailure",
         "violations": [{"quotaId": quota_id}]},
    ]}}
    return SimpleNamespace(code=429, details=details)


def test_map_persistent_error_defaults_429_to_daily():
    mapped = llm.map_persistent_error(SimpleNamespace(code=429, details=None))
    assert isinstance(mapped, llm.QuotaExhaustedError)
    assert "tomorrow" in str(mapped).lower()


def test_map_persistent_error_per_minute_says_wait_not_tomorrow():
    mapped = llm.map_persistent_error(
        _quota_exc("GenerateRequestsPerMinutePerProjectPerModel-FreeTier")
    )
    assert isinstance(mapped, llm.QuotaExhaustedError)
    msg = str(mapped).lower()
    assert "minute" in msg and "tomorrow" not in msg


def test_map_persistent_error_per_day_says_tomorrow():
    mapped = llm.map_persistent_error(
        _quota_exc("GenerateRequestsPerDayPerProjectPerModel-FreeTier")
    )
    assert "tomorrow" in str(mapped).lower()


def test_map_persistent_error_passes_other_through():
    err = SimpleNamespace(code=503)
    assert llm.map_persistent_error(err) is err