# -*- coding: utf-8 -*- """标签写入(mutagen)与内容嗅探(filetype)的单元测试。 覆盖: - build_tags 的字段勾选、name/title 别名、LRC 时间标签剔除、曲号/年份缺省 - write_tags 的失败路径(不支持的扩展名 / 文件不存在) - MP3 与 FLAC 的真实往返(写入后按标签读回) - _sniff_audio_header 对音频容器放行、对 CDN 错误页/JSON/空文件拒绝 """ from __future__ import annotations from typing import Any import pytest from music_downloader import MusicInfo from music_downloader._http import _sniff_audio_header from music_downloader.tagging import TagSet, build_tags, write_tags # 最小合法 FLAC:fLaC 魔数 + 一个 34 字节的 STREAMINFO(last-block 标志置位)。 # STREAMINFO 不能全零——mutagen 会拒绝采样率为 0 的流;此处填 44100Hz/2ch/16bit: # min_blocksize=4096 max_blocksize=4096 min_framesize=1000 max_framesize=2000 # 随后 8 字节打包 (sample_rate 20bit | channels-1 3bit | bps-1 5bit | total 36bit) _STREAMINFO = bytes.fromhex( '1000' '1000' '0003e8' '0007d0' '0ac442f00000ac44' + '00' * 16) assert len(_STREAMINFO) == 34 _MINIMAL_FLAC = b'fLaC' + bytes([0x80, 0x00, 0x00, 0x22]) + _STREAMINFO # 最小 JPEG 头(仅用于标签写入,不参与解码) _JPEG = b'\xff\xd8\xff\xe0' + b'\x00' * 32 def _info(**kw) -> MusicInfo: base: dict[str, Any] = { 'id': 1, 'name': '歌名', 'artists': '歌手', 'album': '专辑', 'pic_url': '', 'duration': 200, 'track_number': 3, 'download_url': 'http://example.invalid/x', 'file_type': 'mp3', 'file_size': 0, 'quality': 'standard', 'lyric': '[00:12.34]第一行\n[00:15.00]第二行', 'tlyric': '', 'publish_year': 2024, } base.update(kw) return MusicInfo(**base) # ---------- build_tags ---------- def test_build_tags_all_fields(): tags = build_tags(_info(), None) assert tags.title == '歌名' assert tags.artist == '歌手' assert tags.album_artist == '歌手' assert tags.album == '专辑' assert tags.date == '2024' assert tags.track == 3 assert tags.lyrics == '第一行\n第二行' # LRC 时间标签已剔除 def test_build_tags_respects_selection(): tags = build_tags(_info(), ['title']) assert tags.title == '歌名' assert tags.artist is None assert tags.album is None assert tags.lyrics is None assert tags.track is None def test_build_tags_accepts_name_alias(): assert build_tags(_info(), ['name']).title == '歌名' @pytest.mark.parametrize('kw, field', [ ({'track_number': 0}, 'track'), ({'publish_year': None}, 'date'), ({'lyric': ''}, 'lyrics'), ({'lyric': '[00:01.00]'}, 'lyrics'), # 只有时间标签 → 清洗后为空 ]) def test_build_tags_omits_empty_values(kw, field): assert getattr(build_tags(_info(**kw), None), field) is None # ---------- write_tags ---------- def test_write_tags_rejects_unsupported_extension(tmp_path): p = tmp_path / 'a.wav' p.write_bytes(b'RIFF' + b'\x00' * 8) assert write_tags(p, TagSet(title='x')) is False def test_write_tags_returns_false_when_missing(tmp_path): assert write_tags(tmp_path / 'nope.mp3', TagSet(title='x')) is False def test_write_tags_mp3_roundtrip(tmp_path): from mutagen.id3 import ID3 p = tmp_path / 'a.mp3' ID3().save(str(p)) # 构造一个只含 ID3 标签区的合法 mp3 tags = TagSet(title='T', artist='A', album='B', album_artist='A2', date='2020', track=2, lyrics='L') assert write_tags(p, tags, cover=_JPEG) is True got = ID3(str(p)) assert got['TIT2'].text[0] == 'T' assert got['TPE1'].text[0] == 'A' assert got['TPE2'].text[0] == 'A2' assert got['TALB'].text[0] == 'B' assert got['TRCK'].text[0] == '2' # USLT/APIC 的 HashKey 带语言/描述后缀,须用 getall 取 assert got.getall('USLT')[0].text == 'L' assert got.getall('APIC')[0].mime == 'image/jpeg' def test_write_tags_mp3_keeps_untouched_frames(tmp_path): from mutagen.id3 import ID3, TXXX p = tmp_path / 'a.mp3' audio = ID3() audio['TXXX:keep'] = TXXX(encoding=3, desc='keep', text=['v']) audio.save(str(p)) assert write_tags(p, TagSet(title='T')) is True assert ID3(str(p))['TXXX:keep'].text[0] == 'v' # 未触及的帧被保留 def test_write_tags_flac_roundtrip(tmp_path): from mutagen.flac import FLAC p = tmp_path / 'a.flac' p.write_bytes(_MINIMAL_FLAC) assert write_tags(p, TagSet(title='T', artist='A', track=4), cover=_JPEG) is True got = FLAC(str(p)) assert got['title'][0] == 'T' assert got['artist'][0] == 'A' assert got['tracknumber'][0] == '4' assert len(got.pictures) == 1 assert got.pictures[0].mime == 'image/jpeg' def test_write_tags_failure_is_swallowed(tmp_path): """写坏文件不应抛异常(主流程靠返回值判断)。""" p = tmp_path / 'broken.flac' p.write_bytes(b'fLaC') # 截断的 FLAC,解析必然失败 assert write_tags(p, TagSet(title='T')) is False # ---------- 内容嗅探 ---------- def test_sniff_accepts_id3(tmp_path): p = tmp_path / 'x.mp3' p.write_bytes(b'ID3\x03\x00\x00\x00' + b'\x00' * 64) assert _sniff_audio_header(p) is True def test_sniff_accepts_flac(tmp_path): p = tmp_path / 'x.flac' p.write_bytes(_MINIMAL_FLAC) assert _sniff_audio_header(p) is True def test_sniff_accepts_ogg(tmp_path): p = tmp_path / 'x.ogg' p.write_bytes(b'OggS' + b'\x00' * 64) assert _sniff_audio_header(p) is True def test_sniff_rejects_risk_json(tmp_path): """CDN 风控 JSON 不能被当成音频交付。""" p = tmp_path / 'x.mp3' p.write_bytes(b'{"code": 403, "msg": "risk"}') assert _sniff_audio_header(p) is False def test_sniff_rejects_html_error_page(tmp_path): p = tmp_path / 'x.mp3' p.write_bytes(b'403') assert _sniff_audio_header(p) is False def test_sniff_rejects_empty_file(tmp_path): p = tmp_path / 'x.mp3' p.write_bytes(b'') assert _sniff_audio_header(p) is False def test_sniff_rejects_missing_file(tmp_path): assert _sniff_audio_header(tmp_path / 'nope.mp3') is False