snowy-yang
bb60137ac7
代码测试 / ruff + pytest (push) Canceled after 0s
bili 的 songmid 形如「BV号|cid」，与回调 data 的分隔符撞车：拆段 从 5 变 6，handle_button 判为非本 bot 按钮直接丢弃，且发生在回应 互动之前——客户端按钮一直转圈、日志无任何记录，表象即「点歌卡死」。 _song_data 给歌曲 id 一并 % 转义（_q），_song_of_data 对应 unquote； 旧按钮 id 不含 %，unquote 幂等不受影响。附 bili 往返回归测试。
1898 lines
77 KiB
Python
1898 lines
77 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, proxy
|
||
from musicbot.config import BotConfig
|
||
from musicbot.music import (
|
||
handle_button,
|
||
handle_command,
|
||
register,
|
||
search,
|
||
send_menu,
|
||
)
|
||
from musicbot.panel import sync_command_panel
|
||
from musicbot.proxy import FileProxy
|
||
|
||
|
||
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 的统一信封接口（路径式 /search/{source}/{keyword}），
|
||
记录搜索/榜单请求参数。
|
||
|
||
/url 返回指向自身的直链；tx_denied=True 时 tx 的直链指向 403
|
||
路由，模拟 CDN 风控（测直链探测与候选源回退）。
|
||
"""
|
||
|
||
def __init__(
|
||
self,
|
||
tx_denied: bool = False,
|
||
serve_sources: bool = True,
|
||
no_url_sources: frozenset[str] = frozenset(),
|
||
extra_platforms: tuple[dict, ...] = (),
|
||
album_total: int = 23,
|
||
) -> None:
|
||
self.tx_denied = tx_denied
|
||
self.serve_sources = serve_sources
|
||
self.no_url_sources = no_url_sources
|
||
self.extra_platforms = extra_platforms
|
||
self.album_total = album_total
|
||
self.album_empty_ids: frozenset[str] = frozenset() # 已下线的节目
|
||
self.runner: web.AppRunner | None = None
|
||
self.port = 0
|
||
self.queries: list[dict] = []
|
||
self.board_queries: list[dict] = []
|
||
self.album_queries: list[dict] = []
|
||
|
||
async def start(self) -> None:
|
||
app = web.Application()
|
||
app.router.add_get("/search/{source}/{keyword}/{page}", self._search)
|
||
app.router.add_get("/search/{source}/{keyword}", 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("/hot-search/{source}", self._hot)
|
||
app.router.add_get("/boards/{source}/{board_id}/{page}", self._board_detail)
|
||
app.router.add_get("/boards/{source}", self._boards)
|
||
app.router.add_get("/album/{source}/{album_id}/{page}", self._album)
|
||
app.router.add_get("/album/{source}/{album_id}", self._album)
|
||
app.router.add_get("/login/status", self._login_status)
|
||
if self.serve_sources:
|
||
app.router.add_get("/sources", self._sources)
|
||
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:
|
||
info = request.match_info
|
||
source, keyword = info["source"], info["keyword"]
|
||
self.queries.append({
|
||
"keyword": keyword,
|
||
"source": source,
|
||
"page": info.get("page", "1"),
|
||
"limit": request.query.get("limit"),
|
||
})
|
||
items: list[dict] = []
|
||
if "空" not in keyword:
|
||
if "兜底" in keyword:
|
||
# tx 搜不到、聚合（all）有 —— 测兜底路径
|
||
items = [] if source == "tx" else [
|
||
{"source": "kw", "songmid": "S1",
|
||
"name": keyword, "singer": "歌手A"},
|
||
]
|
||
elif source == "kw":
|
||
items = [
|
||
{"source": "kw", "songmid": "S1",
|
||
"name": keyword, "singer": "歌手A"},
|
||
]
|
||
elif source == "kg":
|
||
items = [
|
||
# 不带 source 字段：真机 kg 带、xm/me/qt/lr 不带，
|
||
# 这里顺带覆盖 _songs_of 的请求级补源
|
||
{"songmid": "332", "hash": "G1",
|
||
"name": keyword, "singer": "歌手G"}
|
||
]
|
||
elif source in ("tx", "all"):
|
||
items = [
|
||
{
|
||
"source": "tx",
|
||
"songmid": "T1",
|
||
"name": keyword,
|
||
"singer": "歌手T",
|
||
"albumName": "专辑T",
|
||
"interval": "04:00",
|
||
"img": "https://img.test/cover.jpg",
|
||
},
|
||
{"source": "kw", "songmid": "S1",
|
||
"name": keyword, "singer": "歌手A"},
|
||
]
|
||
return web.json_response({"code": 0, "data": {"list": items}})
|
||
|
||
async def _hot(self, request: web.Request) -> web.Response:
|
||
return web.json_response({
|
||
"code": 0,
|
||
"data": {
|
||
"source": request.match_info["source"],
|
||
"list": ["热词一", "热词二", "热词三"],
|
||
},
|
||
})
|
||
|
||
async def _boards(self, request: web.Request) -> web.Response:
|
||
source = request.match_info["source"]
|
||
return web.json_response({
|
||
"code": 0,
|
||
"data": {"list": [
|
||
{"id": f"{source}__93", "name": "飙升榜", "bangid": "93"},
|
||
{"id": f"{source}__16", "name": "热歌榜", "bangid": "16"},
|
||
]},
|
||
})
|
||
|
||
async def _board_detail(self, request: web.Request) -> web.Response:
|
||
info = request.match_info
|
||
limit = request.query.get("limit", "10")
|
||
self.board_queries.append({
|
||
"board": info["board_id"], "page": info["page"], "limit": limit,
|
||
})
|
||
page = int(info["page"])
|
||
songs = [
|
||
{
|
||
"source": "tx",
|
||
"songmid": f"{info['board_id']}-P{page}-{i}",
|
||
"name": f"榜歌{i}",
|
||
"singer": f"歌手{i}",
|
||
"interval": "03:00",
|
||
"img": "https://img.test/board.jpg",
|
||
}
|
||
for i in range(1, int(limit) + 1)
|
||
]
|
||
return web.json_response({
|
||
"code": 0,
|
||
"data": {"list": songs, "total": 30, "page": page, "limit": int(limit)},
|
||
})
|
||
|
||
async def _album(self, request: web.Request) -> web.Response:
|
||
"""选集接口，仿真真机差异：me 忽略 page 参数、一次给足 limit；
|
||
lr/qt 按服务端翻页切片。album_empty_ids 里的节目给空列表（已下线）。"""
|
||
info = request.match_info
|
||
source, album_id = info["source"], info["album_id"]
|
||
page = int(info.get("page", 1))
|
||
limit = int(request.query.get("limit", 20))
|
||
self.album_queries.append({
|
||
"source": source, "album_id": album_id, "page": page, "limit": limit,
|
||
})
|
||
if album_id in self.album_empty_ids:
|
||
return web.json_response({"code": 0, "data": {"list": [], "total": 0}})
|
||
eps = [
|
||
{
|
||
"songmid": f"{album_id}-{i}",
|
||
"albumId": album_id,
|
||
"name": f"第{i}期",
|
||
"singer": "节目名",
|
||
"interval": "0",
|
||
"img": "https://img.test/ep.jpg",
|
||
}
|
||
for i in range(1, self.album_total + 1)
|
||
]
|
||
if source == "me":
|
||
window = eps[:limit]
|
||
else:
|
||
window = eps[(page - 1) * limit: page * limit]
|
||
return web.json_response({
|
||
"code": 0, "data": {"list": window, "total": self.album_total},
|
||
})
|
||
|
||
async def _login_status(self, _: web.Request) -> web.Response:
|
||
return web.json_response({
|
||
"code": 0,
|
||
"data": {
|
||
"kg": {"logged": True, "nickname": "测试号"},
|
||
"tx": {"logged": False},
|
||
},
|
||
})
|
||
|
||
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 _sources(self, _: web.Request) -> web.Response:
|
||
platforms = [
|
||
{
|
||
"id": pid,
|
||
"name": name,
|
||
"capabilities": {
|
||
"search": True,
|
||
"url": pid not in self.no_url_sources,
|
||
"hotSearch": pid in ("kw", "kg", "tx", "wy"),
|
||
"boards": pid in ("kw", "kg", "tx", "wy"),
|
||
},
|
||
}
|
||
for pid, name in (
|
||
("kw", "酷我音乐"),
|
||
("kg", "酷狗音乐"),
|
||
("tx", "QQ音乐"),
|
||
("wy", "网易音乐"),
|
||
("qs", "汽水音乐"),
|
||
)
|
||
]
|
||
platforms.extend(self.extra_platforms)
|
||
return web.json_response({"code": 0, "data": {"platforms": platforms}})
|
||
|
||
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():
|
||
"""菜单/卡片消息记录、消息/回调/跨事件去重表、平台缓存、下载别名与
|
||
文件代理是模块级状态，测试间清空/复位避免串扰。"""
|
||
tables = (music._seen_messages, music._recent_clicks, music._msg_dup_keys)
|
||
|
||
def reset() -> None:
|
||
music._last_msg.clear()
|
||
for t in tables:
|
||
t.clear()
|
||
music._file_proxy = None
|
||
music._platforms_cache.clear()
|
||
music._apply_aliases(music._STATIC_PLATFORMS)
|
||
|
||
reset()
|
||
yield
|
||
reset()
|
||
|
||
|
||
# 回调按钮 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 == [
|
||
{"keyword": "稻香", "source": "tx", "page": "1", "limit": "5"}
|
||
]
|
||
|
||
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", "all"]
|
||
|
||
|
||
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 # 每首歌后面不再标音源
|
||
# 有封面的条目（tx）编号行下跟一张小封面图，无封面的（kw）不加
|
||
assert f"1. 稻香 - 歌手T\n" in md
|
||
assert md.count("![") == 1
|
||
# 菜单里除封面图外没有任何链接
|
||
assert md.replace(IMG_URL, "").count("http") == 0
|
||
# 编号回调按钮（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 "![" not 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,
|
||
]
|
||
# 进门先回应互动（避免 QQ 客户端等超时），再发卡片
|
||
assert (calls[0]["method"], calls[0]["path"]) == ("PUT", "/interactions/I1")
|
||
assert posts(calls)[0]["json"]["msg_type"] == 2
|
||
|
||
|
||
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_card_for_qs_song_shows_source_and_cover(env):
|
||
"""qs（汽水音乐）歌曲：来源显示「汽水音乐」，qs 搜索带封面 → 卡片带图。"""
|
||
http, config = env
|
||
api, calls = make_api()
|
||
cover = "https://p3-luna.douyinpic.com/img/x~c5_500x500.jpg"
|
||
data = music._song_data(
|
||
"pick", music.Song("qs", "7679694801201334312", "稻香", "歌手Q", img=cover)
|
||
)
|
||
await handle_button(api, http, config, "G1", "I1", data)
|
||
md = posts(calls)[0]["json"]["markdown"]["content"]
|
||
assert "## 稻香 - 歌手Q" in md
|
||
assert "来源：汽水音乐" in md
|
||
assert f"" in md
|
||
|
||
|
||
async def test_bili_songmid_pipe_round_trips(env):
|
||
"""bili 的 songmid 是「BV号|cid」，自带分隔符：按钮 data 里的 id 必须
|
||
% 转义，否则拆段变 6 段、整个回调被当成非本 bot 按钮静默丢弃（点歌
|
||
无反应）。转义后 pick 正常弹卡片，播放/下载按钮携带完整 id。"""
|
||
http, config = env
|
||
api, calls = make_api()
|
||
data = music._song_data(
|
||
"pick",
|
||
music.Song("bili", "BV1vg2AYdE4J|26216173455", "听", "UP主"),
|
||
)
|
||
assert data.count("|") == 6 # 7 段：头+action+source+id+name+singer+img
|
||
await handle_button(api, http, config, "G1", "I1", data)
|
||
card = posts(calls)[0]["json"]
|
||
assert "## 听 - UP主" in card["markdown"]["content"]
|
||
assert "来源：哔哩哔哩" in card["markdown"]["content"]
|
||
datas = [b["action"]["data"] for b in buttons(card)[0]["buttons"]]
|
||
assert "BV1vg2AYdE4J%7C26216173455" in datas[0] # ▶ 播放带完整 id
|
||
|
||
|
||
async def test_card_cover_url_is_encoded(env):
|
||
"""卡片封面 URL 走标准百分号编码：空格/中文转义、Markdown 语法里的
|
||
括号也要转（否则会截断 ![]() 链接），查询串结构保留。"""
|
||
http, config = env
|
||
api, calls = make_api()
|
||
img = "https://img.test/封面 图(1).jpg?q=1&x=2"
|
||
data = music._song_data("pick", music.Song("kw", "S1", "稻香", "歌手A", img=img))
|
||
await handle_button(api, http, config, "G1", "I1", data)
|
||
md = posts(calls)[0]["json"]["markdown"]["content"]
|
||
assert (
|
||
"![封面 #240px #240px]"
|
||
"(https://img.test/%E5%B0%81%E9%9D%A2%20%E5%9B%BE%281%29.jpg?q=1&x=2)"
|
||
) in md
|
||
|
||
|
||
async def test_direct_link_url_is_encoded(env):
|
||
"""源站直链带空格/中文时，发出的文本与上传 URL 都是标准编码形态。"""
|
||
http, config = env
|
||
encoded = "/media/kw/%E7%A8%BB%20%E9%A6%99(1).mp3" # 稻 香(1)
|
||
weird = music._song_data("dl", music.Song("kw", "稻 香(1)", "稻香", "歌手A"))
|
||
|
||
api, calls = make_api() # link 模式：直链文本
|
||
await handle_button(api, http, config, "G1", "I1", weird)
|
||
assert encoded in posts(calls)[-1]["json"]["content"]
|
||
|
||
api, calls = make_api() # file 模式：上传给 QQ 的 url 同样编码
|
||
music._recent_clicks.clear() # 同一按钮 data，避开连点防抖
|
||
await handle_button(api, http, replace(config, send_mode="file"), "G1", "I2", weird)
|
||
upload = [c for c in calls if c.get("file_type")][0]
|
||
assert upload["url"].endswith(encoded)
|
||
|
||
|
||
def test_encode_url_standard_and_idempotent():
|
||
"""URL 统一标准编码：只转义非法字符（空格/中文/引号），结构字符与已有
|
||
%XX 原样保留，重复调用结果一致（不二次编码）。"""
|
||
raw = 'https://cdn.test/目录/歌 曲(1)"x".mp3?a=1&b=x y#片段'
|
||
enc = (
|
||
"https://cdn.test/%E7%9B%AE%E5%BD%95/%E6%AD%8C%20%E6%9B%B2(1)%22x%22"
|
||
".mp3?a=1&b=x%20y#%E7%89%87%E6%AE%B5"
|
||
)
|
||
assert music._encode_url(raw) == enc
|
||
assert music._encode_url(enc) == enc # 幂等
|
||
md = music._encode_url("https://img.test/a b(1).jpg", music._MD_URL_SAFE)
|
||
assert md == "https://img.test/a%20b%281%29.jpg"
|
||
|
||
|
||
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 start_file_proxy(http: ClientSession, config: BotConfig) -> FileProxy:
|
||
"""起一个绑临时端口的 FileProxy（与 music.register 同款解析接线），
|
||
起好后把对外基址改成实际端口，claim 出的 URL 才打得通。"""
|
||
p = FileProxy(
|
||
"http://127.0.0.1:1",
|
||
http,
|
||
lambda song, keyword: music._fetchable_play(http, config, song, keyword),
|
||
port=0,
|
||
)
|
||
await p.start()
|
||
p._base = f"http://127.0.0.1:{p.port}"
|
||
return p
|
||
|
||
|
||
async def test_file_proxy_delivers_and_falls_back(server, env):
|
||
"""端到端：claim 换链 → 请求时现场重解析并转发字节。tx 直链被风控
|
||
（403）也能交付——解析自带候选源回退落到 kw，正是本功能要解决的场景。"""
|
||
http, config = env
|
||
server.tx_denied = True # 直链探测 403，逼解析走 kw
|
||
p = await start_file_proxy(http, config)
|
||
try:
|
||
url = p.claim(
|
||
music.Song("tx", "T1", "稻香", "歌手T"), "稻香 歌手T", "稻香 - 歌手T.mp3"
|
||
)
|
||
assert url.startswith(f"http://127.0.0.1:{p.port}/file/")
|
||
assert url.endswith("/" + quote("稻香 - 歌手T.mp3")) # 尾段是下载文件名
|
||
async with http.get(url) as resp:
|
||
assert resp.status == 200
|
||
assert await resp.read() == b"fake-audio-bytes" # kw 媒体字节
|
||
assert resp.headers["Content-Type"] == "application/octet-stream"
|
||
assert resp.content_length == 16 # 上游 Content-Length 透传
|
||
assert resp.headers["Content-Disposition"] == (
|
||
"attachment; filename*=UTF-8''" + quote("稻香 - 歌手T.mp3")
|
||
)
|
||
finally:
|
||
await p.stop()
|
||
|
||
|
||
async def test_file_proxy_key_lifecycle(server, env, monkeypatch):
|
||
"""key 不可猜、会过期；登记总量封顶、按插入序淘汰最旧。"""
|
||
http, config = env
|
||
p = await start_file_proxy(http, config)
|
||
try:
|
||
async with http.get(
|
||
f"http://127.0.0.1:{p.port}/file/no-such-key/x.mp3"
|
||
) as resp:
|
||
assert resp.status == 404 # 未知 key
|
||
|
||
monkeypatch.setattr(proxy, "_TTL", 0.0) # 登记即过期
|
||
url = p.claim(music.Song("kw", "S1", "稻香", "歌手A"), "稻香", "a.mp3")
|
||
async with http.get(url) as resp:
|
||
assert resp.status == 404
|
||
|
||
monkeypatch.setattr(proxy, "_TTL", 7200.0)
|
||
monkeypatch.setattr(proxy, "_MAX_ENTRIES", 2)
|
||
first = p.claim(music.Song("kw", "S1", "稻香", "歌手A"), "稻香", "a.mp3")
|
||
p.claim(music.Song("kw", "S2", "晴天", "歌手B"), "晴天", "b.mp3")
|
||
third = p.claim(music.Song("kw", "S3", "夜曲", "歌手C"), "夜曲", "c.mp3")
|
||
async with http.get(first) as resp: # 超量淘汰最旧
|
||
assert resp.status == 404
|
||
async with http.get(third) as resp:
|
||
assert resp.status == 200
|
||
finally:
|
||
await p.stop()
|
||
|
||
|
||
async def _resolve_none(song, keyword):
|
||
return None
|
||
|
||
|
||
async def _resolve_boom(song, keyword):
|
||
raise RuntimeError("解析坏了")
|
||
|
||
|
||
async def test_file_proxy_resolve_failure_502(server, env):
|
||
"""解析失败（返回 None 或抛错）→ 502，请求方拿到可读的失败而非 500 栈。"""
|
||
http, config = env
|
||
for resolver in (_resolve_none, _resolve_boom):
|
||
p = FileProxy("http://127.0.0.1:1", http, resolver, port=0)
|
||
await p.start()
|
||
p._base = f"http://127.0.0.1:{p.port}"
|
||
try:
|
||
url = p.claim(music.Song("kw", "S1", "稻香", "歌手A"), "稻香", "a.mp3")
|
||
async with http.get(url) as resp:
|
||
assert resp.status == 502
|
||
finally:
|
||
await p.stop()
|
||
|
||
|
||
async def test_send_song_swaps_urls_to_proxy(server, env):
|
||
"""开了代理后：上传给 QQ 的与直链文本里的 URL 都是本机代理地址，
|
||
QQ 服务器/群友随后取这个地址能拿到字节（请求时重解析）。"""
|
||
http, config = env
|
||
p = await start_file_proxy(http, config)
|
||
music._file_proxy = p
|
||
try:
|
||
api, calls = make_api()
|
||
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"].startswith(f"http://127.0.0.1:{p.port}/file/")
|
||
async with http.get(upload["url"]) as resp: # QQ 侧抓这个地址
|
||
assert resp.status == 200
|
||
assert await resp.read() == b"fake-audio-bytes"
|
||
|
||
api, calls = make_api() # link 模式下载文本同样换链
|
||
await handle_button(api, http, config, "G1", "I3", DL_TX)
|
||
body = posts(calls)[-1]["json"]
|
||
assert f"http://127.0.0.1:{p.port}/file/" in body["content"]
|
||
assert "/media/" not in body["content"]
|
||
finally:
|
||
music._file_proxy = None
|
||
await p.stop()
|
||
|
||
|
||
async def test_file_proxy_not_ready_keeps_direct_link(server, env):
|
||
"""代理实例在但服务没起来（端口被占等）→ ready=False 不换链，
|
||
发送链路退回原直链、不受影响。"""
|
||
http, config = env
|
||
p = FileProxy(
|
||
"http://127.0.0.1:1",
|
||
http,
|
||
lambda song, keyword: music._fetchable_play(http, config, song, keyword),
|
||
port=1,
|
||
) # 不 start：ready=False
|
||
music._file_proxy = p
|
||
api, calls = make_api()
|
||
await handle_button(api, http, config, "G1", "I2", PLAY_TX)
|
||
upload = [c for c in calls if c.get("file_type")][0]
|
||
assert upload["url"].endswith("/media/tx/T1.mp3")
|
||
|
||
|
||
async def test_register_wires_file_proxy(server, env):
|
||
"""register 按配置创建代理（接线与解析闭包），未配置时为 None 不启用。"""
|
||
http, config = env
|
||
on = replace(
|
||
config, file_proxy=True, public_url="http://proxy.test:16689",
|
||
proxy_port=16689,
|
||
)
|
||
register(EventEmitter(), on, http)
|
||
assert music._file_proxy is not None
|
||
assert music._file_proxy._base == "http://proxy.test:16689"
|
||
# 解析闭包已接上 _fetchable_play：能取到可抓取直链
|
||
found = await music._file_proxy._resolve(
|
||
music.Song("tx", "T1", "稻香", "歌手T"), "稻香"
|
||
)
|
||
assert found is not None and found[1].endswith("/media/tx/T1.mp3")
|
||
|
||
register(EventEmitter(), config, http) # 未配置 PUBLIC_URL → 不启用
|
||
assert music._file_proxy is None
|
||
|
||
|
||
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_fallback_sources_follow_server_platforms():
|
||
"""回退链由 /sources 动态解析：具备 search+url 能力的平台按优先序
|
||
排、其余追加在后，并排除原曲所在源与不可作候选的源（tx/qs）——
|
||
tx 的候选 = kw, wy, kg（qs 有 url 能力也被排除）。"""
|
||
s = FakeMusicServer()
|
||
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",
|
||
)
|
||
got = await music._fallback_sources(
|
||
http, config, music.Song("tx", "T1", "稻香", "歌手T")
|
||
)
|
||
assert got == ["kw", "wy", "kg"]
|
||
got = await music._fallback_sources(
|
||
http, config, music.Song("kg", "K1", "稻香", "歌手K")
|
||
)
|
||
assert got == ["kw", "wy"]
|
||
got = await music._fallback_sources(
|
||
http, config, music.Song("qs", "Q1", "稻香", "歌手Q")
|
||
)
|
||
assert got == ["kw", "wy", "kg"]
|
||
finally:
|
||
await s.stop()
|
||
|
||
|
||
async def test_fallback_sources_skip_platforms_without_url():
|
||
"""/sources 里 url 能力缺失的平台不进回退链。"""
|
||
s = FakeMusicServer(no_url_sources=frozenset({"kg"}))
|
||
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",
|
||
)
|
||
got = await music._fallback_sources(
|
||
http, config, music.Song("tx", "T1", "稻香", "歌手T")
|
||
)
|
||
assert got == ["kw", "wy"]
|
||
finally:
|
||
await s.stop()
|
||
|
||
|
||
async def test_fallback_sources_degrade_without_sources_api():
|
||
"""/sources 不可用时退回内置优先序（排除原曲源），回退不失效。"""
|
||
s = FakeMusicServer(serve_sources=False)
|
||
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",
|
||
)
|
||
got = await music._fallback_sources(
|
||
http, config, music.Song("tx", "T1", "稻香", "歌手T")
|
||
)
|
||
assert got == ["kw", "wy"]
|
||
finally:
|
||
await s.stop()
|
||
|
||
|
||
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[0]["method"], calls[0]["path"]) == ("PUT", "/interactions/I1")
|
||
|
||
api, calls = make_api()
|
||
assert not await handle_button(api, http, config, "G1", "I1", "别的插件|xx")
|
||
assert calls == [] # 非音乐按钮不处理也不回应
|
||
|
||
|
||
async def test_source_download_command(server, env):
|
||
"""「酷狗下载 <歌名>」：在指定平台搜歌出选歌菜单（与点歌同款交互），
|
||
标题带平台名、点编号后才播放/下载；全名「酷狗音乐下载」同样命中。"""
|
||
http, config = env
|
||
api, calls = make_api()
|
||
assert await handle_command(api, http, config, "G1", "m1", "酷狗下载 稻香")
|
||
assert server.queries[-1]["source"] == "kg" # 指定源搜索，不走聚合
|
||
body = posts(calls)[0]["json"]
|
||
assert body["msg_type"] == 2
|
||
assert "## 酷狗音乐：稻香" in body["markdown"]["content"]
|
||
assert "1. 稻香 - 歌手G" in body["markdown"]["content"]
|
||
assert labels(body) == [["1"]] # 一首候选：一个编号按钮、无翻页行
|
||
|
||
api, calls = make_api()
|
||
assert await handle_command(api, http, config, "G1", "m2", "/酷狗音乐下载 稻香")
|
||
assert "## 酷狗音乐：稻香" in posts(calls)[0]["json"]["markdown"]["content"]
|
||
|
||
|
||
async def test_source_menu_pages_within_source(server, env):
|
||
"""「<平台>下载」菜单的翻页按钮走 spage 动作、携带平台，翻页仍搜同一
|
||
平台且标题不变。"""
|
||
http, config = env
|
||
config = replace(config, menu_page_size=1) # 一首即满员，逼出下一页按钮
|
||
api, calls = make_api()
|
||
assert await handle_command(api, http, config, "G1", "m1", "酷狗下载 稻香")
|
||
body = posts(calls)[0]["json"]
|
||
nav = buttons(body)[-1]["buttons"]
|
||
assert [b["render_data"]["label"] for b in nav] == ["下一页"]
|
||
assert nav[0]["action"]["data"] == f"音乐|spage|kg|{quote('稻香', safe='')}|2"
|
||
|
||
api, calls = make_api()
|
||
assert await handle_button(
|
||
api, http, config, "G1", "I1", nav[0]["action"]["data"],
|
||
event_id="INTERACTION_CREATE:I1",
|
||
)
|
||
assert server.queries[-1] == {
|
||
"keyword": "稻香", "source": "kg", "page": "2", "limit": "1",
|
||
}
|
||
menu = posts(calls)[0]["json"]
|
||
assert "## 酷狗音乐：稻香" in menu["markdown"]["content"]
|
||
# 第 2 页有回翻；假服务器页页满员，「下一页」按本页满员启发式保留
|
||
assert labels(menu) == [["1"], ["上一页", "下一页"]]
|
||
|
||
|
||
async def test_pick_show_opens_track_menu(server, env):
|
||
"""节目级条目（me 的 d: 前缀）点编号先出选集菜单：标题带总集数、集条目
|
||
抹掉与节目名重复的歌手与 0 时长；选集再点编号弹的是该集卡片。"""
|
||
http, config = env
|
||
api, calls = make_api()
|
||
show = music.Song("me", "d:44753", "节目名", "", img=IMG_URL)
|
||
await handle_button(
|
||
api, http, config, "G1", "I1", music._song_data("pick", show),
|
||
event_id="E1",
|
||
)
|
||
body = posts(calls)[0]["json"]
|
||
assert "## 选集：节目名（共23集）" in body["markdown"]["content"]
|
||
assert "1. 第1期" in body["markdown"]["content"]
|
||
assert " - 节目名" not in body["markdown"]["content"] # 冗余歌手已抹
|
||
assert labels(body) == [
|
||
["1", "2", "3", "4", "5"], ["6", "7", "8", "9", "10"], ["下一页"],
|
||
]
|
||
nav = buttons(body)[-1]["buttons"][0]["action"]["data"]
|
||
assert nav == f"音乐|track|me|44753|{quote('节目名', safe='')}|2"
|
||
|
||
# 选集编号 → 该集卡片（普通单曲，照常播放/下载）
|
||
api, calls = make_api()
|
||
episode = music.Song("me", "44753-1", "第1期", "", img=IMG_URL)
|
||
await handle_button(
|
||
api, http, config, "G1", "I2", music._song_data("pick", episode),
|
||
event_id="E2",
|
||
)
|
||
assert "## 第1期" in posts(calls)[0]["json"]["markdown"]["content"]
|
||
|
||
|
||
async def test_pick_lr_book_opens_track_menu(server, env):
|
||
"""懒人/蜻蜓搜索返回的整条都是书/频道（直接取链 502）：点编号同样先出
|
||
选集菜单。"""
|
||
http, config = env
|
||
api, calls = make_api()
|
||
book = music.Song("lr", "50678018", "三体广播剧", "")
|
||
await handle_button(
|
||
api, http, config, "G1", "I1", music._song_data("pick", book),
|
||
event_id="E1",
|
||
)
|
||
assert "## 选集：三体广播剧（共23集）" in posts(calls)[0]["json"]["markdown"]["content"]
|
||
|
||
|
||
async def test_track_menu_local_paging_for_me(server, env):
|
||
"""me 的 /album 忽略 page 参数：一次拉大 limit 本地切片翻页；第 3 页
|
||
只剩 3 集、无下一页。"""
|
||
http, config = env
|
||
api, calls = make_api()
|
||
await music.send_track_menu(
|
||
api, http, config, "G1", "me", "44753", "节目名", 3, event_id="E1",
|
||
)
|
||
assert server.album_queries[-1] == {
|
||
"source": "me", "album_id": "44753", "page": 1, "limit": 50,
|
||
} # 不依赖 page 参数，单次拉 50
|
||
body = posts(calls)[0]["json"]
|
||
assert "1. 第21期" in body["markdown"]["content"]
|
||
assert "第20期" not in body["markdown"]["content"]
|
||
assert labels(body) == [["1", "2", "3"], ["上一页"]]
|
||
|
||
|
||
async def test_track_menu_server_paging_for_lr(server, env):
|
||
"""lr/qt 服务端翻页可用：按页直取当页。"""
|
||
http, config = env
|
||
api, calls = make_api()
|
||
await music.send_track_menu(
|
||
api, http, config, "G1", "lr", "50678018", "三体", 2, event_id="E1",
|
||
)
|
||
assert server.album_queries[-1] == {
|
||
"source": "lr", "album_id": "50678018", "page": 2, "limit": 10,
|
||
}
|
||
body = posts(calls)[0]["json"]
|
||
assert "1. 第11期" in body["markdown"]["content"] # 服务端切好的第 11-20 集
|
||
assert labels(body) == [
|
||
["1", "2", "3", "4", "5"], ["6", "7", "8", "9", "10"],
|
||
["上一页", "下一页"],
|
||
]
|
||
|
||
|
||
async def test_track_menu_empty_and_delisted(server, env):
|
||
"""空选集：第 1 页空（已下线/需登录）给明确提示，翻过头给「没有更多了」。"""
|
||
http, config = env
|
||
server.album_empty_ids = frozenset({"62806"})
|
||
api, calls = make_api()
|
||
await music.send_track_menu(
|
||
api, http, config, "G1", "me", "62806", "三体", 1, event_id="E1",
|
||
)
|
||
assert "拿不到剧集列表" in posts(calls)[0]["json"]["content"]
|
||
|
||
api, calls = make_api()
|
||
await music.send_track_menu(
|
||
api, http, config, "G1", "me", "62806", "三体", 2, event_id="E1",
|
||
)
|
||
assert posts(calls)[0]["json"]["content"] == "没有更多了"
|
||
|
||
|
||
async def test_source_download_usage_and_miss(server, env):
|
||
"""只敲「酷狗下载」给用法提示；指定源搜不到给源名提示。"""
|
||
http, config = env
|
||
api, calls = make_api()
|
||
assert await handle_command(api, http, config, "G1", "m1", "酷狗下载")
|
||
assert posts(calls)[0]["json"]["content"] == "用法：酷狗下载 <歌名>"
|
||
|
||
api, calls = make_api()
|
||
assert await handle_command(api, http, config, "G1", "m2", "酷我下载 空白歌")
|
||
assert posts(calls)[-1]["json"]["content"] == "在酷我音乐没找到「空白歌」"
|
||
|
||
|
||
async def test_hot_search_flow(server, 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
|
||
assert "## QQ音乐热搜" in body["markdown"]["content"] # 配置源 tx 支持
|
||
assert labels(body) == [["热词一", "热词二", "热词三"]]
|
||
hot_data = buttons(body)[0]["buttons"][0]["action"]["data"]
|
||
assert hot_data == f"音乐|hot|{quote('热词一', safe='')}"
|
||
|
||
api, calls = make_api()
|
||
assert await handle_button(
|
||
api, http, config, "G1", "I1", hot_data, event_id="INTERACTION_CREATE:I1"
|
||
)
|
||
menu = posts(calls)[0]["json"]
|
||
assert "点歌：热词一" in menu["markdown"]["content"]
|
||
assert (calls[0]["method"], calls[0]["path"]) == ("PUT", "/interactions/I1")
|
||
|
||
|
||
async def test_boards_flow(server, env):
|
||
"""榜单：点榜单名出榜单选歌菜单，翻页携带榜单上下文（总数决定下一页）。"""
|
||
http, config = env
|
||
api, calls = make_api()
|
||
assert await handle_command(api, http, config, "G1", "m1", "榜单")
|
||
body = posts(calls)[0]["json"]
|
||
assert "## QQ音乐榜单" in body["markdown"]["content"]
|
||
assert labels(body) == [["飙升榜", "热歌榜"]]
|
||
board_data = buttons(body)[0]["buttons"][1]["action"]["data"]
|
||
assert board_data == (
|
||
f"音乐|board|tx|tx__16|{quote('热歌榜', safe='')}|1"
|
||
)
|
||
|
||
api, calls = make_api()
|
||
assert await handle_button(
|
||
api, http, config, "G1", "I1", board_data, event_id="INTERACTION_CREATE:I1"
|
||
)
|
||
menu = posts(calls)[0]["json"]
|
||
md = menu["markdown"]["content"]
|
||
assert "榜单：热歌榜" in md
|
||
assert "1. 榜歌1 - 歌手1" in md
|
||
nav = buttons(menu)[-1]["buttons"]
|
||
assert [b["render_data"]["label"] for b in nav] == ["下一页"] # total 30 > 1×10
|
||
next_data = nav[0]["action"]["data"]
|
||
|
||
api, calls = make_api()
|
||
assert await handle_button(
|
||
api, http, config, "G1", "I2", next_data, event_id="INTERACTION_CREATE:I2"
|
||
)
|
||
assert server.board_queries[-1] == {"board": "tx__16", "page": "2", "limit": "10"}
|
||
assert "榜单：热歌榜" in posts(calls)[0]["json"]["markdown"]["content"]
|
||
|
||
|
||
async def test_panel_items_follow_server(server, env):
|
||
"""面板条目按 /sources 现算：基础指令 + 热搜/榜单 + 各平台「<平台>下载」
|
||
（指令名取惯用/短名：酷狗、QQ音乐、网易云），帮助垫底。"""
|
||
http, config = env
|
||
items = await music.panel_items(http, config)
|
||
assert [i["name"] for i in items] == [
|
||
"点歌", "歌词", "热搜", "榜单",
|
||
"酷我下载", "酷狗下载", "QQ音乐下载", "网易云下载", "汽水下载",
|
||
"帮助",
|
||
]
|
||
assert all(i["type"] == "command" for i in items)
|
||
kg = next(i for i in items if i["name"] == "酷狗下载")
|
||
assert kg["desc"] == "在酷狗音乐搜歌出选歌菜单"
|
||
|
||
|
||
async def test_panel_items_degrade_without_sources():
|
||
"""/sources 不可用时退回静态表：xm 等新平台也有下载指令，热搜/榜单在。"""
|
||
s = FakeMusicServer(serve_sources=False)
|
||
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",
|
||
)
|
||
names = [i["name"] for i in await music.panel_items(http, config)]
|
||
finally:
|
||
await s.stop()
|
||
assert "酷狗下载" in names and "喜马拉雅下载" in names
|
||
assert "热搜" in names and "榜单" in names
|
||
|
||
|
||
async def test_panel_items_fit_qq_width_limits():
|
||
"""/v2/panels 长度按宽度算（ASCII 1、汉字 2）而非字符数：「Apple Music
|
||
下载」全名超宽（15>14）借短名「Apple下载」入面板；换无可换的平台只能
|
||
放弃；指令满 20 条时「帮助」仍垫底不被挤掉。宽度超限曾致 PUT 30013。"""
|
||
caps = {"search": True, "url": True}
|
||
extras = (
|
||
{"id": "apple", "name": "Apple Music", "capabilities": caps},
|
||
{"id": "toolong", "name": "名字特别长的平台呀", "capabilities": caps},
|
||
*[{"id": f"p{i}", "name": f"平台{i}号", "capabilities": caps} for i in range(14)],
|
||
)
|
||
s = FakeMusicServer(extra_platforms=extras)
|
||
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",
|
||
)
|
||
items = await music.panel_items(http, config)
|
||
finally:
|
||
await s.stop()
|
||
width = lambda t: sum(1 if ord(c) < 128 else 2 for c in t) # noqa: E731
|
||
assert all(width(i["name"]) <= 14 and width(i["desc"]) <= 30 for i in items)
|
||
names = [i["name"] for i in items]
|
||
assert "Apple下载" in names
|
||
assert not any("名字特别长" in i["desc"] for i in items)
|
||
# 基础 4 + 下载 20 截到 19，再垫「帮助」：总数 20、帮助在最后
|
||
assert len(items) == 20 and items[-1]["name"] == "帮助"
|
||
|
||
|
||
def test_songs_of_injects_source_and_drops_unknown():
|
||
"""无 source 字段的条目（xm/me/qt/lr 真机如此）按请求源补上；
|
||
聚合请求（all）无从判断来源，这类条目只能丢弃。"""
|
||
items = [{"songmid": "X1", "name": "歌", "singer": "手"}]
|
||
assert [s.source for s in music._songs_of(items, "xm")] == ["xm"]
|
||
assert music._songs_of(items, "all") == []
|
||
assert [s.source for s in music._songs_of(
|
||
[{"source": "kw", "songmid": "S1", "name": "歌", "singer": "手"}], "xm"
|
||
)] == ["kw"] # 自带 source 的不覆盖
|
||
|
||
|
||
def test_alias_rules():
|
||
"""别名规则：剥「音乐」短名、纯拉丁词干不剥、wy 惯用名「网易云」；
|
||
/sources 增删平台后正则即时重建。"""
|
||
music._apply_aliases({
|
||
"kg": {"name": "酷狗音乐"},
|
||
"tx": {"name": "QQ音乐"},
|
||
"wy": {"name": "网易音乐"},
|
||
})
|
||
try:
|
||
assert music._alias_to_source["酷狗"] == "kg"
|
||
assert music._alias_to_source["酷狗音乐"] == "kg"
|
||
assert music._alias_to_source["网易云"] == "wy"
|
||
assert "QQ" not in music._alias_to_source # 纯拉丁词干不剥
|
||
m = music._dl_re.search("QQ音乐下载 稻香")
|
||
assert m is not None and m.group(1) == "QQ音乐"
|
||
m = music._dl_re.search("网易云下载 稻香")
|
||
assert m is not None and m.group(1) == "网易云"
|
||
assert music._dl_re.search("喜马拉雅下载 三体") is None # 表里没有
|
||
finally:
|
||
music._apply_aliases(music._STATIC_PLATFORMS)
|
||
|
||
|
||
async def test_login_status_smoke(server, env):
|
||
"""登录态日志不抛错（老版服务器无此接口时静默跳过）。"""
|
||
http, config = env
|
||
await music._log_login_status(http, config)
|
||
broken = replace(config, music_server_url="http://127.0.0.1:1")
|
||
await music._log_login_status(http, broken)
|
||
|
||
|
||
async def test_button_debounce_skips_repeats(env):
|
||
"""防抖：同一互动 uuid 的重复推送、同群同按钮的连点（新 uuid），10 秒
|
||
窗口内只处理第一次；连点仍要应答互动，把客户端的 loading 关掉。"""
|
||
http, config = env
|
||
api, calls = make_api()
|
||
assert await handle_button(api, http, config, "G1", "I1", PICK_1)
|
||
assert await handle_button(api, http, config, "G1", "I1", PICK_1) # 重复推送
|
||
assert len(posts(calls)) == 1 # 卡片只发一次
|
||
assert len([c for c in calls if c["method"] == "PUT"]) == 1 # 不重复应答
|
||
|
||
music._recent_clicks.clear() # 只测连点，不受上一段记下的钥匙影响
|
||
api, calls = make_api()
|
||
assert await handle_button(api, http, config, "G1", "I3", PICK_1)
|
||
assert await handle_button(api, http, config, "G1", "I4", PICK_1) # 连点
|
||
assert len(posts(calls)) == 1
|
||
assert (calls[-1]["method"], calls[-1]["path"]) == ("PUT", "/interactions/I4")
|
||
|
||
# 不同群的同一按钮互不影响
|
||
music._recent_clicks.clear()
|
||
api, calls = make_api()
|
||
assert await handle_button(api, http, config, "G2", "I5", PICK_1)
|
||
assert len(posts(calls)) == 1
|
||
|
||
|
||
async def test_button_debounce_expires(env, monkeypatch):
|
||
"""防抖窗口（TTL）过后，同一按钮可再次处理。"""
|
||
http, config = env
|
||
api, calls = make_api()
|
||
assert await handle_button(api, http, config, "G1", "I1", PICK_1)
|
||
monkeypatch.setattr(music, "_CLICK_TTL", 0.0)
|
||
assert await handle_button(api, http, config, "G1", "I1", PICK_1)
|
||
assert len(posts(calls)) == 2
|
||
|
||
|
||
async def test_play_replies_placeholder_before_fetching(env):
|
||
"""播放进门先应答互动、回「正在获取」占位（msg_seq=1），结果以 msg_seq=2
|
||
补发——同一事件相同 msg_seq 的第二次回复会被 QQ 拒绝。"""
|
||
http, config = env
|
||
api, calls = make_api()
|
||
await handle_button(
|
||
api, http, config, "G1", "I2", PLAY_TX, event_id="INTERACTION_CREATE:I2"
|
||
)
|
||
assert (calls[0]["method"], calls[0]["path"]) == ("PUT", "/interactions/I2")
|
||
ack, result = posts(calls)[0], posts(calls)[-1]
|
||
assert "正在获取" in ack["json"]["content"]
|
||
assert ack["json"]["msg_seq"] == 1
|
||
assert result["json"]["msg_type"] == 7 # 语音照常
|
||
assert result["json"]["msg_seq"] == 2
|
||
assert result["json"]["event_id"] == "INTERACTION_CREATE:I2"
|
||
|
||
|
||
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
|
||
datas = [b["action"]["data"] for b in rows[0]["buttons"]]
|
||
assert [b["action"]["type"] for b in rows[0]["buttons"]] == [2, 2, 2, 2]
|
||
assert datas == ["/点歌 ", "/歌词 ", "/热搜", "/榜单"]
|
||
|
||
|
||
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}
|
||
|
||
|
||
_PANEL_ITEMS = [
|
||
{"name": "点歌", "type": "command", "desc": "出选歌菜单，点编号选歌"},
|
||
{"name": "歌词", "type": "command", "desc": "查看歌词"},
|
||
{"name": "酷狗下载", "type": "command", "desc": "在酷狗音乐搜歌出选歌菜单"},
|
||
{"name": "帮助", "type": "command", "desc": "查看帮助"},
|
||
]
|
||
|
||
|
||
async def test_panel_created_when_missing():
|
||
api = PanelRecording()
|
||
await sync_command_panel(api, _PANEL_ITEMS)
|
||
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():
|
||
record = {"panel_id": "P9", "panel": {"items": _PANEL_ITEMS, "remark": "音乐bot"}}
|
||
api = PanelRecording(records=[record])
|
||
await sync_command_panel(api, _PANEL_ITEMS)
|
||
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, _PANEL_ITEMS)
|
||
assert (api.calls[1]["method"], api.calls[1]["path"]) == ("PUT", "/v2/panels/P9")
|
||
body = api.calls[1]["json"]
|
||
assert [i["name"] for i in body["panel"]["items"]] == [
|
||
"点歌", "歌词", "酷狗下载", "帮助",
|
||
]
|
||
|
||
|
||
async def test_panel_failure_swallowed():
|
||
api = PanelRecording(fail_get=True)
|
||
await sync_command_panel(api, _PANEL_ITEMS) # 失败只记日志，绝不抛出
|
||
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_file_proxy(monkeypatch):
|
||
"""文件代理开关：默认关；配了 PUBLIC_URL 即开（FILE_PROXY=off 可关）；
|
||
监听端口缺省跟 PUBLIC_URL 里的端口，无端口用默认。"""
|
||
from musicbot.config import BotConfig
|
||
|
||
for var in ("PUBLIC_URL", "FILE_PROXY", "PROXY_PORT"):
|
||
monkeypatch.delenv(var, raising=False)
|
||
cfg = BotConfig.load()
|
||
assert cfg.file_proxy is False and cfg.public_url == ""
|
||
|
||
monkeypatch.setenv("PUBLIC_URL", "http://1.2.3.4:16690/")
|
||
cfg = BotConfig.load()
|
||
assert cfg.file_proxy is True
|
||
assert cfg.public_url == "http://1.2.3.4:16690" # 去掉尾斜杠
|
||
assert cfg.proxy_port == 16690 # 缺省跟 URL 里的端口
|
||
|
||
monkeypatch.setenv("PUBLIC_URL", "http://music.example.com")
|
||
assert BotConfig.load().proxy_port == 16689 # 无端口用默认
|
||
|
||
monkeypatch.setenv("PUBLIC_URL", "http://1.2.3.4:16690")
|
||
monkeypatch.setenv("PROXY_PORT", "17777") # 监听端口可独立指定
|
||
assert BotConfig.load().proxy_port == 17777
|
||
|
||
monkeypatch.setenv("FILE_PROXY", "off") # 显式关闭
|
||
assert BotConfig.load().file_proxy is False
|
||
|
||
|
||
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 会在 bot.start()
|
||
装配时登记的 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, user: str = "U1"
|
||
) -> dict:
|
||
return {
|
||
"id": msg_id,
|
||
"content": content,
|
||
"group_openid": "G1",
|
||
"author": {"member_openid": user, "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 重复推送、或 @消息同时命中两个事件（两事件 msg_id 不同，
|
||
2026-10-03 线上因此双占位双直链）时只处理一次。"""
|
||
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)
|
||
# 双事件的另一帧：msg_id 不同、文本归一后一致（@ 前缀 + 斜杠都有）
|
||
await push(
|
||
ee, "GROUP_AT_MESSAGE_CREATE", group_payload("m2", "<@!BOT9> /点歌 稻香")
|
||
)
|
||
assert len(posts(calls)) == 1
|
||
|
||
|
||
async def test_cross_event_dedup_is_scoped(env, monkeypatch):
|
||
"""内容级去重只吞同群同人同文本的短窗重复：换人、换文本、窗口过期
|
||
都照常处理。"""
|
||
http, config = env
|
||
api, calls = make_api()
|
||
ee = make_emitter(api, http, config)
|
||
await push(ee, "GROUP_MESSAGE_CREATE", group_payload("m1", "点歌 稻香"))
|
||
# 同群同人同文本、不同 msg_id（双事件的另一帧）→ 只处理第一次
|
||
await push(ee, "GROUP_AT_MESSAGE_CREATE", group_payload("m2", "点歌 稻香"))
|
||
assert len(posts(calls)) == 1
|
||
# 换用户同文本 → 照常
|
||
await push(ee, "GROUP_MESSAGE_CREATE", group_payload("m3", "点歌 稻香", user="U2"))
|
||
# 同用户换文本 → 照常
|
||
await push(ee, "GROUP_MESSAGE_CREATE", group_payload("m4", "点歌 晴天"))
|
||
assert len(posts(calls)) == 3
|
||
# 窗口过期后同人同文本再来 → 照常
|
||
monkeypatch.setattr(music, "_MSG_DUP_TTL", 0.0)
|
||
await push(ee, "GROUP_MESSAGE_CREATE", group_payload("m5", "点歌 稻香"))
|
||
assert len(posts(calls)) == 4
|
||
|
||
|
||
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
|