hansjone/oclaw
1
0
Fork 0
mirror of https://github.com/hansjone/oclaw.git synced 2026-10-15 22:13:23 +08:00
Code Issues Projects Releases Packages Wiki Activity Actions

feat(weixin): native official runner and OSS install docs

Browse source 
Default Weixin to official native runner, add /weixin/native/reply host endpoint, and harden gateway stop to clean stale port listeners.
Document open-source installation flow for gateway + Personal WeChat.

Made-with: Cursor
This commit is contained in:
oliver 2026-04-28 19:26:26 +08:00
parent 31e3962c74
commit 88b1fdc9d3
12 changed files with 823 additions and 42 deletions
Download patch file Download diff file Expand all files Collapse all files

48
tests/test_weixin_ilink_api.py
View file

@ -30,7 +30,7 @@ class WeixinIlinkApiTests(unittest.TestCase):
self.assertEqual((r.json() or {}).get("ret"), 400)
def test_sendmessage_enqueues_reply_for_getupdates(self) -> None:
old_usecase = weixin_ilink_api.process_inbound_payload_usecase
old_usecase = weixin_ilink_api._process_inbound_payload_usecase
def _fake_usecase(payload: dict[str, object]) -> dict[str, object]:
text = str(payload.get("text") or "")
@ -45,7 +45,7 @@ class WeixinIlinkApiTests(unittest.TestCase):
}
try:
weixin_ilink_api.process_inbound_payload_usecase = _fake_usecase # type: ignore[assignment]
weixin_ilink_api._process_inbound_payload_usecase = _fake_usecase # type: ignore[assignment]
s = self.client.post(
"/ilink/bot/sendmessage",
headers=self.headers,
@ -77,7 +77,49 @@ class WeixinIlinkApiTests(unittest.TestCase):
self.assertTrue(msgs, data)
self.assertEqual(str((msgs[0] or {}).get("text") or ""), "echo:ping")
finally:
weixin_ilink_api.process_inbound_payload_usecase = old_usecase # type: ignore[assignment]
weixin_ilink_api._process_inbound_payload_usecase = old_usecase # type: ignore[assignment]
def test_native_reply_returns_sync_replies(self) -> None:
old_usecase = weixin_ilink_api._process_inbound_payload_usecase
def _fake_usecase(payload: dict[str, object]) -> dict[str, object]:
text = str(payload.get("text") or "")
self.assertEqual(str(payload.get("channel") or ""), "wechat")
self.assertEqual(str(payload.get("user_id") or ""), "wxid_u3")
return {
"ok": True,
"replies": [
{
"chat_id": str(payload.get("chat_id") or ""),
"text": f"native:{text}",
}
],
}
try:
weixin_ilink_api._process_inbound_payload_usecase = _fake_usecase # type: ignore[assignment]
r = self.client.post(
"/weixin/native/reply",
headers=self.headers,
json={
"channel": "wechat",
"account_id": "bot-1",
"ctx": {
"From": "wxid_u3",
"To": "wxid_u3",
"Body": "hello native",
"context_token": "ctx-1",
},
},
)
self.assertEqual(r.status_code, 200, r.text)
data = r.json() or {}
self.assertTrue(data.get("ok"), data)
replies = data.get("replies") if isinstance(data.get("replies"), list) else []
self.assertEqual(len(replies), 1, data)
self.assertEqual(str((replies[0] or {}).get("text") or ""), "native:hello native")
finally:
weixin_ilink_api._process_inbound_payload_usecase = old_usecase # type: ignore[assignment]
if __name__ == "__main__":