mirror of
https://github.com/snowy-yang/qqbotsdk.git
synced 2026-10-21 12:24:49 +00:00
37 lines
1.0 KiB
Python
37 lines
1.0 KiB
Python
import pytest
|
|
|
|
from qqbotsdk.emitter import EventEmitter
|
|
from qqbotsdk.model import Opcode
|
|
|
|
|
|
@pytest.fixture
|
|
def emitter() -> EventEmitter:
|
|
return EventEmitter() # 无参构造，覆盖 queue 默认自建路径
|
|
|
|
|
|
def test_route_dispatch_uses_t(emitter: EventEmitter):
|
|
routed = emitter._route({"op": Opcode.DISPATCH, "t": "READY", "d": {}})
|
|
assert routed is not None
|
|
op, name, event = routed
|
|
assert op == Opcode.DISPATCH
|
|
assert name == "READY"
|
|
assert event.type == "READY"
|
|
|
|
|
|
def test_route_non_dispatch_uses_opcode_name(emitter: EventEmitter):
|
|
routed = emitter._route({"op": Opcode.HELLO, "d": {"heartbeat_interval": 1}})
|
|
assert routed is not None
|
|
assert routed[1] == "HELLO"
|
|
|
|
|
|
def test_route_rejects_missing_op(emitter: EventEmitter):
|
|
assert emitter._route({"d": {}}) is None
|
|
|
|
|
|
def test_route_rejects_unknown_opcode(emitter: EventEmitter):
|
|
assert emitter._route({"op": 999}) is None
|
|
|
|
|
|
def test_route_rejects_dispatch_without_t(emitter: EventEmitter):
|
|
assert emitter._route({"op": Opcode.DISPATCH, "d": {}}) is None
|