"""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")