"""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) def test_maindata_uses_rid_and_merges_partial_update(self) -> None: """First call is a full fetch (no rid); later calls send rid and merge the diff.""" self.client._logged_in = True self.client._maindata_ttl = 0 # force a real fetch each call full = { "rid": 10, "full_update": True, "server_state": {"dl_info_speed": 100}, "torrents": {"a": {"name": "A", "state": "downloading"}}, } partial = { "rid": 11, "full_update": False, "server_state": {"dl_info_speed": 200}, "torrents": {"a": {"name": "A", "state": "pausedDL"}}, } self.session.get.side_effect = [self._get_response(full), self._get_response(partial)] r1 = self.client.maindata() self.assertNotIn("rid", self.session.get.call_args_list[0].kwargs["params"]) self.assertEqual(r1["server_state"]["dl_info_speed"], 100) self.assertEqual(r1["torrents"]["a"]["state"], "downloading") r2 = self.client.maindata() self.assertEqual(self.session.get.call_args_list[1].kwargs["params"].get("rid"), 10) self.assertEqual(r2["server_state"]["dl_info_speed"], 200) # merged self.assertEqual(r2["torrents"]["a"]["state"], "pausedDL") # merged def test_maindata_caches_concurrent_calls_within_ttl(self) -> None: """Two calls within the TTL collapse to a single HTTP fetch.""" self.client._logged_in = True payload = { "rid": 1, "full_update": True, "server_state": {"dl_info_speed": 5}, "torrents": {"a": {"state": "downloading"}}, } self.session.get.return_value = self._get_response(payload) self.client.maindata() r2 = self.client.maindata() # served from cache — no extra HTTP self.assertEqual(self.session.get.call_count, 1) self.assertEqual(r2["server_state"]["dl_info_speed"], 5) def test_maindata_backoff_after_failure_does_not_pile_on(self) -> None: """A failed fetch arms backoff; the next call skips the network entirely.""" self.client._logged_in = True self.client._maindata_ttl = 0 bad = MagicMock() bad.status_code = 503 bad.raise_for_status.side_effect = requests.HTTPError("503 Server Error") bad.text = "" self.session.get.return_value = bad with self.assertRaises(RuntimeError): # no snapshot yet -> raises + arms backoff self.client.maindata() self.assertEqual(self.session.get.call_count, 1) with self.assertRaises(RuntimeError): # within backoff -> no new HTTP self.client.maindata() self.assertEqual(self.session.get.call_count, 1) def test_maindata_serves_stale_snapshot_during_backoff(self) -> None: """After a good fetch, a later failure serves stale data instead of erroring.""" self.client._logged_in = True self.client._maindata_ttl = 0 good = { "rid": 1, "full_update": True, "server_state": {"dl_info_speed": 7}, "torrents": {"a": {"state": "downloading"}}, } bad = MagicMock() bad.status_code = 503 bad.raise_for_status.side_effect = requests.HTTPError("503") bad.text = "" self.session.get.side_effect = [self._get_response(good), bad] r1 = self.client.maindata() self.assertEqual(r1["server_state"]["dl_info_speed"], 7) r2 = self.client.maindata() # fetch fails -> serves stale snapshot, no raise self.assertEqual(r2["server_state"]["dl_info_speed"], 7) @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_accepts_qbt_sid_cookie_newer_versions(self) -> None: """Newer qBittorrent names the session cookie QBT_SID_; recognize it.""" resp = self._login_response("") resp.status_code = 204 resp.cookies = {"QBT_SID_5080": "abc123"} 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()