This repository has been archived on 2026-10-02. You can view files and clone it. You cannot open issues or pull requests or push a commit.
Files
NeteaseUrl/tests/test_zip_workers.py
T
snowy-yang 095171e0b7 refactor: 两个根级游离模块并入 music_api 包,消除分层倒置
为什么:podcast_api.py(560 行) 与 qr_login.py(397 行) 既不属于任何包、也不是
入口脚本,却同时被 services 层与 web 层引用——`services/downloads.py` 直接
`from qr_login import ...`、`services/podcast.py` 直接 `import podcast_api`,
业务层反向依赖顶层游离模块,是本次架构体检里唯一的无争议分层倒置。
两者本就依赖 music_api(前者取 url_v1/APIConstants/APIException,后者取
QRLoginManager),与 music_api 下已有的 qr.py / compat.py 是同一层的同类物。

迁移(git 保留重命名历史):
- podcast_api.py -> music_api/podcast.py
- qr_login.py    -> music_api/qr_client.py

**命名避让**:不叫 music_api/qr_login.py——music_api/__init__ 已经从 .compat
导出了一个**同名函数** qr_login,同名子模块会与之互相遮蔽。改叫 qr_client
(其核心导出即 QRLoginClient),并在验证里专门确认 music_api.qr_login 仍是
可调用函数、music_api.podcast 是模块,两者不冲突。

内部导入顺带改为精确子模块,不再绕道包门面:
- podcast.py:from music_api import APIConstants/APIException/url_v1
  → from .compat import url_v1 / from .constants import ... / from .errors import ...
- qr_client.py:删掉那层 try/except ImportError 包装(它给的是「请确保
  music_api.py 文件存在」这类顶层脚本时代的提示,作为包内模块已无意义),
  改 from .cookies/.errors/.qr 相对导入
- CLI 用法文本 python qr_login.py → python -m music_api.qr_client
  (相对导入后直接跑脚本会失败,必须走 -m;全仓确认无脚本/文档引用旧入口)
- 路径安全性已核对:qr_client 用的是 CWD 相对的 cookie.txt,不是 __file__ 相对,
  迁移不改其解析结果

引用点全部更新(AST 扫描确认覆盖全集,替换后残留 0):
- podcast:services/podcast.py(1 + 5 处调用) · web/routers/podcast.py(1 + 6)
  · web/routers/download.py(2 处函数内 + 3),导入名统一为 `podcast`
- qr_login:services/downloads.py · web/routers/admin.py · web/routers/user.py(×2)
- services/__init__.py 依赖自检列表去掉这两个名字(已并入 music_api)
- pyproject include 去掉两个顶层文件名(music_api 目录已覆盖)

验证:
- 11 + 3 个符号在新位置全部 hasattr 通过;podcast_api / qr_login 已
  ModuleNotFoundError;6 个调用方模块均可 import;services 依赖自检仍生效
- 门禁:ruff 0 · pyright 0 errors/0 warnings · pytest 184 passed
- 真实 uvicorn 冒烟 10/10,含 /podcast/parse 与 /podcast/hot 返回 401
  (模块导入失败会是 500,路由缺失会是 404,401 证明两者都正常)
- 生产 data/ 指纹 before == after
2026-09-26 15:03:47 +08:00

