From 8716617d5dfcdd64c621fa71605162c41ded19b6 Mon Sep 17 00:00:00 2001
From: snowy-yang <102860110+snowy-yang@users.noreply.github.com>
Date: Sun, 27 Sep 2026 07:56:01 +0800
Subject: [PATCH] =?UTF-8?q?refactor:=20download=5Fmusic=5Ffile=20=E6=8B=86?=
 =?UTF-8?q?=E8=A7=A3=EF=BC=88205=20=E8=A1=8C=20/=20=E5=9C=88=2047=20?=
 =?UTF-8?q?=E2=86=92=20125=20=E8=A1=8C=20/=20=E5=9C=88=2020=EF=BC=89?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

为什么：download_music_file 是**唯一**的音频落盘路径（单曲 worker、音乐批量 ZIP
都经它），205 行 / 圈复杂度 47 / 嵌套 7 的单体里叠着缓存校验、断点续传、内容嗅探、
体积上限、取消语义、重试退避、元数据限流七件事，却**零直接测试覆盖**
（test_zip_workers 里只有一个 stub）。任何一处被改坏都不会有测试报警。

改法：拆成 8 个具名方法，主函数只留「5 步管线 + 重试策略」：
  _cached_result          锁内复查缓存（命中即返回，坏缓存删除重下）
  _promote_complete_part  已下完未改名的 .part 直接转正，不发起网络请求
  _resume_offset          断点续传起点，并清理没有续传价值的残片
  _transfer               占全局名额 + 选通道（aria2c / 内置 aiohttp）
  _size_mismatch_error    完整性判定（已知期望精确相等 / 未知期望至少非空）
  _publish                嗅探 → 原子转正 → 写元数据
  _write_metadata_locked  标签信号量内的元数据写入（原三处重复的 with 收敛）
  _backoff_wait           退避秒数（原五处重复的 0.5*(attempt+1) 收敛）

复杂度（同口径 AST 实测）：
  download_music_file  205 行 / 圈 47 / 嵌套 7  →  125 行 / 圈 20 / 嵌套 4
                       （125 行里 26 行是 docstring，正文 87 行）
  文件内嵌套 >=5 层     2 → 1（余下的是本次未动的 _fetch_via_http）
  文件内最大圈复杂度     47 → 35（同样落在未动的 _fetch_via_http 上）
  文件行数             614 → 710（+96，全是新签名与 docstring）

等价性证据（两道，互相独立）：
1. 29 例行为契约测试**在未改动的原始实现上一次性全绿**（先写测试、后动代码；改完
   再跑仍全绿）——这批测试因此忠实编码了旧行为，构成等价性验证，而不是「照着新
   实现写的新行为测试」。断言只依赖返回值 / 磁盘上的文件 / 发往网络的请求，不引用
   任何内部函数名，故可跨后续重构复用。
2. AST 字面量多重集比对 + **逐条位置定位**：数值/布尔 27 种、字符串 80 种。差异
   全部归因，且**没有一条用户可见文案或阈值被改动**——
     0   +2：_resume_offset 的三个 return 0 取代原 resume_from = 0
     0.0 +1、0.5 −3、1 −3：四处重复的 0.5*(attempt+1) 收敛进 _backoff_wait
     None +16：新增方法的 6 处类型标注 + 4 处 return None + 2 处 is not None
     ''  +1：_size_mismatch_error 的内部哨兵（非用户可见）
   续传起点逻辑另做逐例核对：promote 命中 / existing<=0 / 超尺寸 / 部分 四路等价。

刻意保留的语义（是决定，不是疏漏）：
- 退避睡眠仍在文件锁外执行——同名并发任务不陪睡。
- 缓存命中时若 write_meta 仍按本次请求重跑一次标签写入（幂等）；write_meta=False
  则不写。
- 已转正（os.replace 之后）才发现取消时**不回滚文件**，只跳过元数据——原行为如此，
  已用 test_cancel_after_promote_keeps_file_but_skips_metadata 固化。
- 未知期望尺寸（file_size=0）走「非空即通过」+ 内容嗅探兜底；已知期望尺寸要求
  **精确相等**（放宽到容差会交付截断的坏文件还照扣次数）。
- 缓存有效性用 st_size != file_size（而非 >=）：跨账号音质降级 / 上游变更都会让
  同 ID 同音质的实际文件不同，命中即返回会把旧文件当本次结果交付。

测试：新增 tests/test_downloader.py（29 例）——正常落盘与 .part 转正、空分块跳过、
进度回调（起点 / 每块 / 异常吞掉）、缓存命中不发请求、坏缓存重下、0 字节缓存拒绝、
续传发 Range 并追加、服务器忽略 Range 时重写、完整残片免网络转正、超尺寸残片丢弃、
206 起点错位 / 206 缺 Content-Range / 416 三路从头重下、CDN 4xx 不重试、408 可重试、
不完整保留 .part、下轮续传、未知尺寸非音频拒绝、体积上限、五种取消路径、元数据开关。

验证：ruff 全过 · pyright 0/0 · pytest 274 passed（245→274）· 生产 data/ 未被触碰
（全部用例走 tmp_path 沙箱）。
---
 music_downloader/downloader.py | 312 ++++++++++++------
 tests/test_downloader.py       | 581 +++++++++++++++++++++++++++++++++
 2 files changed, 785 insertions(+), 108 deletions(-)
 create mode 100644 tests/test_downloader.py

