From 80a4a88cb10574021bcd3cc06feb8704270ddc6c Mon Sep 17 00:00:00 2001 From: AdrianoDev Date: Thu, 30 Apr 2026 18:17:15 +0200 Subject: [PATCH] feat(V2): error envelope module estratto da server.py --- src/cerbero_mcp/common/errors.py | 51 ++++++++++++++++++++++++++++++++ tests/unit/common/test_errors.py | 29 ++++++++++++++++++ 2 files changed, 80 insertions(+) create mode 100644 src/cerbero_mcp/common/errors.py create mode 100644 tests/unit/common/test_errors.py diff --git a/src/cerbero_mcp/common/errors.py b/src/cerbero_mcp/common/errors.py new file mode 100644 index 0000000..2089a6a --- /dev/null +++ b/src/cerbero_mcp/common/errors.py @@ -0,0 +1,51 @@ +"""Error envelope standard per tutti i tool MCP.""" +from __future__ import annotations + +import uuid +from datetime import UTC, datetime +from typing import Any + + +def error_envelope( + *, + type_: str, + code: str, + message: str, + retryable: bool, + suggested_fix: str | None = None, + details: dict | None = None, + request_id: str | None = None, +) -> dict: + env: dict[str, Any] = { + "error": { + "type": type_, + "code": code, + "message": message, + "retryable": retryable, + }, + "request_id": request_id or uuid.uuid4().hex, + "data_timestamp": datetime.now(UTC).isoformat(), + } + if suggested_fix: + env["error"]["suggested_fix"] = suggested_fix + if details: + env["error"]["details"] = details + return env + + +HTTP_CODE_MAP = { + 400: "BAD_REQUEST", + 401: "UNAUTHORIZED", + 403: "FORBIDDEN", + 404: "NOT_FOUND", + 408: "TIMEOUT", + 409: "CONFLICT", + 422: "VALIDATION_ERROR", + 429: "RATE_LIMIT", + 500: "INTERNAL_ERROR", + 502: "UPSTREAM_ERROR", + 503: "UNAVAILABLE", + 504: "GATEWAY_TIMEOUT", +} + +RETRYABLE_STATUSES = frozenset({408, 429, 502, 503, 504}) diff --git a/tests/unit/common/test_errors.py b/tests/unit/common/test_errors.py new file mode 100644 index 0000000..0991660 --- /dev/null +++ b/tests/unit/common/test_errors.py @@ -0,0 +1,29 @@ +from __future__ import annotations + +from cerbero_mcp.common.errors import error_envelope, HTTP_CODE_MAP + + +def test_envelope_minimal(): + e = error_envelope(type_="x", code="C", message="m", retryable=False) + assert e["error"]["code"] == "C" + assert e["error"]["retryable"] is False + assert "request_id" in e + assert "data_timestamp" in e + + +def test_envelope_with_suggested_fix_and_details(): + e = error_envelope( + type_="validation_error", + code="INVALID_INPUT", + message="bad", + retryable=False, + suggested_fix="check field x", + details={"field": "x"}, + ) + assert e["error"]["suggested_fix"] == "check field x" + assert e["error"]["details"] == {"field": "x"} + + +def test_http_code_map_has_common_codes(): + assert HTTP_CODE_MAP[401] == "UNAUTHORIZED" + assert HTTP_CODE_MAP[502] == "UPSTREAM_ERROR"