qqbotsdk/tests/test_payloads.py

from qqbotsdk.emitter import EventEmitter
from qqbotsdk.model import Opcode
from qqbotsdk.payloads import (
    C2CMessage,
    GroupAtMessage,
    Interaction,
    Ready,
    parse_event,
)
from qqbotsdk.queue import EventQueue


def test_parse_group_at_message():
    msg = parse_event(
        "GROUP_AT_MESSAGE_CREATE",
        {
            "author": {"id": None, "member_openid": "M1"},
            "content": "hi",
            "group_openid": "G1",
            "id": "msgid-1",
            "timestamp": "2026-01-01",
        },
    )
    assert isinstance(msg, GroupAtMessage)
    assert msg.group_openid == "G1"
    assert msg.user_openid == "M1"
    assert msg.id == "msgid-1"
    assert msg.author is not None and msg.author.member_openid == "M1"


def test_parse_group_full_message():
    msg = parse_event(
        "GROUP_MESSAGE_CREATE",
        {
            "author": {"member_openid": "M2"},
            "content": "大家好",
            "group_openid": "G1",
            "id": "msgid-2",
            "timestamp": "2026-01-01",
            "message_type": 0,
            "message_scene": {"unknown": False},
            "attachments": [{"content_type": 1, "url": "https://x/img.png"}],
            "mentions": [{"id": "U1"}],
            "msg_elements": [{"type": 1, "text_element": {"content": "大家好"}}],
        },
    )
    assert isinstance(msg, GroupAtMessage)
    assert msg.group_openid == "G1"
    assert msg.user_openid == "M2"
    assert msg.message_type == 0
    assert msg.message_scene == {"unknown": False}
    assert msg.attachments == [{"content_type": 1, "url": "https://x/img.png"}]
    assert msg.mentions == [{"id": "U1"}]
    assert msg.msg_elements == [{"type": 1, "text_element": {"content": "大家好"}}]


async def test_group_full_message_reaches_typed_handler():
    queue = EventQueue()
    emitter = EventEmitter(queue)
    received: list[GroupAtMessage] = []

    @emitter.on("GROUP_MESSAGE_CREATE")
    async def handler(msg: GroupAtMessage):
        received.append(msg)

    await emitter.emit(
        {
            "op": Opcode.DISPATCH,
            "t": "GROUP_MESSAGE_CREATE",
            "d": {"content": "hi", "group_openid": "G1"},
        }
    )
    assert len(received) == 1
    assert received[0].content == "hi"
    assert received[0].group_openid == "G1"


def test_parse_c2c_message():
    msg = parse_event(
        "C2C_MESSAGE_CREATE",
        {"author": {"user_openid": "U1"}, "content": "hello", "id": "m2"},
    )
    assert isinstance(msg, C2CMessage)
    assert msg.user_openid == "U1"


def test_parse_ready():
    ready = parse_event(
        "READY", {"session_id": "s1", "version": 1, "user": {"id": "b"}}
    )
    assert isinstance(ready, Ready)
    assert ready.session_id == "s1"


def test_parse_unknown_type_returns_dict():
    d = {"whatever": 1}
    assert parse_event("SOME_FUTURE_EVENT", d) is d
    assert parse_event(None, d) is d


def test_parse_ignores_extra_and_missing_fields():
    msg = parse_event("GROUP_AT_MESSAGE_CREATE", {"content": "x", "future_field": 9})
    assert isinstance(msg, GroupAtMessage)
    assert msg.content == "x"
    assert msg.group_openid is None


def test_parse_non_dict_d_returns_dict():
    d = 42  # 心跳等 opcode 的 d 是数字
    assert parse_event("GROUP_AT_MESSAGE_CREATE", d) == 42


def test_interaction_button_data():
    inter = parse_event(
        "INTERACTION_CREATE",
        {
            "id": "i1",
            "chat_type": 1,
            "group_openid": "G1",
            "data": {"resolved": {"button_data": "btn|like"}},
        },
    )
    assert isinstance(inter, Interaction)
    assert inter.button_data == "btn|like"


def test_interaction_button_data_defaults_to_empty():
    inter = Interaction(id="i2", data=None)
    assert inter.button_data == ""
    inter = Interaction(id="i3", data={"resolved": None})
    assert inter.button_data == ""


async def test_typed_annotation_receives_dataclass():
    queue = EventQueue()
    emitter = EventEmitter(queue)
    received: list[GroupAtMessage] = []

    @emitter.on("GROUP_AT_MESSAGE_CREATE")
    async def handler(msg: GroupAtMessage):
        received.append(msg)

    await emitter.emit(
        {
            "op": Opcode.DISPATCH,
            "t": "GROUP_AT_MESSAGE_CREATE",
            "d": {"content": "hi", "group_openid": "G1"},
        }
    )
    assert len(received) == 1
    assert received[0].content == "hi"
    assert received[0].group_openid == "G1"


async def test_untyped_annotation_still_receives_raw_dict():
    queue = EventQueue()
    emitter = EventEmitter(queue)
    received: list[dict] = []

    @emitter.on("GROUP_AT_MESSAGE_CREATE")
    async def handler(d):
        received.append(d)

    await emitter.emit(
        {"op": Opcode.DISPATCH, "t": "GROUP_AT_MESSAGE_CREATE", "d": {"content": "x"}}
    )
    assert received == [{"content": "x"}]