diff --git a/music_api/api.py b/music_api/api.py index 873e003..2c3944d 100644 --- a/music_api/api.py +++ b/music_api/api.py @@ -6,6 +6,7 @@ import json import time +from collections.abc import Callable from random import randrange from typing import Any @@ -28,6 +29,81 @@ from .errors import ( ) +def _map_song_item(item: dict[str, Any]) -> dict[str, Any]: + """歌曲条目映射。""" + return { + 'type': 'song', + 'id': item['id'], + 'name': item.get('name') or '未知歌曲', + 'artists': '/'.join(filter(None, (a.get('name') for a in item.get('ar') or []))) or '未知歌手', + 'album': (item.get('al') or {}).get('name') or '未知专辑', + 'picUrl': (item.get('al') or {}).get('picUrl') or '' + } + + +def _map_album_item(item: dict[str, Any]) -> dict[str, Any]: + """专辑条目映射(artists 为空**不**兜底成"未知歌手"——与歌曲路不同口径)。""" + return { + 'type': 'album', + 'id': item['id'], + 'name': item.get('name') or '未知名称', + 'artists': '/'.join(filter(None, (a.get('name') for a in item.get('artists') or []))), + 'alias': '', + 'album': '', + 'picUrl': item.get('picUrl') + } + + +def _map_artist_item(item: dict[str, Any]) -> dict[str, Any]: + """歌手条目映射(alias 缺失/为空 → '')。""" + return { + 'type': 'artist', + 'id': item['id'], + 'name': item.get('name') or '未知名称', + 'artists': '', + 'alias': '/'.join(item.get('alias', [])) if item.get('alias') else '', + 'album': '', + 'picUrl': item.get('picUrl') + } + + +def _map_playlist_item(item: dict[str, Any]) -> dict[str, Any]: + """歌单条目映射。""" + return { + 'type': 'playlist', + 'id': item['id'], + 'name': item.get('name') or '未知名称', + 'artists': '', + 'creator': (item.get('creator') or {}).get('nickname', ''), + 'trackCount': item.get('trackCount', 0), + 'album': '', + 'picUrl': item.get('coverImgUrl') + } + + +# search_type → (上游结果键, 条目映射);未知类型不在表内 → 空结果 +_SEARCH_RESULT_KEYS: dict[int, str] = { + 1: 'songs', 10: 'albums', 100: 'artists', 1000: 'playlists', +} +_SEARCH_ITEM_MAPPERS: dict[int, Callable[[dict[str, Any]], dict[str, Any]]] = { + 1: _map_song_item, 10: _map_album_item, 100: _map_artist_item, 1000: _map_playlist_item, +} + + +def _collect_search_items(result_data: dict[str, Any], search_type: int) -> list[dict[str, Any]]: + """按类型取上游结果键并逐条映射;残缺条目(无 id,下架/屏蔽)跳过,不拖死整页。""" + key = _SEARCH_RESULT_KEYS.get(search_type) + mapper = _SEARCH_ITEM_MAPPERS.get(search_type) + if key is None or mapper is None: + return [] + items: list[dict[str, Any]] = [] + for item in result_data.get(key, []): + if not item.get('id'): + continue + items.append(mapper(item)) + return items + + class NeteaseAPI: """网易云音乐API主类""" @@ -196,66 +272,7 @@ class NeteaseAPI: raise APIException(f"搜索失败: {result.get('message', '未知错误')}") result_data = result.get('result') or {} - items: list[dict[str, Any]] = [] - - if search_type == 1: - # 歌曲(id 改 .get——上游单条残缺(下架/屏蔽)不再拖死整页搜索) - for item in result_data.get('songs', []): - if not item.get('id'): - continue - items.append({ - 'type': 'song', - 'id': item['id'], - 'name': item.get('name') or '未知歌曲', - 'artists': '/'.join(filter(None, (a.get('name') for a in item.get('ar') or []))) or '未知歌手', - 'album': (item.get('al') or {}).get('name') or '未知专辑', - 'picUrl': (item.get('al') or {}).get('picUrl') or '' - }) - elif search_type == 10: - # 专辑(同上) - for item in result_data.get('albums', []): - if not item.get('id'): - continue - items.append({ - 'type': 'album', - 'id': item['id'], - 'name': item.get('name') or '未知名称', - 'artists': '/'.join(filter(None, (a.get('name') for a in item.get('artists') or []))), - 'alias': '', - 'album': '', - 'picUrl': item.get('picUrl') - }) - elif search_type == 100: - # 歌手(同上) - for item in result_data.get('artists', []): - if not item.get('id'): - continue - items.append({ - 'type': 'artist', - 'id': item['id'], - 'name': item.get('name') or '未知名称', - 'artists': '', - 'alias': '/'.join(item.get('alias', [])) if item.get('alias') else '', - 'album': '', - 'picUrl': item.get('picUrl') - }) - elif search_type == 1000: - # 歌单(同上) - for item in result_data.get('playlists', []): - if not item.get('id'): - continue - items.append({ - 'type': 'playlist', - 'id': item['id'], - 'name': item.get('name') or '未知名称', - 'artists': '', - 'creator': (item.get('creator') or {}).get('nickname', ''), - 'trackCount': item.get('trackCount', 0), - 'album': '', - 'picUrl': item.get('coverImgUrl') - }) - - return items + return _collect_search_items(result_data, search_type) except requests.RequestException as e: raise APIException(f"搜索请求失败: {e}") from e except (json.JSONDecodeError, KeyError) as e: diff --git a/tests/test_search_music.py b/tests/test_search_music.py new file mode 100644 index 0000000..d7267a8 --- /dev/null +++ b/tests/test_search_music.py @@ -0,0 +1,212 @@ +# -*- 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', '', 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