admin/ComfyUI-OpenClaw
1
0
Fork 0
mirror of https://github.com/rookiestar28/ComfyUI-OpenClaw.git synced 2026-08-14 00:48:07 +00:00
Code Issues Packages Projects Releases Wiki Activity

feat(api): implement R61/R62 unified error contract and lazy aiohttp queue degrade with regression tests

Browse Source
This commit is contained in:
rookiestar28
2026-02-14 00:15:35 +08:00
parent fec211d603
commit 1798b65b90
11 changed files with 474 additions and 53 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.pre-commit-config.yaml
+1 -2
View File
@@ -85,5 +85,4 @@ repos:
rev: 5.13.2
hooks:
- id: isort
args: ['--check-only', '--diff']
# Remove --check-only --diff to auto-sort
# Auto-sort imports (same behavior style as black-single autofix hook).
api/errors.py
+96
View File
@@ -0,0 +1,96 @@
"""
R61: Unified API Error Contract.
Provides a shared local exception schema and response serializer.
"""
import json
from enum import Enum
from typing import Any, Dict, Optional
# CI guard: Keep importable even without aiohttp installed
try:
from aiohttp import web
except ImportError:
web = None
class ErrorCode(str, Enum):
"""Standard machine-readable error codes."""
# Dependency / Runtime
DEPENDENCY_UNAVAILABLE = "dependency_unavailable"
INTERNAL_ERROR = "internal_error"
# Validation
VALIDATION_ERROR = "validation_error"
INVALID_REQUEST = "invalid_request"
INVALID_JSON = "invalid_json"
# Queue
QUEUE_SUBMIT_FAILED = "queue_submit_failed"
QUEUE_FULL = "queue_full"
# Auth
AUTH_FAILED = "auth_failed"
FORBIDDEN = "forbidden"
# Request / IO
RATE_LIMIT_EXCEEDED = "rate_limit_exceeded"
UNSUPPORTED_MEDIA_TYPE = "unsupported_media_type"
PAYLOAD_TOO_LARGE = "payload_too_large"
READ_ERROR = "read_error"
class APIError(Exception):
"""
Base class for API-contract exceptions.
Carries status code, machine error code, and human message.
"""
def __init__(
self,
message: str,
code: str = ErrorCode.INTERNAL_ERROR.value,
status: int = 500,
detail: Optional[Dict[str, Any]] = None,
):
super().__init__(message)
self.message = message
self.code = code
self.status = status
self.detail = detail or {}
def to_dict(self) -> Dict[str, Any]:
"""Serialize to contract JSON."""
return {
"ok": False,
"error": self.message, # Legacy field
"code": self.code,
"message": self.message,
"detail": self.detail,
}
def to_response(error: APIError) -> Any:
"""
Convert APIError to aiohttp.web.Response.
Gracefully handles missing aiohttp by returning dict (for tests/fallback).
"""
payload = error.to_dict()
if web is not None:
return web.json_response(payload, status=error.status)
# Fallback for environments without aiohttp (e.g. some unit tests)
return payload
def create_error_response(
message: str,
code: str = ErrorCode.INTERNAL_ERROR.value,
status: int = 500,
detail: Optional[Dict[str, Any]] = None,
) -> Any:
"""Helper to create a response directly without raising."""
err = APIError(message, code, status, detail)
return to_response(err)
api/webhook.py
+54 -38
View File
@@ -14,6 +14,12 @@ import logging
from aiohttp import web
try:
from .errors import APIError, ErrorCode, create_error_response
except ImportError:
# Build-time / Test fallback
from api.errors import APIError, ErrorCode, create_error_response
try:
from ..models.schemas import MAX_BODY_SIZE, WebhookJobRequest
from ..services.metrics import metrics
@@ -36,16 +42,6 @@ except ImportError:
logger = diagnostics.get_logger("ComfyUI-OpenClaw.api.webhook", "webhook")
def safe_error_response(status: int, error: str, detail: str = "") -> web.Response:
"""
Return a safe error response (no secrets, no stack traces).
"""
body = {"ok": False, "error": error}
if detail:
body["detail"] = detail
return web.json_response(body, status=status)
async def webhook_handler(request: web.Request) -> web.Response:
"""
POST /moltbot/webhook
@@ -55,10 +51,11 @@ async def webhook_handler(request: web.Request) -> web.Response:
# S17: Rate Limit
if not check_rate_limit(request, "webhook"):
metrics.inc("webhook_denied")
return web.json_response(
{"ok": False, "error": "rate_limit_exceeded"},
return create_error_response(
message="Rate limit exceeded",
code=ErrorCode.RATE_LIMIT_EXCEEDED,
status=429,
headers={"Retry-After": "60"},
detail={"retry_after": "60"},
)
try:
@@ -66,8 +63,10 @@ async def webhook_handler(request: web.Request) -> web.Response:
content_type = request.headers.get("Content-Type", "")
if not content_type.startswith("application/json"):
metrics.inc("webhook_denied")
return safe_error_response(
415, "unsupported_media_type", "Content-Type must be application/json"
return create_error_response(
message="Content-Type must be application/json",
code=ErrorCode.UNSUPPORTED_MEDIA_TYPE,
status=415,
)
# Read raw body with size limit
@@ -75,13 +74,19 @@ async def webhook_handler(request: web.Request) -> web.Response:
raw_body = await request.content.read(MAX_BODY_SIZE + 1)
if len(raw_body) > MAX_BODY_SIZE:
metrics.inc("webhook_denied")
return safe_error_response(
413, "payload_too_large", f"Max body size: {MAX_BODY_SIZE} bytes"
return create_error_response(
message=f"Payload too large (max {MAX_BODY_SIZE} bytes)",
code=ErrorCode.PAYLOAD_TOO_LARGE,
status=413,
)
except Exception as e:
logger.error(f"Failed to read request body: {e}")
metrics.inc("errors")
return safe_error_response(400, "read_error")
return create_error_response(
message="Failed to read request body",
code=ErrorCode.READ_ERROR,
status=400,
)
# Require auth
valid, error = require_auth(request, raw_body)
@@ -92,46 +97,55 @@ async def webhook_handler(request: web.Request) -> web.Response:
metrics.inc("webhook_denied")
# Map error to appropriate status code
if error in (
"auth_not_configured",
"bearer_not_configured",
"hmac_not_configured",
):
return safe_error_response(403, error)
else:
return safe_error_response(401, error)
status = (
403
if error
in (
"auth_not_configured",
"bearer_not_configured",
"hmac_not_configured",
)
else 401
)
return create_error_response(
message=error, code=ErrorCode.AUTH_FAILED, status=status
)
# Parse JSON
try:
data = json.loads(raw_body.decode("utf-8"))
# R46: Log payload if validation diagnostics enabled
if diagnostics.is_enabled("webhook.validate"):
# Use a separate logger for validation if we want specific granularity,
# or just reuse the main one but check the flag dynamically?
# Since logger is scoped to "webhook", we can check specific sub-flag here manually.
# Actually, let's just log to the main scoped logger, but with a clear prefix.
# Or create a sub-scope logger?
# For simplicity, reuse main logger but only call debug if specific intent matches?
# The 'diagnostics.get_logger' wraps 'debug' with 'is_enabled("webhook")'.
# If we want detailed validation logs only on "webhook.validate", we can do:
diagnostics.get_logger(
"ComfyUI-OpenClaw.api.webhook.validate", "webhook.validate"
).debug("Incoming Payload", data=data)
except (json.JSONDecodeError, UnicodeDecodeError) as e:
metrics.inc("webhook_denied")
return safe_error_response(400, "invalid_json")
return create_error_response(
message="Invalid JSON", code=ErrorCode.INVALID_JSON, status=400
)
# Validate schema
try:
job_request = WebhookJobRequest.from_dict(data)
except ValueError as e:
metrics.inc("webhook_denied")
return safe_error_response(400, "validation_error", str(e))
return create_error_response(
message="Validation Error",
code=ErrorCode.VALIDATION_ERROR,
status=400,
detail={"error": str(e)},
)
except Exception as e:
logger.error(f"Unexpected validation error: {e}")
metrics.inc("errors")
return safe_error_response(400, "validation_error")
return create_error_response(
message="Validation system error",
code=ErrorCode.VALIDATION_ERROR,
status=400,
)
# R25: Trace Context Extraction
trace_id = get_effective_trace_id(request.headers, data)
@@ -165,4 +179,6 @@ async def webhook_handler(request: web.Request) -> web.Response:
# Catch-all for unexpected errors - log but don't expose details
logger.exception(f"Unexpected webhook error: {e}")
metrics.inc("errors")
return safe_error_response(500, "internal_error")
return create_error_response(
message="Internal Server Error", code=ErrorCode.INTERNAL_ERROR, status=500
)
scripts/pre_push_checks.sh
+17 -2
View File
@@ -279,8 +279,23 @@ echo "[pre-push] Node version: $(node -v)"
echo "[pre-push] 1/4 detect-secrets"
run_pre_commit_safe run detect-secrets --all-files
echo "[pre-push] 2/4 pre-commit all hooks"
run_pre_commit_safe run --all-files
echo "[pre-push] 2/4 pre-commit all hooks (pass 1)"
if run_pre_commit_safe run --all-files --show-diff-on-failure; then
:
else
echo "[pre-push] INFO: pre-commit reported changes/issues; running pass 2 verification..." >&2
if run_pre_commit_safe run --all-files --show-diff-on-failure; then
if ! git diff --quiet -- .; then
echo "[pre-push] ERROR: hooks auto-fixed files during pre-push." >&2
echo "[pre-push] Please review, commit the fixes, then push again." >&2
git status --short
exit 1
fi
echo "[pre-push] WARN: first pass failed but second pass succeeded without local file changes." >&2
else
exit 1
fi
fi
echo "[pre-push] 3/4 backend unit tests"
MOLTBOT_STATE_DIR="$ROOT_DIR/moltbot_state/_pre_push_unit" \
scripts/run_full_tests_linux.sh
+7 -2
View File
@@ -108,8 +108,13 @@ echo "[tests] Node version: $(node -v)"
echo "[tests] 1/4 detect-secrets"
"$VENV_PY" -m pre_commit run detect-secrets --all-files
echo "[tests] 2/4 pre-commit all hooks"
"$VENV_PY" -m pre_commit run --all-files --show-diff-on-failure
echo "[tests] 2/4 pre-commit all hooks (pass 1: autofix)"
if "$VENV_PY" -m pre_commit run --all-files --show-diff-on-failure; then
:
else
echo "[tests] INFO: pre-commit reported changes/issues; running pass 2 verification..."
"$VENV_PY" -m pre_commit run --all-files --show-diff-on-failure
fi
echo "[tests] 3/4 backend unit tests"
MOLTBOT_STATE_DIR="$ROOT_DIR/moltbot_state/_local_unit" "$VENV_PY" scripts/run_unittests.py --start-dir tests --pattern "test_*.py"
scripts/run_full_tests_windows.ps1
+6 -2
View File
@@ -129,8 +129,12 @@ Write-Host "[tests] Node version: $(node -v)"
Write-Host "[tests] 1/4 detect-secrets"
Invoke-Checked "detect-secrets" { & $venvPython -m pre_commit run detect-secrets --all-files }
Write-Host "[tests] 2/4 pre-commit all hooks"
Invoke-Checked "pre-commit all hooks" { & $venvPython -m pre_commit run --all-files --show-diff-on-failure }
Write-Host "[tests] 2/4 pre-commit all hooks (pass 1: autofix)"
& $venvPython -m pre_commit run --all-files --show-diff-on-failure
if ($LASTEXITCODE -ne 0) {
Write-Host "[tests] INFO: pre-commit reported changes/issues; running pass 2 verification..."
Invoke-Checked "pre-commit all hooks (pass 2 verify)" { & $venvPython -m pre_commit run --all-files --show-diff-on-failure }
}
Write-Host "[tests] 3/4 backend unit tests"
$env:MOLTBOT_STATE_DIR = "$root\moltbot_state\_local_unit"
services/queue_submit.py
+48 -3
View File
@@ -12,7 +12,23 @@ import logging
import uuid
from typing import Any, Dict, Optional
import aiohttp
try:
from api.errors import APIError, ErrorCode, create_error_response
except ImportError:
# Fallback if api module not found (e.g. some test environments)
# Define minimal mocks to avoid crash
class ErrorCode:
DEPENDENCY_UNAVAILABLE = "dependency_unavailable"
QUEUE_SUBMIT_FAILED = "queue_submit_failed"
INTERNAL_ERROR = "internal_error"
class APIError(Exception):
def __init__(self, message, code="internal_error", status=500, detail=None):
super().__init__(message)
self.code = code
self.status = status
self.detail = detail or {}
logger = logging.getLogger("ComfyUI-OpenClaw.services.queue")
@@ -100,6 +116,19 @@ async def submit_prompt(
# unless we can hook internal server entry point.
# For MVP, HTTP loopback is safest and standard.
# R62: Lazy import aiohttp to avoid hard dependency crash at startup
try:
import aiohttp
except ImportError:
msg = "aiohttp is required for queue submission but not installed."
logger.error(msg)
raise APIError(
message=msg,
code=ErrorCode.DEPENDENCY_UNAVAILABLE,
status=503,
detail={"package": "aiohttp"},
)
url = f"{COMFYUI_URL}/prompt"
try:
@@ -116,9 +145,25 @@ async def submit_prompt(
logger.error(
f"Failed to queue prompt: {resp.status} - {text} (source={source}, trace_id={trace_id})"
)
raise RuntimeError(f"Queue submission failed: {resp.status}")
# R61: Use APIError for queue failure
raise APIError(
message=f"Queue submission failed: {resp.status}",
code=ErrorCode.QUEUE_SUBMIT_FAILED,
status=502,
detail={
"upstream_status": resp.status,
"upstream_response": text[:200],
},
)
except APIError:
raise
except Exception as e:
logger.error(
f"Error submitting to queue: {e} (source={source}, trace_id={trace_id})"
)
raise
# R61: Wrap generic exceptions too
raise APIError(
message=f"Queue submission error: {str(e)}",
code=ErrorCode.INTERNAL_ERROR,
status=500,
)
tests/test_r61_api_errors.py
+76
View File
@@ -0,0 +1,76 @@
"""
Tests for R61 API Error Contract.
"""
import os
import sys
import unittest
from unittest.mock import MagicMock, patch
# Add project root to path
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
from api.errors import APIError, ErrorCode, create_error_response, to_response
class TestR61APIErrors(unittest.TestCase):
def test_apierror_serialization(self):
err = APIError(
"Something went wrong",
code=ErrorCode.VALIDATION_ERROR.value,
status=400,
detail={"field": "prompt"},
)
data = err.to_dict()
self.assertFalse(data["ok"])
self.assertEqual(data["error"], "Something went wrong") # Legacy
self.assertEqual(data["message"], "Something went wrong")
self.assertEqual(data["code"], "validation_error")
self.assertEqual(data["detail"], {"field": "prompt"})
def test_default_values(self):
err = APIError("Basic error")
data = err.to_dict()
self.assertEqual(data["code"], "internal_error")
self.assertEqual(data["detail"], {})
self.assertEqual(err.status, 500)
@patch("api.errors.web")
def test_to_response_aiohttp(self, mock_web):
"""Test conversion to aiohttp response when safe."""
mock_response = MagicMock()
mock_web.json_response.return_value = mock_response
err = APIError("Test", status=418)
resp = to_response(err)
mock_web.json_response.assert_called_once()
args, kwargs = mock_web.json_response.call_args
self.assertEqual(kwargs["status"], 418)
self.assertEqual(args[0]["error"], "Test")
self.assertEqual(resp, mock_response)
@patch("api.errors.web", None)
def test_to_response_fallback(self):
"""Test fallback when aiohttp is missing."""
err = APIError("Fallback")
resp = to_response(err)
self.assertIsInstance(resp, dict)
self.assertEqual(resp["error"], "Fallback")
def test_create_error_response_helper(self):
# With active mock for web to return a distinct object
with patch("api.errors.web") as mock_web:
mock_web.json_response.return_value = "RESPONSE_OBJ"
resp = create_error_response("Helper test", status=404)
self.assertEqual(resp, "RESPONSE_OBJ")
args, kwargs = mock_web.json_response.call_args
self.assertEqual(kwargs["status"], 404)
self.assertEqual(args[0]["error"], "Helper test")
if __name__ == "__main__":
unittest.main()
tests/test_r61_webhook_adoption.py
+81
View File
@@ -0,0 +1,81 @@
"""
Tests for R61 Webhook Adoption.
Verifies that webhook_handler returns R61-compliant error responses.
"""
import json
import os
import sys
import unittest
from unittest.mock import AsyncMock, MagicMock, patch
# Add project root to path
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
from api.errors import ErrorCode
from api.webhook import webhook_handler
class TestR61WebhookAdoption(unittest.IsolatedAsyncioTestCase):
async def test_rate_limit_error(self):
"""Test rate limit error uses R61 contract."""
with patch("api.webhook.check_rate_limit", return_value=False):
request = MagicMock()
with patch("aiohttp.web.json_response") as mock_json_response:
await webhook_handler(request)
mock_json_response.assert_called_once()
args, kwargs = mock_json_response.call_args
body = args[0]
status = kwargs["status"]
self.assertEqual(status, 429)
self.assertFalse(body["ok"])
self.assertEqual(body["code"], "rate_limit_exceeded")
async def test_content_type_error(self):
"""Test content type error."""
with patch("api.webhook.check_rate_limit", return_value=True):
request = MagicMock()
request.headers.get.return_value = "text/plain"
with patch("aiohttp.web.json_response") as mock_json_response:
await webhook_handler(request)
args, kwargs = mock_json_response.call_args
body = args[0]
status = kwargs["status"]
self.assertEqual(status, 415)
self.assertEqual(body["code"], "unsupported_media_type")
async def test_auth_failed(self):
"""Test auth failure."""
with patch("api.webhook.check_rate_limit", return_value=True):
with patch(
"api.webhook.require_auth", return_value=(False, "auth_failed_test")
):
request = MagicMock()
request.headers.get.return_value = "application/json"
# Mock content read
request.content.read = AsyncMock(return_value=b"{}")
# Mock MAX_BODY_SIZE if needed during read?
# Real module is imported, so it uses real Int. That's fine.
with patch("aiohttp.web.json_response") as mock_json_response:
await webhook_handler(request)
args, kwargs = mock_json_response.call_args
body = args[0]
status = kwargs["status"]
self.assertEqual(status, 401)
self.assertEqual(body["code"], ErrorCode.AUTH_FAILED)
self.assertEqual(body["message"], "auth_failed_test")
if __name__ == "__main__":
unittest.main()
tests/test_r62_queue_degrade.py
+84
View File
@@ -0,0 +1,84 @@
"""
Tests for R62 Queue Submit Degrade and R61 Error Contract.
"""
import os
import sys
import unittest
from unittest.mock import AsyncMock, MagicMock, patch
# Add project root to path
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
from api.errors import APIError, ErrorCode
# We import submit_prompt inside tests to ensure mocks applied before import if needed,
# although the lazy import inside the function makes it easier.
from services.queue_submit import submit_prompt
class TestR62QueueDegrade(unittest.IsolatedAsyncioTestCase):
async def test_dependency_missing(self):
"""Test R62: specific error when aiohttp is missing."""
# Clean sys.modules to ensure we can control import
with patch.dict(sys.modules, {"aiohttp": None}):
with self.assertRaises(APIError) as cm:
await submit_prompt({"test": "workflow"})
err = cm.exception
self.assertEqual(err.code, ErrorCode.DEPENDENCY_UNAVAILABLE)
self.assertEqual(err.status, 503)
self.assertIn("required for queue submission", err.message)
async def test_submit_success(self):
"""Test successful submission when aiohttp is present."""
mock_response = MagicMock()
mock_response.status = 200
mock_response.json = AsyncMock(return_value={"prompt_id": "123", "number": 1})
mock_session_inst = MagicMock()
mock_session_inst.post.return_value.__aenter__.return_value = mock_response
mock_session_cls = MagicMock()
mock_session_cls.return_value.__aenter__.return_value = mock_session_inst
# Mock aiohttp module
mock_aiohttp = MagicMock()
mock_aiohttp.ClientSession = mock_session_cls
with patch.dict(sys.modules, {"aiohttp": mock_aiohttp}):
result = await submit_prompt({"test": "workflow"})
self.assertEqual(result["prompt_id"], "123")
mock_session_inst.post.assert_called_once()
async def test_upstream_failure(self):
"""Test standard APIError when upstream returns non-200."""
mock_response = MagicMock()
mock_response.status = 500
mock_response.text = AsyncMock(return_value="Internal Server Error")
mock_session_inst = MagicMock()
mock_session_inst.post.return_value.__aenter__.return_value = mock_response
mock_session_cls = MagicMock()
mock_session_cls.return_value.__aenter__.return_value = mock_session_inst
mock_aiohttp = MagicMock()
mock_aiohttp.ClientSession = mock_session_cls
with patch.dict(sys.modules, {"aiohttp": mock_aiohttp}):
with self.assertRaises(APIError) as cm:
await submit_prompt({"test": "workflow"})
err = cm.exception
self.assertEqual(err.code, ErrorCode.QUEUE_SUBMIT_FAILED)
self.assertEqual(err.status, 502)
self.assertIn("Queue submission failed: 500", err.message)
if __name__ == "__main__":
unittest.main()
tests/test_webhook_contract.py
+4 -4
View File
@@ -59,7 +59,7 @@ class TestWebhookContract(AioHTTPTestCase):
)
self.assertEqual(resp.status, 415)
body = await resp.json()
self.assertEqual(body["error"], "unsupported_media_type")
self.assertEqual(body["code"], "unsupported_media_type")
async def test_payload_too_large(self):
"""Test that payloads > 64KB are rejected."""
@@ -75,7 +75,7 @@ class TestWebhookContract(AioHTTPTestCase):
)
self.assertEqual(resp.status, 413)
body = await resp.json()
self.assertEqual(body["error"], "payload_too_large")
self.assertEqual(body["code"], "payload_too_large")
async def test_missing_auth_returns_401(self):
"""Test that missing auth returns 401."""
@@ -137,7 +137,7 @@ class TestWebhookContract(AioHTTPTestCase):
)
self.assertEqual(resp.status, 400)
body = await resp.json()
self.assertEqual(body["error"], "invalid_json")
self.assertEqual(body["code"], "invalid_json")
async def test_schema_validation_error_returns_400(self):
"""Test that schema validation errors return 400."""
@@ -152,7 +152,7 @@ class TestWebhookContract(AioHTTPTestCase):
)
self.assertEqual(resp.status, 400)
body = await resp.json()
self.assertEqual(body["error"], "validation_error")
self.assertEqual(body["code"], "validation_error")
if __name__ == "__main__":