dbdb43faff
- Add plugin discovery via Python entry_points (group: nanobot.channels) - Move 11 channel Config classes from schema.py into their own channel modules - ChannelsConfig now only keeps send_progress + send_tool_hints (extra=allow) - Each built-in channel parses dict->Pydantic in __init__, zero internal changes - All channels implement default_config() for onboard auto-population - nanobot onboard injects defaults for all discovered channels (built-in + plugins) - Add nanobot plugins list CLI command - Add Channel Plugin Guide (docs/CHANNEL_PLUGIN_GUIDE.md) - Fully backward compatible: existing config.json and sessions work as-is - 340 tests pass, zero regressions
97 lines
2.6 KiB
Python
97 lines
2.6 KiB
Python
from types import SimpleNamespace
|
|
|
|
import pytest
|
|
|
|
from nanobot.bus.events import OutboundMessage
|
|
from nanobot.bus.queue import MessageBus
|
|
from nanobot.channels.qq import QQChannel
|
|
from nanobot.channels.qq import QQConfig
|
|
|
|
|
|
class _FakeApi:
|
|
def __init__(self) -> None:
|
|
self.c2c_calls: list[dict] = []
|
|
self.group_calls: list[dict] = []
|
|
|
|
async def post_c2c_message(self, **kwargs) -> None:
|
|
self.c2c_calls.append(kwargs)
|
|
|
|
async def post_group_message(self, **kwargs) -> None:
|
|
self.group_calls.append(kwargs)
|
|
|
|
|
|
class _FakeClient:
|
|
def __init__(self) -> None:
|
|
self.api = _FakeApi()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_on_group_message_routes_to_group_chat_id() -> None:
|
|
channel = QQChannel(QQConfig(app_id="app", secret="secret", allow_from=["user1"]), MessageBus())
|
|
|
|
data = SimpleNamespace(
|
|
id="msg1",
|
|
content="hello",
|
|
group_openid="group123",
|
|
author=SimpleNamespace(member_openid="user1"),
|
|
)
|
|
|
|
await channel._on_message(data, is_group=True)
|
|
|
|
msg = await channel.bus.consume_inbound()
|
|
assert msg.sender_id == "user1"
|
|
assert msg.chat_id == "group123"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_send_group_message_uses_plain_text_group_api_with_msg_seq() -> None:
|
|
channel = QQChannel(QQConfig(app_id="app", secret="secret", allow_from=["*"]), MessageBus())
|
|
channel._client = _FakeClient()
|
|
channel._chat_type_cache["group123"] = "group"
|
|
|
|
await channel.send(
|
|
OutboundMessage(
|
|
channel="qq",
|
|
chat_id="group123",
|
|
content="hello",
|
|
metadata={"message_id": "msg1"},
|
|
)
|
|
)
|
|
|
|
assert len(channel._client.api.group_calls) == 1
|
|
call = channel._client.api.group_calls[0]
|
|
assert call == {
|
|
"group_openid": "group123",
|
|
"msg_type": 0,
|
|
"content": "hello",
|
|
"msg_id": "msg1",
|
|
"msg_seq": 2,
|
|
}
|
|
assert not channel._client.api.c2c_calls
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_send_c2c_message_uses_plain_text_c2c_api_with_msg_seq() -> None:
|
|
channel = QQChannel(QQConfig(app_id="app", secret="secret", allow_from=["*"]), MessageBus())
|
|
channel._client = _FakeClient()
|
|
|
|
await channel.send(
|
|
OutboundMessage(
|
|
channel="qq",
|
|
chat_id="user123",
|
|
content="hello",
|
|
metadata={"message_id": "msg1"},
|
|
)
|
|
)
|
|
|
|
assert len(channel._client.api.c2c_calls) == 1
|
|
call = channel._client.api.c2c_calls[0]
|
|
assert call == {
|
|
"openid": "user123",
|
|
"msg_type": 0,
|
|
"content": "hello",
|
|
"msg_id": "msg1",
|
|
"msg_seq": 2,
|
|
}
|
|
assert not channel._client.api.group_calls
|