Archived
snowy-yang
e00ac14460
为什么：站内搜索的唯一上游通道，101 行里 if/elif 四路条目映射占了 2/3，错误映射与字段容错口径（歌曲 artists 空缺兜底未知歌手、专辑 不兜底、alias 缺省 '' 等）全埋在分支里，改一路要通读全函数。 拆出（模块级，与既有 _ensure_dict 同风格）： - _map_song_item / _map_album_item / _map_artist_item / _map_playlist_item ——四路口径原样，docstring 标注各路容错差异 - _SEARCH_RESULT_KEYS + _SEARCH_ITEM_MAPPERS 派发表（未知类型不在表内 → 空结果，原 if/elif 无 else 行为一致） - _collect_search_items 统一「取键 → 无 id 跳过 → 映射」 先写测试后动代码：新增 tests/test_search_music.py 12 例，在未改动 实现上跑绿（修正 1 例替身形状：网络异常必须抛垫片自己的 RequestException，aiohttp.ClientError 由 aiohttp_compat 内部翻译， 直接抛底层是不存在的形状）。契约测试不为违反 offset: int 签名的 None 入参背书（防御性 int(offset or 0) 保留，不测）。 门禁：ruff 全过 · pyright 0/0 · pytest 330 passed（319→330）。
213 lines
8.8 KiB
Python
213 lines
8.8 KiB
Python
# -*- coding: utf-8 -*-
|
||
"""search_music 的行为契约测试（沙箱，不触真实上游）。
|
||
|
||
为什么：``music_api/api.py::search_music`` 是站内搜索的唯一上游通道
|
||
（101 行 / 四路条目映射 / 嵌套 7），此前**零测试覆盖**。它的错误映射
|
||
（RequestException/JSONDecodeError→APIException）与四路字段的容错口径
|
||
（残缺单条跳过、空 artists 不兜底、alias 缺省 '' 等）都是对外可观察行为，
|
||
拆解时改坏任意一路都不会有测试报警。
|
||
|
||
按惯例：先在未改动的原始实现上跑绿，重构时这批测试即等价性验证器。
|
||
替身边界：monkeypatch ``music_api.api.requests.post``（aiohttp_compat 垫片
|
||
的同形入口），响应替身只实现 ``raise_for_status`` / ``json`` 两个同步方法——
|
||
``search_music`` 只用到这两个。
|
||
"""
|
||
import json
|
||
from typing import Any
|
||
|
||
import pytest
|
||
|
||
from music_api import APIException
|
||
from music_api import api as api_mod
|
||
from music_api.api import NeteaseAPI
|
||
|
||
# 与生产一致的请求入口（搜索走 SEARCH_API）
|
||
_API = NeteaseAPI()
|
||
|
||
|
||
class _FakeResp:
|
||
"""``requests.post`` 响应替身：json 载荷或注入异常。"""
|
||
|
||
def __init__(self, payload: Any = None, *,
|
||
status_error: Exception | None = None,
|
||
json_error: Exception | None = None):
|
||
self.payload = payload
|
||
self.status_error = status_error
|
||
self.json_error = json_error
|
||
|
||
def raise_for_status(self) -> None:
|
||
if self.status_error is not None:
|
||
raise self.status_error
|
||
|
||
def json(self) -> Any:
|
||
if self.json_error is not None:
|
||
raise self.json_error
|
||
return self.payload
|
||
|
||
|
||
@pytest.fixture()
|
||
def post_capture(monkeypatch):
|
||
"""替换 requests.post，捕获调用参数并按用例注入响应。"""
|
||
holder: dict[str, Any] = {'resp': _FakeResp({'code': 200, 'result': {}})}
|
||
|
||
async def _fake_post(url: str, **kw: Any):
|
||
holder['url'] = url
|
||
holder['data'] = kw.get('data')
|
||
return holder['resp']
|
||
|
||
monkeypatch.setattr(api_mod.requests, 'post', _fake_post)
|
||
return holder
|
||
|
||
|
||
def _result(*items: dict[str, Any], key: str = 'songs') -> dict[str, Any]:
|
||
return {'code': 200, 'result': {key: list(items)}}
|
||
|
||
|
||
# ============ 四路映射 ============
|
||
|
||
async def test_search_songs_maps_fields(post_capture):
|
||
"""type=1：歌曲字段映射（ar 拼接、al 取名/封面、type 标记）。"""
|
||
post_capture['resp'] = _FakeResp(_result(
|
||
{'id': 1, 'name': '歌A', 'ar': [{'name': '歌手X'}, {'name': '歌手Y'}],
|
||
'al': {'name': '专辑Z', 'picUrl': 'https://pic/1.jpg'}},
|
||
{'id': 2, 'name': '歌B', 'ar': [], 'al': {}},
|
||
))
|
||
out = await _API.search_music('测试', {}, limit=30, search_type=1)
|
||
|
||
assert out == [
|
||
{'type': 'song', 'id': 1, 'name': '歌A', 'artists': '歌手X/歌手Y',
|
||
'album': '专辑Z', 'picUrl': 'https://pic/1.jpg'},
|
||
{'type': 'song', 'id': 2, 'name': '歌B', 'artists': '未知歌手',
|
||
'album': '未知专辑', 'picUrl': ''},
|
||
], out
|
||
# 请求形状：表单字段 + 分页透传
|
||
assert post_capture['data'] == {'s': '测试', 'type': 1, 'limit': 30, 'offset': 0}
|
||
|
||
|
||
async def test_search_songs_skips_items_without_id(post_capture):
|
||
"""上游单条残缺（无 id，下架/屏蔽）：跳过该条，不拖死整页搜索。"""
|
||
post_capture['resp'] = _FakeResp(_result(
|
||
{'name': '残缺条目'}, # 无 id
|
||
{'id': 9, 'name': '正常'},
|
||
))
|
||
out = await _API.search_music('测试', {}, search_type=1)
|
||
assert [i['id'] for i in out] == [9]
|
||
|
||
|
||
async def test_search_albums_keeps_empty_artists(post_capture):
|
||
"""type=10：专辑 artists 为空时**不**兜底成"未知歌手"（与歌曲路不同口径）。"""
|
||
post_capture['resp'] = _FakeResp(_result(
|
||
{'id': 5, 'name': '专辑A', 'artists': [{'name': '歌手X'}],
|
||
'picUrl': 'https://pic/a.jpg'},
|
||
{'id': 6, 'name': '专辑B', 'artists': [], 'picUrl': None},
|
||
key='albums'))
|
||
out = await _API.search_music('测试', {}, search_type=10)
|
||
|
||
assert out == [
|
||
{'type': 'album', 'id': 5, 'name': '专辑A', 'artists': '歌手X',
|
||
'alias': '', 'album': '', 'picUrl': 'https://pic/a.jpg'},
|
||
{'type': 'album', 'id': 6, 'name': '专辑B', 'artists': '',
|
||
'alias': '', 'album': '', 'picUrl': None},
|
||
], out
|
||
|
||
|
||
async def test_search_artists_alias_join_and_default(post_capture):
|
||
"""type=100：alias 列表拼接；alias 缺失/为空 → ''（不是"未知"兜底）。"""
|
||
post_capture['resp'] = _FakeResp(_result(
|
||
{'id': 7, 'name': '歌手A', 'alias': ['A1', 'A2'], 'picUrl': 'https://pic/p.png'},
|
||
{'id': 8, 'name': '歌手B'},
|
||
key='artists'))
|
||
out = await _API.search_music('测试', {}, search_type=100)
|
||
|
||
assert out == [
|
||
{'type': 'artist', 'id': 7, 'name': '歌手A', 'artists': '',
|
||
'alias': 'A1/A2', 'album': '', 'picUrl': 'https://pic/p.png'},
|
||
{'type': 'artist', 'id': 8, 'name': '歌手B', 'artists': '',
|
||
'alias': '', 'album': '', 'picUrl': None},
|
||
], out
|
||
|
||
|
||
async def test_search_playlists_maps_creator_and_cover(post_capture):
|
||
"""type=1000：歌单取 creator.nickname / trackCount / coverImgUrl。"""
|
||
post_capture['resp'] = _FakeResp(_result(
|
||
{'id': 11, 'name': '歌单X', 'creator': {'nickname': '阿村'},
|
||
'trackCount': 42, 'coverImgUrl': 'https://cover/1.jpg'},
|
||
key='playlists'))
|
||
out = await _API.search_music('测试', {}, search_type=1000)
|
||
|
||
assert out == [{'type': 'playlist', 'id': 11, 'name': '歌单X', 'artists': '',
|
||
'creator': '阿村', 'trackCount': 42, 'album': '',
|
||
'picUrl': 'https://cover/1.jpg'}], out
|
||
|
||
|
||
async def test_unknown_search_type_returns_empty(post_capture):
|
||
"""未知 search_type：返回空列表，不抛（历史行为：前端枚举外类型静默）。"""
|
||
out = await _API.search_music('测试', {}, search_type=999)
|
||
assert out == []
|
||
|
||
|
||
async def test_missing_result_key_returns_empty(post_capture):
|
||
"""响应缺对应结果键：空列表，不抛 KeyError。"""
|
||
post_capture['resp'] = _FakeResp({'code': 200, 'result': {}})
|
||
out = await _API.search_music('测试', {}, search_type=1)
|
||
assert out == []
|
||
|
||
|
||
# ============ 错误映射 ============
|
||
|
||
async def test_non_200_code_raises_with_message(post_capture):
|
||
"""code!=200：APIException 透出上游 message；缺 message 用"未知错误"。"""
|
||
post_capture['resp'] = _FakeResp({'code': 400, 'message': '关键词太长'})
|
||
with pytest.raises(APIException, match=r'搜索失败: 关键词太长'):
|
||
await _API.search_music('测试', {}, search_type=1)
|
||
|
||
post_capture['resp'] = _FakeResp({'code': 400})
|
||
with pytest.raises(APIException, match=r'搜索失败: 未知错误'):
|
||
await _API.search_music('测试', {}, search_type=1)
|
||
|
||
|
||
async def test_non_dict_response_raises_format_error(post_capture):
|
||
"""响应是 null/数组（合法 JSON 非对象）：明确的"响应格式异常"。"""
|
||
post_capture['resp'] = _FakeResp(None)
|
||
with pytest.raises(APIException, match='响应格式异常'):
|
||
await _API.search_music('测试', {}, search_type=1)
|
||
|
||
post_capture['resp'] = _FakeResp([{'code': 200}])
|
||
with pytest.raises(APIException, match='响应格式异常'):
|
||
await _API.search_music('测试', {}, search_type=1)
|
||
|
||
|
||
async def test_request_error_maps_to_api_exception(post_capture, monkeypatch):
|
||
"""网络层异常（垫片 RequestException 族）：转"搜索请求失败"。
|
||
|
||
注意替身抛的必须是垫片自己的 ``RequestException``——真实链路里
|
||
aiohttp.ClientError 由 aiohttp_compat 内部翻译后才冒出来，直接抛底层
|
||
异常是不存在的形状（test_request_* 相关层次见 core.http_errors）。
|
||
"""
|
||
async def _boom(url: str, **kw: Any):
|
||
raise api_mod.requests.RequestException('connection reset')
|
||
|
||
monkeypatch.setattr(api_mod.requests, 'post', _boom)
|
||
with pytest.raises(APIException, match='搜索请求失败'):
|
||
await _API.search_music('测试', {}, search_type=1)
|
||
|
||
|
||
async def test_json_decode_error_maps_to_api_exception(post_capture):
|
||
"""JSON 解析失败：转"解析搜索响应失败"。"""
|
||
bad = _FakeResp(None, json_error=json.JSONDecodeError('Expecting value', '<html>', 0))
|
||
post_capture['resp'] = bad
|
||
with pytest.raises(APIException, match='解析搜索响应失败'):
|
||
await _API.search_music('测试', {}, search_type=1)
|
||
|
||
|
||
# ============ 入参整形 ============
|
||
|
||
async def test_negative_offset_clamped_to_zero(post_capture):
|
||
"""offset 为负：夹到 0（max(0, int(offset or 0))），不透传负数。
|
||
|
||
（实现里还有 int(offset or 0) 对 None/字符串的防御，但签名声明 offset: int，
|
||
契约测试不为违反签名的形状背书。）
|
||
"""
|
||
await _API.search_music('测试', {}, search_type=1, offset=-5)
|
||
assert post_capture['data']['offset'] == 0
|