diff --git a/backend/src/module/downloader/__init__.py b/backend/src/module/downloader/__init__.py index 46e59cc3..0aa01bbb 100644 --- a/backend/src/module/downloader/__init__.py +++ b/backend/src/module/downloader/__init__.py @@ -3,11 +3,12 @@ from .base import ( DownloaderCapabilities, DownloaderClient, ) -from .download_client import DownloadClient +from .download_client import DownloadClient, shutdown __all__ = [ "CoreDownloaderClient", "DownloadClient", "DownloaderCapabilities", "DownloaderClient", + "shutdown", ] diff --git a/backend/src/module/downloader/client/aria2_downloader.py b/backend/src/module/downloader/client/aria2_downloader.py index b5c53942..4c457d47 100644 --- a/backend/src/module/downloader/client/aria2_downloader.py +++ b/backend/src/module/downloader/client/aria2_downloader.py @@ -24,6 +24,7 @@ class Aria2Downloader: self.host = host self.secret = password self._client: httpx.AsyncClient | None = None + self._authed = False self._rpc_url = f"{host}/jsonrpc" self._id = 0 @@ -46,13 +47,17 @@ class Aria2Downloader: return result.get("result") async def auth(self, retry=3): - self._client = httpx.AsyncClient( - timeout=httpx.Timeout(connect=3.1, read=10.0, write=10.0, pool=10.0) - ) + if self._client is not None and self._authed: + return True + if self._client is None: + self._client = httpx.AsyncClient( + timeout=httpx.Timeout(connect=3.1, read=10.0, write=10.0, pool=10.0) + ) times = 0 while times < retry: try: await self._call("getVersion") + self._authed = True return True except Exception as e: logger.warning( @@ -63,6 +68,7 @@ class Aria2Downloader: return False async def logout(self): + self._authed = False if self._client: await self._client.aclose() self._client = None diff --git a/backend/src/module/downloader/client/mock_downloader.py b/backend/src/module/downloader/client/mock_downloader.py index 77bb9eea..45fbe8ec 100644 --- a/backend/src/module/downloader/client/mock_downloader.py +++ b/backend/src/module/downloader/client/mock_downloader.py @@ -31,6 +31,7 @@ class MockDownloader: self._rules: dict[str, dict] = {} self._feeds: dict[str, dict] = {} self._categories: set[str] = {"Bangumi", "BangumiCollection"} + self._authed = False self._prefs = { "save_path": "/tmp/mock-downloads", "rss_auto_downloading_enabled": True, @@ -41,10 +42,14 @@ class MockDownloader: logger.debug("[MockDownloader] Initialized") async def auth(self, retry=3) -> bool: + # No real session; idempotent by construction, kept for parity with + # QbDownloader so session-reuse tests stay meaningful. + self._authed = True logger.debug("[MockDownloader] Auth successful (mocked)") return True async def logout(self): + self._authed = False logger.debug("[MockDownloader] Logout (mocked)") async def check_host(self) -> bool: diff --git a/backend/src/module/downloader/client/qb_downloader.py b/backend/src/module/downloader/client/qb_downloader.py index 9fd6977f..411fda97 100644 --- a/backend/src/module/downloader/client/qb_downloader.py +++ b/backend/src/module/downloader/client/qb_downloader.py @@ -29,13 +29,12 @@ class QbDownloader: self.password = password self.ssl = ssl self._client: httpx.AsyncClient | None = None + self._authed = False def _url(self, endpoint: str) -> str: return f"{self.host}/api/v2/{endpoint}" - async def auth(self, retry=3): - times = 0 - use_https = self.host.startswith("https://") + def _new_client(self) -> httpx.AsyncClient: timeout = httpx.Timeout(connect=5.0, read=10.0, write=10.0, pool=10.0) # Keepalive_expiry keeps idle TCP sockets short-lived so they can't # outlive a proxy / NAS idle-reap, which would otherwise surface as @@ -49,7 +48,17 @@ class QbDownloader: ) # Never verify certificates - self-signed certs are the norm for # home-server / NAS / Docker qBittorrent setups. - self._client = httpx.AsyncClient(timeout=timeout, limits=limits, verify=False) + return httpx.AsyncClient(timeout=timeout, limits=limits, verify=False) + + async def auth(self, retry=3): + # Session reuse: a live, already-authenticated client short-circuits so + # repeated operations don't re-login every cycle (#1039 / #900). + if self._client is not None and self._authed: + return True + times = 0 + use_https = self.host.startswith("https://") + if self._client is None: + self._client = self._new_client() while times < retry: try: resp = await self._client.post( @@ -64,6 +73,7 @@ class QbDownloader: if ( resp.status_code == 200 and resp.text.startswith("Ok") ) or resp.status_code == 204: + self._authed = True return True elif resp.status_code == 403: logger.error("Login refused by qBittorrent Server") @@ -98,6 +108,26 @@ class QbDownloader: break return False + async def _request(self, method: str, endpoint: str, **kwargs) -> httpx.Response: + """Issue a request, re-authenticating once if the session expired. + + qBittorrent answers 403 when the session cookie is no longer valid + (e.g. the server restarted or reaped the session). Since we now keep one + long-lived session, force a single re-login and retry the request once. + """ + resp = await self._client.request(method, self._url(endpoint), **kwargs) + if resp.status_code == 403: + self._authed = False + await self.auth() + resp = await self._client.request(method, self._url(endpoint), **kwargs) + return resp + + async def _get(self, endpoint: str, **kwargs) -> httpx.Response: + return await self._request("GET", endpoint, **kwargs) + + async def _post(self, endpoint: str, **kwargs) -> httpx.Response: + return await self._request("POST", endpoint, **kwargs) + async def logout(self): if self._client: try: @@ -110,10 +140,11 @@ class QbDownloader: logger.debug("[Downloader] Logout request failed (non-critical): %s", e) await self._client.aclose() self._client = None + self._authed = False async def check_host(self): try: - resp = await self._client.get(self._url("app/version")) + resp = await self._get("app/version") return resp.status_code == 200 except (httpx.ConnectError, httpx.RequestError): return False @@ -123,20 +154,20 @@ class QbDownloader: @qb_connect_failed_wait async def prefs_init(self, prefs): - resp = await self._client.post( - self._url("app/setPreferences"), + resp = await self._post( + "app/setPreferences", data={"json": json.dumps(prefs)}, ) return resp @qb_connect_failed_wait async def get_app_prefs(self): - resp = await self._client.get(self._url("app/preferences")) + resp = await self._get("app/preferences") return resp.json() async def add_category(self, category): - await self._client.post( - self._url("torrents/createCategory"), + await self._post( + "torrents/createCategory", data={"category": category, "savePath": ""}, ) @@ -149,14 +180,12 @@ class QbDownloader: params["category"] = category if tag: params["tag"] = tag - resp = await self._client.get(self._url("torrents/info"), params=params) + resp = await self._get("torrents/info", params=params) return resp.json() @qb_connect_failed_wait async def torrents_files(self, torrent_hash: str): - resp = await self._client.get( - self._url("torrents/files"), params={"hash": torrent_hash} - ) + resp = await self._get("torrents/files", params={"hash": torrent_hash}) return resp.json() async def add_torrents( @@ -195,8 +224,8 @@ class QbDownloader: max_retries = 3 for attempt in range(max_retries): try: - resp = await self._client.post( - self._url("torrents/add"), + resp = await self._post( + "torrents/add", data=data, files=files if files else None, ) @@ -214,15 +243,15 @@ class QbDownloader: raise async def get_torrents_by_tag(self, tag: str) -> list[dict]: - resp = await self._client.get(self._url("torrents/info"), params={"tag": tag}) + resp = await self._get("torrents/info", params={"tag": tag}) return resp.json() async def torrents_delete(self, hash, delete_files: bool = True) -> bool: # qBittorrent expects one pipe-joined "hashes" field; a Python list would # be form-encoded as repeated fields and silently ignored (#1046). hashes = "|".join(hash) if isinstance(hash, (list, tuple)) else hash - resp = await self._client.post( - self._url("torrents/delete"), + resp = await self._post( + "torrents/delete", data={"hashes": hashes, "deleteFiles": str(delete_files).lower()}, ) if resp.status_code != 200: @@ -235,14 +264,14 @@ class QbDownloader: return True async def torrents_pause(self, hashes: str): - await self._client.post( - self._url("torrents/pause"), + await self._post( + "torrents/pause", data={"hashes": hashes}, ) async def torrents_resume(self, hashes: str): - await self._client.post( - self._url("torrents/resume"), + await self._post( + "torrents/resume", data={"hashes": hashes}, ) @@ -250,8 +279,8 @@ class QbDownloader: self, torrent_hash, old_path, new_path, verify: bool = True ) -> bool: try: - resp = await self._client.post( - self._url("torrents/renameFile"), + resp = await self._post( + "torrents/renameFile", data={"hash": torrent_hash, "oldPath": old_path, "newPath": new_path}, ) if resp.status_code == 409: @@ -291,75 +320,73 @@ class QbDownloader: return False async def rss_add_feed(self, url, item_path): - resp = await self._client.post( - self._url("rss/addFeed"), + resp = await self._post( + "rss/addFeed", data={"url": url, "path": item_path}, ) if resp.status_code == 409: logger.warning(f"[Downloader] RSS feed {url} already exists") async def rss_remove_item(self, item_path): - resp = await self._client.post( - self._url("rss/removeItem"), + resp = await self._post( + "rss/removeItem", data={"path": item_path}, ) if resp.status_code == 409: logger.warning(f"[Downloader] RSS item {item_path} does not exist") async def rss_get_feeds(self): - resp = await self._client.get(self._url("rss/items")) + resp = await self._get("rss/items") return resp.json() async def rss_set_rule(self, rule_name, rule_def): - await self._client.post( - self._url("rss/setRule"), + await self._post( + "rss/setRule", data={"ruleName": rule_name, "ruleDef": json.dumps(rule_def)}, ) async def move_torrent(self, hashes, new_location): - await self._client.post( - self._url("torrents/setLocation"), + await self._post( + "torrents/setLocation", data={"hashes": hashes, "location": new_location}, ) async def get_download_rule(self): - resp = await self._client.get(self._url("rss/rules")) + resp = await self._get("rss/rules") return resp.json() async def get_torrent_path(self, _hash): - resp = await self._client.get( - self._url("torrents/info"), params={"hashes": _hash} - ) + resp = await self._get("torrents/info", params={"hashes": _hash}) torrents = resp.json() if torrents: return torrents[0].get("save_path", "") return "" async def set_category(self, _hash, category): - resp = await self._client.post( - self._url("torrents/setCategory"), + resp = await self._post( + "torrents/setCategory", data={"hashes": _hash, "category": category}, ) if resp.status_code == 409: logger.warning(f"[Downloader] Category {category} does not exist") await self.add_category(category) - await self._client.post( - self._url("torrents/setCategory"), + await self._post( + "torrents/setCategory", data={"hashes": _hash, "category": category}, ) async def check_connection(self): - resp = await self._client.get(self._url("app/version")) + resp = await self._get("app/version") return resp.text async def remove_rule(self, rule_name): - await self._client.post( - self._url("rss/removeRule"), + await self._post( + "rss/removeRule", data={"ruleName": rule_name}, ) async def add_tag(self, _hash, tag): - await self._client.post( - self._url("torrents/addTags"), + await self._post( + "torrents/addTags", data={"hashes": _hash, "tags": tag}, ) diff --git a/backend/src/module/downloader/download_client.py b/backend/src/module/downloader/download_client.py index 2c06278b..d2941bfe 100644 --- a/backend/src/module/downloader/download_client.py +++ b/backend/src/module/downloader/download_client.py @@ -10,21 +10,79 @@ from .rules import build_rss_rule logger = logging.getLogger(__name__) + +# --------------------------------------------------------------------------- +# Module-level concrete-client cache (session reuse, #1039 / #900) +# +# Every ``async with DownloadClient()`` used to spin up a fresh concrete client +# and log in on enter / log out on exit -- one qB login+logout per operation, +# roughly once a minute from the rename loop. The cache keeps a single concrete +# client alive across operations, keyed by the connection-relevant settings. +# When those settings change the old client is retired and re-closed lazily +# inside the next ``__aenter__`` (so aclose runs in an async context). +# --------------------------------------------------------------------------- +_client_cache: tuple[tuple, object] | None = None +_stale_client: object | None = None + # Warn at most once per (client type, operation) when a backend cannot perform # an operation, so aria2 users are not spammed every rename cycle. _warned_unsupported: set[tuple[str, str]] = set() +def _settings_key() -> tuple: + d = settings.downloader + return (d.type, d.host, d.username, d.password, d.ssl) + + +def _reset_client_cache() -> None: + """Drop the cached/stale concrete clients (used by tests).""" + global _client_cache, _stale_client + _client_cache = None + _stale_client = None + + +async def shutdown() -> None: + """Log out and close the cached concrete client. + + Not called anywhere yet -- the Phase 5 composition root will own the + downloader lifecycle and invoke this on application shutdown. + """ + global _client_cache, _stale_client + clients = [] + if _stale_client is not None: + clients.append(_stale_client) + if _client_cache is not None: + clients.append(_client_cache[1]) + _client_cache = None + _stale_client = None + for client in clients: + try: + await client.logout() + except Exception as e: # pragma: no cover - best-effort cleanup + logger.debug("[Downloader] Error closing client on shutdown: %s", e) + + class DownloadClient: """Unified async download client. Wraps qBittorrent, Aria2, or MockDownloader behind a common interface. Intended to be used as an async context manager; authentication is - performed on ``__aenter__`` and the session is closed on ``__aexit__``. + performed on ``__aenter__``. The concrete client's session is reused across + context-manager blocks (see the module-level cache above) and only torn + down by :func:`shutdown`. """ def __init__(self): - self.client = self.__getClient() + global _client_cache, _stale_client + key = _settings_key() + if _client_cache is not None and _client_cache[0] == key: + self.client = _client_cache[1] + else: + if _client_cache is not None: + # Settings changed: retire the previous client, close it later. + _stale_client = _client_cache[1] + self.client = self.__getClient() + _client_cache = (key, self.client) self.authed = False @staticmethod @@ -70,22 +128,30 @@ class DownloadClient: return False async def __aenter__(self): + global _stale_client, _client_cache + if _stale_client is not None: + stale, _stale_client = _stale_client, None + try: + await stale.logout() + except Exception as e: # pragma: no cover - best-effort cleanup + logger.debug("[Downloader] Error closing stale client: %s", e) if not self.authed: await self.auth() if not self.authed: # __aexit__ never runs when we raise here, so close the # concrete client's connection pool now or it leaks on every - # failed connect (#1043). + # failed connect (#1043). Also drop it from the cache so the + # next construction rebuilds a fresh client. await self.client.logout() + if _client_cache is not None and _client_cache[1] is self.client: + _client_cache = None raise ConnectionError("Download client authentication failed") - else: - logger.error("[Downloader] Already authed.") return self async def __aexit__(self, exc_type, exc_val, exc_tb): - if self.authed: - await self.client.logout() - self.authed = False + # The concrete session is reused across operations; do NOT log out here. + # Teardown is deferred to shutdown() (Phase 5 composition root). + self.authed = False async def auth(self): self.authed = await self.client.auth() diff --git a/backend/src/test/conftest.py b/backend/src/test/conftest.py index 61c13c73..3f49b88b 100644 --- a/backend/src/test/conftest.py +++ b/backend/src/test/conftest.py @@ -32,6 +32,21 @@ def _clear_bangumi_cache(): _invalidate_bangumi_cache() +@pytest.fixture(autouse=True) +def _reset_downloader_client_cache(): + """Clear the module-level concrete-client cache between tests. + + DownloadClient reuses one concrete client across context-manager blocks + (session reuse, #1039 / #900); the cache would otherwise leak a mock from + one test into the next. + """ + from module.downloader.download_client import _reset_client_cache + + _reset_client_cache() + yield + _reset_client_cache() + + @pytest.fixture def db_engine(): """Create an in-memory SQLite engine for testing.""" diff --git a/backend/src/test/test_download_client.py b/backend/src/test/test_download_client.py index bc10e2e5..0e1e0620 100644 --- a/backend/src/test/test_download_client.py +++ b/backend/src/test/test_download_client.py @@ -365,9 +365,15 @@ class TestContextManagerAuth: assert result is download_client assert download_client.authed is True - async def test_aexit_calls_logout_when_authed(self, download_client, mock_qb_client): - """__aexit__ calls logout and resets authed when session was active.""" + async def test_aexit_reuses_session_without_logout( + self, download_client, mock_qb_client + ): + """__aexit__ resets authed but keeps the session (no logout). + + The concrete client is reused across operations (#1039 / #900); + teardown is deferred to shutdown(). + """ download_client.authed = True await download_client.__aexit__(None, None, None) - mock_qb_client.logout.assert_called_once() + mock_qb_client.logout.assert_not_called() assert download_client.authed is False diff --git a/backend/src/test/test_download_session_reuse.py b/backend/src/test/test_download_session_reuse.py new file mode 100644 index 00000000..3f62d876 --- /dev/null +++ b/backend/src/test/test_download_session_reuse.py @@ -0,0 +1,148 @@ +"""Regression tests for qBittorrent session reuse (#1039 / #900). + +The concrete client is cached across ``async with DownloadClient()`` blocks so +we log in once, re-authenticate only when the session expires, and rebuild when +connection settings change. +""" + +from unittest.mock import AsyncMock, MagicMock, patch + +import pytest + +from module.downloader.client.qb_downloader import QbDownloader +from module.downloader.download_client import DownloadClient + + +def _patch_qb_settings(mock_settings, host="localhost:8080"): + mock_settings.downloader.type = "qbittorrent" + mock_settings.downloader.host = host + mock_settings.downloader.username = "admin" + mock_settings.downloader.password = "admin" + mock_settings.downloader.ssl = False + + +class _FakeHTTP: + """Minimal httpx.AsyncClient stand-in counting logins and data requests.""" + + def __init__(self, login_calls: list, request_calls: list): + self._login_calls = login_calls + self._request_calls = request_calls + + async def post(self, url, data=None): + # auth/login and auth/logout both arrive via post(); count logins only. + if url.endswith("auth/login"): + self._login_calls.append(url) + resp = MagicMock() + resp.status_code = 200 + resp.text = "Ok." + return resp + + async def request(self, method, url, **kwargs): + self._request_calls.append((method, url)) + resp = MagicMock() + resp.status_code = 200 + resp.text = "Ok." + resp.json = lambda: [] + return resp + + async def aclose(self): + pass + + +class TestSessionReuse: + async def test_two_blocks_share_one_login(self): + """Two sequential ``async with DownloadClient()`` blocks log in once.""" + login_calls: list = [] + request_calls: list = [] + + def make_http(*args, **kwargs): + return _FakeHTTP(login_calls, request_calls) + + with patch("module.downloader.download_client.settings") as mock_settings: + _patch_qb_settings(mock_settings) + with patch( + "module.downloader.client.qb_downloader.httpx.AsyncClient", + side_effect=make_http, + ): + async with DownloadClient(): + pass + async with DownloadClient(): + pass + + assert len(login_calls) == 1 + + async def test_expired_session_reauths_once_and_retries(self): + """A 403 mid-operation forces exactly one re-login, then the op succeeds.""" + qb = QbDownloader("localhost:8080", "admin", "admin", ssl=False) + login_calls: list = [] + request_calls: list = [] + + class _Fake403Once: + async def post(self, url, data=None): + login_calls.append(url) + resp = MagicMock() + resp.status_code = 200 + resp.text = "Ok." + return resp + + async def request(self, method, url, **kwargs): + request_calls.append((method, url)) + resp = MagicMock() + # First data request hits an expired session; retry succeeds. + resp.status_code = 403 if len(request_calls) == 1 else 200 + resp.json = lambda: [] + return resp + + async def aclose(self): + pass + + with patch( + "module.downloader.client.qb_downloader.httpx.AsyncClient", + return_value=_Fake403Once(), + ): + await qb.auth() + resp = await qb._get("torrents/info") + + assert resp.status_code == 200 + assert len(login_calls) == 2 # initial login + one forced re-login + assert len(request_calls) == 2 # original request + one retry + + async def test_settings_change_creates_new_client(self): + """Changing the connection settings retires the cached client.""" + with patch("module.downloader.download_client.settings") as mock_settings: + _patch_qb_settings(mock_settings, host="host-a:8080") + first = DownloadClient().client + reused = DownloadClient().client # same settings -> same client + _patch_qb_settings(mock_settings, host="host-b:8080") + rebuilt = DownloadClient().client # settings changed -> new client + + assert first is reused + assert first is not rebuilt + + async def test_no_login_idempotency_on_cheap_reauth(self): + """QbDownloader.auth() short-circuits when already authenticated.""" + qb = QbDownloader("localhost:8080", "admin", "admin", ssl=False) + login_calls: list = [] + + def make_http(*args, **kwargs): + client = MagicMock() + + async def _post(url, data=None): + login_calls.append(url) + resp = MagicMock() + resp.status_code = 200 + resp.text = "Ok." + return resp + + client.post = _post + client.aclose = AsyncMock() + return client + + with patch( + "module.downloader.client.qb_downloader.httpx.AsyncClient", + side_effect=make_http, + ): + assert await qb.auth() is True + assert await qb.auth() is True # cheap, no second login + + assert len(login_calls) == 1 diff --git a/backend/src/test/test_qb_downloader.py b/backend/src/test/test_qb_downloader.py index 40d7ed72..15d81b30 100644 --- a/backend/src/test/test_qb_downloader.py +++ b/backend/src/test/test_qb_downloader.py @@ -625,12 +625,13 @@ class TestTorrentsDelete: qb._client = AsyncMock() mock_resp = MagicMock() mock_resp.status_code = 200 - qb._client.post = AsyncMock(return_value=mock_resp) + # Data ops now go through _request -> _client.request (session reuse). + qb._client.request = AsyncMock(return_value=mock_resp) result = await qb.torrents_delete(["aaa", "bbb"], delete_files=True) assert result is True - sent = qb._client.post.call_args.kwargs["data"] + sent = qb._client.request.call_args.kwargs["data"] assert sent["hashes"] == "aaa|bbb" assert sent["deleteFiles"] == "true" @@ -639,8 +640,8 @@ class TestTorrentsDelete: qb = QbDownloader(host="localhost:8080", username="u", password="p", ssl=False) qb._client = AsyncMock() mock_resp = MagicMock() - mock_resp.status_code = 403 - qb._client.post = AsyncMock(return_value=mock_resp) + mock_resp.status_code = 500 + qb._client.request = AsyncMock(return_value=mock_resp) result = await qb.torrents_delete("aaa", delete_files=True)