Archived
snowy-yang
046225e38d
为什么：aria2p 内部是 requests + websocket-client 的同步实现，每个 RPC 都是一次 「同步网络 IO 藏在 to_thread 里」——线程池跑得再满也不产生真正的并发，每次轮询 还要额外付一次线程跳转。这是本项目最后一处此类残留，删掉后整个进程的网络 IO 都走同一个 aiohttp 连接池。 改动： - music_downloader/_aria2.py 整体重写：新增 _Aria2Rpc，直接向 aria2 的 JSON-RPC 2.0 端点发 getVersion / addUri / tellStatus / forceRemove / removeDownloadResult，一律走 core.http_client.request_once（不重试，与 aria2p 口径一致）；新增 Aria2RpcError 承载 HTTP 200 + error 对象的协议错误。 - _abort 逐条复刻 aria2p.remove(force=True, files=?, clean=True) 语义：终态走 removeDownloadResult；在途先 forceRemove 再清结果表；控制文件 .aria2 无条件 删除（clean=True 默认）；仅当 files 且主移除成功才删数据文件。 - get_backend / _try_start 改原生异步：asyncio.Lock 单飞 + await rpc.get_version() 探活，不再用 to_thread 包 subprocess.Popen；超时路径改 to_thread(_reap) 回收。 - 新增 Protocol _Rpc / _ProcessHandle（补 returncode 声明）——只声明本模块真正 调用的方法，测试据此用轻量替身驱动状态机，不必依赖真实 aria2 进程。 - pyproject.toml 删 aria2p 依赖并同步更新过时注释；uv lock 连带移除 8 个传递 依赖：requests / urllib3 / websocket-client / charset-normalizer / loguru / platformdirs / win32-setctime。至此全仓无任何 requests 栈。 - tests/test_aria2.py 重写为 21 项用例（_FakeRpc / _FakeProc / _LiveProc / _FakeHTTP 替身），新增终态体积守卫、_try_start 三分支（就绪重试 / 超时不就绪 杀进程 / 进程提前退出）与 JSON-RPC 信封四例。 - tests/test_downloader.py 的 get_backend 替身由同步 lambda 改为协程——调用点 已直接 await，同步替身会被 await 成 None 而报 TypeError。 验证（在 aria2p 已物理移除的环境上复核，故能真正证明无残留依赖）： - uv run ruff check . → All checks passed! - uv run pyright → 0 errors, 0 warnings, 0 informations - uv run pytest → 392 项，0 失败 / 0 error / 0 skipped（基线 382，净增 10） - data/site.db sha256 前后一致（a19f5dbf…6487），无 -wal/-shm 残留
427 lines
15 KiB
Python
427 lines
15 KiB
Python
# -*- coding: utf-8 -*-
|
|
"""aria2 下载后端的单元测试（用假 RPC 驱动，不需要真实 aria2c 二进制）。
|
|
|
|
覆盖：
|
|
- aria2c 缺失时 get_backend 静默降级且失败结果被记忆
|
|
- 正常完成路径的进度回调、控制文件兜底清理与选项下发
|
|
- 取消 / 体积超限 连带删除残片；任务失败 / 总超时 保留残片供续传
|
|
- 提交失败与 RPC 抖动统一上报为可重试异常
|
|
- 已终结任务只清结果表（removeDownloadResult），在途任务才 forceRemove
|
|
- JSON-RPC 信封（方法名 / token / params）与 error 对象的收敛
|
|
|
|
替身边界：本模块已不再依赖 aria2p——RPC 由 ``_Aria2Rpc`` 直接打
|
|
core.http_client。故替身只替换 ``_Aria2Rpc`` 这一层的四个方法，
|
|
以及（信封测试里）``_http.request_once``。
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import threading
|
|
|
|
import pytest
|
|
|
|
import music_downloader._aria2 as a2
|
|
from core.http_client import HTTPResult, UpstreamNetworkError
|
|
from music_downloader import DownloadCanceled, DownloadException
|
|
|
|
|
|
def _st(**kw):
|
|
"""构造一条 tellStatus 结果（aria2 的数值字段都是字符串）。"""
|
|
base = {'status': 'active', 'completedLength': '0', 'totalLength': '100'}
|
|
base.update(kw)
|
|
return base
|
|
|
|
|
|
class _FakeRpc:
|
|
"""按预设步进产出 tellStatus 结果，并记录移除类调用。"""
|
|
|
|
def __init__(self, steps, *, fail_add=False, fail_tell=False):
|
|
self._steps = list(steps)
|
|
self._i = 0
|
|
self._fail_add = fail_add
|
|
self._fail_tell = fail_tell
|
|
self.added: tuple[str, dict[str, str]] | None = None
|
|
self.removed: list[tuple[str, str]] = [] # ('forceRemove'|'removeDownloadResult', gid)
|
|
|
|
async def get_version(self) -> str:
|
|
return "1.36.0"
|
|
|
|
async def add_uri(self, url: str, options: dict[str, str]) -> str:
|
|
if self._fail_add:
|
|
raise RuntimeError('rpc down')
|
|
self.added = (url, options)
|
|
return 'g1'
|
|
|
|
async def tell_status(self, gid: str) -> dict:
|
|
if self._fail_tell:
|
|
raise RuntimeError('rpc jitter')
|
|
idx = min(self._i, len(self._steps) - 1)
|
|
self._i += 1
|
|
return dict(self._steps[idx])
|
|
|
|
async def force_remove(self, gid: str) -> str:
|
|
self.removed.append(('forceRemove', gid))
|
|
return gid
|
|
|
|
async def remove_download_result(self, gid: str) -> str:
|
|
self.removed.append(('removeDownloadResult', gid))
|
|
return 'OK'
|
|
|
|
|
|
class _FakeProc:
|
|
"""已退出的假进程：poll() 返回非 None，故 _try_start 会立即判定「提前退出」。"""
|
|
|
|
returncode: int | None = 0
|
|
|
|
def poll(self):
|
|
return 0
|
|
|
|
def terminate(self):
|
|
pass
|
|
|
|
def kill(self):
|
|
pass
|
|
|
|
def wait(self, timeout=None):
|
|
return 0
|
|
|
|
|
|
class _LiveProc:
|
|
"""仍在运行的假进程：poll() 返回 None（用于 _try_start 的就绪等待）。"""
|
|
|
|
returncode: int | None = None
|
|
|
|
def poll(self):
|
|
return None
|
|
|
|
def terminate(self):
|
|
pass
|
|
|
|
def kill(self):
|
|
pass
|
|
|
|
def wait(self, timeout=None):
|
|
return 0
|
|
|
|
|
|
def _backend(monkeypatch, steps, **kw):
|
|
# fetch 的轮询改 asyncio.sleep（不再 time.sleep）：把间隔压到 0 让状态机快速走完
|
|
monkeypatch.setattr(a2, '_POLL_INTERVAL', 0.0)
|
|
rpc = _FakeRpc(steps, fail_add=kw.get('fail_add', False),
|
|
fail_tell=kw.get('fail_tell', False))
|
|
return a2.Aria2Backend(rpc, _FakeProc(), 6800), rpc
|
|
|
|
|
|
# ---------- 降级 ----------
|
|
|
|
async def test_get_backend_degrades_without_binary(monkeypatch):
|
|
monkeypatch.setattr(a2.shutil, 'which', lambda _n: None)
|
|
monkeypatch.setattr(a2, '_backend', None)
|
|
monkeypatch.setattr(a2, '_unavailable', False)
|
|
assert await a2.get_backend() is None
|
|
assert await a2.get_backend() is None # 失败结果被记忆，不反复拉起进程
|
|
|
|
|
|
def test_free_port_is_in_valid_range():
|
|
assert 0 < a2._free_port() < 65536
|
|
|
|
|
|
async def test_try_start_waits_for_rpc_ready(monkeypatch):
|
|
"""RPC 就绪前的探活失败要被容忍并重试；就绪后返回可用后端（不需真 aria2c）。
|
|
|
|
这条覆盖的是「首次调用最长阻塞 8s」那条路径——它已从「同步 sleep + to_thread」
|
|
改成「await asyncio.sleep 让出事件循环」，用例钉住状态机不变。
|
|
"""
|
|
probes = {'n': 0}
|
|
|
|
class _RpcStub:
|
|
def __init__(self, port, secret):
|
|
self.port = port
|
|
self.secret = secret
|
|
|
|
async def get_version(self):
|
|
probes['n'] += 1
|
|
if probes['n'] < 2:
|
|
raise RuntimeError('not ready yet')
|
|
return '1.36.0'
|
|
|
|
monkeypatch.setattr(a2.shutil, 'which', lambda _n: 'aria2c')
|
|
monkeypatch.setattr(a2, '_free_port', lambda: 6800)
|
|
monkeypatch.setattr(a2, '_Aria2Rpc', _RpcStub)
|
|
monkeypatch.setattr(a2.subprocess, 'Popen', lambda *a, **kw: _LiveProc())
|
|
monkeypatch.setattr(a2.atexit, 'register', lambda _f: None)
|
|
monkeypatch.setattr(a2, '_RPC_READY_TIMEOUT', 3.0)
|
|
|
|
be = await a2._try_start()
|
|
assert be is not None
|
|
assert be.port == 6800
|
|
assert probes['n'] == 2 # 第一次没就绪、重试后才成功
|
|
|
|
|
|
async def test_try_start_gives_up_when_rpc_never_ready(monkeypatch):
|
|
"""超过上限仍未就绪：返回 None 走降级，并回收子进程（不留下孤儿 aria2c）。"""
|
|
killed = []
|
|
proc = _LiveProc()
|
|
monkeypatch.setattr(proc, 'kill', lambda: killed.append(1))
|
|
|
|
class _NeverReady:
|
|
def __init__(self, port, secret):
|
|
pass
|
|
|
|
async def get_version(self):
|
|
raise RuntimeError('not ready')
|
|
|
|
monkeypatch.setattr(a2.shutil, 'which', lambda _n: 'aria2c')
|
|
monkeypatch.setattr(a2, '_free_port', lambda: 6800)
|
|
monkeypatch.setattr(a2, '_Aria2Rpc', _NeverReady)
|
|
monkeypatch.setattr(a2.subprocess, 'Popen', lambda *a, **kw: proc)
|
|
monkeypatch.setattr(a2.atexit, 'register', lambda _f: None)
|
|
monkeypatch.setattr(a2, '_RPC_READY_TIMEOUT', 0.0)
|
|
|
|
assert await a2._try_start() is None
|
|
assert killed == [1] # 已强杀回收
|
|
|
|
|
|
async def test_try_start_returns_none_when_process_exits_early(monkeypatch):
|
|
"""aria2c 进程提前退出：立即降级，不等满超时。"""
|
|
|
|
class _RpcStub:
|
|
def __init__(self, port, secret):
|
|
pass
|
|
|
|
async def get_version(self):
|
|
raise RuntimeError('should not matter')
|
|
|
|
monkeypatch.setattr(a2.shutil, 'which', lambda _n: 'aria2c')
|
|
monkeypatch.setattr(a2, '_free_port', lambda: 6800)
|
|
monkeypatch.setattr(a2, '_Aria2Rpc', _RpcStub)
|
|
monkeypatch.setattr(a2.subprocess, 'Popen', lambda *a, **kw: _FakeProc())
|
|
monkeypatch.setattr(a2.atexit, 'register', lambda _f: None)
|
|
|
|
assert await a2._try_start() is None # _FakeProc.poll() != None → 提前退出
|
|
|
|
|
|
# ---------- 完成路径 ----------
|
|
|
|
async def test_fetch_reports_progress_and_completes(monkeypatch, tmp_path):
|
|
be, rpc = _backend(monkeypatch, [
|
|
_st(completedLength='50'),
|
|
_st(status='complete', completedLength='100'),
|
|
])
|
|
target = tmp_path / 'y.part'
|
|
ctrl = tmp_path / 'y.part.aria2'
|
|
ctrl.write_bytes(b'aria2 ctrl') # 预置控制文件，验证兜底清理
|
|
|
|
seen = []
|
|
await be.fetch('http://x/y.mp3', target,
|
|
progress_cb=lambda done, total: seen.append((done, total)))
|
|
|
|
assert seen == [(50, 100), (100, 100)]
|
|
assert rpc.removed == [] # 完成不需要中止
|
|
assert not ctrl.exists() # 控制文件已兜底清理
|
|
assert rpc.added is not None
|
|
url, options = rpc.added
|
|
assert url == 'http://x/y.mp3'
|
|
assert options['continue'] == 'true' # 续传选项已下发
|
|
assert options['max-connection-per-server'] == str(a2._SPLIT)
|
|
assert options['out'] == 'y.part'
|
|
assert options['auto-file-renaming'] == 'false' # 与 .part 语义不冲突
|
|
|
|
|
|
async def test_progress_callback_exception_is_swallowed(monkeypatch, tmp_path):
|
|
be, _rpc = _backend(monkeypatch, [_st(status='complete', completedLength='100')])
|
|
|
|
def boom(_d, _t):
|
|
raise RuntimeError('cb')
|
|
|
|
# 不应抛出
|
|
await be.fetch('http://x/y.mp3', tmp_path / 'y.part', progress_cb=boom)
|
|
|
|
|
|
async def test_referer_is_forwarded_when_given(monkeypatch, tmp_path):
|
|
be, rpc = _backend(monkeypatch, [_st(status='complete', completedLength='100')])
|
|
await be.fetch('http://x/y.mp3', tmp_path / 'y.part', referer='http://ref/')
|
|
assert rpc.added is not None
|
|
assert rpc.added[1]['referer'] == 'http://ref/'
|
|
|
|
|
|
# ---------- 取消 / 守卫 / 失败 ----------
|
|
|
|
async def test_fetch_maps_cancel_and_drops_partial(monkeypatch, tmp_path):
|
|
be, rpc = _backend(monkeypatch, [_st(completedLength='10')])
|
|
target = tmp_path / 'y.part'
|
|
target.write_bytes(b'partial')
|
|
ctrl = tmp_path / 'y.part.aria2'
|
|
ctrl.write_bytes(b'aria2 ctrl')
|
|
|
|
ev = threading.Event()
|
|
ev.set()
|
|
with pytest.raises(DownloadCanceled):
|
|
await be.fetch('http://x/y.mp3', target, cancel_event=ev)
|
|
|
|
# 在途分支：先 forceRemove 停掉，再清结果表（与 aria2p 的 453-454 行一致）
|
|
assert rpc.removed == [('forceRemove', 'g1'),
|
|
('removeDownloadResult', 'g1')]
|
|
assert not target.exists() # 残片一并删除
|
|
assert not ctrl.exists() # 控制文件总是删除
|
|
|
|
|
|
async def test_fetch_guards_size_and_drops_partial(monkeypatch, tmp_path):
|
|
be, rpc = _backend(monkeypatch, [_st(completedLength='700')])
|
|
target = tmp_path / 'y.part'
|
|
target.write_bytes(b'x' * 10)
|
|
|
|
with pytest.raises(DownloadException):
|
|
await be.fetch('http://x/y.mp3', target, max_bytes=500)
|
|
|
|
assert rpc.removed == [('forceRemove', 'g1'),
|
|
('removeDownloadResult', 'g1')]
|
|
assert not target.exists()
|
|
|
|
|
|
async def test_size_guard_after_complete_clears_result_only(monkeypatch, tmp_path):
|
|
"""体积守卫在「完成」判定之前：任务已终结时只能清结果表，不是 forceRemove。
|
|
|
|
这条分支与 aria2p.remove() 的 if download.is_complete 分支对齐——
|
|
对已终结的 GID 调 forceRemove 会报错，故必须走 removeDownloadResult。
|
|
"""
|
|
be, rpc = _backend(monkeypatch, [_st(status='complete', completedLength='700')])
|
|
target = tmp_path / 'y.part'
|
|
target.write_bytes(b'x' * 10)
|
|
|
|
with pytest.raises(DownloadException):
|
|
await be.fetch('http://x/y.mp3', target, max_bytes=500)
|
|
|
|
assert rpc.removed == [('removeDownloadResult', 'g1')]
|
|
assert not target.exists()
|
|
|
|
|
|
async def test_fetch_failure_is_retryable_and_keeps_partial(monkeypatch, tmp_path):
|
|
be, rpc = _backend(monkeypatch, [
|
|
_st(status='error', completedLength='10', errorMessage='boom')])
|
|
target = tmp_path / 'y.part'
|
|
target.write_bytes(b'partial')
|
|
ctrl = tmp_path / 'y.part.aria2'
|
|
ctrl.write_bytes(b'aria2 ctrl')
|
|
|
|
with pytest.raises(UpstreamNetworkError):
|
|
await be.fetch('http://x/y.mp3', target)
|
|
|
|
assert rpc.removed == [] # 不调 _abort：残片与控制文件都保留，供下轮续传
|
|
assert target.exists()
|
|
assert ctrl.exists()
|
|
|
|
|
|
async def test_fetch_timeout_is_retryable_and_keeps_partial(monkeypatch, tmp_path):
|
|
be, rpc = _backend(monkeypatch, [_st(completedLength='10')]) # 永不完成
|
|
target = tmp_path / 'y.part'
|
|
target.write_bytes(b'partial')
|
|
ctrl = tmp_path / 'y.part.aria2'
|
|
ctrl.write_bytes(b'aria2 ctrl')
|
|
|
|
with pytest.raises(UpstreamNetworkError):
|
|
await be.fetch('http://x/y.mp3', target, timeout=-1)
|
|
|
|
assert rpc.removed == [('forceRemove', 'g1'),
|
|
('removeDownloadResult', 'g1')]
|
|
assert target.exists() # 超时保留残片（与 aria2p 的 files=False 对齐）
|
|
assert not ctrl.exists() # 但控制文件仍删（aria2p 的 clean=True 默认）
|
|
|
|
|
|
async def test_fetch_submit_failure_is_retryable(monkeypatch, tmp_path):
|
|
be, _rpc = _backend(monkeypatch, [], fail_add=True)
|
|
with pytest.raises(UpstreamNetworkError):
|
|
await be.fetch('http://x/y.mp3', tmp_path / 'y.part')
|
|
|
|
|
|
async def test_fetch_update_failure_is_retryable(monkeypatch, tmp_path):
|
|
be, _rpc = _backend(monkeypatch, [], fail_tell=True)
|
|
with pytest.raises(UpstreamNetworkError):
|
|
await be.fetch('http://x/y.mp3', tmp_path / 'y.part')
|
|
|
|
|
|
async def test_abort_failure_does_not_delete_partial(monkeypatch, tmp_path):
|
|
"""主移除动作失败时**不**删数据文件（与 aria2p 的 `files and result[-1]` 对齐）。"""
|
|
be, rpc = _backend(monkeypatch, [_st(completedLength='10')])
|
|
target = tmp_path / 'y.part'
|
|
target.write_bytes(b'partial')
|
|
|
|
async def boom(_gid):
|
|
raise RuntimeError('rpc down')
|
|
|
|
monkeypatch.setattr(rpc, 'force_remove', boom)
|
|
with pytest.raises(UpstreamNetworkError):
|
|
await be.fetch('http://x/y.mp3', target, timeout=-1)
|
|
|
|
assert target.exists()
|
|
|
|
|
|
def test_close_is_idempotent(monkeypatch):
|
|
be, _rpc = _backend(monkeypatch, [])
|
|
be.close()
|
|
be.close()
|
|
|
|
|
|
# ---------- JSON-RPC 信封 ----------
|
|
|
|
class _FakeHTTP:
|
|
"""捕获 request_once 的入参，并按预设响应体返回。"""
|
|
|
|
def __init__(self, payload):
|
|
self.payload = payload
|
|
self.calls: list[dict] = []
|
|
|
|
async def __call__(self, method, url, **kw):
|
|
self.calls.append({'method': method, 'url': url, **kw})
|
|
return HTTPResult(status=200, url=url, text=json.dumps(self.payload))
|
|
|
|
|
|
async def test_rpc_envelope_shape(monkeypatch):
|
|
"""信封必须是 aria2 要求的 {"jsonrpc","id","method","params":["token:…",…]}。"""
|
|
fake = _FakeHTTP({'jsonrpc': '2.0', 'id': '1', 'result': 'gid-1'})
|
|
monkeypatch.setattr(a2._http, 'request_once', fake)
|
|
|
|
rpc = a2._Aria2Rpc(6800, 'SEC')
|
|
gid = await rpc.add_uri('http://u/a.mp3', {'out': 'a.part'})
|
|
|
|
assert gid == 'gid-1'
|
|
assert len(fake.calls) == 1
|
|
call = fake.calls[0]
|
|
assert call['method'] == 'POST'
|
|
assert call['url'] == 'http://127.0.0.1:6800/jsonrpc'
|
|
body = call['json_body']
|
|
assert body['jsonrpc'] == '2.0'
|
|
assert body['method'] == 'aria2.addUri'
|
|
assert body['params'][0] == 'token:SEC' # secret 以 token: 前缀入参
|
|
assert body['params'][1] == ['http://u/a.mp3']
|
|
assert body['params'][2] == {'out': 'a.part'}
|
|
|
|
|
|
async def test_rpc_call_impl_does_not_retry(monkeypatch):
|
|
"""变更类 RPC 必须走 request_once（不重试），避免 addUri 被重复提交。"""
|
|
fake = _FakeHTTP({'jsonrpc': '2.0', 'id': '1', 'result': 'OK'})
|
|
monkeypatch.setattr(a2._http, 'request_once', fake)
|
|
rpc = a2._Aria2Rpc(6800, 'SEC')
|
|
await rpc.force_remove('g1')
|
|
assert len(fake.calls) == 1
|
|
|
|
|
|
async def test_rpc_error_object_becomes_rpc_error(monkeypatch):
|
|
"""aria2 的协议层错误是 HTTP 200 + error 对象，必须收敛成 Aria2RpcError。"""
|
|
fake = _FakeHTTP({'jsonrpc': '2.0', 'id': '1',
|
|
'error': {'code': 1, 'message': 'GID not found'}})
|
|
monkeypatch.setattr(a2._http, 'request_once', fake)
|
|
|
|
rpc = a2._Aria2Rpc(6800, 'SEC')
|
|
with pytest.raises(a2.Aria2RpcError) as ei:
|
|
await rpc.tell_status('nope')
|
|
assert 'GID not found' in str(ei.value)
|
|
|
|
|
|
async def test_rpc_non_object_response_is_rpc_error(monkeypatch):
|
|
fake = _FakeHTTP([1, 2, 3]) # 合法 JSON 但不是对象
|
|
monkeypatch.setattr(a2._http, 'request_once', fake)
|
|
rpc = a2._Aria2Rpc(6800, 'SEC')
|
|
with pytest.raises(a2.Aria2RpcError):
|
|
await rpc.get_version()
|