Archived
snowy-yang
96af56cc27
拆掉两类垫片的最后一处：auth_system 原为同步 sqlite3 栈，web 层与 services 的
worker 只能靠 asyncio.to_thread 逐个桥接才不阻塞事件循环——每次 DB 访问都白付一次
线程池调度，且「同步调异步」的边界散落在 100+ 个调用点。B1 已把 worker 变成协程，
本批把访问层本身也迁到原生异步，两条线在此合拢。
改动（按子系统）：
- auth_system：db.py 照 core/db.py 现有范式重建为原生异步访问层（aiosqlite + WAL）
——读路径 _q 每次独立连接、无锁、可并发；写路径 _x / write_transaction 用
asyncio.Lock 串行化，写者不阻塞读者。users / cards / credits / ledger / logs /
settings / mail / geo / backup 九模块随之全部 async/await；security.py 是纯计算
（pbkdf2/normalize）不动。__init__ 去掉 _DB_LOCK 与 get_conn 导出。
- web：app / deps / billing / tasks_registry 与 routers/{admin_site,auth,music,
register,user} 直接 await，拆掉包裹的 to_thread。deps.secret_key 改 async，
连带 _parse_cookie 也 async（会话解析要读密钥）。
- services：_taskcore / tasks / downloads / podcast 的账本调用与查询直接 await。
4 处 start_task 调用点（download.py）确认已 await。
设计取舍（为什么这么做，而不是直接包一层）：
- 锁按**事件循环**隔离（_LOOP_LOCKS 以 id(loop) 为键）：asyncio.Lock 绑定创建它的
loop，应用主 loop 与测试各自的 loop 并存，跨 loop 复用会抛 "bound to a different
event loop"。
- WAL 切换用**独立锁** _wal_lock 而非 _write_lock：execute() 已持写锁，再在锁内
connect→_configure 会自锁死（asyncio.Lock 非重入）。切换 + 标记在专用锁内串行并
二次复查，否则并发首连会同时执行 PRAGMA journal_mode=WAL 争排他锁报
database is locked（core/db.py 同款教训，实测复现）。
- force_wal 保留：journal_mode 持久化在库文件头，库文件被删重建时新文件不继承，
必须重设。exists() 判定须在 aiosqlite.connect **之前**（connect 会创建空文件）。
- connect() 用 DB_PATH.parent 兜底建目录，优于 DATA_DIR：生产两者相同，但测试
monkeypatch 只换 DB_PATH 时兜底仍能命中正确目录。
留在 to_thread 上的仅剩**正当的阻塞卸载**，不是同步门面：PBKDF2 的 10 万次迭代
（登录链路每次都要跑）、smtplib 发信、备份文件的 copy2 / os.replace、secret.key
的文件 IO。原有 aria2 / downloader / zip 侧的 to_thread 未动。
不变量逐条保持（迁移只改执行模型，不改语义）：
- ledger_settle 幂等：账本非 open 直接 return 0，重启后重放不重复退款；
- 终态先到先得：_mark_task_terminal 对已 done/error/canceled 的任务直接 return，
堵住 canceled→error 的非法迁移边；
- settle 必须在任务锁**外**调用（ledger_settle 要抢 SQLite 写锁，busy 最长 15s，
持任务锁会让全站进度/提交/取消随库停摆）；
- reserve_and_open 的预扣与开账同事务，消除原两步之间的崩溃丢账窗口。
路径真源红线：DATA_DIR / DB_PATH / BACKUP_DIR / SECRET_KEY_PATH 仍是 auth_system.db
的**模块全局**（用与 core.db 相同的表达式，但**不是**它的别名）。测试只 patch
auth_system.db.*，若改成 core.db 的别名会静默把测试写到生产库。已核 data/site.db
读写前后 sha256 完全一致。
顺带清掉 get_conn：迁移后它只剩「转调 connect()」一层空壳、全仓零消费者（测试
直接开裸连接、备份用 db.connect()），保留等于留一条「拿裸连接自己写」的口子。
测试：19 个文件同步迁移——10 个文件的 web.deps.secret_key 替身改 async；
test_db_conn.py 的多线程首连用例按异步语义拆成 2 条（单 loop gather(16) 覆盖批量
并发首连、WAL 就绪后 8 线程各自 loop 建连覆盖跨 loop）；test_billing_scope.py 去掉
多余的 @pytest.mark.anyio；被 await 且**正常返回**的 monkeypatch 替身一律改 async
（仅 raise 的同步替身仍可用）。
门禁：ruff 全过 / pyright 0-0-0（standard 模式含 tests 与顶层模块）/ pytest
384 项全绿（0 失败 0 error 0 skipped；383→384 因 test_db_conn 拆用例）。
data/site.db 读写前后 sha256 一致（a19f5db…6487），无 -wal/-shm 边车残留。
下一步 B3：删 core/sync_http.py、core/async_bridge.py、core/http_errors.py 及
tests/test_async_bridge.py——三者在生产侧已零消费者（http_errors 仅 sync_http
自用，requests 已非声明依赖）。
592 lines
23 KiB
Python
592 lines
23 KiB
Python
# -*- 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/<tid>/ 下，与共享缓存分离
|
||
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']}"
|