457 lines
19 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
# -*- coding: utf-8 -*-
"""两个 ZIP 打包 worker 的行为契约测试(沙箱,不触碰生产 data/)。
为什么:``_batch_zip_worker`` / ``_podcast_zip_worker`` 合计约 642 行此前**零测试覆盖**,
而两者高度重复——连 ``_wait_if_paused`` 那 16 行都逐字相同(上一次修的
``zip_valid`` 可能未绑定缺陷就是同型出现在这两处,重复代码在持续制造同类 bug)。
抽公共实现之前,先用**行为级**测试钉住可观察契约:任务终态、ZIP 内容、
账本守恒、暂停/恢复/取消语义。断言只依赖「调用 worker 后世界里发生了什么」,
不引用任何内部函数名——这样重构时测试不必跟着改,才具备等价性验证能力。
账本口径:一律用 ``reserve_and_open`` 真预扣(``ledger_open`` 只记账不扣款,
会让「退款」凭空给余额加次数,守恒断言就失去意义)。
"""
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 batch as batch_svc
from services import podcast as podcast_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:
"""可编程下载器替身:记录调用序,支持预设失败集与按序触发钩子。"""
def __init__(self, cache_dir: Path, *, fail_ids: tuple[int, ...] = (),
hooks: dict[int, Any] | None = None):
self.cache_dir = cache_dir
self.fail_ids = set(fail_ids)
self.hooks = hooks or {}
self.calls: list[int] = []
self._lock = threading.Lock()
def get_music_info(self, mid: Any, quality: str, cookies: Any) -> MusicInfo:
return MusicInfo(
id=int(mid), name=f'歌{mid}', artists='歌手A', album='专辑A',
pic_url='', duration=200, track_number=1, download_url='',
file_type='mp3', file_size=1024, quality='standard',
lyric='[00:01.00]第一行\n[00:02.00]第二行',
)
def download_music_file(self, mid: Any, quality: str, **kw: Any) -> DownloadResult:
with self._lock:
self.calls.append(int(mid))
seq = len(self.calls)
hook = self.hooks.get(seq)
if hook is not None:
hook()
cancel_ev = kw.get('cancel_event')
if cancel_ev is not None and cancel_ev.is_set():
raise DownloadCanceled('任务已取消')
if int(mid) in self.fail_ids:
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)
auth.init_db()
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())
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 = auth.get_user_by_name(username)
assert u is not None
return int(u['id'])
def _balance(username: str = 'payer') -> int:
u = auth.get_user_by_name(username)
assert u is not None
return int(u['duration_seconds'] or 0)
def _wait_for(pred, timeout: float = 30.0, interval: float = 0.05) -> bool:
deadline = time.time() + timeout
while time.time() < deadline:
if pred():
return True
time.sleep(interval)
return False
def _status(task_id: str) -> str:
with DOWNLOAD_TASKS_LOCK:
return str(DOWNLOAD_TASKS.get(task_id, {}).get('status') or '')
def _resume_like_heartbeat(task: dict[str, Any]) -> None:
"""完全对齐生产心跳(web/routers/tasks.py:task_heartbeat)的恢复动作:
置 status='queued' 且 clear pause 事件——两者缺一 worker 都不会醒来。"""
with DOWNLOAD_TASKS_LOCK:
task['status'] = 'queued'
task['pause'].clear()
def _register_batch(sandbox, ids: list[int], *, reserved: int, user_id: int,
concurrency: int = 1) -> tuple[str, dict[str, Any]]:
tid = 't' + str(int(time.time() * 1000000) % 10 ** 10)
# 必须走 reserve_and_open(真预扣+开账同事务)——ledger_open 只记账不扣款,
# 用它会让「退款」凭空给余额加次数,账本守恒断言失去意义
left = auth.reserve_and_open(user_id, reserved, tid)
assert left is not None, '预扣失败(余额不足,检查 fixture 的 credits)'
task: dict[str, Any] = {
'type': 'batch', 'task_id': tid,
'user_id': user_id, 'billing_user_id': user_id, 'billing_reserved': reserved,
'total': len(ids), 'done': 0, 'failed': 0,
'status': 'queued', 'start': time.time(), 'msg': '排队中...',
'cancel': threading.Event(), 'pause': threading.Event(),
'device_id': 'dev-test', 'result': None,
'params': {
'ids': list(ids), 'quality': 'standard', 'filename_format': 'artist_song',
'write_meta': False, 'meta_fields': None, 'lyric_format': 'lrc',
'concurrency': concurrency, 'cookies': {}, 'remote_ip': '-',
'want_zip': True,
},
}
with DOWNLOAD_TASKS_LOCK:
DOWNLOAD_TASKS[tid] = task
return tid, task
def _register_podcast(sandbox, pids: list[int], *, reserved: int, user_id: int) -> tuple[str, dict[str, Any]]:
tid = 'p' + str(int(time.time() * 1000000) % 10 ** 10)
left = auth.reserve_and_open(user_id, reserved, tid)
assert left is not None, '预扣失败(余额不足)'
task: dict[str, Any] = {
'type': 'podcast_batch', 'task_id': tid,
'user_id': user_id, 'billing_user_id': user_id, 'billing_reserved': reserved,
'total': len(pids), 'done': 0, 'failed': 0,
'status': 'queued', 'start': time.time(), 'msg': '排队中...',
'cancel': threading.Event(), 'pause': threading.Event(),
'device_id': 'dev-test', 'result': None,
'params': {
'radio_id': None, 'ids': list(pids), 'limit': max(1, len(pids)),
'quality': 'standard', 'cookies': {}, 'remote_ip': '-',
'radio_name': '测试电台', 'billing_uid': user_id, 'billed': reserved,
},
}
with DOWNLOAD_TASKS_LOCK:
DOWNLOAD_TASKS[tid] = task
return tid, task
def _run(worker, task_id: str) -> threading.Thread:
th = threading.Thread(target=worker, args=(task_id,), daemon=True)
th.start()
return th
def _zip_names(path: str) -> list[str]:
with zipfile.ZipFile(path) as zf:
return zf.namelist()
# ============ batch worker ============
def test_batch_zip_completes(sandbox, monkeypatch):
"""3 首全部成功:终态 done、ZIP 含音频与歌词、账本按交付结算。"""
uid = _uid()
before = _balance()
monkeypatch.setattr(api_service, 'downloader', _FakeDownloader(sandbox['cache']))
tid, task = _register_batch(sandbox, [101, 102, 103], reserved=3, user_id=uid)
th = _run(batch_svc._batch_zip_worker, 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['done'] == 3 and task['failed'] == 0
assert task['result'] is not None
names = _zip_names(task['result']['zip_path'])
assert sum(1 for n in names if n.endswith('.mp3')) == 3, names
assert sum(1 for n in names if n.endswith('.lrc')) == 3, names
row = auth.ledger_get(tid)
assert row is not None and row['status'] != 'open'
assert row['delivered'] == 3
assert _balance() == before - 3, '全部成功:只扣预扣的 3 次,无退款'
def test_batch_zip_pause_then_resume(sandbox, monkeypatch):
"""断线暂停 → 心跳恢复:ZIP 最终完整,任务走完 done。"""
uid = _uid()
before = _balance()
holder: dict[str, Any] = {}
monkeypatch.setattr(api_service, 'downloader',
_FakeDownloader(sandbox['cache'],
hooks={1: lambda: holder['task']['pause'].set()}))
tid, task = _register_batch(sandbox, [201, 202, 203], reserved=3, user_id=uid)
holder['task'] = task
th = _run(batch_svc._batch_zip_worker, tid)
assert _wait_for(lambda: _status(tid) == 'paused'), f"未进入 paused(当前 {_status(tid)})"
assert '已暂停' in task['msg']
assert task['done'] >= 1, '暂停前应至少完成 1 首'
_resume_like_heartbeat(task)
th.join(timeout=30)
assert _status(tid) == 'done', f"恢复后终态={_status(tid)} msg={task.get('msg')}"
names = _zip_names(task['result']['zip_path'])
assert sum(1 for n in names if n.endswith('.mp3')) == 3, names
assert _balance() == before - 3
def test_batch_zip_pause_then_cancel(sandbox, monkeypatch):
"""暂停期间取消:任务收敛 canceled,未交付则不发布 ZIP、账本全额退费。"""
uid = _uid()
before = _balance()
holder: dict[str, Any] = {}
monkeypatch.setattr(api_service, 'downloader',
_FakeDownloader(sandbox['cache'],
hooks={1: lambda: holder['task']['pause'].set()}))
tid, task = _register_batch(sandbox, [301, 302, 303], reserved=3, user_id=uid)
holder['task'] = task
th = _run(batch_svc._batch_zip_worker, tid)
assert _wait_for(lambda: _status(tid) == 'paused'), f"未进入 paused({_status(tid)})"
task['cancel'].set()
th.join(timeout=30)
assert _status(tid) == 'canceled', f"终态={_status(tid)} msg={task.get('msg')}"
assert task['result'] is None, '未交付不应发布 ZIP'
assert _balance() == before, '取消后应全额退回预扣'
row = auth.ledger_get(tid)
assert row is not None and row['delivered'] == 0 and row['status'] != 'open'
def test_batch_zip_all_fail_refunds(sandbox, monkeypatch):
"""全部失败:不发布空 ZIP、转 error、预扣全额退回。"""
uid = _uid()
before = _balance()
monkeypatch.setattr(api_service, 'downloader',
_FakeDownloader(sandbox['cache'], fail_ids=(401, 402)))
tid, task = _register_batch(sandbox, [401, 402], reserved=2, user_id=uid)
th = _run(batch_svc._batch_zip_worker, tid)
th.join(timeout=30)
assert _status(tid) == 'error', f"终态={_status(tid)} msg={task.get('msg')}"
assert task['failed'] == 2
assert task['result'] is None, '全失败不应发布 ZIP'
assert _balance() == before, '全失败应全额退回预扣'
row = auth.ledger_get(tid)
assert row is not None and row['delivered'] == 0 and row['status'] != 'open'
def test_batch_zip_midway_cancel_refunds_all(sandbox, monkeypatch):
"""循环中途取消:按「不发布 + 全额退费」收敛(记录现状)。
注意这与 ``test_batch_zip_cancel_at_publish_checkpoint_delivers`` 的差别:
worker 的 ``zip_valid = True`` 位于 ``with zipfile`` 块**之后**,而 ``while ids``
循环内的任何取消都以异常**穿过**该块(``zipfile.__exit__`` 仍会写出中央目录,
但那行赋值不会执行)→ ``zip_valid`` 保持 False → ``delivered = ok_count if
zip_valid else 0`` 恒为 0。因此源码注释所称的「已入包部分照常交付(成果守恒)」
在循环内取消时并不可达,只在发布前最后一个检查点可达。
"""
uid = _uid()
before = _balance()
holder: dict[str, Any] = {}
monkeypatch.setattr(api_service, 'downloader',
_FakeDownloader(sandbox['cache'],
hooks={2: lambda: holder['task']['cancel'].set()}))
tid, task = _register_batch(sandbox, [501, 502, 503, 504], reserved=4, user_id=uid)
holder['task'] = task
th = _run(batch_svc._batch_zip_worker, tid)
th.join(timeout=30)
assert _status(tid) == 'canceled', f"终态={_status(tid)} msg={task.get('msg')}"
assert task['result'] is None, '当前实现:循环内取消不发布'
assert _balance() == before, '未交付部分全额退回'
row = auth.ledger_get(tid)
assert row is not None and row['delivered'] == 0
def test_batch_zip_cancel_at_publish_checkpoint_delivers(sandbox, monkeypatch):
"""发布前最后一刻取消:已打包部分照常交付、按交付数结算(成果守恒)。
这是「部分成功后取消仍交付」唯一的可达路径,用替身包一层结算函数
在其返回后立即置取消位,从而精确命中该检查点。
"""
uid = _uid()
before = _balance()
holder: dict[str, Any] = {}
real_settle = batch_svc._settle_batch_billing
def settle_then_cancel(task_id: str, ok_count: int, final_status: str = 'settled') -> int:
refunded = real_settle(task_id, ok_count, final_status)
holder['task']['cancel'].set() # 结算已完成、发布前一刻取消
return refunded
monkeypatch.setattr(batch_svc, '_settle_batch_billing', settle_then_cancel)
monkeypatch.setattr(api_service, 'downloader', _FakeDownloader(sandbox['cache']))
tid, task = _register_batch(sandbox, [601, 602, 603], reserved=3, user_id=uid)
holder['task'] = task
th = _run(batch_svc._batch_zip_worker, tid)
th.join(timeout=30)
assert _status(tid) == 'canceled', f"终态={_status(tid)} msg={task.get('msg')}"
assert task['result'] is not None, '已打包部分应交付'
names = _zip_names(task['result']['zip_path'])
assert sum(1 for n in names if n.endswith('.mp3')) == 3, names
assert _balance() == before - 3, '交付 3 首即不退款'
# ============ podcast worker ============
@pytest.fixture()
def podcast_stub(monkeypatch, sandbox):
"""把播客上游与流式下载替换为本地确定性实现。"""
def fake_detail(pid: Any, cookies: Any) -> dict[str, Any]:
n = int(pid)
return {'id': n, 'name': f'单集{n}', 'main_song_id': 1,
'serial_num': n, 'cover_url': '', 'songs': []}
def fake_resolve(prog: dict[str, Any], quality: str, cookies: Any) -> dict[str, Any]:
return {'url': f"http://example.invalid/{prog['id']}.mp3", 'size': 2048, 'type': 'mp3'}
def fake_stream(url: str, target: Path, total_size: int, cancel_ev: Any,
task_id: str = '', progress_label: str = '') -> int:
if cancel_ev is not None and cancel_ev.is_set():
raise DownloadCanceled('任务已取消')
target.write_bytes(b'\xff\xfb\x90\x00' + b'\x00' * 128)
return target.stat().st_size
monkeypatch.setattr(podcast_svc, '_podcast_resolve_audio', fake_resolve)
monkeypatch.setattr(podcast_svc, '_podcast_stream_download', fake_stream)
monkeypatch.setattr(podcast_svc, 'fetch_cover_bytes', lambda _u: None)
monkeypatch.setattr(podcast_svc.podcast, 'get_program_detail', fake_detail)
return sandbox
def test_podcast_zip_completes(podcast_stub):
"""整台打包正常路径:终态 done、ZIP 含全部单集、账本按交付结算。"""
uid = _uid()
before = _balance()
tid, task = _register_podcast(podcast_stub, [11, 12, 13], reserved=3, user_id=uid)
th = _run(podcast_svc._podcast_zip_worker, tid)
th.join(timeout=30)
assert _status(tid) == 'done', f"终态={_status(tid)} msg={task.get('msg')}"
assert task['done'] == 3 and task['failed'] == 0
names = _zip_names(task['result']['zip_path'])
assert sum(1 for n in names if n.endswith('.mp3')) == 3, names
row = auth.ledger_get(tid)
assert row is not None and row['delivered'] == 3
assert _balance() == before - 3
def test_podcast_zip_pause_then_resume(podcast_stub, monkeypatch):
"""播客打包同样支持断线暂停 → 心跳恢复,最终完整交付。"""
uid = _uid()
before = _balance()
holder: dict[str, Any] = {}
fired = {'n': 0}
def _pause_once(url, target, total_size, cancel_ev, task_id='', progress_label=''):
target.write_bytes(b'\xff\xfb\x90\x00' + b'\x00' * 128)
fired['n'] += 1
if fired['n'] == 1: # 只在第一集触发——否则恢复后立刻又被暂停
holder['task']['pause'].set()
return target.stat().st_size
monkeypatch.setattr(podcast_svc, '_podcast_stream_download', _pause_once)
tid, task = _register_podcast(podcast_stub, [21, 22, 23], reserved=3, user_id=uid)
holder['task'] = task
th = _run(podcast_svc._podcast_zip_worker, tid)
assert _wait_for(lambda: _status(tid) == 'paused'), f"未进入 paused({_status(tid)})"
assert '已暂停' in task['msg']
_resume_like_heartbeat(task)
th.join(timeout=30)
assert _status(tid) == 'done', f"恢复后终态={_status(tid)} msg={task.get('msg')}"
names = _zip_names(task['result']['zip_path'])
assert sum(1 for n in names if n.endswith('.mp3')) == 3, names
assert _balance() == before - 3
def test_podcast_zip_all_fail_refunds(podcast_stub, monkeypatch):
"""全部单集失败:转 error 并全额退费,不发布空包。"""
uid = _uid()
before = _balance()
def _boom(prog, quality, cookies):
raise RuntimeError('上游不可用')
monkeypatch.setattr(podcast_svc, '_podcast_resolve_audio', _boom)
tid, task = _register_podcast(podcast_stub, [31, 32], reserved=2, user_id=uid)
th = _run(podcast_svc._podcast_zip_worker, tid)
th.join(timeout=30)
assert _status(tid) == 'error', f"终态={_status(tid)} msg={task.get('msg')}"
assert _balance() == before, '全失败应全额退回'
row = auth.ledger_get(tid)
assert row is not None and row['delivered'] == 0