File size: 2,931 Bytes
58a7a73 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 | """A provider refusal must reach the user, not read as silence."""
import httpx
import numpy as np
import pytest
from ante.zerog.errors import ZeroGRequestError
from ante.zerog.language_model import _raise_for_error_body
from ante.zerog.speech_to_text import ZeroGSpeechToText, raise_for_provider_error
UNFUNDED_BODY = {
"error": (
"validate request: insufficient balance: your locked balance is 0.000000 0G, "
"but the required minimum is 1.000000 0G"
)
}
def test_clean_payload_passes_through() -> None:
"""A normal transcription result is not mistaken for an error."""
raise_for_provider_error({"text": "hello robot"})
def test_error_in_a_successful_body_is_raised() -> None:
"""Providers report refusals inside HTTP 200, where raise_for_status cannot see them."""
with pytest.raises(ZeroGRequestError, match="insufficient balance"):
raise_for_provider_error(UNFUNDED_BODY)
def test_structured_error_object_is_unwrapped() -> None:
"""Some providers nest the message inside an error object."""
with pytest.raises(ZeroGRequestError, match="model overloaded"):
raise_for_provider_error({"error": {"message": "model overloaded"}})
@pytest.mark.asyncio
async def test_unfunded_transcription_raises_rather_than_returning_empty() -> None:
"""An unfunded sub-account must not look like an utterance nobody could hear."""
def respond(request: httpx.Request) -> httpx.Response:
return httpx.Response(200, json=UNFUNDED_BODY)
client = httpx.AsyncClient(transport=httpx.MockTransport(respond))
speech_to_text = ZeroGSpeechToText("http://0g.local/v1", "openai/whisper-large-v3", client)
with pytest.raises(ZeroGRequestError, match="insufficient balance"):
await speech_to_text.transcribe(np.zeros(1600, dtype=np.int16), 16000)
@pytest.mark.asyncio
async def test_successful_transcription_returns_text() -> None:
"""The happy path still works."""
def respond(request: httpx.Request) -> httpx.Response:
return httpx.Response(200, json={"text": " what is the capital of France? "})
client = httpx.AsyncClient(transport=httpx.MockTransport(respond))
speech_to_text = ZeroGSpeechToText("http://0g.local/v1", "openai/whisper-large-v3", client)
assert await speech_to_text.transcribe(np.zeros(1600, dtype=np.int16), 16000) == ("what is the capital of France?")
def test_chat_error_body_is_raised() -> None:
"""A refusal returned in place of an event stream must not be parsed as content."""
with pytest.raises(ZeroGRequestError, match="insufficient balance"):
_raise_for_error_body('{"error": "insufficient balance"}')
def test_unreadable_chat_body_is_reported() -> None:
"""An HTML error page must produce a clear failure, not a JSON traceback."""
with pytest.raises(ZeroGRequestError, match="unreadable"):
_raise_for_error_body("<html>502 Bad Gateway</html>")
|