# -*- coding: utf-8 -*- """单曲下载 worker 的行为契约测试(沙箱,不触碰生产 data/)。 为什么:``services/downloads.py::_single_download_worker`` 是单曲下载的**唯一**后台 执行路径(``/download`` 的 file 模式走 ``start_task`` 启动它),177 行 / 圈复杂度 30, 此前**零测试覆盖**。它的三个 except 分支各写一遍「结算 + 删任务目录」,正文里又散着 四处取消检查——改坏任意一处都不会有测试报警。 按惯例:**先在未改动的原始实现上跑绿**,这批测试因此忠实编码了旧行为,重构时本身 即等价性验证器。断言只依赖「调用 worker 后世界里发生了什么」(任务终态、磁盘上的 文件、ZIP 内容、账本行、余额),不引用任何内部函数名,故可跨后续重构复用。 账本口径:一律用 ``reserve_and_open`` 真预扣(``ledger_open`` 只记账不扣款, 会让「退款」凭空给余额加次数,守恒断言就失去意义)。 """ import asyncio import threading import time import zipfile from pathlib import Path from typing import Any import pytest import auth_system as auth import core.db as core_db from core.state import DOWNLOAD_TASKS, DOWNLOAD_TASKS_LOCK from music_downloader import DownloadCanceled, DownloadResult, MusicInfo from services import downloads as downloads_svc from services.api_service import api_service PW = 'Passw0rd12' CREDITS = 50 # ============ 测试替身 ============ class _NullLogger: """吞掉 api_service.logger 输出,避免测试往项目根 music_api.log 追加。""" def debug(self, *a: Any, **k: Any) -> None: ... def info(self, *a: Any, **k: Any) -> None: ... def warning(self, *a: Any, **k: Any) -> None: ... def error(self, *a: Any, **k: Any) -> None: ... class _FakeDownloader: """可编程下载器替身。 ``on_download`` 在 ``download_music_file`` 内部被调用,拿到调用时的 kwargs, 用于在「下载中途」注入取消 / 调用进度回调 / 抛异常。 ``info_error`` 让 ``get_music_info`` 抛异常(验证未知异常路径)。 """ def __init__(self, cache_dir: Path, *, download_ok: bool = True, info_error: Exception | None = None, on_download: Any = None, lyric: str = '[00:01.00]第一行\n[00:02.00]第二行'): self.cache_dir = cache_dir self.download_ok = download_ok self.info_error = info_error self.on_download = on_download self.lyric = lyric self.info_calls = 0 self.download_calls = 0 async def get_music_info(self, mid: Any, quality: str, cookies: Any) -> MusicInfo: self.info_calls += 1 if self.info_error is not None: raise self.info_error return MusicInfo( id=int(mid), name='歌名X', artists='歌手A', album='专辑A', pic_url='https://pic.example/x.jpg', duration=200, track_number=1, download_url='', file_type='mp3', file_size=1024, quality='standard', lyric=self.lyric, ) async def download_music_file(self, mid: Any, quality: str, **kw: Any) -> DownloadResult: self.download_calls += 1 if self.on_download is not None: self.on_download(kw) cancel_ev = kw.get('cancel_event') if cancel_ev is not None and cancel_ev.is_set(): raise DownloadCanceled('任务已取消') if not self.download_ok: return DownloadResult(success=False, error_message='上游 404') path = self.cache_dir / f'{mid}.mp3' path.write_bytes(b'ID3\x03\x00\x00\x00' + b'\x00' * 64) return DownloadResult(success=True, file_path=str(path), file_size=path.stat().st_size, music_info=kw.get('music_info')) # ============ 沙箱 ============ @pytest.fixture() def sandbox(tmp_path, monkeypatch): """临时站点库 + 临时下载目录 + 空任务表。""" db_path = tmp_path / 'site.db' monkeypatch.setattr(auth.db, 'DATA_DIR', tmp_path) monkeypatch.setattr(auth.db, 'DB_PATH', db_path) monkeypatch.setattr(auth.db, 'BACKUP_DIR', tmp_path / 'backups') monkeypatch.setattr(auth.db, 'SECRET_KEY_PATH', tmp_path / 'secret.key') monkeypatch.setattr(core_db, 'DATA_DIR', tmp_path) monkeypatch.setattr(core_db, 'DB_PATH', db_path) asyncio.run(auth.init_db()) asyncio.run(auth.create_user('payer', PW, email='p@example.com', credits=CREDITS)) downloads = tmp_path / 'downloads' downloads.mkdir() cache = downloads / 'cache' cache.mkdir() monkeypatch.setattr(api_service, 'downloads_path', downloads) monkeypatch.setattr(api_service, 'logger', _NullLogger()) # 封面走网络——测试里默认返回固定字节,个例可覆盖 async def _cover(_url: str) -> bytes: return b'\xff\xd8\xff cover' monkeypatch.setattr(downloads_svc, 'fetch_cover_bytes', _cover) # 操作记录写进程内共享列表——隔离掉,断言改由个例捕获 monkeypatch.setattr(downloads_svc, '_record_log', lambda *a: None) with DOWNLOAD_TASKS_LOCK: DOWNLOAD_TASKS.clear() yield {'root': tmp_path, 'downloads': downloads, 'cache': cache} with DOWNLOAD_TASKS_LOCK: DOWNLOAD_TASKS.clear() # ============ 工具 ============ def _uid(username: str = 'payer') -> int: u = asyncio.run(auth.get_user_by_name(username)) assert u is not None return int(u['id']) def _balance(username: str = 'payer') -> int: u = asyncio.run(auth.get_user_by_name(username)) assert u is not None return int(u['duration_seconds'] or 0) def _status(task_id: str) -> str: with DOWNLOAD_TASKS_LOCK: return str(DOWNLOAD_TASKS.get(task_id, {}).get('status') or '') def _register_single(sandbox, *, music_id: int = 1, user_id: int, reserved: int = 1, **params: Any) -> tuple[str, dict[str, Any]]: tid = 's' + str(int(time.time() * 1000000) % 10 ** 10) # reserve_and_open:真预扣 + 开账同事务(ledger_open 只记账不扣款,见模块 docstring) left = asyncio.run(auth.reserve_and_open(user_id, reserved, tid)) assert left is not None, '预扣失败(余额不足,检查 fixture 的 credits)' task: dict[str, Any] = { 'type': 'single', 'task_id': tid, 'user_id': user_id, 'billing_user_id': user_id, 'billing_reserved': reserved, 'total': 1, 'done': 0, 'failed': 0, 'status': 'queued', 'start': time.time(), 'msg': '排队中...', 'cancel': threading.Event(), 'pause': threading.Event(), 'device_id': 'dev-test', 'result': None, 'params': { 'music_id': music_id, 'quality': 'standard', 'filename_format': 'artist_song', 'write_meta': False, 'meta_fields': None, 'lyric_format': 'lrc', 'need_zip': False, 'cookies': {}, 'remote_ip': '1.2.3.4', **params, }, } with DOWNLOAD_TASKS_LOCK: DOWNLOAD_TASKS[tid] = task return tid, task def _run(task_id: str) -> threading.Thread: """把 worker 协程跑在独立线程的私有事件循环里(等价于生产的任务循环)。""" def _target() -> None: asyncio.run(downloads_svc._single_download_worker(task_id)) th = threading.Thread(target=_target, daemon=True) th.start() return th def _zip_names(path: str) -> list[str]: with zipfile.ZipFile(path) as zf: return zf.namelist() # ============ 正常路径 ============ def test_single_download_completes(sandbox, monkeypatch): """常规单曲(不打包):终态 done、交付文件落在任务目录、账本按交付结算。""" uid = _uid() before = _balance() monkeypatch.setattr(api_service, 'downloader', _FakeDownloader(sandbox['cache'])) tid, task = _register_single(sandbox, music_id=101, user_id=uid) th = _run(tid) th.join(timeout=30) assert not th.is_alive(), 'worker 未在 30s 内结束' assert _status(tid) == 'done', f"终态={_status(tid)} msg={task.get('msg')}" assert task['msg'] == '下载完成' assert task['done'] == 1 and task['completed_at'] > 0 res = task['result'] assert res is not None assert res['filename'] == '歌手A - 歌名X.mp3' assert res['mimetype'] == 'audio/mp3' assert res['audio_files'] == [res['file_path']] assert 'zip_path' not in res # 交付副本在 tasks// 下,与共享缓存分离 delivered = Path(res['file_path']) assert delivered.exists() assert delivered.parent == sandbox['downloads'] / 'tasks' / tid assert delivered.read_bytes() == (sandbox['cache'] / '101.mp3').read_bytes() assert (sandbox['cache'] / '101.mp3').exists(), '共享缓存不应被搬走' row = asyncio.run(auth.ledger_get(tid)) assert row is not None and row['status'] != 'open' assert row['delivered'] == 1 assert _balance() == before - 1, '交付成功:只扣预扣的 1 次,无退款' def test_single_zip_contains_audio_lyric_cover_and_info(sandbox, monkeypatch): """zip=true:包内含音频 + 歌词 + 封面 + info.txt(信息栏四行)。""" uid = _uid() monkeypatch.setattr(api_service, 'downloader', _FakeDownloader(sandbox['cache'])) tid, task = _register_single(sandbox, music_id=102, user_id=uid, need_zip=True) th = _run(tid) th.join(timeout=30) assert _status(tid) == 'done', f"终态={_status(tid)} msg={task.get('msg')}" res = task['result'] assert res is not None assert res['filename'] == '歌手A - 歌名X.zip' assert res['mimetype'] == 'application/zip' names = sorted(_zip_names(res['zip_path'])) assert names == ['info.txt', '歌手A - 歌名X.jpg', '歌手A - 歌名X.lrc', '歌手A - 歌名X.mp3'], names with zipfile.ZipFile(res['zip_path']) as zf: info = zf.read('info.txt').decode('utf-8') assert '歌曲: 歌名X' in info assert '歌手: 歌手A' in info assert '专辑: 专辑A' in info assert info.startswith('歌曲: ') def test_single_zip_lyric_format_srt(sandbox, monkeypatch): """lyric_format=srt:歌词条目扩展名随之改变。""" uid = _uid() monkeypatch.setattr(api_service, 'downloader', _FakeDownloader(sandbox['cache'])) tid, task = _register_single(sandbox, music_id=103, user_id=uid, need_zip=True, lyric_format='srt') th = _run(tid) th.join(timeout=30) assert _status(tid) == 'done' assert '歌手A - 歌名X.srt' in _zip_names(task['result']['zip_path']) def test_empty_lyric_skips_lyric_entry(sandbox, monkeypatch): """上游无歌词(mi.lyric 为空):不写歌词条目,其余照常。""" uid = _uid() monkeypatch.setattr(api_service, 'downloader', _FakeDownloader(sandbox['cache'], lyric='')) tid, task = _register_single(sandbox, music_id=104, user_id=uid, need_zip=True) th = _run(tid) th.join(timeout=30) assert _status(tid) == 'done' names = _zip_names(task['result']['zip_path']) assert not [n for n in names if n.endswith('.lrc')], names assert [n for n in names if n.endswith('.mp3')], names def test_cover_failure_keeps_zip_without_cover(sandbox, monkeypatch): """取封面失败:ZIP 仍发布,只是没有封面条目(封面非交付必要条件)。 注意 ``fetch_cover_bytes`` 的契约是**失败返回 None**(内部已 except Exception), 所以这里注入的是返回 None,不是抛异常——后者是不存在的形状。 """ uid = _uid() monkeypatch.setattr(api_service, 'downloader', _FakeDownloader(sandbox['cache'])) async def _no_cover(_url: str) -> None: return None monkeypatch.setattr(downloads_svc, 'fetch_cover_bytes', _no_cover) tid, task = _register_single(sandbox, music_id=105, user_id=uid, need_zip=True) th = _run(tid) th.join(timeout=30) assert _status(tid) == 'done', f"终态={_status(tid)} msg={task.get('msg')}" names = _zip_names(task['result']['zip_path']) assert not [n for n in names if n.endswith('.jpg')], names def test_filename_format_song_only(sandbox, monkeypatch): """filename_format=song:交付名只用歌名。""" uid = _uid() monkeypatch.setattr(api_service, 'downloader', _FakeDownloader(sandbox['cache'])) tid, task = _register_single(sandbox, music_id=106, user_id=uid, filename_format='song') th = _run(tid) th.join(timeout=30) assert _status(tid) == 'done' assert task['result']['filename'] == '歌名X.mp3' def test_record_log_gets_song_name(sandbox, monkeypatch): """成功交付后按歌名记一条操作记录(IP 取 params.remote_ip)。""" uid = _uid() logs: list[tuple] = [] monkeypatch.setattr(api_service, 'downloader', _FakeDownloader(sandbox['cache'])) monkeypatch.setattr(downloads_svc, '_record_log', lambda *a: logs.append(a)) tid, _ = _register_single(sandbox, music_id=107, user_id=uid) th = _run(tid) th.join(timeout=30) assert logs == [('下载', '1.2.3.4', '歌名X')], logs def test_remote_ip_defaults_when_missing(sandbox, monkeypatch): """params 无 remote_ip:按 '-' 记录,不因缺字段报错。""" uid = _uid() logs: list[tuple] = [] monkeypatch.setattr(api_service, 'downloader', _FakeDownloader(sandbox['cache'])) monkeypatch.setattr(downloads_svc, '_record_log', lambda *a: logs.append(a)) tid, task = _register_single(sandbox, music_id=108, user_id=uid) del task['params']['remote_ip'] th = _run(tid) th.join(timeout=30) assert _status(tid) == 'done' assert logs == [('下载', '-', '歌名X')] # ============ 失败与取消 ============ def test_download_failure_marks_error_and_refunds(sandbox, monkeypatch): """下载器返回 success=False:转 error、全额退回预扣、任务目录不残留。""" uid = _uid() before = _balance() monkeypatch.setattr(api_service, 'downloader', _FakeDownloader(sandbox['cache'], download_ok=False)) tid, task = _register_single(sandbox, music_id=201, user_id=uid) th = _run(tid) th.join(timeout=30) assert _status(tid) == 'error', f"终态={_status(tid)} msg={task.get('msg')}" assert task['msg'] == '上游 404' assert task['result'] is None assert _balance() == before, '失败应全额退回预扣' row = asyncio.run(auth.ledger_get(tid)) assert row is not None and row['delivered'] == 0 and row['status'] != 'open' assert not (sandbox['downloads'] / 'tasks' / tid).exists(), '失败应清掉任务目录' def test_unexpected_exception_uses_generic_message(sandbox, monkeypatch): """非预期异常:用户可见文案泛化(不泄上游细节),原文进日志、余额退回。""" uid = _uid() before = _balance() monkeypatch.setattr(api_service, 'downloader', _FakeDownloader(sandbox['cache'], info_error=RuntimeError('upstream host=10.0.0.9 refused'))) tid, task = _register_single(sandbox, music_id=202, user_id=uid) th = _run(tid) th.join(timeout=30) assert _status(tid) == 'error', f"终态={_status(tid)}" assert task['msg'] == '下载失败,请稍后重试' assert '10.0.0.9' not in task['msg'], '上游细节不得出现在用户可见文案里' assert _balance() == before assert not (sandbox['downloads'] / 'tasks' / tid).exists() def test_cancel_mid_download(sandbox, monkeypatch): """下载中被取消:worker 收敛 canceled、全额退费、任务目录清掉。""" uid = _uid() before = _balance() holder: dict[str, Any] = {} def _cancel_inside(kw): holder['task']['cancel'].set() monkeypatch.setattr(api_service, 'downloader', _FakeDownloader(sandbox['cache'], on_download=_cancel_inside)) tid, task = _register_single(sandbox, music_id=203, user_id=uid) holder['task'] = task th = _run(tid) th.join(timeout=30) assert _status(tid) == 'canceled', f"终态={_status(tid)} msg={task.get('msg')}" assert task['msg'] == '任务已取消' assert task['result'] is None assert _balance() == before, '取消应全额退回预扣' row = asyncio.run(auth.ledger_get(tid)) assert row is not None and row['delivered'] == 0 assert not (sandbox['downloads'] / 'tasks' / tid).exists() def test_cancel_after_download_before_publish(sandbox, monkeypatch): """下载已落盘但取消已置位:不发布交付(不发布 result),转 canceled 并退费。""" uid = _uid() before = _balance() holder: dict[str, Any] = {} def _cancel_after(kw): # 下载返回之后再置位——命中「下载完成后的取消检查」 def _later(): holder['task']['cancel'].set() threading.Timer(0.0, _later).start() monkeypatch.setattr(api_service, 'downloader', _FakeDownloader(sandbox['cache'], on_download=_cancel_after)) tid, task = _register_single(sandbox, music_id=204, user_id=uid) holder['task'] = task th = _run(tid) th.join(timeout=30) assert _status(tid) == 'canceled', f"终态={_status(tid)} msg={task.get('msg')}" assert task['result'] is None, '取消后不得发布交付' assert _balance() == before assert not (sandbox['downloads'] / 'tasks' / tid).exists() def test_cancel_during_zip_stage(sandbox, monkeypatch): """打包阶段被取消:ZIP 不发布、转 canceled、退费。""" uid = _uid() before = _balance() holder: dict[str, Any] = {} def _cancel_after(kw): threading.Timer(0.0, lambda: holder['task']['cancel'].set()).start() monkeypatch.setattr(api_service, 'downloader', _FakeDownloader(sandbox['cache'], on_download=_cancel_after)) tid, task = _register_single(sandbox, music_id=205, user_id=uid, need_zip=True) holder['task'] = task th = _run(tid) th.join(timeout=30) assert _status(tid) == 'canceled', f"终态={_status(tid)} msg={task.get('msg')}" assert task['result'] is None assert _balance() == before assert not (sandbox['downloads'] / 'tasks' / tid).exists() def test_terminal_state_not_overwritten_by_late_success(sandbox, monkeypatch): """任务已被外部置终态(canceled)时,worker 的成功路径不得覆盖为 done。""" uid = _uid() before = _balance() holder: dict[str, Any] = {} def _mark_canceled(kw): with DOWNLOAD_TASKS_LOCK: holder['task']['status'] = 'canceled' holder['task']['completed_at'] = time.time() monkeypatch.setattr(api_service, 'downloader', _FakeDownloader(sandbox['cache'], on_download=_mark_canceled)) tid, task = _register_single(sandbox, music_id=206, user_id=uid) holder['task'] = task th = _run(tid) th.join(timeout=30) assert _status(tid) == 'canceled', f"终态={_status(tid)}" assert task['result'] is None, '已终态不得被覆盖为 done' assert _balance() == before # ============ 进度上报 ============ def test_progress_callback_writes_live_percent(sandbox, monkeypatch): """进度回调写 live_percent 与「下载中 x% · 已收/总量」文案。""" uid = _uid() seen: dict[str, Any] = {} def _emit_progress(kw): cb = kw.get('progress_cb') if cb is not None: cb(512, 1024) with DOWNLOAD_TASKS_LOCK: t = DOWNLOAD_TASKS.get(seen['tid']) or {} seen['live'] = t.get('live_percent') seen['msg'] = t.get('msg') monkeypatch.setattr(api_service, 'downloader', _FakeDownloader(sandbox['cache'], on_download=_emit_progress)) tid, _task = _register_single(sandbox, music_id=301, user_id=uid) seen['tid'] = tid th = _run(tid) th.join(timeout=30) assert seen.get('live') == 50.0, seen assert '下载中 50%' in str(seen.get('msg')), seen assert _status(tid) == 'done', '进度上报不应影响最终交付' def test_progress_callback_without_total_uses_received_bytes(sandbox, monkeypatch): """总量未知:文案退化为「已接收 <大小>」,不写 live_percent。""" uid = _uid() seen: dict[str, Any] = {} def _emit_progress(kw): cb = kw.get('progress_cb') if cb is not None: cb(2048, 0) with DOWNLOAD_TASKS_LOCK: t = DOWNLOAD_TASKS.get(seen['tid']) or {} seen['live'] = t.get('live_percent') seen['msg'] = t.get('msg') monkeypatch.setattr(api_service, 'downloader', _FakeDownloader(sandbox['cache'], on_download=_emit_progress)) tid, _task = _register_single(sandbox, music_id=302, user_id=uid) seen['tid'] = tid th = _run(tid) th.join(timeout=30) assert '已接收' in str(seen.get('msg')), seen assert seen.get('live') is None, seen def test_progress_callback_errors_do_not_break_download(sandbox, monkeypatch): """进度回调自身抛错:被吞掉,下载照常完成。""" uid = _uid() def _bad_progress(kw): cb = kw.get('progress_cb') if cb is not None: # 回调内部访问非法对象——必须被 except 吞掉 cb('not-a-number', None) monkeypatch.setattr(api_service, 'downloader', _FakeDownloader(sandbox['cache'], on_download=_bad_progress)) tid, task = _register_single(sandbox, music_id=303, user_id=uid) th = _run(tid) th.join(timeout=30) assert _status(tid) == 'done', f"终态={_status(tid)} msg={task.get('msg')}" # ============ 结算兜底 ============ def test_settle_failure_does_not_flip_done_to_error(sandbox, monkeypatch): """交付已落定后结算抛错:任务仍为 done(钱由重启对账兜底,不坑已付费用户)。""" uid = _uid() def _boom(*a, **k): raise RuntimeError('库锁超时') monkeypatch.setattr(api_service, 'downloader', _FakeDownloader(sandbox['cache'])) monkeypatch.setattr(downloads_svc.auth_system, 'ledger_settle', _boom) tid, task = _register_single(sandbox, music_id=401, user_id=uid) th = _run(tid) th.join(timeout=30) assert _status(tid) == 'done', f"终态={_status(tid)} msg={task.get('msg')}" assert task['result'] is not None def test_ledger_open_when_worker_crashes_before_settle(sandbox, monkeypatch): """worker 在结算前崩掉:账本行仍为 open(留给重启对账兜底,不是静默丢失)。""" uid = _uid() monkeypatch.setattr(api_service, 'downloader', _FakeDownloader(sandbox['cache'], info_error=RuntimeError('boom'))) tid, _ = _register_single(sandbox, music_id=402, user_id=uid) # 让结算也失败,模拟「连退款都没走成」的极端情况 monkeypatch.setattr(downloads_svc.auth_system, 'ledger_settle', lambda *a, **k: (_ for _ in ()).throw(RuntimeError('db busy'))) th = _run(tid) th.join(timeout=30) row = asyncio.run(auth.ledger_get(tid)) assert row is not None assert row['status'] == 'open', f"应留给对账兜底,实际 status={row['status']}"