iomgaa
e302247022
Issue #10 Task 1: a rejected call's reason had nowhere to live. The field goes on the base class because these errors all come from one HTTP response - which class it is and what the peer said are orthogonal.
180 lines
7.1 KiB
Python
180 lines
7.1 KiB
Python
"""errors.py 错误四分类与 scope 级不可用语义测试(M1 设计 §3)。"""
|
|
|
|
import pytest
|
|
|
|
from polygateway.errors import (
|
|
GOVERNANCE_BACKEND_RETRY_AFTER_S,
|
|
SCOPE_REASONS,
|
|
AllSourcesExhausted,
|
|
CircuitOpenError,
|
|
GatewayUnavailableError,
|
|
GovernanceBackendError,
|
|
PolyGatewayError,
|
|
RequestRejectedError,
|
|
ResultInvalidError,
|
|
SourceDeadError,
|
|
SourceNotConfiguredError,
|
|
TransientError,
|
|
)
|
|
|
|
|
|
class TestBaseShape:
|
|
def test_base_carries_source_context(self):
|
|
exc = PolyGatewayError("boom", source_name="qwen_1", status_code=502, operation="chat")
|
|
assert exc.source_name == "qwen_1"
|
|
assert exc.status_code == 502
|
|
assert exc.operation == "chat"
|
|
|
|
def test_four_way_taxonomy_inherits_base(self):
|
|
for cls in (TransientError, SourceDeadError, RequestRejectedError, ResultInvalidError):
|
|
assert issubclass(cls, PolyGatewayError)
|
|
|
|
def test_transient_retry_after_optional(self):
|
|
assert TransientError("429").retry_after_s is None
|
|
assert TransientError("429", retry_after_s=2.5).retry_after_s == 2.5
|
|
|
|
|
|
class TestBodyText:
|
|
"""issue #10: 非 2xx 的响应体摘要必须有承载处,否则拒绝理由事后不可查。"""
|
|
|
|
@pytest.mark.parametrize(
|
|
"cls", (PolyGatewayError, TransientError, SourceDeadError, RequestRejectedError)
|
|
)
|
|
def test_defaults_empty_and_accepts_summary(self, cls):
|
|
assert cls("boom").body_text == ""
|
|
assert cls("boom", body_text='{"error":{"code":"bad"}}').body_text == (
|
|
'{"error":{"code":"bad"}}'
|
|
)
|
|
|
|
def test_result_invalid_keeps_both_fields_apart(self):
|
|
"""`body_text`(非 2xx 的拒绝理由)与 `raw_text`(2xx 的不可解析输出)不得混用。"""
|
|
exc = ResultInvalidError("bad json", raw_text="{oops", body_text="")
|
|
assert exc.raw_text == "{oops"
|
|
assert exc.body_text == ""
|
|
|
|
@pytest.mark.parametrize(
|
|
"exc",
|
|
(
|
|
AllSourcesExhausted(scope="llm", reason="stalled", retry_after_s=1.0),
|
|
CircuitOpenError(scope="llm", retry_after_s=1.0),
|
|
GovernanceBackendError("redis down", scope="llm"),
|
|
),
|
|
)
|
|
def test_scope_level_errors_carry_no_body(self, exc):
|
|
"""scope 级失败没有单一响应体可言,空串是如实表达而非噪音。"""
|
|
assert exc.body_text == ""
|
|
|
|
def test_body_text_does_not_leak_into_str(self):
|
|
"""字段是旁路数据: 加了它不得改变任何既有异常的 str() 输出。"""
|
|
assert str(RequestRejectedError("qwen_1 请求被拒: 400", body_text="whatever")) == (
|
|
"qwen_1 请求被拒: 400"
|
|
)
|
|
|
|
|
|
class TestResultInvalid:
|
|
def test_carries_diagnosis(self):
|
|
exc = ResultInvalidError(
|
|
"bad json",
|
|
raw_text="{oops",
|
|
repair_error="unterminated",
|
|
validation_errors=("field x missing",),
|
|
)
|
|
assert exc.raw_text == "{oops"
|
|
assert exc.repair_error == "unterminated"
|
|
assert exc.validation_errors == ("field x missing",)
|
|
|
|
|
|
class TestGatewayUnavailable:
|
|
def test_fields_and_inheritance(self):
|
|
exc = AllSourcesExhausted(
|
|
scope="LLM",
|
|
reason="retry_exhausted",
|
|
retry_after_s=4.0,
|
|
per_source_reasons={"qwen_1": "timeout"},
|
|
)
|
|
assert isinstance(exc, GatewayUnavailableError)
|
|
assert exc.scope == "llm" # CHS 同款: scope 归一化小写
|
|
assert exc.reason == "retry_exhausted"
|
|
assert exc.retry_after_s == 4.0
|
|
assert exc.per_source_reasons == {"qwen_1": "timeout"}
|
|
|
|
def test_circuit_open_reason_fixed(self):
|
|
exc = CircuitOpenError(scope="LLM", retry_after_s=30.0)
|
|
assert exc.reason == "circuit_open"
|
|
assert isinstance(exc, GatewayUnavailableError)
|
|
|
|
def test_scope_reason_domain_enforced(self):
|
|
with pytest.raises(ValueError):
|
|
AllSourcesExhausted(scope="LLM", reason="bad_reason", retry_after_s=0.0)
|
|
|
|
def test_per_source_reason_domain_enforced(self):
|
|
with pytest.raises(ValueError):
|
|
AllSourcesExhausted(
|
|
scope="LLM",
|
|
reason="no_sources",
|
|
retry_after_s=0.0,
|
|
per_source_reasons={"qwen_1": "weird"},
|
|
)
|
|
|
|
def test_retry_after_non_negative_and_scope_non_empty(self):
|
|
with pytest.raises(ValueError):
|
|
AllSourcesExhausted(scope="LLM", reason="stalled", retry_after_s=-1.0)
|
|
with pytest.raises(ValueError):
|
|
AllSourcesExhausted(scope=" ", reason="stalled", retry_after_s=0.0)
|
|
|
|
|
|
class TestBackendFailure:
|
|
def test_governance_backend_error_is_not_transient(self):
|
|
"""限流/熔断后端故障必须报错不放行,且不落入可重试分类。"""
|
|
exc = GovernanceBackendError("redis down", scope="llm")
|
|
assert isinstance(exc, PolyGatewayError)
|
|
assert not isinstance(exc, TransientError)
|
|
|
|
def test_is_scope_level_unavailability(self):
|
|
"""fail-closed 时整个 scope 一个请求都发不出去,调用方一条 except 应覆盖(issue #7)。"""
|
|
exc = GovernanceBackendError("限流后端 try_acquire 失败: boom", scope="LLM")
|
|
assert isinstance(exc, GatewayUnavailableError)
|
|
assert exc.reason == "governance_backend_down"
|
|
assert exc.scope == "llm" # 与既有 scope 级异常同款: 归一化小写
|
|
assert exc.retry_after_s == GOVERNANCE_BACKEND_RETRY_AFTER_S
|
|
|
|
def test_diagnostic_message_survives_reparenting(self):
|
|
"""父类把 message 覆写为模板串,而各构造点的诊断串是排障主线索(§3.5)。"""
|
|
exc = GovernanceBackendError("熔断后端 try_enter 失败: boom", scope="llm")
|
|
assert str(exc) == "熔断后端 try_enter 失败: boom"
|
|
|
|
def test_retry_after_overridable(self):
|
|
exc = GovernanceBackendError("redis down", scope="llm", retry_after_s=30.0)
|
|
assert exc.retry_after_s == 30.0
|
|
|
|
|
|
class TestSourceNotConfigured:
|
|
"""装配缺陷有意留在 scope 级家族之外(issue #7 §3.4,Q1 人类拍板)。"""
|
|
|
|
def test_is_domain_error_but_not_scope_level(self):
|
|
exc = SourceNotConfiguredError("未知源 'nope'(scope=llm)")
|
|
assert isinstance(exc, PolyGatewayError)
|
|
# 关键断言: 归入可重投家族会让配置写错的任务永远重投、永不进死信
|
|
assert not isinstance(exc, GatewayUnavailableError)
|
|
|
|
def test_exported_at_package_top_level(self):
|
|
import polygateway
|
|
|
|
assert polygateway.SourceNotConfiguredError is SourceNotConfiguredError
|
|
assert "SourceNotConfiguredError" in polygateway.__all__
|
|
|
|
|
|
class TestGovernanceBackendReason:
|
|
"""新 scope 级 reason 值域(issue #7 §3.1)。"""
|
|
|
|
def test_reason_admitted_to_scope_domain(self):
|
|
assert "governance_backend_down" in SCOPE_REASONS
|
|
|
|
def test_gateway_unavailable_accepts_the_new_reason(self):
|
|
exc = AllSourcesExhausted(scope="LLM", reason="governance_backend_down", retry_after_s=0.0)
|
|
assert exc.reason == "governance_backend_down"
|
|
|
|
def test_retry_after_default_is_non_zero(self):
|
|
"""取 0 会让积压任务零延迟冲击已挂掉的后端(§3.2)。"""
|
|
assert GOVERNANCE_BACKEND_RETRY_AFTER_S > 0
|