diff --git a/app/modules/transmission/__init__.py b/app/modules/transmission/__init__.py index a28cb3c43..d4363b060 100644 --- a/app/modules/transmission/__init__.py +++ b/app/modules/transmission/__init__.py @@ -254,7 +254,7 @@ class TransmissionModule(_DownloaderModuleBase[Transmission]): server: Transmission, torrent_hash: str, episodes: Set[int], custom_words: Optional[str], ) -> Tuple[bool, str]: """ - 先确认任务暂停再按来源集数映射选集，兼容 RPC 返回正在运行的重复任务。 + 先确认任务暂停再按来源集数映射选集，确认启动状态后才返回成功。 :param server: 已配置的 Transmission 客户端 :param torrent_hash: 需要暂停、选集并恢复的任务哈希 @@ -262,7 +262,7 @@ class TransmissionModule(_DownloaderModuleBase[Transmission]): :param custom_words: 来源识别词，按行参与物理文件名识别 :return: 是否确认选集及启动成功，以及失败时的下载器核查说明 """ - if not server.stop_torrents(torrent_hash): + if not server.ensure_torrent_state(torrent_hash, paused=True): return False, f"选集失败，任务 {torrent_hash} 无法确认暂停，未执行选集，请在下载器核实" failure = f"选集失败，任务 {torrent_hash} 保持暂停，请在下载器检查" torrent_files = server.get_files(torrent_hash) @@ -286,8 +286,8 @@ class TransmissionModule(_DownloaderModuleBase[Transmission]): return False, f"{failure}：未找到目标集数 {sorted(set(episodes) - matched)}" if not server.set_files(torrent_hash, wanted) or not server.set_unwanted_files(torrent_hash, unwanted): return False, f"{failure}：文件选择设置未确认" - if not server.start_torrents(torrent_hash): - if server.stop_torrents(torrent_hash): + if not server.ensure_torrent_state(torrent_hash, paused=False): + if server.ensure_torrent_state(torrent_hash, paused=True): return False, f"{failure}：启动任务未确认，未记录成功" return False, f"选集失败，任务 {torrent_hash} 启动任务未确认，请在下载器核实，未记录成功" return True, "添加下载任务成功" diff --git a/app/modules/transmission/transmission.py b/app/modules/transmission/transmission.py index 25ff17c45..41709b35b 100755 --- a/app/modules/transmission/transmission.py +++ b/app/modules/transmission/transmission.py @@ -2,6 +2,7 @@ Transmission 客户端连接与下载任务操作封装。 """ +import time from typing import List, Optional, Sequence, Tuple, Union import transmission_rpc @@ -13,6 +14,14 @@ from app.runtime.log import logger TorrentIds = Union[str, int, Sequence[Union[str, int]]] +# 每次启停的状态确认窗口与轮询间隔，单位均为秒。 +_TORRENT_STATE_TIMEOUT = 5.0 +_TORRENT_STATE_POLL_INTERVAL = 0.5 +# 启动后允许等待校验、下载或做种队列，不要求已经产生下载流量。 +_TORRENT_STARTED_STATES = { + "check pending", "checking", "download pending", "downloading", "seed pending", "seeding", +} + class Transmission: """ @@ -293,6 +302,43 @@ class Transmission: logger.error(f"停止种子出错：{str(err)}") return False + def ensure_torrent_state(self, tid: str, *, paused: bool) -> bool: + """ + 在选集边界确认任务实际暂停或启动，避免异步停止覆盖后续启动。 + + 已暂停的任务不再发送停止请求。启动后只在仍停止时重试启动， + 校验、下载、做种及对应排队状态均视为已接管，不强制跳过队列。 + + :param tid: 需要确认状态的单个种子哈希 + :param paused: 为 True 时等待实际停止，为 False 时请求启动并确认状态 + :return: 状态已确认时返回 True；未连接、RPC 失败或五秒内未确认时返回 False + """ + if not self.trc: + return False + if not paused and not self.start_torrents(tid): + return False + expected = {"stopped"} if paused else _TORRENT_STARTED_STATES + deadline = time.monotonic() + _TORRENT_STATE_TIMEOUT + stop_requested = False + while (remaining := deadline - time.monotonic()) > 0: + try: + status = self.trc.get_torrent(tid, arguments=["status"], timeout=remaining).status + except Exception as err: + logger.error(f"确认种子 {tid} 启停状态出错：{str(err)}") + return False + if status in expected: + return True + if paused: + if not stop_requested and not self.stop_torrents(tid): + return False + stop_requested = True + elif status == "stopped" and not self.start_torrents(tid): + return False + time.sleep(min(_TORRENT_STATE_POLL_INTERVAL, max(0, deadline - time.monotonic()))) + action = "暂停" if paused else "启动" + logger.warning(f"种子 {tid} 在 {_TORRENT_STATE_TIMEOUT} 秒内未确认{action}状态") + return False + def delete_torrents(self, delete_file: Optional[bool], ids: TorrentIds) -> bool: """ 删除指定种子任务，并按选项删除下载文件。 diff --git a/tests/test_downloader_submission.py b/tests/test_downloader_submission.py index ac90192bb..f4f6c713c 100644 --- a/tests/test_downloader_submission.py +++ b/tests/test_downloader_submission.py @@ -10,6 +10,7 @@ from unittest.mock import MagicMock, Mock import pytest from torrentool.bencode import Bencode +import app.modules.transmission.transmission as transmission_client from app.chain.download.facade import DownloadChain from app.domain.context import Context, MediaInfo, TorrentInfo from app.domain.metainfo import MetaInfo @@ -68,6 +69,10 @@ def downloader(request): ) for method in ("set_files", "set_unwanted_files", "start_torrents", "stop_torrents"): getattr(server, method).return_value = True + server.ensure_torrent_state.side_effect = ( + lambda torrent_hash, *, paused: + server.stop_torrents(torrent_hash) if paused else server.start_torrents(torrent_hash) + ) module = request.param.__new__(request.param) module.get_instance = Mock(return_value=server) module.normalize_path = Mock(side_effect=lambda path, _name: path) @@ -151,12 +156,140 @@ def test_transmission_all_files_wanted_starts_without_unwanted_rpc(downloader, f server.start_torrents.assert_called_once_with(torrent_hash) +def _use_transmission_rpc(module, server) -> Transmission: + """ + 保留真实客户端的启停逻辑，仅替换 RPC 和种子文件读取边界。 + """ + client = Transmission.__new__(Transmission) + client.trc = Mock() + client.add_torrent = server.add_torrent + client.get_files = server.get_files + module.get_instance.return_value = client + return client + + +@pytest.fixture +def transmission_clock(monkeypatch): + """ + 在真实模块和客户端的联动测试中推进虚拟轮询时间。 + """ + clock = SimpleNamespace(now=0.0) + + def advance(seconds: float) -> None: + """ + 按客户端要求推进等待时间。 + """ + clock.now += seconds + + monkeypatch.setattr(transmission_client, "time", SimpleNamespace(monotonic=lambda: clock.now, sleep=advance)) + return clock + + +def test_transmission_new_paused_task_starts_without_stop_rpc(downloader) -> None: + """ + 已暂停的新任务无需再次停止，防止停止请求覆盖选集后的启动请求。 + """ + module, server, content, torrent_hash = downloader + if not isinstance(module, TransmissionModule): + pytest.skip("Transmission 的异步启停状态确认") + client = _use_transmission_rpc(module, server) + state = {"status": "stopped"} + client.trc.get_torrent.side_effect = lambda *_args, **_kwargs: SimpleNamespace(status=state["status"]) + client.trc.start_torrent.side_effect = lambda **_kwargs: state.update(status="download pending") + + result = module.download(content, Path("/downloads"), "", episodes={170}, custom_words=_CUSTOM_WORDS) + + assert result[1] == torrent_hash + assert client.add_torrent.call_args.kwargs["is_paused"] is True + client.trc.stop_torrent.assert_not_called() + client.trc.start_torrent.assert_called_once_with(ids=torrent_hash) + + +def test_transmission_stopped_after_accepted_start_never_settles_success(downloader, transmission_clock) -> None: + """ + 启动 RPC 成功而任务持续停止时，下载链必须失败结算，不能记录订阅下载成功。 + """ + module, server, content, _torrent_hash = downloader + if not isinstance(module, TransmissionModule): + pytest.skip("Transmission 的异步启停状态确认") + client = _use_transmission_rpc(module, server) + client.trc.get_torrent.return_value = SimpleNamespace(status="stopped") + chain = DownloadChain.__new__(DownloadChain) + chain.run_module = lambda method, **kwargs: getattr(module, method)(**kwargs) + chain._settle_accepted_download = Mock() + chain._record_rejected_download = Mock() + + result, message = _submit(chain, content, {170}, _CUSTOM_WORDS) + + assert result is None + assert "启动任务未确认" in message and "未记录成功" in message + assert transmission_clock.now == transmission_client._TORRENT_STATE_TIMEOUT + chain._settle_accepted_download.assert_not_called() + chain._record_rejected_download.assert_called_once() + + +@pytest.mark.parametrize("pause_completes", [True, False]) +def test_transmission_async_pause_guards_file_selection(downloader, transmission_clock, pause_completes) -> None: + """ + 重复任务异步暂停前不得选集；暂停超时也不能启动或记录订阅成功。 + """ + module, server, content, torrent_hash = downloader + if not isinstance(module, TransmissionModule): + pytest.skip("Transmission 的异步启停状态确认") + client = _use_transmission_rpc(module, server) + state = {"status": "downloading", "stop_queries": 0} + list_files = client.get_files.side_effect + + def read_status(*_args, **_kwargs): + """ + 模拟暂停 RPC 返回后，任务仍需两次查询才能实际停止。 + """ + if client.trc.stop_torrent.called: + state["stop_queries"] += 1 + if pause_completes and state["stop_queries"] >= 2: + state["status"] = "stopped" + if client.trc.start_torrent.called: + state["status"] = "download pending" + return SimpleNamespace(status=state["status"]) + + def read_paused_files(*args, **kwargs): + """ + 文件选择只能发生在实际暂停和下一次启动之间。 + """ + assert state["status"] == "stopped" + client.trc.start_torrent.assert_not_called() + return list_files(*args, **kwargs) + + client.trc.get_torrent.side_effect = read_status + client.get_files.side_effect = read_paused_files + chain = DownloadChain.__new__(DownloadChain) + chain.run_module = lambda method, **kwargs: getattr(module, method)(**kwargs) + chain._settle_accepted_download = Mock() + chain._record_rejected_download = Mock() + + result, message = _submit(chain, content, {170}, _CUSTOM_WORDS) + + client.trc.stop_torrent.assert_called_once_with(ids=torrent_hash) + assert transmission_clock.now > 0 + if pause_completes: + assert result == torrent_hash + chain._settle_accepted_download.assert_called_once() + client.trc.start_torrent.assert_called_once_with(ids=torrent_hash) + else: + assert result is None and "无法确认暂停" in message + client.get_files.assert_not_called() + client.trc.change_torrent.assert_not_called() + client.trc.start_torrent.assert_not_called() + chain._settle_accepted_download.assert_not_called() + + def test_transmission_empty_episode_selection_never_starts_task() -> None: """ 选集入口没有目标集数时应保留暂停状态，不把空数组转换为全部文件。 """ server = Mock() server.stop_torrents.return_value = True + server.ensure_torrent_state.return_value = True server.get_files.return_value = [SimpleNamespace(id=0, name="Show.S01E01.mkv")] selected, message = TransmissionModule._select_download_files(server, "hash", set(), None) diff --git a/tests/test_transmission_client_state.py b/tests/test_transmission_client_state.py index 53d30fe64..c61ce359f 100644 --- a/tests/test_transmission_client_state.py +++ b/tests/test_transmission_client_state.py @@ -2,14 +2,34 @@ Transmission 未配置状态和 SDK 参数转发的兼容边界。 """ +from types import SimpleNamespace from unittest.mock import Mock import pytest +from transmission_rpc import Torrent import app.modules.transmission.transmission as client_module from app.modules.transmission.transmission import Transmission +@pytest.fixture +def state_clock(monkeypatch): + """ + 使用虚拟时钟验证启停等待上限，不让回归测试实际等待五秒。 + """ + clock = SimpleNamespace(now=0.0) + + def advance(seconds: float) -> None: + """ + 记录每次轮询等待的时间。 + """ + assert seconds > 0 + clock.now += seconds + + monkeypatch.setattr(client_module, "time", SimpleNamespace(monotonic=lambda: clock.now, sleep=advance)) + return clock + + @pytest.mark.parametrize("host", [None, "", "invalid-address"]) @pytest.mark.parametrize("rename_partial_files", [None, False, True]) def test_missing_address_keeps_state_and_reconnect_safe(monkeypatch, host, rename_partial_files): @@ -113,3 +133,126 @@ def test_file_selection_without_connection_keeps_failure(method, file_ids) -> No client.trc = None assert getattr(client, method)("hash", file_ids) is False + + +@pytest.mark.parametrize("paused", [False, True]) +def test_state_confirmation_without_connection_fails(paused) -> None: + """ + 未连接时不能把启停状态解释为成功。 + """ + client = Transmission.__new__(Transmission) + client.trc = None + + assert client.ensure_torrent_state("hash", paused=paused) is False + + +@pytest.mark.parametrize("status", [1, 2, 3, 4, 5, 6]) +def test_started_state_accepts_sdk_active_and_queued_statuses(state_clock, status) -> None: + """ + 启动请求后校验、下载及做种的活动或排队状态均可确认，不要求实际流量。 + """ + client = Transmission.__new__(Transmission) + client.trc = Mock() + client.trc.get_torrent.return_value = Torrent(fields={"id": 1, "status": status}) + + assert client.ensure_torrent_state("hash", paused=False) is True + + client.trc.start_torrent.assert_called_once_with(ids="hash") + client.trc.stop_torrent.assert_not_called() + assert state_clock.now == 0 + assert client.trc.get_torrent.call_args.kwargs["arguments"] == ["status"] + + +def test_pause_waits_for_async_stop_to_complete(state_clock) -> None: + """ + 停止请求已接受但仍在运行时继续等待，确认暂停前不能返回成功。 + """ + client = Transmission.__new__(Transmission) + client.trc = Mock() + client.trc.get_torrent.side_effect = [ + SimpleNamespace(status=status) for status in ["downloading", "downloading", "stopped"] + ] + + assert client.ensure_torrent_state("hash", paused=True) is True + + client.trc.stop_torrent.assert_called_once_with(ids="hash") + client.trc.start_torrent.assert_not_called() + assert state_clock.now > 0 + + +def test_start_retries_stopped_task_until_queue_accepts_it(state_clock) -> None: + """ + 启动后仍停止时重发启动，进入下载队列后立即确认，保留正常排队语义。 + """ + client = Transmission.__new__(Transmission) + client.trc = Mock() + client.trc.get_torrent.side_effect = [ + SimpleNamespace(status=status) for status in ["stopped", "download pending"] + ] + + assert client.ensure_torrent_state("hash", paused=False) is True + + assert client.trc.start_torrent.call_count == 2 + assert state_clock.now > 0 + + +def test_start_retry_rpc_failure_never_confirms_success(state_clock) -> None: + """ + 首次启动已接受但重试断连时，仍须返回失败而非沿用首次 RPC 的成功。 + """ + client = Transmission.__new__(Transmission) + client.trc = Mock() + client.trc.get_torrent.return_value = SimpleNamespace(status="stopped") + client.trc.start_torrent.side_effect = [None, ConnectionError("重试启动失败")] + + assert client.ensure_torrent_state("hash", paused=False) is False + assert state_clock.now == 0 + + +@pytest.mark.parametrize("paused,status", [(True, "downloading"), (False, "stopped")]) +def test_state_confirmation_timeout_is_bounded(state_clock, paused, status) -> None: + """ + RPC 接受命令但状态始终未改变时，在确认窗口内结束并报告失败。 + """ + client = Transmission.__new__(Transmission) + client.trc = Mock() + client.trc.get_torrent.return_value = SimpleNamespace(status=status) + + assert client.ensure_torrent_state("hash", paused=paused) is False + + assert state_clock.now == client_module._TORRENT_STATE_TIMEOUT + assert all(0 < call.kwargs["timeout"] <= client_module._TORRENT_STATE_TIMEOUT + for call in client.trc.get_torrent.call_args_list) + if paused: + client.trc.stop_torrent.assert_called_once_with(ids="hash") + else: + assert client.trc.start_torrent.call_count > 1 + + +@pytest.mark.parametrize("paused", [False, True]) +@pytest.mark.parametrize("error", [KeyError("任务不存在"), ConnectionError("RPC 查询失败")]) +def test_state_query_failure_never_confirms_success(state_clock, paused, error) -> None: + """ + 查询失败或任务已消失时返回失败，不能把 RPC 接受命令当成状态确认。 + """ + client = Transmission.__new__(Transmission) + client.trc = Mock() + client.trc.get_torrent.side_effect = error + + assert client.ensure_torrent_state("hash", paused=paused) is False + assert state_clock.now == 0 + + +@pytest.mark.parametrize("paused", [False, True]) +def test_state_change_rpc_failure_never_confirms_success(state_clock, paused) -> None: + """ + 启停命令失败时立即返回，暂停失败不应继续查询或重发启动。 + """ + client = Transmission.__new__(Transmission) + client.trc = Mock() + client.trc.get_torrent.return_value = SimpleNamespace(status="downloading") + command = client.trc.stop_torrent if paused else client.trc.start_torrent + command.side_effect = ConnectionError("RPC 启停失败") + + assert client.ensure_torrent_state("hash", paused=paused) is False + assert state_clock.now == 0