Developer
7e53edfcc6
qBittorrent (or its reverse proxy) was returning 204 No Content with an SID
cookie and no body on a successful login, but the client only treated the
response as success if body == "Ok." or the cookie was in the parsed
requests cookie jar (resp.cookies.get("SID")). In the reported case the
Set-Cookie header was present (set-cookie=yes) yet the jar was empty —
requests doesn't always populate the jar from such headers (proxy-set/oddly-
attributed cookies) — so a valid login was reported as "Unexpected response".
The user re-entering correct credentials never helped.
Detect a successful login from EITHER the parsed jar OR the raw Set-Cookie
header (cookie name == SID). qBittorrent only sets SID on a valid login, so
this remains authoritative. The diagnostic now lists the cookie names it saw
for future-proofing.
New regression test reproduces the exact 204 + Set-Cookie SID + empty jar
case. 387/387 backend tests pass; ruff clean.
215 lines
8.7 KiB
Python
215 lines
8.7 KiB
Python
"""Unit tests for the QbittorrentClient."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import unittest
|
|
from unittest.mock import MagicMock, patch
|
|
|
|
import requests
|
|
|
|
from media_library_viewer_api.clients.qbittorrent import QbittorrentClient
|
|
|
|
|
|
class QbittorrentClientTests(unittest.TestCase):
|
|
def setUp(self) -> None:
|
|
self.client = QbittorrentClient("https://qb.example.com", "admin", "secret", timeout=5)
|
|
self.session = MagicMock()
|
|
self.client._session = self.session
|
|
|
|
def _login_response(self, text: str = "Ok.") -> MagicMock:
|
|
resp = MagicMock()
|
|
resp.text = text
|
|
resp.raise_for_status.return_value = None
|
|
resp.status_code = 200
|
|
resp.headers = {}
|
|
resp.cookies = {}
|
|
return resp
|
|
|
|
def _get_response(self, json_data: dict, status_code: int = 200) -> MagicMock:
|
|
resp = MagicMock()
|
|
resp.json.return_value = json_data
|
|
resp.raise_for_status.return_value = None
|
|
resp.status_code = status_code
|
|
resp.text = ""
|
|
return resp
|
|
|
|
def test_base_url_appends_api_v2(self) -> None:
|
|
c = QbittorrentClient("https://qb.example.com", "u", "p")
|
|
self.assertEqual(c.base_url, "https://qb.example.com/api/v2")
|
|
|
|
def test_base_url_keeps_existing_api_v2(self) -> None:
|
|
c = QbittorrentClient("https://qb.example.com/api/v2", "u", "p")
|
|
self.assertEqual(c.base_url, "https://qb.example.com/api/v2")
|
|
|
|
def test_base_url_strips_trailing_slash(self) -> None:
|
|
c = QbittorrentClient("https://qb.example.com/", "u", "p")
|
|
self.assertEqual(c.base_url, "https://qb.example.com/api/v2")
|
|
|
|
def test_empty_base_url_raises(self) -> None:
|
|
with self.assertRaises(ValueError):
|
|
QbittorrentClient("", "u", "p")
|
|
|
|
def test_empty_username_raises(self) -> None:
|
|
with self.assertRaises(ValueError):
|
|
QbittorrentClient("https://qb.example.com", "", "p")
|
|
|
|
def test_login_posts_credentials(self) -> None:
|
|
self.session.post.return_value = self._login_response("Ok.")
|
|
self.client._login()
|
|
self.session.post.assert_called_once()
|
|
call_args = self.session.post.call_args
|
|
self.assertIn("/auth/login", call_args.args[0])
|
|
self.assertEqual(call_args.kwargs["data"], {"username": "admin", "password": "secret"})
|
|
self.assertTrue(self.client._logged_in)
|
|
|
|
def test_login_failure_raises_runtime_error(self) -> None:
|
|
self.session.post.return_value = self._login_response("Fails.")
|
|
with self.assertRaises(RuntimeError):
|
|
self.client._login()
|
|
|
|
def test_get_auto_logs_in_on_first_call(self) -> None:
|
|
"""First _get triggers login, then fetches data."""
|
|
self.session.post.return_value = self._login_response("Ok.")
|
|
self.session.get.return_value = self._get_response({"server_state": {}, "torrents": {}})
|
|
|
|
result = self.client._get("/sync/maindata")
|
|
|
|
self.session.post.assert_called_once() # login happened
|
|
self.assertEqual(result, {"server_state": {}, "torrents": {}})
|
|
|
|
def test_cookie_reuse_does_not_re_login(self) -> None:
|
|
"""After login, subsequent _get calls do NOT re-login."""
|
|
self.client._logged_in = True # simulate already logged in
|
|
self.session.get.return_value = self._get_response({"data": 1})
|
|
|
|
self.client._get("/some/path")
|
|
|
|
self.session.post.assert_not_called() # no re-login
|
|
|
|
def test_403_triggers_re_login(self) -> None:
|
|
"""A 403 response triggers re-login and retries the GET."""
|
|
self.client._logged_in = True # already logged in from a prior call
|
|
forbidden = MagicMock()
|
|
forbidden.status_code = 403
|
|
ok = self._get_response({"server_state": {}, "torrents": {}})
|
|
self.session.get.side_effect = [forbidden, ok]
|
|
self.session.post.return_value = self._login_response("Ok.")
|
|
|
|
result = self.client._get("/sync/maindata")
|
|
|
|
self.assertEqual(self.session.get.call_count, 2) # initial + retry
|
|
self.session.post.assert_called_once() # re-login happened
|
|
self.assertEqual(result, {"server_state": {}, "torrents": {}})
|
|
|
|
def test_maindata_returns_full_payload(self) -> None:
|
|
self.client._logged_in = True
|
|
payload = {
|
|
"server_state": {"dl_info_speed": 12345, "up_info_speed": 6789},
|
|
"torrents": {
|
|
"abc": {"name": "Movie.mkv", "state": "downloading", "progress": 0.5},
|
|
"def": {"name": "Show.mkv", "state": "uploading", "progress": 1.0},
|
|
},
|
|
}
|
|
self.session.get.return_value = self._get_response(payload)
|
|
|
|
result = self.client.maindata()
|
|
|
|
self.assertEqual(result["server_state"]["dl_info_speed"], 12345)
|
|
self.assertEqual(len(result["torrents"]), 2)
|
|
|
|
@patch("media_library_viewer_api.clients.qbittorrent.requests.Session")
|
|
def test_login_http_error_propagates(self, mock_session_cls: MagicMock) -> None:
|
|
"""A network error during login propagates as requests exception."""
|
|
mock_session = MagicMock()
|
|
mock_session_cls.return_value = mock_session
|
|
bad_resp = MagicMock()
|
|
bad_resp.raise_for_status.side_effect = requests.ConnectionError("refused")
|
|
bad_resp.text = ""
|
|
mock_session.post.return_value = bad_resp
|
|
|
|
client = QbittorrentClient("https://qb.example.com", "u", "p")
|
|
with self.assertRaises(requests.ConnectionError):
|
|
client._login()
|
|
|
|
def test_qbittorrent_client_uses_tuple_timeout(self) -> None:
|
|
"""The client passes a (connect, read) tuple to requests, not an int."""
|
|
# self.client was constructed with timeout=5 in setUp() and has a mocked session.
|
|
assert isinstance(self.client.timeout, tuple)
|
|
assert len(self.client.timeout) == 2
|
|
assert self.client.timeout[0] == 5.0 # connect timeout
|
|
assert self.client.timeout[1] == 5.0 # read timeout (what we passed)
|
|
|
|
# Verify it's actually passed to requests as-is.
|
|
self.session.post.return_value = self._login_response()
|
|
self.client._login()
|
|
call_kwargs = self.session.post.call_args.kwargs
|
|
assert call_kwargs["timeout"] == (5.0, 5.0)
|
|
assert not isinstance(call_kwargs["timeout"], int)
|
|
|
|
def test_login_fails_message_names_bad_credentials(self) -> None:
|
|
"""'Fails.' body yields a clear 'invalid username or password' error."""
|
|
self.session.post.return_value = self._login_response("Fails.")
|
|
with self.assertRaises(RuntimeError) as ctx:
|
|
self.client._login()
|
|
self.assertIn("invalid username or password", str(ctx.exception))
|
|
|
|
def test_login_accepts_sid_cookie_when_body_mangled(self) -> None:
|
|
"""A reverse proxy that strips the body but forwards SID still logs in."""
|
|
resp = self._login_response("") # empty body — the reported symptom
|
|
resp.cookies = {"SID": "abc123"}
|
|
self.session.post.return_value = resp
|
|
|
|
self.client._login()
|
|
|
|
self.assertTrue(self.client._logged_in)
|
|
|
|
def test_login_accepts_sid_cookie_in_header_on_204(self) -> None:
|
|
"""204 + Set-Cookie SID with no body (no jar entry) is a valid login.
|
|
|
|
Reproduces the reported case: qBittorrent (or its reverse proxy) returns
|
|
204 No Content with an SID cookie, and requests doesn't always populate
|
|
the cookie jar from such a header, so the cookie must be detected from
|
|
the raw Set-Cookie header.
|
|
"""
|
|
resp = self._login_response("")
|
|
resp.status_code = 204
|
|
resp.cookies = {} # NOT in the parsed jar
|
|
resp.headers = {"Set-Cookie": "SID=abc123; HttpOnly; path=/"}
|
|
self.session.post.return_value = resp
|
|
|
|
self.client._login()
|
|
|
|
self.assertTrue(self.client._logged_in)
|
|
|
|
def test_login_empty_body_without_cookie_is_diagnostic(self) -> None:
|
|
"""Empty 200 body with no SID surfaces a URL/proxy diagnostic hint."""
|
|
resp = self._login_response("")
|
|
resp.cookies = {} # no SID cookie forwarded
|
|
self.session.post.return_value = resp
|
|
|
|
with self.assertRaises(RuntimeError) as ctx:
|
|
self.client._login()
|
|
|
|
message = str(ctx.exception)
|
|
self.assertIn("HTTP 200", message)
|
|
self.assertIn("base_url", message)
|
|
|
|
def test_login_gateway_error_is_diagnostic(self) -> None:
|
|
"""A 502/503/504 from the reverse proxy surfaces a clear gateway message."""
|
|
resp = MagicMock()
|
|
resp.status_code = 504
|
|
resp.text = ""
|
|
resp.cookies = {}
|
|
self.session.post.return_value = resp
|
|
|
|
with self.assertRaises(RuntimeError) as ctx:
|
|
self.client._login()
|
|
|
|
message = str(ctx.exception)
|
|
self.assertIn("HTTP 504", message)
|
|
self.assertIn("reverse proxy", message.lower())
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|