diff --git a/services/_taskcore.py b/services/_taskcore.py index 92b6bc7..e9bb119 100644 --- a/services/_taskcore.py +++ b/services/_taskcore.py @@ -5,6 +5,7 @@ import threading import time import traceback +from pathlib import Path from typing import Any import auth_system @@ -12,6 +13,9 @@ from core.state import ( DOWNLOAD_TASKS, DOWNLOAD_TASKS_LOCK, ) +from music_downloader import ( + DownloadCanceled, +) from .api_service import ( api_service, @@ -108,3 +112,45 @@ def _settle_batch_billing(task_id: str, ok_count: int, final_status: str = 'sett except Exception: pass return refund + + +def _task_work_dir(task_id: str) -> Path: + """任务专属工作目录（交付 ZIP 与中间文件都落这里）。 + + 原实现把临时包建在 downloads 根目录，会被缓存清扫线程按 mtime 回收 + （Linux 上已打开文件同样能被 unlink），排队超时的任务于是出现 + 「扣费后领取 404」的悬空交付；任务目录从此对根目录清扫免疫。 + 音乐歌单打包与播客整台打包共用本目录约定。 + """ + path = api_service.downloads_path / 'tasks' / task_id + path.mkdir(parents=True, exist_ok=True) + return path + + +def _wait_if_paused(task: dict[str, Any], pause_ev: threading.Event, + cancel_ev: threading.Event, done: int) -> None: + """断线暂停闸门：置 paused 态并等待恢复或取消，恢复后回到 running。 + + 两个打包 worker（音乐歌单 / 播客整台）共用同一套暂停语义——此前各自持有 + 一份逐字相同的副本（连注释都只差一行），是「同型缺陷同时出现在两处」的温床。 + + 调用点位于窗口循环顶部，等待期间**不持有** BATCH_ZIP_SEM：名额交还队列 + 给下一位，这正是「断线不占坑」的实现方式。 + + ``done`` 由调用方按当前进度传入（ok_count + len(failed)）——写成参数而非 + 捕获闭包，是为了让本函数脱离两个 worker 各自的局部状态。 + """ + if not (pause_ev.is_set() or task.get('status') == 'paused'): + return + with DOWNLOAD_TASKS_LOCK: + task['status'] = 'paused' + task['done'] = done + task['msg'] = '已暂停（重新打开页面自动恢复）' + while not cancel_ev.is_set() and (pause_ev.is_set() or task.get('status') == 'paused'): + time.sleep(1) + if cancel_ev.is_set(): + raise DownloadCanceled("任务已取消") + pause_ev.clear() + with DOWNLOAD_TASKS_LOCK: + task['status'] = 'running' + task['msg'] = '继续下载...' diff --git a/services/batch.py b/services/batch.py index e967f45..4c248ad 100644 --- a/services/batch.py +++ b/services/batch.py @@ -27,6 +27,8 @@ from ._shared import ( from ._taskcore import ( _mark_task_terminal, _settle_batch_billing, + _task_work_dir, + _wait_if_paused, ) from .api_service import ( api_service, @@ -110,30 +112,10 @@ def _batch_zip_worker(task_id: str) -> None: DOWNLOAD_TASKS[task_id]['done'] = ok_count + len(failed) DOWNLOAD_TASKS[task_id]['failed'] = len(failed) - def _wait_if_paused() -> None: - """暂停时等待恢复或取消（不持有 BATCH_ZIP_SEM）""" - if not (pause_ev.is_set() or task['status'] == 'paused'): - return - with DOWNLOAD_TASKS_LOCK: - task['status'] = 'paused' - task['done'] = ok_count + len(failed) - task['msg'] = '已暂停（重新打开页面自动恢复）' - while not cancel_ev.is_set() and (pause_ev.is_set() or task['status'] == 'paused'): - time.sleep(1) - if cancel_ev.is_set(): - raise DownloadCanceled("任务已取消") - pause_ev.clear() - with DOWNLOAD_TASKS_LOCK: - task['status'] = 'running' - task['msg'] = '继续下载...' - # ZIP 构建放 downloads 卷（原实现落系统 %TEMP%——500 首无损可达 GB 级， # 小系统盘一次大单即塞爆；单曲/播客早已指定 dir，唯批量漏了） - # F-A：临时包建在任务专属目录 tasks//——原放根目录会被 - # 缓存清扫线程按 mtime 回收（Linux 上已打开文件也能被 unlink，Docker 部署 - # 排队超 35 分钟即出现"扣费后领取 404"的悬空交付），根目录从此碰不到交付物 - _task_dir_for_zip = api_service.downloads_path / 'tasks' / task_id - _task_dir_for_zip.mkdir(parents=True, exist_ok=True) + # 且建在任务专属目录（见 _task_work_dir）：根目录会被缓存清扫按 mtime 回收 + _task_dir_for_zip = _task_work_dir(task_id) # 中央目录完整写出后（下方）才置 True，否则一律按失败全额退费 with _tempfile.NamedTemporaryFile(prefix='batch_zip_', suffix='.zip', dir=str(_task_dir_for_zip), @@ -149,7 +131,7 @@ def _batch_zip_worker(task_id: str) -> None: if cancel_ev.is_set(): raise DownloadCanceled("任务已取消") # 断线暂停：释放名额等待恢复/取消，恢复后重新进入排队 - _wait_if_paused() + _wait_if_paused(task, pause_ev, cancel_ev, ok_count + len(failed)) # 每个窗口持有 BATCH_ZIP_SEM（同一时刻全局最多 3 个批量任务在下载） # 等待名额期间可被取消，不再无限阻塞线程与退款 if not BATCH_ZIP_SEM.acquire_cancellable(cancel_ev): diff --git a/services/podcast.py b/services/podcast.py index 5f97896..0696fbb 100644 --- a/services/podcast.py +++ b/services/podcast.py @@ -33,6 +33,8 @@ from ._shared import ( from ._taskcore import ( _mark_task_terminal, _settle_batch_billing, + _task_work_dir, + _wait_if_paused, ) from .api_service import ( api_service, @@ -359,23 +361,6 @@ def _podcast_zip_worker(task_id: str) -> None: except Exception as e: raise DownloadException(f"补扣下载次数异常: {e}") from e - def _wait_if_paused() -> None: - """暂停时等待恢复或取消""" - if not (pause_ev.is_set() or task['status'] == 'paused'): - return - with DOWNLOAD_TASKS_LOCK: - task['status'] = 'paused' - task['done'] = ok_count + len(failed) - task['msg'] = '已暂停（重新打开页面自动恢复）' - while not cancel_ev.is_set() and (pause_ev.is_set() or task['status'] == 'paused'): - time.sleep(1) - if cancel_ev.is_set(): - raise DownloadCanceled("任务已取消") - pause_ev.clear() - with DOWNLOAD_TASKS_LOCK: - task['status'] = 'running' - task['msg'] = '继续下载...' - def _update_progress(msg: str = '') -> None: with DOWNLOAD_TASKS_LOCK: t = DOWNLOAD_TASKS[task_id] @@ -389,8 +374,7 @@ def _podcast_zip_worker(task_id: str) -> None: # 未 valid 一律不发布、不按 ok_count 结算 # 同音乐批量——ZIP 构建放 downloads 卷，不落系统 %TEMP% # 与音乐批量打包同口径：建在任务专属目录 tasks//，根目录缓存清扫不再触及交付物 - _podcast_task_dir = api_service.downloads_path / 'tasks' / task_id - _podcast_task_dir.mkdir(parents=True, exist_ok=True) + _podcast_task_dir = _task_work_dir(task_id) with _tempfile.NamedTemporaryFile(prefix='podcast_batch_', suffix='.zip', dir=str(_podcast_task_dir), delete=False) as zip_tmp: @@ -400,7 +384,7 @@ def _podcast_zip_worker(task_id: str) -> None: for idx, ep in enumerate(episodes, 1): if cancel_ev.is_set(): raise DownloadCanceled("任务已取消") - _wait_if_paused() + _wait_if_paused(task, pause_ev, cancel_ev, ok_count + len(failed)) # 并发对齐音乐批量：播客打包同样占用全局批量名额（BATCH_ZIP_SEM，全局 3）， # 逐集 acquire/release——暂停等待/取消时不长期占名额；音频本身仍受 # 全局下载信号量约束。原实现不受批量并发控制，多用户同时打包会无限并发 diff --git a/tests/test_zip_workers.py b/tests/test_zip_workers.py new file mode 100644 index 0000000..3885ddb --- /dev/null +++ b/tests/test_zip_workers.py @@ -0,0 +1,456 @@ +# -*- 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_api, '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