133 lines
4.2 KiB
Python
133 lines
4.2 KiB
Python
import httpx
|
|
import pytest
|
|
from openai import APITimeoutError, AuthenticationError, InternalServerError, RateLimitError
|
|
|
|
from apps.modelhub.crypto import encrypt_secret
|
|
from apps.modelhub.models import ModelDeployment, ModelInvocation, ModelProvider
|
|
from apps.modelhub.services import CompletionResult, execute_invocation
|
|
|
|
|
|
def create_invocation():
|
|
provider = ModelProvider.objects.create(
|
|
name="模型渠道",
|
|
base_url="https://models.example.com/v1",
|
|
api_key_encrypted=encrypt_secret("secret"),
|
|
)
|
|
deployment = ModelDeployment.objects.create(
|
|
provider=provider,
|
|
display_name="支持模型",
|
|
model_name="support-model",
|
|
)
|
|
return ModelInvocation.objects.create(
|
|
deployment=deployment,
|
|
input_snapshot={"messages": [{"role": "user", "content": "hello"}]},
|
|
)
|
|
|
|
|
|
def test_transient_error_retries_once(db, monkeypatch):
|
|
invocation = create_invocation()
|
|
calls = []
|
|
retries = []
|
|
|
|
def request(*args, **kwargs):
|
|
calls.append(1)
|
|
if len(calls) == 1:
|
|
raise APITimeoutError(request=None)
|
|
return CompletionResult("ok", 3, 1)
|
|
|
|
monkeypatch.setattr("apps.modelhub.services._request_completion", request)
|
|
monkeypatch.setattr("apps.modelhub.services.sleep", lambda *_: None)
|
|
|
|
execute_invocation(
|
|
invocation,
|
|
invocation.input_snapshot["messages"],
|
|
on_retry=lambda retry_number, code: retries.append((retry_number, code)),
|
|
)
|
|
|
|
invocation.refresh_from_db()
|
|
assert len(calls) == 2
|
|
assert invocation.status == "succeeded"
|
|
assert invocation.retry_count == 1
|
|
assert retries == [(1, "timeout")]
|
|
|
|
|
|
def test_non_transient_error_fails_without_exposing_raw_error(db, monkeypatch):
|
|
invocation = create_invocation()
|
|
|
|
def request(*args, **kwargs):
|
|
raise ValueError("secret provider detail")
|
|
|
|
monkeypatch.setattr("apps.modelhub.services._request_completion", request)
|
|
|
|
execute_invocation(invocation, invocation.input_snapshot["messages"])
|
|
|
|
invocation.refresh_from_db()
|
|
assert invocation.status == "failed"
|
|
assert invocation.error_code == "unexpected_error"
|
|
assert "secret provider detail" not in invocation.error_message
|
|
|
|
|
|
def test_agent_time_budget_does_not_repeat_a_full_timeout(db, monkeypatch):
|
|
invocation = create_invocation()
|
|
calls = []
|
|
clock = iter([0.0, 0.0, 19.5, 20.0])
|
|
|
|
def request(*args, **kwargs):
|
|
calls.append(kwargs["timeout_seconds"])
|
|
raise APITimeoutError(request=None)
|
|
|
|
monkeypatch.setattr("apps.modelhub.services._request_completion", request)
|
|
monkeypatch.setattr("apps.modelhub.services.monotonic", lambda: next(clock))
|
|
|
|
execute_invocation(
|
|
invocation,
|
|
invocation.input_snapshot["messages"],
|
|
total_timeout_seconds=20,
|
|
)
|
|
|
|
invocation.refresh_from_db()
|
|
assert calls == [20]
|
|
assert invocation.status == "failed"
|
|
assert invocation.error_code == "timeout"
|
|
assert invocation.retry_count == 0
|
|
|
|
|
|
def provider_error(error_type, status_code):
|
|
request = httpx.Request("POST", "https://models.example.com/v1/chat/completions")
|
|
response = httpx.Response(status_code, request=request)
|
|
return error_type("provider raw error", response=response, body={})
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("error", "expected_code", "expected_calls"),
|
|
[
|
|
(provider_error(AuthenticationError, 401), "authentication_failed", 1),
|
|
(provider_error(RateLimitError, 429), "rate_limited", 2),
|
|
(provider_error(InternalServerError, 500), "provider_unavailable", 2),
|
|
],
|
|
)
|
|
def test_provider_error_classification_and_retry(
|
|
db,
|
|
monkeypatch,
|
|
error,
|
|
expected_code,
|
|
expected_calls,
|
|
):
|
|
invocation = create_invocation()
|
|
calls = []
|
|
|
|
def request(*args, **kwargs):
|
|
calls.append(1)
|
|
raise error
|
|
|
|
monkeypatch.setattr("apps.modelhub.services._request_completion", request)
|
|
monkeypatch.setattr("apps.modelhub.services.sleep", lambda *_: None)
|
|
|
|
execute_invocation(invocation, invocation.input_snapshot["messages"])
|
|
|
|
invocation.refresh_from_db()
|
|
assert invocation.status == "failed"
|
|
assert invocation.error_code == expected_code
|
|
assert len(calls) == expected_calls
|
|
assert "provider raw error" not in invocation.error_message
|