snowy-yang
94ee3298b7
- panel.py 调开放平台 2026-08 新增的 /v2/panels 接口，把 点歌/歌词/帮助 三条指令同步到输入框 / 菜单：READY 时幂等同步（没有则创建、内容有变 则更新、一致则跳过，重连自愈），target_type=all 对全部群生效；失败 只记日志不影响运行；COMMAND_PANEL=off 可关闭 - 斜杠指令与 @机器人 消息归一解析：_normalize 剥开头 / 与 @ 前缀， 与裸文本命令共用 handle_command（/点歌 稻香、@bot 点歌 稻香、 点歌 稻香 三种写法等价） - 只敲命令不带参数（斜杠指令参数可为空）时回复用法提示 - 帮助消息带 [🎤 点歌][📝 歌词] 指令按钮（action type=2，点击把 /点歌 填进输入框），键盘无权限自动回退纯文本；帮助文案改以 / 指令呈现
1010 lines
39 KiB
Python
1010 lines
39 KiB
Python
"""点歌业务测试：假 MusicServer + Recording BotApi，覆盖选歌菜单与分页、
|
||
歌曲卡片、按钮回调、默认 tx 源与聚合兜底、直链候选源回退、错误路径。"""
|
||
|
||
from collections.abc import AsyncIterator
|
||
from dataclasses import replace
|
||
from pathlib import Path
|
||
from typing import Any
|
||
from urllib.parse import quote
|
||
|
||
import pytest
|
||
from aiohttp import ClientSession, web
|
||
from qqbotsdk import EventEmitter
|
||
from qqbotsdk.api import ApiError, BotApi
|
||
|
||
from musicbot import music
|
||
from musicbot.config import BotConfig
|
||
from musicbot.music import (
|
||
handle_button,
|
||
handle_command,
|
||
register,
|
||
search,
|
||
send_menu,
|
||
)
|
||
from musicbot.panel import sync_command_panel
|
||
|
||
|
||
def make_api(
|
||
fail_types: set[int] | None = None,
|
||
fail_md: bool = False,
|
||
fail_event: bool = False,
|
||
fail_upload: bool = False,
|
||
fail_file_perm: bool = False,
|
||
fail_keyboard: bool = False,
|
||
):
|
||
"""替换 upload_group_file / Markdown 消息以模拟失败，request 记录其余调用。
|
||
|
||
fail_event=True 时拒绝一切带 event_id 的被动回复（模拟互动回执失效）；
|
||
fail_upload=True 时上传一律抛 TimeoutError（模拟 QQ 侧抓流超过请求超时）；
|
||
fail_file_perm=True 时 file_type=4（文件）返回 40093005（仅群主/管理员可传）；
|
||
fail_keyboard=True 时拒绝一切带 keyboard 的消息（模拟键盘无权限）。
|
||
"""
|
||
calls: list[dict] = []
|
||
|
||
class Recording(BotApi):
|
||
async def request(self, method: str, path: str, **kwargs: Any) -> Any:
|
||
body = kwargs.get("json", {})
|
||
if fail_md and body.get("msg_type") == 2:
|
||
raise ApiError(400, 304036, "无 Markdown 权限")
|
||
if fail_keyboard and "keyboard" in body:
|
||
raise ApiError(400, 304003, "无键盘权限")
|
||
if fail_event and "event_id" in body:
|
||
raise ApiError(400, 40034025, "请求参数event_id无效")
|
||
if path.endswith("/files"):
|
||
if fail_upload:
|
||
raise TimeoutError()
|
||
if fail_file_perm and body.get("file_type") == 4:
|
||
raise ApiError(400, 40093005, "仅允许群主和管理员上传文件")
|
||
if fail_types and body.get("file_type") in fail_types:
|
||
raise ApiError(400, "40054002", "上传失败")
|
||
calls.append(
|
||
{
|
||
"method": method,
|
||
"path": path,
|
||
"file_type": body.get("file_type"),
|
||
"url": body.get("url"),
|
||
"timeout": kwargs.get("timeout"),
|
||
}
|
||
)
|
||
return {"file_info": f"f{body.get('file_type')}"}
|
||
calls.append({"method": method, "path": path, **kwargs})
|
||
return {"id": "MSG"} if path.endswith("/messages") else {}
|
||
|
||
return Recording.__new__(Recording), calls
|
||
|
||
|
||
class FakeMusicServer:
|
||
"""模拟 MusicServer 的统一信封接口，记录搜索参数。
|
||
|
||
/url 返回指向自身的直链；tx_denied=True 时 tx 的直链指向 403
|
||
路由，模拟 CDN 风控（测直链探测与候选源回退）。
|
||
"""
|
||
|
||
def __init__(self, tx_denied: bool = False) -> None:
|
||
self.tx_denied = tx_denied
|
||
self.runner: web.AppRunner | None = None
|
||
self.port = 0
|
||
self.queries: list[dict] = []
|
||
|
||
async def start(self) -> None:
|
||
app = web.Application()
|
||
app.router.add_get("/search", self._search)
|
||
app.router.add_get("/url/{source}/{song_id}", self._url)
|
||
app.router.add_get("/lyric/{source}/{song_id}", self._lyric)
|
||
app.router.add_get("/media/{tail:.+}", self._media)
|
||
app.router.add_get("/denied/{tail:.+}", self._denied)
|
||
self.runner = web.AppRunner(app)
|
||
await self.runner.setup()
|
||
site = web.TCPSite(self.runner, "127.0.0.1", 0)
|
||
await site.start()
|
||
# sockets 未列入 aiohttp 的类型存根，只能从私有属性取实际端口
|
||
assert site._server is not None
|
||
self.port = site._server.sockets[0].getsockname()[1] # type: ignore[attr-defined]
|
||
|
||
async def stop(self) -> None:
|
||
assert self.runner is not None
|
||
await self.runner.cleanup()
|
||
|
||
async def _search(self, request: web.Request) -> web.Response:
|
||
name = request.query["name"]
|
||
source = request.query.get("source", "")
|
||
self.queries.append({"name": name, "source": source})
|
||
items: list[dict] = []
|
||
if "空" not in name:
|
||
if "兜底" in name:
|
||
# tx 搜不到、聚合有 —— 测兜底路径
|
||
items = [] if source == "tx" else [
|
||
{"source": "kw", "songmid": "S1", "name": name, "singer": "歌手A"}
|
||
]
|
||
elif source == "kw":
|
||
items = [
|
||
{"source": "kw", "songmid": "S1", "name": name, "singer": "歌手A"}
|
||
]
|
||
elif source == "tx" or not source:
|
||
items = [
|
||
{
|
||
"source": "tx",
|
||
"songmid": "T1",
|
||
"name": name,
|
||
"singer": "歌手T",
|
||
"albumName": "专辑T",
|
||
"interval": "04:00",
|
||
"img": "https://img.test/cover.jpg",
|
||
},
|
||
{"source": "kw", "songmid": "S1", "name": name, "singer": "歌手A"},
|
||
]
|
||
return web.json_response({"code": 0, "data": {"list": items}})
|
||
|
||
async def _url(self, request: web.Request) -> web.Response:
|
||
info = request.match_info
|
||
prefix = "denied" if info["source"] == "tx" and self.tx_denied else "media"
|
||
return web.json_response(
|
||
{
|
||
"code": 0,
|
||
"data": {
|
||
"type": "128k",
|
||
"url": f"http://127.0.0.1:{self.port}/{prefix}/{info['source']}/{info['song_id']}.mp3",
|
||
},
|
||
}
|
||
)
|
||
|
||
async def _media(self, _: web.Request) -> web.Response:
|
||
return web.Response(body=b"fake-audio-bytes")
|
||
|
||
async def _denied(self, _: web.Request) -> web.Response:
|
||
return web.Response(status=403, text="forbidden")
|
||
|
||
async def _lyric(self, request: web.Request) -> web.Response:
|
||
lyric = "[kuwo:113]\n[ver:v1.0]\n[00:01.00]第一行\n[00:05.00]第二行\n"
|
||
return web.json_response({"code": 0, "data": {"lyric": lyric}})
|
||
|
||
|
||
@pytest.fixture
|
||
async def server():
|
||
s = FakeMusicServer()
|
||
await s.start()
|
||
yield s
|
||
await s.stop()
|
||
|
||
|
||
@pytest.fixture
|
||
async def env(server) -> AsyncIterator[tuple]:
|
||
async with ClientSession() as http:
|
||
config = BotConfig(
|
||
music_server_url=f"http://127.0.0.1:{server.port}",
|
||
music_quality="128k",
|
||
music_source="tx",
|
||
)
|
||
yield http, config
|
||
|
||
|
||
@pytest.fixture(autouse=True)
|
||
def _reset_menu_state():
|
||
"""菜单/卡片消息记录与消息去重表是模块级状态，测试间清空避免串扰。"""
|
||
music._last_msg.clear()
|
||
music._seen_messages.clear()
|
||
yield
|
||
music._last_msg.clear()
|
||
music._seen_messages.clear()
|
||
|
||
|
||
# 回调按钮 data：音乐|动作|源|song_id|歌名|歌手|封面（歌名/歌手/封面 % 转义）
|
||
IMG_URL = "https://img.test/cover.jpg"
|
||
PICK_1 = music._song_data("pick", music.Song("tx", "T1", "稻香", "歌手T", img=IMG_URL))
|
||
PLAY_TX = music._song_data("play", music.Song("tx", "T1", "稻香", "歌手T", img=IMG_URL))
|
||
DL_TX = music._song_data("dl", music.Song("tx", "T1", "稻香", "歌手T", img=IMG_URL))
|
||
# 旧版按钮 data（无封面段）—— 兼容性用例
|
||
PICK_1_LEGACY = "音乐|pick|tx|T1|%E7%A8%BB%E9%A6%99|%E6%AD%8C%E6%89%8BT"
|
||
|
||
|
||
def posts(calls: list[dict]) -> list[dict]:
|
||
return [c for c in calls if c.get("path", "").endswith("/messages")]
|
||
|
||
|
||
def buttons(body: dict) -> list[dict]:
|
||
"""键盘的行列表；每行是 {"buttons": [...]}。"""
|
||
return body["keyboard"]["content"]["rows"]
|
||
|
||
|
||
def labels(body: dict) -> list[list[str]]:
|
||
return [[b["render_data"]["label"] for b in r["buttons"]] for r in buttons(body)]
|
||
|
||
|
||
async def test_search_prefers_tx_and_falls_back(server, env):
|
||
http, config = env
|
||
song = await search(http, config, "稻香")
|
||
assert song is not None and (song.source, song.song_id) == ("tx", "T1")
|
||
assert server.queries == [{"name": "稻香", "source": "tx"}]
|
||
|
||
server.queries.clear()
|
||
fallback = await search(http, config, "兜底")
|
||
assert fallback is not None
|
||
assert (fallback.source, fallback.song_id) == ("kw", "S1") # tx 没有结果 → 聚合兜底
|
||
assert [q["source"] for q in server.queries] == ["tx", ""]
|
||
|
||
|
||
async def test_menu_only_selects_songs(env):
|
||
"""搜索菜单只负责选歌：编号回调按钮，不放播放/下载、不放链接、不带音源。"""
|
||
http, config = env
|
||
api, calls = make_api()
|
||
assert await handle_command(api, http, config, "G1", "m1", "点歌 稻香")
|
||
body = posts(calls)[0]["json"]
|
||
assert body["msg_type"] == 2
|
||
md = body["markdown"]["content"]
|
||
assert "## 点歌：稻香" in md
|
||
assert "1. 稻香 - 歌手T" in md
|
||
assert "QQ音乐" not in md and "酷我" not in md # 每首歌后面不再标音源
|
||
assert "http" not in md # 菜单里没有任何链接
|
||
# 编号回调按钮（type=1）：点一下弹卡片，data 自包含歌曲（含封面）
|
||
rows = buttons(body)
|
||
assert [b["render_data"]["label"] for b in rows[0]["buttons"]] == ["1", "2"]
|
||
assert [b["action"]["type"] for b in rows[0]["buttons"]] == [1, 1]
|
||
assert rows[0]["buttons"][0]["action"]["data"] == PICK_1
|
||
# 首页无上一页，本页未满一页（2 首 < 10）→ 也不给下一页
|
||
assert labels(body)[-1] == ["1", "2"]
|
||
|
||
|
||
async def test_menu_pagination_buttons(env):
|
||
"""满页给「下一页」，非首页给「上一页」，data 带页码与关键词。"""
|
||
http, config = env
|
||
api, calls = make_api()
|
||
full = replace(config, menu_page_size=2)
|
||
await send_menu(api, http, full, "G1", "稻香", page=2)
|
||
body = posts(calls)[0]["json"]
|
||
nav = [b["render_data"]["label"] for b in buttons(body)[-1]["buttons"]]
|
||
assert nav == ["上一页", "下一页"]
|
||
data = [b["action"]["data"] for b in buttons(body)[-1]["buttons"]]
|
||
assert data[0] == f"音乐|page|{quote('稻香', safe='')}|1"
|
||
assert data[1] == f"音乐|page|{quote('稻香', safe='')}|3"
|
||
|
||
|
||
async def test_menu_page_size_row_layout(env):
|
||
"""每行最多 5 个按钮，行数不超过键盘上限。"""
|
||
http, config = env
|
||
api, calls = make_api()
|
||
big = replace(config, menu_page_size=20)
|
||
await send_menu(api, http, big, "G1", "稻香")
|
||
rows = buttons(posts(calls)[0]["json"])
|
||
assert len(rows) <= 5
|
||
for row in rows[:-1]:
|
||
assert 1 <= len(row["buttons"]) <= 5
|
||
|
||
|
||
async def test_menu_withdraws_previous(env):
|
||
"""同群出新菜单时撤回上一张菜单；不同群互不影响。"""
|
||
http, config = env
|
||
api, calls = make_api()
|
||
assert await handle_command(api, http, config, "G1", "m1", "点歌 稻香")
|
||
assert await handle_command(api, http, config, "G2", "m2", "点歌 稻香")
|
||
assert not [c for c in calls if c["method"] == "DELETE"]
|
||
assert await handle_command(api, http, config, "G1", "m3", "点歌 稻香")
|
||
assert [c["path"] for c in calls if c["method"] == "DELETE"] == [
|
||
"/v2/groups/G1/messages/MSG"
|
||
]
|
||
|
||
|
||
async def test_menu_fallback_keeps_keyboard(env):
|
||
http, config = env
|
||
api, calls = make_api(fail_md=True)
|
||
await handle_command(api, http, config, "G1", "m1", "点歌 稻香")
|
||
sent = posts(calls)
|
||
assert len(sent) == 1 # Markdown 失败后回退文本 + 按钮仍带上
|
||
body = sent[0]["json"]
|
||
assert body["msg_type"] == 0
|
||
assert "稻香 - 歌手T" in body["content"]
|
||
assert "（点下方编号选歌）" in body["content"]
|
||
assert "keyboard" in body
|
||
|
||
|
||
async def test_pick_button_sends_card(env):
|
||
"""点编号 → 弹歌曲卡片：封面图 + 歌名/来源 + [▶播放][⬇下载] 回调按钮。
|
||
|
||
卡片由回调 data 还原（自包含 song 字段），故不含专辑/时长等菜单外信息。"""
|
||
http, config = env
|
||
api, calls = make_api()
|
||
await handle_button(
|
||
api, http, config, "G1", "I1", PICK_1, event_id="INTERACTION_CREATE:I1"
|
||
)
|
||
card = posts(calls)[0]["json"]
|
||
assert card["msg_type"] == 2
|
||
md = card["markdown"]["content"]
|
||
assert "## 稻香 - 歌手T" in md
|
||
assert "" in md # 专辑封面
|
||
assert "来源：QQ音乐" in md # 卡片保留音源
|
||
assert labels(card) == [["▶ 播放", "⬇ 下载"]]
|
||
assert [b["action"]["type"] for b in buttons(card)[0]["buttons"]] == [1, 1]
|
||
assert [b["action"]["data"] for b in buttons(card)[0]["buttons"]] == [
|
||
PLAY_TX, DL_TX,
|
||
]
|
||
# 已回应互动
|
||
assert (calls[-1]["method"], calls[-1]["path"]) == ("PUT", "/interactions/I1")
|
||
|
||
|
||
async def test_card_without_img_when_source_has_none(env):
|
||
"""无封面源的歌曲（如 kw）：卡片不含图片，其余照常。"""
|
||
http, config = env
|
||
api, calls = make_api()
|
||
no_img = "音乐|pick|kw|S1|%E7%A8%BB%E9%A6%99|%E6%AD%8C%E6%89%8BA|"
|
||
await handle_button(api, http, config, "G1", "I1", no_img)
|
||
md = posts(calls)[0]["json"]["markdown"]["content"]
|
||
assert "![封面" not in md
|
||
assert "## 稻香 - 歌手A" in md
|
||
|
||
|
||
async def test_legacy_button_data_without_img(env):
|
||
"""旧版 4 段 data 仍能处理（封面缺省），保证已发出的菜单按钮不失效。"""
|
||
http, config = env
|
||
api, calls = make_api()
|
||
assert await handle_button(api, http, config, "G1", "I1", PICK_1_LEGACY)
|
||
md = posts(calls)[0]["json"]["markdown"]["content"]
|
||
assert "## 稻香 - 歌手T" in md
|
||
assert "![封面" not in md
|
||
|
||
|
||
async def test_card_play_button_sends_voice(env):
|
||
"""卡片上的 [▶ 播放] 是回调按钮：点一下直接发语音。"""
|
||
http, config = env
|
||
api, calls = make_api()
|
||
await handle_button(
|
||
api, http, config, "G1", "I2", PLAY_TX, event_id="INTERACTION_CREATE:I2"
|
||
)
|
||
upload = [c for c in calls if c.get("file_type")][0]
|
||
assert upload["file_type"] == 3
|
||
assert upload["url"].endswith("/media/tx/T1.mp3")
|
||
assert upload["timeout"].total == 120 # 上传放宽超时：QQ 侧抓流慢
|
||
body = posts(calls)[-1]["json"]
|
||
assert body["msg_type"] == 7
|
||
assert body["event_id"] == "INTERACTION_CREATE:I2"
|
||
|
||
|
||
async def test_card_dl_button_sends_file(env):
|
||
"""SEND_MODE=file 时下载上传群文件（file_type=4）。"""
|
||
http, config = env
|
||
api, calls = make_api()
|
||
file_cfg = replace(config, send_mode="file")
|
||
await handle_button(
|
||
api, http, file_cfg, "G1", "I3", DL_TX, event_id="INTERACTION_CREATE:I3"
|
||
)
|
||
upload = [c for c in calls if c.get("file_type")][0]
|
||
assert upload["file_type"] == 4
|
||
assert posts(calls)[-1]["json"]["msg_type"] == 7
|
||
|
||
|
||
async def test_default_link_mode_skips_upload(env):
|
||
"""默认 SEND_MODE=link：下载直接发直链，根本不上传（避开 40093005）。"""
|
||
http, config = env
|
||
assert config.send_mode == "link"
|
||
api, calls = make_api()
|
||
await handle_button(
|
||
api, http, config, "G1", "I3", DL_TX, event_id="INTERACTION_CREATE:I3"
|
||
)
|
||
assert not [c for c in calls if c.get("file_type")] # 无上传请求
|
||
body = posts(calls)[-1]["json"]
|
||
assert body["msg_type"] == 0
|
||
assert "/media/tx/T1.mp3" in body["content"]
|
||
|
||
|
||
async def test_pick_swaps_menu_for_card(env):
|
||
"""点编号后：撤回旧菜单、发出卡片、记录换成卡片消息 id。"""
|
||
http, config = env
|
||
api, calls = make_api()
|
||
assert await handle_command(api, http, config, "G1", "m1", "点歌 稻香")
|
||
assert music._last_msg.get("G1") == "MSG" # 菜单消息
|
||
await handle_button(api, http, config, "G1", "I1", PICK_1)
|
||
assert ("DELETE", "/v2/groups/G1/messages/MSG") in [
|
||
(c["method"], c["path"]) for c in calls
|
||
]
|
||
assert "## 稻香 - 歌手T" in posts(calls)[-1]["json"]["markdown"]["content"]
|
||
|
||
|
||
async def test_play_after_card_withdraws_card(env):
|
||
"""卡片上点播放成功后，撤回已完成使命的卡片。"""
|
||
http, config = env
|
||
api, calls = make_api()
|
||
await handle_button(api, http, config, "G1", "I1", PICK_1)
|
||
calls.clear()
|
||
await handle_button(api, http, config, "G1", "I2", PLAY_TX)
|
||
upload = [c for c in calls if c.get("file_type")][0]
|
||
assert upload["file_type"] == 3
|
||
assert upload["url"].endswith("/media/tx/T1.mp3")
|
||
assert posts(calls)[-1]["json"]["msg_type"] == 7
|
||
assert ("DELETE", "/v2/groups/G1/messages/MSG") in [
|
||
(c["method"], c["path"]) for c in calls
|
||
]
|
||
assert "G1" not in music._last_msg
|
||
|
||
|
||
async def test_card_withdrawn_even_on_failure(env):
|
||
"""卡片是一次性 UI：即使发送回退文本，卡片也要撤回（原实现只在成功时撤）。"""
|
||
http, config = env
|
||
api, calls = make_api(fail_types={3}) # 语音上传失败 → 回退文本
|
||
await handle_button(api, http, config, "G1", "I1", PICK_1)
|
||
calls.clear()
|
||
await handle_button(api, http, config, "G1", "I2", PLAY_TX)
|
||
assert ("DELETE", "/v2/groups/G1/messages/MSG") in [
|
||
(c["method"], c["path"]) for c in calls
|
||
]
|
||
assert "G1" not in music._last_msg
|
||
|
||
|
||
async def test_card_withdrawn_when_play_url_unavailable(env):
|
||
"""拿不到播放链接（MusicServer 不可达，send_song 直接 reply 后返回）时
|
||
也要撤回卡片。"""
|
||
http, config = env
|
||
api, calls = make_api()
|
||
await handle_button(api, http, config, "G1", "I1", PICK_1)
|
||
calls.clear()
|
||
broken = replace(config, music_server_url="http://127.0.0.1:1")
|
||
await handle_button(api, http, broken, "G1", "I3", DL_TX)
|
||
assert ("DELETE", "/v2/groups/G1/messages/MSG") in [
|
||
(c["method"], c["path"]) for c in calls
|
||
]
|
||
|
||
|
||
async def test_file_upload_permission_denied_falls_back(env):
|
||
"""SEND_MODE=file 且群文件上传被拒（40093005）→ 回退直链文本 + 说明，并撤回卡片。"""
|
||
http, config = env
|
||
api, calls = make_api(fail_file_perm=True)
|
||
file_cfg = replace(config, send_mode="file")
|
||
await handle_button(api, http, file_cfg, "G1", "I1", PICK_1)
|
||
calls.clear()
|
||
await handle_button(api, http, file_cfg, "G1", "I3", DL_TX)
|
||
body = posts(calls)[-1]["json"]
|
||
assert body["msg_type"] == 0
|
||
assert "/media/tx/T1.mp3" in body["content"]
|
||
assert "仅限群主/管理员" in body["content"]
|
||
assert ("DELETE", "/v2/groups/G1/messages/MSG") in [
|
||
(c["method"], c["path"]) for c in calls
|
||
]
|
||
|
||
|
||
async def test_voice_upload_unaffected_by_file_permission(env):
|
||
"""40093005 只针对文件；语音上传不受影响（仍是语音消息）。"""
|
||
http, config = env
|
||
api, calls = make_api(fail_file_perm=True)
|
||
await handle_button(api, http, config, "G1", "I1", PICK_1)
|
||
calls.clear()
|
||
await handle_button(api, http, config, "G1", "I2", PLAY_TX)
|
||
upload = [c for c in calls if c.get("file_type")][0]
|
||
assert upload["file_type"] == 3
|
||
assert posts(calls)[-1]["json"]["msg_type"] == 7
|
||
|
||
|
||
async def test_withdraw_disabled_keeps_messages(env):
|
||
"""WITHDRAW 关闭时不出任何撤回请求（菜单/卡片都留着）。"""
|
||
http, config = env
|
||
api, calls = make_api()
|
||
off = replace(config, withdraw=False)
|
||
await send_menu(api, http, off, "G1", "稻香")
|
||
await send_menu(api, http, off, "G1", "稻香") # 出新菜单也不撤旧
|
||
await handle_button(api, http, off, "G1", "I1", PICK_1)
|
||
await handle_button(api, http, off, "G1", "I2", PLAY_TX)
|
||
assert not [c for c in calls if c["method"] == "DELETE"]
|
||
|
||
|
||
async def test_play_voice_falls_back_to_text(env):
|
||
"""播放只发语音：语音失败直接回退直链文本，不再尝试文件。"""
|
||
http, config = env
|
||
api, calls = make_api(fail_types={3})
|
||
await handle_button(api, http, config, "G1", "I2", PLAY_TX)
|
||
body = posts(calls)[-1]["json"]
|
||
assert body["msg_type"] == 0
|
||
assert "/media/tx/T1.mp3" in body["content"]
|
||
|
||
|
||
async def test_dl_falls_back_to_text(env):
|
||
"""SEND_MODE=file 时文件上传失败 → 回退直链文本。"""
|
||
http, config = env
|
||
api, calls = make_api(fail_types={4})
|
||
file_cfg = replace(config, send_mode="file")
|
||
await handle_button(api, http, file_cfg, "G1", "I3", DL_TX)
|
||
body = posts(calls)[-1]["json"]
|
||
assert body["msg_type"] == 0
|
||
assert "/media/tx/T1.mp3" in body["content"]
|
||
|
||
|
||
async def test_upload_timeout_falls_back_to_text(env):
|
||
"""上传超时（QQ 服务器抓流超过请求超时）→ 回退文本。"""
|
||
http, config = env
|
||
api, calls = make_api(fail_upload=True)
|
||
await handle_button(api, http, config, "G1", "I2", PLAY_TX)
|
||
body = posts(calls)[-1]["json"]
|
||
assert "/media/tx/T1.mp3" in body["content"]
|
||
|
||
|
||
async def test_legacy_pick_text_is_ignored(env):
|
||
"""旧的「点歌播放 …」文字指令已废除，不再匹配（按钮回调取代）。"""
|
||
http, config = env
|
||
api, calls = make_api()
|
||
assert not await handle_command(api, http, config, "G1", "m1", "点歌播放 1")
|
||
assert calls == []
|
||
|
||
|
||
async def test_card_withdrawn_when_send_song_raises(env):
|
||
"""send_song 抛错（如候选源搜索时 MusicServer 挂掉）也要撤回卡片——
|
||
这是 finally 存在的理由，否则卡片会永久残留。"""
|
||
http, config = env
|
||
api, calls = make_api()
|
||
await handle_button(api, http, config, "G1", "I1", PICK_1)
|
||
calls.clear()
|
||
# 换一个不可达的 MusicServer：直链探测与候选源搜索都会失败
|
||
broken = replace(config, music_server_url="http://127.0.0.1:1")
|
||
await handle_button(api, http, broken, "G1", "I2", PLAY_TX)
|
||
assert ("DELETE", "/v2/groups/G1/messages/MSG") in [
|
||
(c["method"], c["path"]) for c in calls
|
||
]
|
||
assert "G1" not in music._last_msg
|
||
|
||
|
||
async def test_card_play_falls_back_to_candidate_source(server):
|
||
"""tx 直链不可抓取时，卡片上的播放同样走候选源探测，用 kw 直链上传。"""
|
||
s = FakeMusicServer(tx_denied=True)
|
||
await s.start()
|
||
try:
|
||
async with ClientSession() as http:
|
||
config = BotConfig(
|
||
music_server_url=f"http://127.0.0.1:{s.port}",
|
||
music_quality="128k",
|
||
music_source="tx",
|
||
)
|
||
api, calls = make_api()
|
||
await handle_button(api, http, config, "G1", "I2", PLAY_TX)
|
||
finally:
|
||
await s.stop()
|
||
upload = [c for c in calls if c.get("file_type")][0]
|
||
assert upload["file_type"] == 3
|
||
assert "/media/kw/" in upload["url"] # 绕开被风控的 tx 直链
|
||
|
||
|
||
async def test_page_button_new_menu(env):
|
||
http, config = env
|
||
api, calls = make_api()
|
||
data = f"音乐|page|{quote('稻香', safe='')}|2"
|
||
assert await handle_button(
|
||
api, http, config, "G1", "I1", data, event_id="INTERACTION_CREATE:I1"
|
||
)
|
||
body = posts(calls)[0]["json"]
|
||
assert body["msg_type"] == 2 # 翻页后发新菜单
|
||
assert "点歌：稻香" in body["markdown"]["content"]
|
||
assert (calls[-1]["method"], calls[-1]["path"]) == ("PUT", "/interactions/I1")
|
||
|
||
api, calls = make_api()
|
||
assert not await handle_button(api, http, config, "G1", "I1", "别的插件|xx")
|
||
assert calls == [] # 非音乐按钮不处理也不回应
|
||
|
||
|
||
async def test_slash_commands_share_text_parsing(env):
|
||
"""斜杠指令与裸文本命令同一套解析：剥掉开头的 / 与 @机器人 前缀。"""
|
||
http, config = env
|
||
for content in ("点歌 稻香", "/点歌 稻香", "<@!BOT123> /点歌 稻香"):
|
||
api, calls = make_api()
|
||
assert await handle_command(api, http, config, "G1", "m1", content)
|
||
body = posts(calls)[0]["json"]
|
||
assert body["msg_type"] == 2
|
||
assert "点歌：稻香" in body["markdown"]["content"]
|
||
|
||
api, calls = make_api()
|
||
assert await handle_command(api, http, config, "G1", "m1", "/歌词 稻香")
|
||
assert posts(calls)[0]["json"]["content"] == "第一行\n第二行"
|
||
|
||
|
||
async def test_slash_help_sends_command_keyboard(env):
|
||
"""/帮助 出帮助正文 + type=2 指令按钮（点一下往输入框填斜杠指令）。"""
|
||
http, config = env
|
||
api, calls = make_api()
|
||
assert await handle_command(api, http, config, "G1", "m1", "/帮助")
|
||
body = posts(calls)[0]["json"]
|
||
assert "点歌" in body["content"] and "歌词" in body["content"]
|
||
rows = buttons(body)
|
||
assert len(rows) == 1
|
||
song_btn, lyric_btn = rows[0]["buttons"]
|
||
assert song_btn["action"]["type"] == 2
|
||
assert song_btn["action"]["data"] == "/点歌 "
|
||
assert lyric_btn["action"]["data"] == "/歌词 "
|
||
|
||
|
||
async def test_slash_help_falls_back_to_plain_text(env):
|
||
http, config = env
|
||
api, calls = make_api(fail_keyboard=True)
|
||
assert await handle_command(api, http, config, "G1", "m1", "/帮助")
|
||
sent = posts(calls)
|
||
assert len(sent) == 1 # 键盘被拒，回退为一条纯文本
|
||
assert "keyboard" not in sent[0]["json"]
|
||
assert "点歌" in sent[0]["json"]["content"]
|
||
|
||
|
||
async def test_bare_command_replies_usage(env):
|
||
"""斜杠菜单的参数可选时可能不带参发出来：只敲命令给用法提示。"""
|
||
http, config = env
|
||
for content, hint in (("点歌", "用法：点歌"), ("/歌词", "用法：歌词")):
|
||
api, calls = make_api()
|
||
assert await handle_command(api, http, config, "G1", "m1", content)
|
||
assert posts(calls)[0]["json"]["content"].startswith(hint)
|
||
|
||
|
||
async def test_lyric_miss_help_and_no_hit(env):
|
||
http, config = env
|
||
api, calls = make_api()
|
||
await handle_command(api, http, config, "G1", "m1", "歌词 稻香")
|
||
assert posts(calls)[0]["json"]["content"] == "第一行\n第二行"
|
||
|
||
api, calls = make_api()
|
||
await handle_command(api, http, config, "G1", "m1", "点歌 空白歌")
|
||
assert posts(calls)[0]["json"]["content"] == "没找到「空白歌」"
|
||
|
||
broken = BotConfig(
|
||
music_server_url="http://127.0.0.1:1",
|
||
music_quality="128k",
|
||
music_source="tx",
|
||
)
|
||
api, calls = make_api()
|
||
await handle_command(api, http, broken, "G1", "m1", "点歌 稻香")
|
||
assert posts(calls)[0]["json"]["content"] == "音乐服务暂时不可用，稍后再试试"
|
||
|
||
api, calls = make_api()
|
||
assert await handle_command(api, http, config, "G1", "m1", "音乐帮助")
|
||
assert "点歌" in posts(calls)[0]["json"]["content"]
|
||
assert not await handle_command(api, http, config, "G1", "m1", "早上好")
|
||
|
||
|
||
def test_trim_strips_tags_and_truncates():
|
||
from musicbot.music import _trim
|
||
|
||
text = "[kuwo:113]\n[ti:稻香]\n[00:01.00]正文一行\n[offset:0]\n"
|
||
assert _trim(text) == "正文一行"
|
||
long = "\n".join(f"[00:{i:02d}.00]第{i}行" for i in range(100))
|
||
trimmed = _trim(long, limit=100)
|
||
assert trimmed.endswith("……(歌词过长已截断)")
|
||
assert len(trimmed) <= 100 + len("\n……(歌词过长已截断)")
|
||
|
||
|
||
class PanelRecording(BotApi):
|
||
"""预设 /v2/panels 响应并记录调用：fail_get=True 时查询抛 403。"""
|
||
|
||
def __init__(self, records: list | None = None, fail_get: bool = False) -> None:
|
||
self.records = records or []
|
||
self.fail_get = fail_get
|
||
self.calls: list[dict] = []
|
||
|
||
async def request(self, method: str, path: str, **kwargs: Any) -> Any:
|
||
if self.fail_get and path == "/v2/panels":
|
||
raise ApiError(403, 11253, "无权限")
|
||
self.calls.append({"method": method, "path": path, **kwargs})
|
||
if method == "GET":
|
||
# 响应整体包一层 {"data": ...}，顺带覆盖 _unwrap 的拆包路径
|
||
return {"data": {"records": self.records, "is_end": True}}
|
||
if method == "POST":
|
||
return {"panel_id": "P1"}
|
||
return {"version": 2}
|
||
|
||
|
||
async def test_panel_created_when_missing():
|
||
api = PanelRecording()
|
||
await sync_command_panel(api)
|
||
assert [(c["method"], c["path"]) for c in api.calls] == [
|
||
("GET", "/v2/panels"),
|
||
("POST", "/v2/panels"),
|
||
]
|
||
body = api.calls[1]["json"]
|
||
assert body["scope"] == "group" and body["target_type"] == "all"
|
||
assert [i["name"] for i in body["panel"]["items"]] == ["点歌", "歌词", "帮助"]
|
||
|
||
|
||
async def test_panel_skipped_when_identical():
|
||
items = [
|
||
{"name": "点歌", "type": "command", "desc": "出选歌菜单，点编号选歌"},
|
||
{"name": "歌词", "type": "command", "desc": "查看歌词"},
|
||
{"name": "帮助", "type": "command", "desc": "查看帮助"},
|
||
]
|
||
api = PanelRecording(
|
||
records=[{"panel_id": "P9", "panel": {"items": items, "remark": "音乐bot"}}]
|
||
)
|
||
await sync_command_panel(api)
|
||
assert [(c["method"], c["path"]) for c in api.calls] == [("GET", "/v2/panels")]
|
||
|
||
|
||
async def test_panel_updated_when_differs():
|
||
# 这条用响应不包 {"data"} 外层的形态，覆盖 _unwrap 的另一分支
|
||
api = PanelRecording(
|
||
records=[
|
||
{
|
||
"panel_id": "P9",
|
||
"panel": {
|
||
"items": [{"name": "点歌", "type": "command", "desc": "旧"}],
|
||
"remark": "音乐bot",
|
||
},
|
||
}
|
||
]
|
||
)
|
||
await sync_command_panel(api)
|
||
assert (api.calls[1]["method"], api.calls[1]["path"]) == ("PUT", "/v2/panels/P9")
|
||
|
||
|
||
async def test_panel_failure_swallowed():
|
||
api = PanelRecording(fail_get=True)
|
||
await sync_command_panel(api) # 失败只记日志，绝不抛出
|
||
assert api.calls == []
|
||
|
||
|
||
def test_config_send_mode(monkeypatch):
|
||
"""SEND_MODE 默认 link（避开文件上传权限），file 可显式开启，非法值回落 link。"""
|
||
from musicbot.config import BotConfig
|
||
|
||
monkeypatch.delenv("SEND_MODE", raising=False)
|
||
assert BotConfig.load().send_mode == "link"
|
||
|
||
monkeypatch.setenv("SEND_MODE", "FILE") # 大小写不敏感
|
||
assert BotConfig.load().send_mode == "file"
|
||
|
||
monkeypatch.setenv("SEND_MODE", "随便写的")
|
||
assert BotConfig.load().send_mode == "link" # 非法值不报错，按默认处理
|
||
|
||
|
||
def test_config_log_level(monkeypatch):
|
||
"""LOG_LEVEL 默认 INFO，可显式设置（大小写不敏感），非法值回落 INFO。"""
|
||
from musicbot.config import BotConfig
|
||
|
||
monkeypatch.delenv("LOG_LEVEL", raising=False)
|
||
assert BotConfig.load().log_level == "INFO"
|
||
|
||
monkeypatch.setenv("LOG_LEVEL", "debug")
|
||
assert BotConfig.load().log_level == "DEBUG"
|
||
|
||
monkeypatch.setenv("LOG_LEVEL", "乱写")
|
||
assert BotConfig.load().log_level == "INFO"
|
||
|
||
|
||
def test_setup_logging_writes_file(tmp_path: Path):
|
||
"""setup_logging 同时写控制台与文件；写不进去时退化为仅控制台不抛错。"""
|
||
from loguru import logger as lg
|
||
|
||
from musicbot.config import BotConfig
|
||
from musicbot.main import setup_logging
|
||
|
||
cfg = BotConfig(
|
||
music_server_url="http://127.0.0.1:1",
|
||
music_quality="128k",
|
||
music_source="tx",
|
||
log_level="INFO",
|
||
log_dir=tmp_path,
|
||
)
|
||
try:
|
||
setup_logging(cfg)
|
||
lg.info("测试日志落盘")
|
||
lg.complete() # 等 enqueue 队列刷完
|
||
logfile = tmp_path / "musicbot.log"
|
||
assert logfile.is_file()
|
||
assert "测试日志落盘" in logfile.read_text(encoding="utf-8")
|
||
finally:
|
||
lg.remove() # 清掉本用例加的 handler，避免影响其他测试
|
||
|
||
# 目录不可写（用一个文件当目录）→ 不抛错，仅告警
|
||
blocked = tmp_path / "not-a-dir"
|
||
blocked.write_text("x", encoding="utf-8")
|
||
setup_logging(replace(cfg, log_dir=blocked))
|
||
lg.remove()
|
||
|
||
|
||
def test_config_reload_flag(monkeypatch):
|
||
"""RELOAD 默认关（生产不受影响），on 可开启，非法值按关处理。"""
|
||
from musicbot.config import BotConfig
|
||
|
||
monkeypatch.delenv("RELOAD", raising=False)
|
||
assert BotConfig.load().reload is False
|
||
|
||
monkeypatch.setenv("RELOAD", "on")
|
||
assert BotConfig.load().reload is True
|
||
|
||
monkeypatch.setenv("RELOAD", "随便写")
|
||
assert BotConfig.load().reload is False
|
||
|
||
|
||
# ---- 热重载：文件监听与进程监督 ----
|
||
|
||
def test_watched_files_and_skip_rules(tmp_path: Path):
|
||
"""监控 .py/.toml 与 .env；跳过 logs/、__pycache__/、.venv/ 与无关文件。"""
|
||
from musicbot import reload
|
||
|
||
(tmp_path / "musicbot").mkdir()
|
||
(tmp_path / "musicbot" / "music.py").write_text("", encoding="utf-8")
|
||
(tmp_path / "tests").mkdir()
|
||
(tmp_path / "tests" / "test_x.py").write_text("", encoding="utf-8")
|
||
(tmp_path / "intents.toml").write_text("", encoding="utf-8")
|
||
(tmp_path / ".env").write_text("", encoding="utf-8")
|
||
(tmp_path / "README.md").write_text("", encoding="utf-8") # 无关后缀
|
||
(tmp_path / "logs").mkdir()
|
||
(tmp_path / "logs" / "musicbot.log").write_text("", encoding="utf-8")
|
||
(tmp_path / "musicbot" / "__pycache__").mkdir()
|
||
(tmp_path / "musicbot" / "__pycache__" / "x.py").write_text("", encoding="utf-8")
|
||
|
||
watched = reload.iter_watched_files(tmp_path)
|
||
names = {p.relative_to(tmp_path).as_posix() for p in watched}
|
||
assert names == {"musicbot/music.py", "tests/test_x.py", "intents.toml", ".env"}
|
||
|
||
|
||
def test_changed_files_detects_edit_add_remove(tmp_path: Path):
|
||
"""快照比对：改动/新增/删除都能识别，未变则返回空。"""
|
||
from musicbot import reload
|
||
|
||
f = tmp_path / "musicbot"
|
||
f.mkdir()
|
||
mod = f / "music.py"
|
||
mod.write_text("a", encoding="utf-8")
|
||
|
||
before = reload.snapshot(tmp_path)
|
||
assert reload.changed_files(before, reload.snapshot(tmp_path)) == []
|
||
|
||
mod.write_text("bb", encoding="utf-8")
|
||
assert reload.changed_files(before, reload.snapshot(tmp_path)) == [mod]
|
||
|
||
|
||
def test_is_child_guard(monkeypatch):
|
||
"""子进程标记生效：带标记的进程不再做监督，避免 --reload 无限自我复制。"""
|
||
from musicbot import reload
|
||
|
||
monkeypatch.delenv(reload.CHILD_ENV, raising=False)
|
||
assert reload.is_child() is False
|
||
monkeypatch.setenv(reload.CHILD_ENV, "1")
|
||
assert reload.is_child() is True
|
||
|
||
|
||
def test_main_child_does_not_supervise(monkeypatch):
|
||
"""子进程即使带 --reload / RELOAD=on 也只跑 bot，不起监督
|
||
（否则 --reload 会无限自我复制）。"""
|
||
from musicbot import main as mb_main
|
||
from musicbot import reload
|
||
|
||
called: dict[str, bool] = {}
|
||
|
||
async def fake_amain() -> None:
|
||
called["bot"] = True
|
||
|
||
monkeypatch.setattr(reload, "is_child", lambda: True)
|
||
monkeypatch.setattr(mb_main, "amain", fake_amain)
|
||
monkeypatch.setattr(
|
||
reload, "run_supervisor",
|
||
lambda *a, **k: called.setdefault("supervisor", True),
|
||
)
|
||
monkeypatch.setattr(mb_main.BotConfig, "load", staticmethod(
|
||
lambda: replace(
|
||
BotConfig(
|
||
music_server_url="http://127.0.0.1:1",
|
||
music_quality="128k",
|
||
music_source="tx",
|
||
),
|
||
reload=True,
|
||
)
|
||
))
|
||
|
||
mb_main.main(["--reload"])
|
||
assert called == {"bot": True} # 跑的是 bot，没有起监督
|
||
|
||
|
||
# ---- 全量模式（不 @ 也能用）与去重 ----
|
||
|
||
def make_emitter(api, http, config) -> EventEmitter:
|
||
"""按 main.py 的装配方式注册 handler，手工注入 SDK 会在 run_loop 里
|
||
登记的 BotApi；http 走闭包（与 main.py 一致）。"""
|
||
ee = EventEmitter()
|
||
register(ee, config, http)
|
||
ee.services.update({BotApi: api})
|
||
return ee
|
||
|
||
|
||
async def push(ee: EventEmitter, event: str, d: dict) -> None:
|
||
await ee.emit({"op": 0, "t": event, "d": d})
|
||
|
||
|
||
def group_payload(msg_id: str, content: str, *, bot: bool = False) -> dict:
|
||
return {
|
||
"id": msg_id,
|
||
"content": content,
|
||
"group_openid": "G1",
|
||
"author": {"member_openid": "U1", "bot": bot},
|
||
}
|
||
|
||
|
||
async def test_no_at_message_sends_menu(env):
|
||
"""全量模式：群里不 @ 机器人也点歌（GROUP_MESSAGE_CREATE），
|
||
被动回复仍用这条消息的 msg_id。"""
|
||
http, config = env
|
||
api, calls = make_api()
|
||
ee = make_emitter(api, http, config)
|
||
await push(ee, "GROUP_MESSAGE_CREATE", group_payload("m1", "点歌 稻香"))
|
||
body = posts(calls)[0]["json"]
|
||
assert body["msg_id"] == "m1"
|
||
# 官方 event_id 支持列表不含 GROUP_MESSAGE_CREATE，只能用 msg_id
|
||
assert "event_id" not in body
|
||
assert "## 点歌：稻香" in body["markdown"]["content"]
|
||
|
||
|
||
async def test_at_message_still_handled(env):
|
||
"""@消息（默认事件）行为不变。"""
|
||
http, config = env
|
||
api, calls = make_api()
|
||
ee = make_emitter(api, http, config)
|
||
await push(ee, "GROUP_AT_MESSAGE_CREATE", group_payload("m1", "点歌 稻香"))
|
||
assert "## 点歌：稻香" in posts(calls)[0]["json"]["markdown"]["content"]
|
||
|
||
|
||
async def test_group_message_ignores_own_and_other_chat(env):
|
||
"""机器人自己的消息（可能被回推）与非命令消息一律不回复。"""
|
||
http, config = env
|
||
api, calls = make_api()
|
||
ee = make_emitter(api, http, config)
|
||
await push(ee, "GROUP_MESSAGE_CREATE", group_payload("m1", "点歌 稻香", bot=True))
|
||
await push(ee, "GROUP_MESSAGE_CREATE", group_payload("m2", ""))
|
||
await push(ee, "GROUP_MESSAGE_CREATE", {
|
||
"id": "m3", "content": "早上好", "group_openid": "G1",
|
||
"author": {"member_openid": "U1"},
|
||
})
|
||
assert calls == []
|
||
|
||
|
||
async def test_duplicate_push_handled_once(env):
|
||
"""同一 msg_id 重复推送、或 @消息同时命中两个事件时只处理一次。"""
|
||
http, config = env
|
||
api, calls = make_api()
|
||
ee = make_emitter(api, http, config)
|
||
payload = group_payload("m1", "点歌 稻香")
|
||
await push(ee, "GROUP_MESSAGE_CREATE", payload)
|
||
await push(ee, "GROUP_MESSAGE_CREATE", payload)
|
||
await push(ee, "GROUP_AT_MESSAGE_CREATE", payload)
|
||
assert len(posts(calls)) == 1
|
||
|
||
|
||
async def test_no_at_command_without_content_is_ignored(env):
|
||
"""纯图片等无正文消息（全量模式同样推送）不触发任何请求。"""
|
||
http, config = env
|
||
api, calls = make_api()
|
||
ee = make_emitter(api, http, config)
|
||
await push(ee, "GROUP_MESSAGE_CREATE", {
|
||
"id": "m1",
|
||
"group_openid": "G1",
|
||
"author": {"member_openid": "U1"},
|
||
"attachments": [{"content_type": 1, "url": "https://x/1.png"}],
|
||
})
|
||
assert calls == []
|
||
|
||
|
||
def test_dedup_expires_and_bounds(monkeypatch):
|
||
"""去重表：过期记录可再次处理，条数达到上限时淘汰最旧。"""
|
||
assert not music._handled("m1")
|
||
assert music._handled("m1")
|
||
|
||
# TTL 过去后同一条消息重新可处理
|
||
monkeypatch.setattr(music, "_DEDUP_TTL", 0.0)
|
||
assert not music._handled("m1")
|
||
|
||
# 记录数超过上限时淘汰最旧的
|
||
monkeypatch.setattr(music, "_DEDUP_TTL", 300.0)
|
||
monkeypatch.setattr(music, "_DEDUP_MAX", 2)
|
||
music._seen_messages.clear()
|
||
for i in range(3):
|
||
music._handled(f"n{i}")
|
||
assert list(music._seen_messages) == ["n1", "n2"]
|
||
|
||
|
||
async def test_bot_own_reply_is_not_self_triggered(env):
|
||
"""bot 自己发的帮助文本里有「点歌」「音乐帮助」，被全量模式回推时不能
|
||
自触发（否则递归出菜单）；作者无 bot 标记时靠发送记录兜底。"""
|
||
http, config = env
|
||
api, calls = make_api()
|
||
ee = make_emitter(api, http, config)
|
||
assert await handle_command(api, http, config, "G1", "m1", "音乐帮助")
|
||
assert len(posts(calls)) == 1
|
||
|
||
# 平台把这个已发出的消息（id=MSG）回推，author 不含 bot 字段
|
||
await push(ee, "GROUP_MESSAGE_CREATE", {
|
||
"id": "MSG", "content": "♪ 音乐 bot\n点歌 <歌名>：出菜单",
|
||
"group_openid": "G1", "author": {"member_openid": "U1"},
|
||
})
|
||
assert len(posts(calls)) == 1 # 未再发任何消息
|
||
|
||
# 同样内容的新消息（不同 id）来自真人 → 正常处理
|
||
await push(ee, "GROUP_MESSAGE_CREATE", {
|
||
"id": "m9", "content": "音乐帮助", "group_openid": "G1",
|
||
"author": {"member_openid": "U2"},
|
||
})
|
||
assert len(posts(calls)) == 2
|