diff --git a/music_downloader/downloader.py b/music_downloader/downloader.py
index 674a671..3b2aa9a 100644
--- a/music_downloader/downloader.py
+++ b/music_downloader/downloader.py
@@ -290,6 +290,10 @@ class MusicDownloader:
             - per-file 锁：同名歌曲并发下载互斥，防止 .part 被交叉写入损坏
             - 连接复用：线程级 Session + keep-alive，避免每首歌重建 TCP/TLS 连接
             - 网络传输与 ffmpeg 元数据写入分离限流，互不抢占对方资源
+
+        本函数只保留编排与重试策略：缓存校验、续传起点、传输通道、完整性判定、
+        元数据限流各自见对应私有方法的 docstring（此前是 205 行 / 圈复杂度 47
+        的单体，任何一处改动都要通读全部步骤）。
         """
         if music_info is None:
             music_info = self.get_music_info(music_id, quality, cookies)
@@ -317,115 +321,34 @@ class MusicDownloader:
             # 同一目标文件的下载者串行化（不同文件之间仍然并行）
             with get_file_lock(file_path):
                 try:
-                    # 其它线程可能刚好完成此文件，锁内复查
-                    if file_path.exists():
-                        # 缓存与本次期望不符（跨账号音质降级/上游变更都会让同 ID 同音质
-                        # 的实际文件不同）：视为缓存 miss 删除重下，而不是把旧文件
-                        # 当本次结果交付（原实现命中即返回、从不校验尺寸）。
-                        # 期望未知（file_size=0）时至少拒绝 0 字节坏文件
-                        st_size = file_path.stat().st_size
-                        cache_bad = ((music_info.file_size and st_size != music_info.file_size)
-                                     or (not music_info.file_size and st_size == 0))
-                        if cache_bad:
-                            self._unlink_quiet(file_path)
-                        else:
-                            # 缓存命中也尊重本次的元数据请求（重跑一次 ffmpeg，幂等）
-                            if write_meta:
-                                with self._tag_semaphore:
-                                    self._write_metadata(file_path, music_info, meta_fields)
-                            return DownloadResult(
-                                success=True,
-                                file_path=str(file_path),
-                                file_size=file_path.stat().st_size,
-                                music_info=music_info,
-                            )
+                    # 其它线程可能刚好完成此文件，锁内复查（命中即返回，不再下载）
+                    cached = self._cached_result(file_path, music_info,
+                                                 write_meta, meta_fields)
+                    if cached is not None:
+                        return cached
 
-                    # part_path 已在文件锁外绑定为 part_path0（同一路径），此处不再重复计算
-                    resume_from = 0
-                    if part_path.exists():
-                        existing = part_path.stat().st_size
-                        if music_info.file_size > 0 and existing == music_info.file_size:
-                            # 上次已下完但未改名：直接转正（转正后立即返回，
-                            # 不再发起网络请求——原实现离线时会把现成完整文件报为失败）
-                            # 改 >= 为 ==——超尺寸残片（上游多发货/坏分片）转正即交付坏文件
-                            os.replace(part_path, file_path)
-                            if write_meta:
-                                with self._tag_semaphore:
-                                    self._write_metadata(file_path, music_info, meta_fields)
-                            return DownloadResult(
-                                success=True,
-                                file_path=str(file_path),
-                                file_size=file_path.stat().st_size,
-                                music_info=music_info,
-                            )
-                        elif existing > 0:
-                            if music_info.file_size > 0 and existing > music_info.file_size:
-                                # 超尺寸残片没有续传价值（上游多发货/坏分片），
-                                # 丢弃从头下——否则下轮被 >= 误转正交付坏文件
-                                self._unlink_quiet(part_path)
-                            else:
-                                resume_from = existing  # 断点续传起点
-                        else:
-                            self._unlink_quiet(part_path)
+                    # 上次已下完但未改名的 .part：直接转正，不发起网络请求
+                    promoted = self._promote_complete_part(
+                        part_path, file_path, music_info, write_meta, meta_fields)
+                    if promoted is not None:
+                        return promoted
 
-                    # 排队改用可取消等待（的 acquire_cancellable）——
-                    # 原阻塞式 with 使取消后仍要等全局 10 名额腾出，任务卡死、退款延迟
-                    if not self._download_semaphore.acquire_cancellable(cancel_event, poll=0.5):
-                        raise DownloadCanceled("下载已取消")
-                    try:
-                        aria2_backend = get_backend()
-                        if aria2_backend is not None:
-                            # aria2c 通道：多连接 + 原生断点续传
-                            self._fetch_via_aria2(aria2_backend, music_info, part_path,
-                                                  cancel_event, progress_cb)
-                        else:
-                            # 内置通道：aiohttp 流式落盘 + Range 起点校验
-                            self._fetch_via_http(music_info, part_path, resume_from,
-                                                 cancel_event, progress_cb)
-                    finally:
-                        # 与 acquire_cancellable 配对释放（原 with 语义）
-                        self._download_semaphore.release()
+                    # part_path 已在文件锁外绑定为 part_path0（同一路径）
+                    self._transfer(music_info, part_path,
+                                   self._resume_offset(part_path, music_info),
+                                   cancel_event, progress_cb)
 
                     # 完整性校验（已知期望大小时必须精确相等；未知期望时至少非空）
-                    actual_size = part_path.stat().st_size
-                    if music_info.file_size > 0:
-                        complete = (actual_size == music_info.file_size)
-                    else:
-                        complete = (actual_size > 0)
-                    if not complete:
+                    mismatch = self._size_mismatch_error(part_path, music_info)
+                    if mismatch:
                         # 保留 .part 以便断点续传；下次续传从现有字节继续
-                        last_error = (
-                            f"下载不完整: 期望 {music_info.file_size or '未知'} 字节, "
-                            f"实际 {actual_size} 字节"
-                        )
-                        if attempt < retries:
-                            next_wait = 0.5 * (attempt + 1)   # 锁外睡眠
+                        last_error = mismatch
+                        next_wait = self._backoff_wait(attempt, retries)
                         continue
 
-                    # 转正前内容嗅探——未知期望尺寸（"非空即通过"口径）
-                    # 时 CDN 错误页/风控 JSON 曾可被存成音频交付
-                    if music_info.file_size <= 0 and not _sniff_audio_header(part_path):
-                        self._unlink_quiet(part_path)
-                        raise DownloadException("下载内容校验失败（非音频数据）")
-
-                    os.replace(part_path, file_path)
-
-                    # 元数据写入移出下载信号量（CPU 操作不占网络名额），
-                    # 但仍在文件锁内执行，保证其它请求看到的是已完成标记的最终文件
-                    if write_meta:
-                        # 元数据段入口补取消检查点——已取消的任务不再
-                        # 进入信号量排队/ffmpeg 等待（有界但可达 120s），收尾不再延迟
-                        if cancel_event is not None and cancel_event.is_set():
-                            raise DownloadCanceled("下载已取消")
-                        with self._tag_semaphore:
-                            self._write_metadata(file_path, music_info, meta_fields)
-
-                    return DownloadResult(
-                        success=True,
-                        file_path=str(file_path),
-                        file_size=file_path.stat().st_size,
-                        music_info=music_info,
-                    )
+                    # 完整性通过 → 嗅探 → 原子转正 → 写元数据
+                    return self._publish(part_path, file_path, music_info,
+                                         cancel_event, write_meta, meta_fields)
 
                 except DownloadCanceled:
                     # 用户明确放弃：清理半截临时文件
@@ -438,8 +361,7 @@ class MusicDownloader:
                 except _RetryFromScratch:
                     # .part 已清理，直接进入下一轮从头下载
                     last_error = "续传起点失效，重新下载"
-                    if attempt < retries:
-                        next_wait = 0.5 * (attempt + 1)   # 锁外睡眠
+                    next_wait = self._backoff_wait(attempt, retries)
                     continue
                 except (requests.RequestException, OSError) as e:
                     # 网络错误：保留 .part 支持断点续传；
@@ -448,13 +370,11 @@ class MusicDownloader:
                         self._unlink_quiet(part_path)
                         raise DownloadCanceled("下载已取消") from e
                     last_error = f"下载请求失败: {e}"
-                    if attempt < retries:
-                        next_wait = 0.5 * (attempt + 1)   # 锁外睡眠
+                    next_wait = self._backoff_wait(attempt, retries)
                     continue
                 except Exception as e:
                     last_error = f"下载过程中发生错误: {e}"
-                    if attempt < retries:
-                        next_wait = 0.5 * (attempt + 1)   # 锁外睡眠
+                    next_wait = self._backoff_wait(attempt, retries)
                     continue
 
         # 最后一次尝试仍失败：若任务已取消按取消收尾
@@ -463,6 +383,182 @@ class MusicDownloader:
             raise DownloadCanceled("下载已取消")
         return DownloadResult(success=False, error_message=last_error or "下载失败")
 
+    # ----------- 下载分步实现（download_music_file 的组成件） -----------
+
+    def _cached_result(
+        self,
+        file_path: Path,
+        music_info: MusicInfo,
+        write_meta: bool,
+        meta_fields: list[str] | None,
+    ) -> DownloadResult | None:
+        """锁内复查目标文件是否已是一份可用缓存；返回 None 表示需要真正下载。
+
+        缓存与本次期望不符时视为 miss 并删除重下，而不是把旧文件当本次结果交付
+        （跨账号音质降级 / 上游变更都会让同 ID 同音质的实际文件不同；原实现
+        命中即返回、从不校验尺寸）。期望未知（file_size=0）时至少拒绝 0 字节坏文件。
+        """
+        if not file_path.exists():
+            return None
+        st_size = file_path.stat().st_size
+        cache_bad = ((music_info.file_size and st_size != music_info.file_size)
+                     or (not music_info.file_size and st_size == 0))
+        if cache_bad:
+            self._unlink_quiet(file_path)
+            return None
+        # 缓存命中也尊重本次的元数据请求（重跑一次 ffmpeg，幂等）
+        if write_meta:
+            self._write_metadata_locked(file_path, music_info, meta_fields)
+        return DownloadResult(
+            success=True,
+            file_path=str(file_path),
+            file_size=file_path.stat().st_size,
+            music_info=music_info,
+        )
+
+    def _promote_complete_part(
+        self,
+        part_path: Path,
+        file_path: Path,
+        music_info: MusicInfo,
+        write_meta: bool,
+        meta_fields: list[str] | None,
+    ) -> DownloadResult | None:
+        """上次已下完但未改名的 .part：直接转正交付；返回 None 表示仍需下载。
+
+        尺寸要求**精确相等**——放宽为 >= 会把超尺寸残片（上游多发货/坏分片）
+        转正成坏文件；期望未知时不走此路（交给下载流程）。转正后立即返回，
+        不再发起网络请求，否则离线时会把现成的完整文件报为失败。
+        """
+        if not part_path.exists():
+            return None
+        if not (music_info.file_size > 0
+                and part_path.stat().st_size == music_info.file_size):
+            return None
+        os.replace(part_path, file_path)
+        if write_meta:
+            self._write_metadata_locked(file_path, music_info, meta_fields)
+        return DownloadResult(
+            success=True,
+            file_path=str(file_path),
+            file_size=file_path.stat().st_size,
+            music_info=music_info,
+        )
+
+    def _resume_offset(self, part_path: Path, music_info: MusicInfo) -> int:
+        """断点续传起点；返回 0 表示从头下载（顺带清理没有续传价值的残片）。"""
+        if not part_path.exists():
+            return 0
+        existing = part_path.stat().st_size
+        if existing <= 0:
+            self._unlink_quiet(part_path)          # 0 字节残片无续传价值
+            return 0
+        if music_info.file_size > 0 and existing > music_info.file_size:
+            # 超尺寸残片（上游多发货/坏分片）丢弃从头下——留着会被
+            # 「尺寸相等即转正」误交付
+            self._unlink_quiet(part_path)
+            return 0
+        return existing
+
+    def _transfer(
+        self,
+        music_info: MusicInfo,
+        part_path: Path,
+        resume_from: int,
+        cancel_event: threading.Event | None,
+        progress_cb,
+    ) -> None:
+        """占全局下载名额后落盘（产物恒为 part_path），异常语义由两条通道对齐。
+
+        aria2c 可用时优先（多连接 + 原生断点续传），否则走内置 aiohttp 流式通道。
+        """
+        # 排队用可取消等待：原阻塞式 with 使取消后仍要等全局 10 名额腾出，
+        # 任务卡死、退款延迟
+        if not self._download_semaphore.acquire_cancellable(cancel_event, poll=0.5):
+            raise DownloadCanceled("下载已取消")
+        try:
+            aria2_backend = get_backend()
+            if aria2_backend is not None:
+                # aria2c 通道：多连接 + 原生断点续传
+                self._fetch_via_aria2(aria2_backend, music_info, part_path,
+                                      cancel_event, progress_cb)
+            else:
+                # 内置通道：aiohttp 流式落盘 + Range 起点校验
+                self._fetch_via_http(music_info, part_path, resume_from,
+                                     cancel_event, progress_cb)
+        finally:
+            # 与 acquire_cancellable 配对释放（原 with 语义）
+            self._download_semaphore.release()
+
+    @staticmethod
+    def _size_mismatch_error(part_path: Path, music_info: MusicInfo) -> str:
+        """完整性判定：已知期望大小时精确相等，未知期望时至少非空。
+
+        通过返回空串；否则返回失败原因（.part 由调用方保留以供下轮续传）。
+        """
+        actual_size = part_path.stat().st_size
+        if music_info.file_size > 0:
+            complete = (actual_size == music_info.file_size)
+        else:
+            complete = (actual_size > 0)
+        if complete:
+            return ''
+        return (f"下载不完整: 期望 {music_info.file_size or '未知'} 字节, "
+                f"实际 {actual_size} 字节")
+
+    def _publish(
+        self,
+        part_path: Path,
+        file_path: Path,
+        music_info: MusicInfo,
+        cancel_event: threading.Event | None,
+        write_meta: bool,
+        meta_fields: list[str] | None,
+    ) -> DownloadResult:
+        """完整性通过后的收尾：内容嗅探 → 原子转正 → 写入元数据。
+
+        嗅探只在期望尺寸未知（"非空即通过"口径）时执行——该口径下 CDN 错误页/
+        风控 JSON 曾可被存成音频交付。转正走 os.replace，半截文件不会被当作已下载。
+        调用点必须在文件锁内（元数据也要求在锁内，其它请求才看到已完成标记的文件）。
+        """
+        if music_info.file_size <= 0 and not _sniff_audio_header(part_path):
+            self._unlink_quiet(part_path)
+            raise DownloadException("下载内容校验失败（非音频数据）")
+
+        os.replace(part_path, file_path)
+
+        # 元数据写入移出下载信号量（CPU 操作不占网络名额），
+        # 但仍在文件锁内执行，保证其它请求看到的是已写入标签的最终文件
+        if write_meta:
+            # 元数据段入口补取消检查点——已取消的任务不再
+            # 进入信号量排队/ffmpeg 等待（有界但可达 120s），收尾不再延迟
+            if cancel_event is not None and cancel_event.is_set():
+                raise DownloadCanceled("下载已取消")
+            self._write_metadata_locked(file_path, music_info, meta_fields)
+
+        return DownloadResult(
+            success=True,
+            file_path=str(file_path),
+            file_size=file_path.stat().st_size,
+            music_info=music_info,
+        )
+
+    def _write_metadata_locked(
+        self,
+        file_path: Path,
+        music_info: MusicInfo,
+        meta_fields: list[str] | None,
+    ) -> None:
+        """在标签信号量内写元数据：写标签会重写整个文件（IO 密集），与下载分离
+        限流——写满时只阻塞后来者，不占下载名额、不拖慢其它文件的传输。"""
+        with self._tag_semaphore:
+            self._write_metadata(file_path, music_info, meta_fields)
+
+    @staticmethod
+    def _backoff_wait(attempt: int, retries: int) -> float:
+        """失败后的退避秒数；已是最后一轮则不等待。锁外睡眠——同名并发任务不陪睡。"""
+        return 0.5 * (attempt + 1) if attempt < retries else 0.0
+
 
     # ----------- 传输通道（二选一，产物都是 part_path） -----------
 
diff --git a/tests/test_downloader.py b/tests/test_downloader.py
new file mode 100644
index 0000000..842f2b9
--- /dev/null
+++ b/tests/test_downloader.py
@@ -0,0 +1,581 @@
+# -*- coding: utf-8 -*-
+"""MusicDownloader.download_music_file 的行为契约测试（沙箱，不触碰生产 data/）。
+
+为什么：download_music_file 是**唯一**的音频落盘路径——单曲 worker 与音乐批量
+ZIP 都经它，约 205 行 / 圈复杂度 47，此前**零直接测试覆盖**（test_zip_workers 里
+只有一个 stub）。它同时承担缓存校验、断点续传、内容嗅探、体积上限、取消语义、
+重试退避与元数据写入，任何一条被拆坏都不会有测试报警。
+
+断言只依赖「调用后世界里发生了什么」——返回值、磁盘上的文件、发往网络的请求——
+不引用任何内部函数名，因此本文件可作为后续拆分的等价性验证器。
+
+替身约定（三处，均为模块级符号，替换点在 downloader 模块命名空间）：
+- 网络：``get_http_session`` / ``get_backend``（后者返回 None = 走内置 HTTP 通道）
+- 内容嗅探：``_sniff_audio_header``（默认放行，按需置 False）
+- 元数据写入：``MusicDownloader._write_metadata``（不真跑 mutagen）
+"""
+import threading
+from pathlib import Path
+from typing import Any
+
+import pytest
+
+from core import sync_http
+from music_downloader import DownloadCanceled, DownloadException, MusicInfo
+from music_downloader.downloader import MusicDownloader
+
+# ============ 替身 ============
+
+
+class _FakeResponse:
+    """core.sync_http.SyncResponse 的替身。
+
+    chunks 保留原始 iterable（**不要** list() 归一化）：生成器的副作用
+    （如中途置取消位）若在构造时刻就求值，「中途取消」会被悄悄测成
+    「开始前取消」——两条不同的路径混成一条，而测试仍是绿的。
+    """
+
+    def __init__(self, chunks, status=200, headers=None):
+        self._chunks = chunks
+        self.status_code = status
+        self.headers = dict(headers or {})
+        self.closed = False
+
+    def iter_content(self, chunk_size=0):
+        yield from self._chunks
+
+    def raise_for_status(self):
+        if self.status_code >= 400:
+            raise sync_http.HTTPError(f'HTTP {self.status_code}', self)
+
+    def close(self):
+        self.closed = True
+
+
+class _FakeSession:
+    """记录每次 GET（含 headers）并按序产出响应。"""
+
+    def __init__(self, responder):
+        self._responder = responder
+        self.calls: list[tuple[str, dict[str, Any]]] = []
+        self.closed = False
+
+    def get(self, url, **kw):
+        self.calls.append((url, kw))
+        return self._responder(len(self.calls), url, kw)
+
+    def close(self):
+        self.closed = True
+
+
+def _body(size, status=200, content_range=None):
+    """固定返回 size 字节的响应工厂（每次调用产出新响应对象）。"""
+
+    def _mk(_attempt, _url, _kw):
+        headers = {'Content-Range': content_range} if content_range else {}
+        return _FakeResponse([b'x' * size], status=status, headers=headers)
+
+    return _mk
+
+
+def _info(mid=101, size=20, file_type='mp3'):
+    return MusicInfo(
+        id=mid, name='测试曲', artists='歌手', album='专辑', pic_url='',
+        duration=180, track_number=1, download_url='http://cdn.test/a.mp3',
+        file_type=file_type, file_size=size, quality='standard',
+    )
+
+
+def _paths(dl, info):
+    """按生产同一函数推算落盘路径与 .part 路径（纯函数，可直接复用）。"""
+    fp = dl._build_file_path(info)
+    return fp, fp.with_name(fp.name + '.part')
+
+
+@pytest.fixture()
+def env(tmp_path, monkeypatch):
+    """返回 (downloader, state, install)。
+
+    - state['sniff']：内容嗅探结果，默认 True
+    - state['meta']：_write_metadata 的调用记录 [(文件路径, fields), ...]
+    - install(responder) -> _FakeSession：安装网络替身，返回该替身以便断言
+    """
+    state: dict[str, Any] = {'sniff': True, 'meta': []}
+    box: dict[str, _FakeSession] = {}
+
+    def install(responder):
+        box['session'] = _FakeSession(responder)
+        return box['session']
+
+    def _record_meta(self, file_path, music_info, fields=None):
+        state['meta'].append((str(file_path), fields))
+        return True
+
+    monkeypatch.setattr('music_downloader.downloader.get_http_session',
+                        lambda: box['session'])
+    monkeypatch.setattr('music_downloader.downloader.get_backend', lambda: None)
+    monkeypatch.setattr('music_downloader.downloader._sniff_audio_header',
+                        lambda _p: state['sniff'])
+    monkeypatch.setattr(MusicDownloader, '_write_metadata', _record_meta)
+
+    install(lambda *a: pytest.fail('本用例不应发出网络请求'))
+    dl = MusicDownloader(download_dir=str(tmp_path))
+    return dl, state, install
+
+
+# ============ 正常落盘 ============
+
+
+def test_success_promotes_part_to_final(env):
+    dl, _state, install = env
+    info = _info(size=20)
+    fp, part = _paths(dl, info)
+    install(_body(20))
+
+    r = dl.download_music_file(101, music_info=info)
+
+    assert r.success and r.error_message == ''
+    assert Path(r.file_path) == fp
+    assert r.file_size == 20 and fp.read_bytes() == b'x' * 20
+    assert not part.exists(), '完成后 .part 应改名转正'
+
+
+def test_empty_chunks_skipped(env):
+    dl, _state, install = env
+    info = _info(size=20)
+    fp, _ = _paths(dl, info)
+    install(lambda a, u, kw: _FakeResponse([b'', b'x' * 20, b'']))
+    seen: list[int] = []
+
+    r = dl.download_music_file(101, music_info=info,
+                               progress_cb=lambda w, _t: seen.append(w))
+
+    assert r.success and fp.stat().st_size == 20
+    assert seen == [0, 20], '空分块不写盘也不触发进度回调'
+
+
+def test_progress_reports_resume_start_then_each_chunk(env):
+    dl, _state, install = env
+    info = _info(size=20)
+    install(lambda a, u, kw: _FakeResponse([b'a' * 8, b'b' * 12]))
+    seen: list[tuple[int, int]] = []
+
+    r = dl.download_music_file(101, music_info=info,
+                               progress_cb=lambda w, t: seen.append((w, t)))
+
+    assert r.success
+    assert seen == [(0, 20), (8, 20), (20, 20)], '起始回调带续传起点，此后每块一次'
+
+
+def test_progress_callback_exception_never_breaks_download(env):
+    dl, _state, install = env
+    info = _info(size=20)
+    install(_body(20))
+
+    def _boom(_w, _t):
+        raise RuntimeError('回调炸了')
+
+    r = dl.download_music_file(101, music_info=info, progress_cb=_boom)
+
+    assert r.success, '进度回调异常绝不影响下载主流程'
+
+
+def test_music_info_fetched_when_not_provided(env, monkeypatch):
+    dl, _state, install = env
+    info = _info(size=20)
+    seen: list[tuple] = []
+    monkeypatch.setattr(dl, 'get_music_info',
+                        lambda mid, q, c=None: (seen.append((mid, q)), info)[1])
+    install(_body(20))
+
+    r = dl.download_music_file(101, 'lossless')
+
+    assert r.success and seen == [(101, 'lossless')]
+
+
+# ============ 缓存校验 ============
+
+
+def test_cache_hit_skips_network(env):
+    dl, _state, install = env
+    info = _info(size=20)
+    fp, _ = _paths(dl, info)
+    fp.write_bytes(b'c' * 20)
+    session = install(lambda *a: pytest.fail('缓存命中不应发网络请求'))
+
+    r = dl.download_music_file(101, music_info=info)
+
+    assert r.success and r.file_size == 20
+    assert session.calls == []
+
+
+def test_cache_size_mismatch_triggers_redownload(env):
+    dl, _state, install = env
+    info = _info(size=20)
+    fp, _ = _paths(dl, info)
+    fp.write_bytes(b'c' * 15)          # 与本次期望不符 = 缓存 miss
+    session = install(_body(20))
+
+    r = dl.download_music_file(101, music_info=info)
+
+    assert r.success and fp.read_bytes() == b'x' * 20
+    assert len(session.calls) == 1, '坏缓存应被删除并重下，而不是当结果交付'
+
+
+def test_zero_byte_cache_rejected_when_expected_size_unknown(env):
+    dl, _state, install = env
+    info = _info(size=0)
+    fp, _ = _paths(dl, info)
+    fp.write_bytes(b'')
+    session = install(lambda a, u, kw: _FakeResponse([b'good']))
+
+    r = dl.download_music_file(101, music_info=info)
+
+    assert r.success and fp.read_bytes() == b'good'
+    assert len(session.calls) == 1, '期望未知时至少拒绝 0 字节坏文件'
+
+
+# ============ 断点续传 ============
+
+
+def test_resume_sends_range_and_appends(env):
+    dl, _state, install = env
+    info = _info(size=20)
+    fp, part = _paths(dl, info)
+    part.write_bytes(b'p' * 10)
+    session = install(lambda a, u, kw: _FakeResponse(
+        [b'q' * 10], status=206, headers={'Content-Range': 'bytes 10-19/20'}))
+    seen: list[tuple[int, int]] = []
+
+    r = dl.download_music_file(101, music_info=info,
+                               progress_cb=lambda w, t: seen.append((w, t)))
+
+    assert r.success
+    assert session.calls[0][1]['headers'].get('Range') == 'bytes=10-'
+    assert fp.read_bytes() == b'p' * 10 + b'q' * 10, '续传应追加而非覆盖'
+    assert seen[0] == (10, 20), '起始进度应带续传起点'
+
+
+def test_server_ignoring_range_rewrites_from_zero(env):
+    dl, _state, install = env
+    info = _info(size=20)
+    fp, part = _paths(dl, info)
+    part.write_bytes(b'p' * 10)
+    install(lambda a, u, kw: _FakeResponse([b'r' * 20], status=200))
+
+    r = dl.download_music_file(101, music_info=info)
+
+    assert r.success and fp.read_bytes() == b'r' * 20, '服务器返回 200 须从头重写'
+
+
+def test_complete_part_promoted_without_network(env):
+    dl, _state, install = env
+    info = _info(size=20)
+    fp, part = _paths(dl, info)
+    part.write_bytes(b'z' * 20)        # 已下完但未改名
+    session = install(lambda *a: pytest.fail('完整残片不应发网络请求'))
+
+    r = dl.download_music_file(101, music_info=info)
+
+    assert r.success and r.file_size == 20
+    assert session.calls == []
+    assert fp.read_bytes() == b'z' * 20 and not part.exists()
+
+
+def test_oversized_part_discarded_starts_from_zero(env):
+    dl, _state, install = env
+    info = _info(size=20)
+    _fp, part = _paths(dl, info)
+    part.write_bytes(b'o' * 30)        # 超尺寸残片没有续传价值
+    session = install(_body(20))
+
+    r = dl.download_music_file(101, music_info=info)
+
+    assert r.success
+    assert 'Range' not in session.calls[0][1]['headers']
+
+
+def test_zero_byte_part_discarded(env):
+    dl, _state, install = env
+    info = _info(size=20)
+    _fp, part = _paths(dl, info)
+    part.write_bytes(b'')
+    session = install(_body(20))
+
+    r = dl.download_music_file(101, music_info=info)
+
+    assert r.success and 'Range' not in session.calls[0][1]['headers']
+
+
+def test_content_range_offset_mismatch_restarts_from_scratch(env):
+    dl, _state, install = env
+    info = _info(size=20)
+    fp, part = _paths(dl, info)
+    part.write_bytes(b'p' * 10)
+
+    def responder(n, _url, _kw):
+        if n == 1:
+            return _FakeResponse([b'q' * 5], status=206,
+                                 headers={'Content-Range': 'bytes 3-7/20'})
+        return _FakeResponse([b'f' * 20], status=200)
+
+    session = install(responder)
+
+    r = dl.download_music_file(101, music_info=info, retries=1)
+
+    assert r.success and len(session.calls) == 2
+    assert 'Range' not in session.calls[1][1]['headers'], '清残片后应从头下'
+    assert fp.read_bytes() == b'f' * 20
+
+
+def test_206_without_content_range_restarts_from_scratch(env):
+    dl, _state, install = env
+    info = _info(size=20)
+    _fp, part = _paths(dl, info)
+    part.write_bytes(b'p' * 10)
+
+    def responder(n, _url, _kw):
+        if n == 1:
+            return _FakeResponse([b'q' * 5], status=206)   # 头被中间层剥离
+        return _FakeResponse([b'f' * 20], status=200)
+
+    session = install(responder)
+
+    r = dl.download_music_file(101, music_info=info, retries=1)
+
+    assert r.success and len(session.calls) == 2
+    assert 'Range' not in session.calls[1][1]['headers']
+
+
+def test_http_416_restarts_from_scratch(env):
+    dl, _state, install = env
+    info = _info(size=20)
+    fp, part = _paths(dl, info)
+    part.write_bytes(b'p' * 10)
+
+    def responder(n, _url, _kw):
+        if n == 1:
+            return _FakeResponse([], status=416)
+        return _FakeResponse([b'f' * 20], status=200)
+
+    session = install(responder)
+
+    r = dl.download_music_file(101, music_info=info, retries=1)
+
+    assert r.success and len(session.calls) == 2
+    assert 'Range' not in session.calls[1][1]['headers']
+    assert fp.read_bytes() == b'f' * 20
+
+
+# ============ 失败与重试 ============
+
+
+def test_cdn_4xx_is_deterministic_and_not_retried(env):
+    dl, _state, install = env
+    info = _info(size=20)
+    session = install(lambda a, u, kw: _FakeResponse([], status=403))
+
+    with pytest.raises(DownloadException) as ei:
+        dl.download_music_file(101, music_info=info, retries=2)
+
+    assert 'HTTP 403' in str(ei.value)
+    assert len(session.calls) == 1, '确定性失败对同一失效直链白重试毫无意义'
+
+
+def test_408_is_retried_then_reported_as_failure(env):
+    dl, _state, install = env
+    info = _info(size=20)
+    session = install(lambda a, u, kw: _FakeResponse([], status=408))
+
+    r = dl.download_music_file(101, music_info=info, retries=1)
+
+    assert not r.success
+    assert '下载请求失败' in r.error_message
+    assert len(session.calls) == 2, '408 请求超时应保留可重试'
+
+
+def test_incomplete_size_keeps_part_and_reports(env):
+    dl, _state, install = env
+    info = _info(size=100)
+    fp, part = _paths(dl, info)
+    install(_body(10))
+
+    r = dl.download_music_file(101, music_info=info, retries=0)
+
+    assert not r.success and '下载不完整' in r.error_message
+    assert part.exists() and part.stat().st_size == 10, '.part 应保留以供续传'
+    assert not fp.exists(), '不完整的下载不得转正交付'
+
+
+def test_incomplete_then_resumes_on_next_attempt(env):
+    dl, _state, install = env
+    info = _info(size=100)
+    _fp, part = _paths(dl, info)
+
+    def responder(n, _url, _kw):
+        if n == 1:
+            return _FakeResponse([b'a' * 10], status=200)
+        return _FakeResponse([b'b' * 5], status=206,
+                             headers={'Content-Range': 'bytes 10-14/100'})
+
+    session = install(responder)
+
+    r = dl.download_music_file(101, music_info=info, retries=1)
+
+    assert not r.success and '下载不完整' in r.error_message
+    assert session.calls[1][1]['headers'].get('Range') == 'bytes=10-', '下轮应续传'
+    assert part.stat().st_size == 15
+
+
+def test_network_error_keeps_part_for_resume(env):
+    dl, _state, install = env
+    info = _info(size=100)
+    fp, part = _paths(dl, info)
+
+    def chunks():
+        yield b'a' * 10
+        raise sync_http.RequestException('连接中断')
+
+    install(lambda a, u, kw: _FakeResponse(chunks()))
+
+    r = dl.download_music_file(101, music_info=info, retries=0)
+
+    assert not r.success and '下载请求失败' in r.error_message
+    assert part.exists() and part.stat().st_size == 10, '网络错误应保留 .part'
+    assert not fp.exists()
+
+
+# ============ 内容校验与体积上限 ============
+
+
+def test_unknown_size_non_audio_content_rejected(env):
+    dl, state, install = env
+    info = _info(size=0)
+    fp, part = _paths(dl, info)
+    state['sniff'] = False
+    install(lambda a, u, kw: _FakeResponse([b'<html>blocked</html>']))
+
+    with pytest.raises(DownloadException) as ei:
+        dl.download_music_file(101, music_info=info)
+
+    assert '非音频数据' in str(ei.value)
+    assert not part.exists() and not fp.exists()
+
+
+def test_unknown_size_audio_content_accepted(env):
+    dl, state, install = env
+    info = _info(size=0)
+    fp, _ = _paths(dl, info)
+    state['sniff'] = True
+    install(lambda a, u, kw: _FakeResponse([b'ID3audio']))
+
+    r = dl.download_music_file(101, music_info=info)
+
+    assert r.success and fp.read_bytes() == b'ID3audio'
+
+
+def test_size_limit_discards_part_and_fails(env, monkeypatch):
+    dl, _state, install = env
+    monkeypatch.setattr('music_downloader.downloader._MAX_DOWNLOAD_BYTES', 16)
+    info = _info(size=0)
+    fp, part = _paths(dl, info)
+    install(lambda a, u, kw: _FakeResponse([b'x' * 100]))
+
+    with pytest.raises(DownloadException) as ei:
+        dl.download_music_file(101, music_info=info)
+
+    assert '大小上限' in str(ei.value)
+    assert not part.exists() and not fp.exists(), '超限残片无续传价值，须清理'
+
+
+# ============ 取消 ============
+
+
+def test_cancel_before_transfer_clears_part_and_skips_network(env):
+    dl, _state, install = env
+    info = _info(size=20)
+    _fp, part = _paths(dl, info)
+    part.write_bytes(b'p' * 5)
+    cancel = threading.Event()
+    cancel.set()
+    session = install(lambda *a: pytest.fail('已取消不应发网络请求'))
+
+    with pytest.raises(DownloadCanceled):
+        dl.download_music_file(101, music_info=info, cancel_event=cancel)
+
+    assert session.calls == []
+    assert not part.exists(), '取消须清掉 .part'
+
+
+def test_cancel_midway_clears_half_written_file(env):
+    dl, _state, install = env
+    info = _info(size=20)
+    fp, part = _paths(dl, info)
+    cancel = threading.Event()
+
+    def chunks():
+        yield b'a' * 5
+        cancel.set()
+        yield b'b' * 15
+
+    install(lambda a, u, kw: _FakeResponse(chunks()))
+
+    with pytest.raises(DownloadCanceled):
+        dl.download_music_file(101, music_info=info, cancel_event=cancel)
+
+    assert not part.exists() and not fp.exists()
+
+
+def test_network_error_while_canceled_reports_canceled(env):
+    dl, _state, install = env
+    info = _info(size=20)
+    fp, part = _paths(dl, info)
+    cancel = threading.Event()
+
+    def chunks():
+        yield b'a' * 5
+        cancel.set()
+        raise OSError('句柄异常')
+
+    install(lambda a, u, kw: _FakeResponse(chunks()))
+
+    with pytest.raises(DownloadCanceled):
+        dl.download_music_file(101, music_info=info, cancel_event=cancel, retries=2)
+
+    assert not part.exists() and not fp.exists(), '已取消时残片没有续传价值'
+
+
+def test_cancel_after_promote_keeps_file_but_skips_metadata(env):
+    dl, state, install = env
+    info = _info(size=20)
+    fp, part = _paths(dl, info)
+    cancel = threading.Event()
+
+    def cb(written, _total):
+        if written >= 20:
+            cancel.set()
+
+    install(_body(20))
+
+    with pytest.raises(DownloadCanceled):
+        dl.download_music_file(101, music_info=info, cancel_event=cancel,
+                               progress_cb=cb)
+
+    assert fp.exists() and not part.exists(), '已转正的文件不回滚'
+    assert state['meta'] == [], '取消后不应再写元数据'
+
+
+# ============ 元数据 ============
+
+
+def test_metadata_written_only_when_requested(env):
+    dl, state, install = env
+    install(_body(20))
+
+    dl.download_music_file(101, music_info=_info(mid=101), write_meta=False)
+    assert state['meta'] == [], 'write_meta=False 不应写元数据'
+
+    r = dl.download_music_file(102, music_info=_info(mid=102), write_meta=True,
+                               meta_fields=['title'])
+    assert r.success
+    assert [f for _p, f in state['meta']] == [['title']], 'fields 应原样透传'