aios
19a2ce8439
- Integrate ActionMonitor into ChatGPT GitOps API for failure tracking - Add ErrorClassifier with permission, git, runtime, system, secret violation error types - Add FailureRouter to route errors to correct recovery agents (openclaw/hermes/windows_codex_app) - Add RecoveryController orchestrating full pipeline: capture→classify→route→plan→execute→validate→record - Update error_classifier to include ClientResponseError, rate limit, and secret_violation patterns - All 8 integration tests passing (ActionMonitor, ErrorClassifier, FailureRouter, RecoveryController, API 500 scenario) - Push baseline to V1R2C2 with tag v1r2c2-baseline
88 lines
3.0 KiB
Python
88 lines
3.0 KiB
Python
from __future__ import annotations
|
|
|
|
import importlib.util
|
|
import sys
|
|
from pathlib import Path
|
|
from types import SimpleNamespace
|
|
from unittest import TestCase
|
|
from unittest.mock import patch
|
|
|
|
|
|
BACKEND_PATH = Path(__file__).resolve().parents[1] / "scripts" / "wechat_send_backend.py"
|
|
|
|
|
|
def _load_backend():
|
|
spec = importlib.util.spec_from_file_location("wechat_send_backend", BACKEND_PATH)
|
|
if spec is None or spec.loader is None:
|
|
raise RuntimeError(f"unable to load backend from {BACKEND_PATH}")
|
|
module = importlib.util.module_from_spec(spec)
|
|
sys.modules[spec.name] = module
|
|
spec.loader.exec_module(module)
|
|
return module
|
|
|
|
|
|
class WechatSendBackendTests(TestCase):
|
|
@classmethod
|
|
def setUpClass(cls):
|
|
cls.backend = _load_backend()
|
|
|
|
def test_select_backend_prefers_ucc_adapter_over_openclaw_channel(self):
|
|
with (
|
|
patch.object(self.backend, "SEND_BACKEND", ""),
|
|
patch.object(self.backend, "ucc_adapter_alive", return_value=True),
|
|
patch.object(self.backend, "discover_openclaw_wechat_channel", return_value="some-wechat-channel"),
|
|
):
|
|
selected = self.backend.select_backend()
|
|
|
|
self.assertEqual(selected["backend"], "ucc-adapter")
|
|
self.assertEqual(selected["reason"], "ucc_adapter_alive")
|
|
|
|
def test_openclaw_chat_only_is_not_wechat_channel(self):
|
|
completed = SimpleNamespace(returncode=0, stdout='{"chat": {}}')
|
|
with (
|
|
patch.object(self.backend, "WECHAT_CHANNEL", ""),
|
|
patch.object(self.backend, "_run_openclaw", return_value=completed),
|
|
):
|
|
self.assertIsNone(self.backend.discover_openclaw_wechat_channel())
|
|
|
|
def test_send_ucc_adapter_uses_canonical_endpoint_and_sent_status(self):
|
|
with (
|
|
patch.object(self.backend, "UCC_SEND_PATH", "/api/internal/wechat/send"),
|
|
patch.object(
|
|
self.backend,
|
|
"_post_ucc",
|
|
return_value=(
|
|
200,
|
|
{
|
|
"ok": True,
|
|
"status": "sent",
|
|
"provider_message_id": "openclaw-weixin:1-abc",
|
|
},
|
|
),
|
|
) as post,
|
|
):
|
|
result = self.backend.send_ucc_adapter(
|
|
target="chat-001",
|
|
content="hello",
|
|
account_id="acct-001",
|
|
message_id="msg-001",
|
|
metadata={"source": "test"},
|
|
).to_dict()
|
|
|
|
self.assertTrue(result["ok"])
|
|
self.assertEqual(result["backend"], "ucc-adapter")
|
|
self.assertEqual(result["status"], "sent")
|
|
self.assertEqual(result["endpoint"], "/api/internal/wechat/send")
|
|
post.assert_called_once()
|
|
args, _ = post.call_args
|
|
self.assertEqual(args[0], "/api/internal/wechat/send")
|
|
self.assertEqual(args[1]["target"], "chat-001")
|
|
self.assertEqual(args[1]["content"], "hello")
|
|
self.assertFalse(args[1]["dry_run"])
|
|
|
|
|
|
if __name__ == "__main__":
|
|
import unittest
|
|
|
|
unittest.main()
|