# -*- coding: utf-8 -*- """get_playlist_detail 的行为契约测试（沙箱，不触真实上游）。 为什么：歌单详情是「歌单解析」功能的唯一上游通道（94 行 / 批次拉取 / 嵌套 3），此前零测试覆盖。它的容错口径都是对外可观察行为： - 上游 code:200 + playlist:null →「歌单不存在或无权访问」（路由统一转 404） - 明细批次业务失败计 failed_batches，不静默缺歌 - trackIds 混入脏 ID（如带圈数字——isdigit() 为真但 int() 抛错）跳过整批 - trackIds 数与实得数不一致时显式标 incomplete/missing 按惯例：先在未改动的原始实现上跑绿，重构时这批测试即等价性验证器。 替身边界：monkeypatch ``music_api.api.requests.post``，按 URL 分流—— PLAYLIST_DETAIL_API 与 SONG_DETAIL_V3 分别给响应替身。 """ 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 from music_api.constants import APIConstants _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 api_capture(monkeypatch): """按 URL 分流的 requests.post 替身：歌单详情响应固定，批次响应按序消费。""" calls: list[dict[str, Any]] = [] state: dict[str, Any] = { 'playlist_resp': _FakeResp({'code': 200, 'playlist': {}}), 'batch_resps': [_FakeResp({'code': 200, 'songs': []})], } async def _fake_post(url: str, **kw: Any): calls.append({'url': url, 'data': kw.get('data')}) if url == APIConstants.PLAYLIST_DETAIL_API: return state['playlist_resp'] assert url == APIConstants.SONG_DETAIL_V3, f'unexpected url: {url}' if not state['batch_resps']: raise AssertionError('批次响应耗尽') return state['batch_resps'].pop(0) monkeypatch.setattr(api_mod.requests, 'post', _fake_post) state['calls'] = calls return state def _playlist_payload(playlist: dict[str, Any]) -> _FakeResp: return _FakeResp({'code': 200, 'playlist': playlist}) def _batch_c_payloads(calls: list[dict[str, Any]]) -> list[list[int]]: """从捕获的调用里提取每批的 c 载荷（id 列表）。""" out = [] for c in calls: if c['url'] == APIConstants.SONG_DETAIL_V3: ids = [e['id'] for e in json.loads(c['data']['c'])] out.append(ids) return out async def _detail(api_capture, playlist: dict[str, Any]) -> dict[str, Any]: api_capture['playlist_resp'] = _playlist_payload(playlist) return await _API.get_playlist_detail(123, {}) # ============ 歌单头 ============ async def test_playlist_null_is_apiexception(api_capture): """上游对不存在/私密歌单返回 code:200 + playlist:null → 显式 APIException。""" api_capture['playlist_resp'] = _playlist_payload({}) with pytest.raises(APIException, match='歌单不存在或无权访问'): await _API.get_playlist_detail(123, {}) async def test_business_code_error(api_capture): api_capture['playlist_resp'] = _FakeResp({'code': 512, 'message': '需要登录'}) with pytest.raises(APIException, match='获取歌单详情失败: 需要登录'): await _API.get_playlist_detail(123, {}) async def test_non_dict_response(api_capture): api_capture['playlist_resp'] = _FakeResp(None) with pytest.raises(APIException, match='响应格式异常'): await _API.get_playlist_detail(123, {}) # ============ 成功主路径 ============ async def test_success_maps_tracks(api_capture): api_capture['playlist_resp'] = _playlist_payload({ 'id': 123, 'name': '歌单A', 'coverImgUrl': 'https://cover/1', 'creator': {'nickname': '作者'}, 'trackCount': 2, 'trackIds': [{'id': 1}, {'id': 2}, {'id': None}], }) api_capture['batch_resps'] = [_FakeResp({'code': 200, 'songs': [ {'id': 1, 'name': '歌A', 'ar': [{'name': '歌手X'}], 'al': {'name': '专辑Z', 'picUrl': 'https://pic/1.jpg'}}, {'id': None, 'name': '残缺'}, {'id': 2, 'name': '歌B', 'ar': [], 'al': {}}, ]})] out = await _API.get_playlist_detail(123, {}) assert out['name'] == '歌单A' assert out['creator'] == '作者' assert out['tracks'] == [ {'id': 1, 'name': '歌A', 'artists': '歌手X', 'album': '专辑Z', 'picUrl': 'https://pic/1.jpg'}, {'id': 2, 'name': '歌B', 'artists': '未知歌手', 'album': '未知专辑', 'picUrl': ''}, ] assert 'incomplete' not in out assert _batch_c_payloads(api_capture['calls']) == [[1, 2]] # 缺 id 的 trackId 不进批次 # ============ 批次容错 ============ async def test_batch_business_failure_marked_incomplete(api_capture): """明细批次业务失败：计 failed_batches，显式标 incomplete，不冒充完整。""" api_capture['batch_resps'] = [_FakeResp({'code': 512, 'message': '爆了'})] out = await _detail(api_capture, {'id': 1, 'trackIds': [{'id': 1}, {'id': 2}]}) assert out['tracks'] == [] assert out['incomplete'] is True assert out['missing'] == 2 assert out['failed_batches'] == 1 async def test_batch_non_dict_counts_as_failed(api_capture): """批次响应体为 null（非 dict）：按失败批次计，不让 AttributeError 失真。""" api_capture['batch_resps'] = [_FakeResp(None)] out = await _detail(api_capture, {'id': 1, 'trackIds': [{'id': 9}]}) assert out['incomplete'] is True assert out['missing'] == 1 assert out['failed_batches'] == 1 async def test_dirty_trackid_skips_whole_batch_without_fail_count(api_capture): """脏 ID（'①'.isdigit() 为真但 int() 抛 ValueError）：跳过整批且不计失败批次。""" api_capture['batch_resps'] = [] # 不应有批次请求发出 out = await _detail(api_capture, {'id': 1, 'trackIds': [{'id': '①'}]}) assert out['tracks'] == [] assert out['incomplete'] is True assert out['missing'] == 1 assert out['failed_batches'] == 0 assert _batch_c_payloads(api_capture['calls']) == [] async def test_empty_trackids_no_batch_no_incomplete(api_capture): """trackIds 为空：不发生批次请求，结果完整。""" out = await _detail(api_capture, {'id': 1, 'trackIds': []}) assert out['tracks'] == [] assert 'incomplete' not in out assert _batch_c_payloads(api_capture['calls']) == [] async def test_batches_of_100(api_capture): """101 首分两批：首批 100、次批 1。""" api_capture['batch_resps'] = [ _FakeResp({'code': 200, 'songs': []}), _FakeResp({'code': 200, 'songs': []}), ] track_ids = [{'id': i} for i in range(1, 102)] out = await _detail(api_capture, {'id': 1, 'trackIds': track_ids}) payloads = _batch_c_payloads(api_capture['calls']) assert len(payloads) == 2 assert len(payloads[0]) == 100 assert payloads[1] == [101] assert out['incomplete'] is True # 明细一首未回，101 全缺 assert out['missing'] == 101 # ============ 错误映射 ============ async def test_network_error_maps_to_request_failed(api_capture, monkeypatch): """网络层异常（含批次阶段的）：统一转「获取歌单详情请求失败」。""" 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.get_playlist_detail(123, {}) async def test_json_decode_error_maps_to_parse_failed(api_capture): api_capture['playlist_resp'] = _FakeResp(json_error=json.JSONDecodeError('bad', 'x', 0)) with pytest.raises(APIException, match='解析歌单详情响应失败'): await _API.get_playlist_detail(123, {})