feat: intents 掩码直传——Bot(intents=) 参数与 INTENTS 环境变量,优先于 intents.toml

Intent 枚举(IntFlag)提升到顶层导出可直接 OR;掩码含未收录位仅告警不拦截。
This commit is contained in:
snowy-yang committed 2026-10-03 20:07:56 +08:00
1 parent bcd69099b6
commit 7e8794bf5f
9 files changed
+146 -5

No files matched your search

+21
View File
@@ -10,12 +10,15 @@ from qqbotsdk import (
Config,
EventEmitter,
EventQueue,
Intent,
Session,
WebhookConnecter,
WebsocketConnecter,
)
from qqbotsdk.payloads import GroupAtMessage
intent_mask = (1 << 25) | (1 << 26)
@pytest.fixture
def env(monkeypatch: pytest.MonkeyPatch) -> None:
@@ -29,6 +32,24 @@ def test_construction_loads_config(env: None) -> None:
assert bot.config.connecter == "websocket"
def test_intents_param_overrides_env(
env: None, monkeypatch: pytest.MonkeyPatch
) -> None:
monkeypatch.setenv("INTENTS", str(1 << 0))
bot = Bot(intents=intent_mask)
assert bot.config.intents == intent_mask # Bot 参数 > INTENTS 环境变量
def test_intents_env_picked_up(env: None, monkeypatch: pytest.MonkeyPatch) -> None:
monkeypatch.setenv("INTENTS", str(intent_mask))
assert Bot().config.intents == intent_mask
def test_intents_accepts_intent_flag(env: None) -> None:
bot = Bot(intents=Intent.GROUP_AND_C2C_EVENT | Intent.INTERACTION)
assert bot.config.intents == (1 << 25) | (1 << 26)
def test_missing_credentials_raises(monkeypatch: pytest.MonkeyPatch) -> None:
monkeypatch.delenv("APPID", raising=False)
monkeypatch.delenv("APPSECRET", raising=False)
+50 -1
View File
@@ -1,7 +1,7 @@
import pytest
from qqbotsdk.config import Config
from qqbotsdk.ws_protocol import get_intents
from qqbotsdk.ws_protocol import get_intents, resolve_intents
@pytest.fixture(autouse=True)
@@ -74,3 +74,52 @@ def test_config_rejects_unknown_connecter(monkeypatch):
monkeypatch.setenv("CONNECTER", "carrier-pigeon")
with pytest.raises(ValueError, match="carrier-pigeon"):
Config.load()
# --- 掩码直传:Config.intents(INTENTS 环境变量 / Bot 参数)优先于 intents.toml ---
def test_config_intents_env_decimal(monkeypatch):
monkeypatch.setenv("INTENTS", str((1 << 25) | (1 << 26)))
assert Config.load().intents == (1 << 25) | (1 << 26)
def test_config_intents_env_hex(monkeypatch):
monkeypatch.setenv("INTENTS", "0x4000000")
assert Config.load().intents == 1 << 26
def test_config_intents_env_invalid(monkeypatch):
monkeypatch.setenv("INTENTS", "abc")
with pytest.raises(ValueError, match="INTENTS"):
Config.load()
def make_config(intents: int | None, path: str) -> Config:
return Config(
app_id="app",
app_secret="secret",
base_url="https://api.bot.qq.com",
timeout=None, # type: ignore[arg-type]
connecter="websocket",
intents_file=path,
webhook_host="0.0.0.0",
webhook_port=8080,
webhook_path="/x",
intents=intents,
)
def test_resolve_intents_mask_wins_over_file(tmp_path):
path = write(tmp_path, "GUILDS = true\n")
assert resolve_intents(make_config(1 << 26, path)) == 1 << 26
def test_resolve_intents_falls_back_to_file(tmp_path):
path = write(tmp_path, "GUILDS = true\nAUDIO_ACTION = true\n")
assert resolve_intents(make_config(None, path)) == (1 << 0) | (1 << 29)
def test_resolve_intents_unknown_bits_pass_through(tmp_path):
mask = (1 << 26) | (1 << 5) # 1<<5 未收录进 Intent 枚举
assert resolve_intents(make_config(mask, str(tmp_path / "nope.toml"))) == mask
+14 -1
View File
@@ -17,7 +17,9 @@ class FakeToken:
return "tok"
def make_protocol(session: Session | None = None) -> tuple[WebsocketProtocol, Session]:
def make_protocol(
session: Session | None = None, intents: int | None = None
) -> tuple[WebsocketProtocol, Session]:
session = session or Session()
config = Config(
app_id="app",
@@ -29,6 +31,7 @@ def make_protocol(session: Session | None = None) -> tuple[WebsocketProtocol, Se
webhook_host="0.0.0.0",
webhook_port=8080,
webhook_path="/x",
intents=intents,
)
return WebsocketProtocol(config, session, FakeToken()), session # type: ignore[arg-type]
@@ -45,6 +48,16 @@ async def test_hello_without_session_identifies():
assert reply["d"]["shard"] == (0, 1)
@pytest.mark.asyncio
async def test_identify_carries_intents_mask():
mask = (1 << 25) | (1 << 26)
protocol, _ = make_protocol(intents=mask)
reply = await protocol.on_frame({"op": Opcode.HELLO, "d": {}})
assert reply is not None
assert reply["op"] == Opcode.IDENTIFY
assert reply["d"]["intents"] == mask
@pytest.mark.asyncio
async def test_hello_with_session_resumes():
protocol, session = make_protocol()