xsl
93b88db2f5
Covers routing, failover, the circuit breaker, structured-output retries, the cache-prefix validator, cost computation including cached-token pricing, the daily budget breaker, the quota protocol, the task state machine and log redaction. HTTP is mocked -- no unit test spends money. The integration test runs the real slice against PostgreSQL and Redis with a faked provider.
97 lines
3.2 KiB
Python
Executable File
97 lines
3.2 KiB
Python
Executable File
"""Unified response and error codes (架构文档 §9.2, FR-SYS-02)."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from packages.shared.errors import (
|
|
ERROR_META,
|
|
AppError,
|
|
ErrorCode,
|
|
QuotaExhaustedError,
|
|
)
|
|
from packages.shared.response import fail, ok
|
|
from packages.shared.security import (
|
|
AccessTokenClaims,
|
|
create_access_token,
|
|
decode_access_token,
|
|
)
|
|
|
|
|
|
def test_success_envelope_shape() -> None:
|
|
body = ok({"a": 1})
|
|
assert body["code"] == 0
|
|
assert body["message"] == "ok"
|
|
assert body["data"] == {"a": 1}
|
|
|
|
|
|
def test_error_envelope_carries_message_and_hint() -> None:
|
|
body = fail(ErrorCode.QUOTA_EXHAUSTED, data={"required": 1, "remaining": 0})
|
|
assert body["code"] == 40301
|
|
assert body["message"] == "今日额度已用完"
|
|
assert body["hint"]
|
|
|
|
|
|
def test_quota_hint_never_mentions_price_or_purchase() -> None:
|
|
"""FR-BILL-07: 提示中不得出现任何价格、购买、会员、升级字样。"""
|
|
forbidden = ("价格", "购买", "会员", "升级", "元", "付费", "充值")
|
|
for code, meta in ERROR_META.items():
|
|
text = f"{meta.message}{meta.hint or ''}"
|
|
assert not any(word in text for word in forbidden), f"{code.name}: {text}"
|
|
|
|
|
|
def test_every_error_code_has_metadata() -> None:
|
|
for code in ErrorCode:
|
|
assert code in ERROR_META, f"{code.name} has no message/hint"
|
|
assert ERROR_META[code].message
|
|
|
|
|
|
def test_error_code_segments_map_to_http_families() -> None:
|
|
for code, meta in ERROR_META.items():
|
|
if code is ErrorCode.OK:
|
|
continue
|
|
segment = int(code) // 10000
|
|
assert segment in (4, 5, 6), code.name
|
|
if segment == 4:
|
|
assert 400 <= meta.http_status < 500, code.name
|
|
else:
|
|
assert meta.http_status >= 500, code.name
|
|
|
|
|
|
def test_app_error_defaults_come_from_the_registry() -> None:
|
|
err = AppError(ErrorCode.NOT_FOUND)
|
|
assert err.http_status == 404
|
|
assert err.message == "内容不存在或已删除"
|
|
|
|
|
|
def test_quota_exhausted_error_carries_actionable_data() -> None:
|
|
err = QuotaExhaustedError(required=1, remaining=0, reset_at="2026-08-14T00:00:00+08:00")
|
|
assert err.code is ErrorCode.QUOTA_EXHAUSTED
|
|
assert err.data["remaining"] == 0
|
|
assert err.data["reset_at"]
|
|
|
|
|
|
def test_jwt_round_trip() -> None:
|
|
token = create_access_token(user_id=123, secret="s" * 32, ttl_seconds=900)
|
|
claims = decode_access_token(token, secret="s" * 32)
|
|
assert isinstance(claims, AccessTokenClaims)
|
|
assert claims.user_id == 123
|
|
|
|
|
|
def test_expired_token_maps_to_a_user_readable_code() -> None:
|
|
token = create_access_token(user_id=1, secret="s" * 32, ttl_seconds=1, now=0)
|
|
try:
|
|
decode_access_token(token, secret="s" * 32)
|
|
except AppError as exc:
|
|
assert exc.code is ErrorCode.TOKEN_EXPIRED
|
|
else: # pragma: no cover
|
|
raise AssertionError("expected AppError")
|
|
|
|
|
|
def test_tampered_token_is_rejected() -> None:
|
|
token = create_access_token(user_id=1, secret="s" * 32)
|
|
try:
|
|
decode_access_token(token + "x", secret="s" * 32)
|
|
except AppError as exc:
|
|
assert exc.code is ErrorCode.TOKEN_INVALID
|
|
else: # pragma: no cover
|
|
raise AssertionError("expected AppError")
|