feat: add frozen core types and error taxonomy

This commit is contained in:
2026-07-20 06:32:14 -04:00
parent fcadd8cd8e
commit 9f177d6d64
4 changed files with 542 additions and 0 deletions
+85
View File
@@ -0,0 +1,85 @@
"""errors.py 错误四分类与 scope 级不可用语义测试(M1 设计 §3)。"""
import pytest
from polygateway.errors import (
AllSourcesExhausted,
CircuitOpenError,
GatewayUnavailableError,
GovernanceBackendError,
PolyGatewayError,
RequestRejectedError,
ResultInvalidError,
SourceDeadError,
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 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")
assert isinstance(exc, PolyGatewayError)
assert not isinstance(exc, TransientError)