416 lines
15 KiB
Python
416 lines
15 KiB
Python
"""Tests for ``alfred.infrastructure.api.qbittorrent.client.QBittorrentClient``.
|
|
|
|
Exercises every public method against a ``MagicMock`` ``requests.Session``
|
|
attached to the client. Auth state (``self._authenticated``) is asserted
|
|
explicitly so the implicit auto-login behavior of mutation methods is
|
|
covered.
|
|
|
|
Scope:
|
|
|
|
- ``TestInit`` — host/credentials wiring + Session attached.
|
|
- ``TestMakeRequest`` — verb dispatch (GET/POST), JSON vs text fallback,
|
|
error translation for timeout/403/5xx/RequestException, invalid verb.
|
|
- ``TestLogin`` — happy path, non-"Ok." rejection, propagation from
|
|
underlying API error.
|
|
- ``TestGetTorrents`` — auto-login, non-list payload safety, per-item parse
|
|
failures.
|
|
- ``TestAddTorrent`` — magnet payload wiring, optional category/save_path,
|
|
paused flag, unexpected response.
|
|
- ``TestMutations`` — pause/resume/delete/recheck/set_location all wire the
|
|
hash and propagate errors.
|
|
- ``TestFindByName`` — exact match, case-insensitive match, save_path fallback,
|
|
no match.
|
|
- ``TestParseTorrent`` — progress percentage conversion, defaults.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from unittest.mock import MagicMock, patch
|
|
|
|
import pytest
|
|
from requests.exceptions import HTTPError, RequestException, Timeout
|
|
|
|
from alfred.infrastructure.api.qbittorrent.client import QBittorrentClient
|
|
from alfred.infrastructure.api.qbittorrent.dto import TorrentInfo
|
|
from alfred.infrastructure.api.qbittorrent.exceptions import (
|
|
QBittorrentAPIError,
|
|
QBittorrentAuthError,
|
|
)
|
|
|
|
|
|
def _resp(body, *, status=200, json_decodable=True):
|
|
r = MagicMock()
|
|
r.status_code = status
|
|
r.raise_for_status.return_value = None
|
|
if json_decodable:
|
|
r.json.return_value = body
|
|
else:
|
|
r.json.side_effect = ValueError("not json")
|
|
r.text = body
|
|
return r
|
|
|
|
|
|
def _http_error(status_code):
|
|
r = MagicMock()
|
|
r.status_code = status_code
|
|
err = HTTPError(f"{status_code}")
|
|
err.response = r
|
|
r.raise_for_status.side_effect = err
|
|
return r
|
|
|
|
|
|
@pytest.fixture
|
|
def client():
|
|
c = QBittorrentClient(
|
|
host="http://qbit.test:8080",
|
|
username="admin",
|
|
password="secret",
|
|
timeout=5,
|
|
)
|
|
# Replace requests.Session with a MagicMock so we control responses
|
|
c.session = MagicMock()
|
|
return c
|
|
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# Init #
|
|
# --------------------------------------------------------------------------- #
|
|
|
|
|
|
class TestInit:
|
|
def test_explicit_args(self):
|
|
c = QBittorrentClient(host="http://x:1", username="u", password="p", timeout=99)
|
|
assert c.host == "http://x:1"
|
|
assert c.username == "u"
|
|
assert c.password == "p"
|
|
assert c.timeout == 99
|
|
assert c._authenticated is False
|
|
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# _make_request #
|
|
# --------------------------------------------------------------------------- #
|
|
|
|
|
|
class TestMakeRequest:
|
|
def test_invalid_verb(self, client):
|
|
with pytest.raises(ValueError, match="HTTP"):
|
|
client._make_request("PATCH", "/api/v2/foo")
|
|
|
|
def test_get_returns_json(self, client):
|
|
client.session.get.return_value = _resp({"k": "v"})
|
|
out = client._make_request("GET", "/x", data={"a": 1})
|
|
assert out == {"k": "v"}
|
|
client.session.get.assert_called_once()
|
|
|
|
def test_post_returns_text_when_not_json(self, client):
|
|
client.session.post.return_value = _resp("Ok.", json_decodable=False)
|
|
out = client._make_request("POST", "/x", data={"a": 1})
|
|
assert out == "Ok."
|
|
|
|
def test_timeout(self, client):
|
|
client.session.get.side_effect = Timeout("slow")
|
|
with pytest.raises(QBittorrentAPIError, match="timeout"):
|
|
client._make_request("GET", "/x")
|
|
|
|
def test_http_403_auth_error(self, client):
|
|
client.session.post.return_value = _http_error(403)
|
|
with pytest.raises(QBittorrentAuthError):
|
|
client._make_request("POST", "/x")
|
|
|
|
def test_http_500_generic(self, client):
|
|
client.session.get.return_value = _http_error(500)
|
|
with pytest.raises(QBittorrentAPIError, match="500"):
|
|
client._make_request("GET", "/x")
|
|
|
|
def test_request_exception(self, client):
|
|
client.session.get.side_effect = RequestException("net down")
|
|
with pytest.raises(QBittorrentAPIError, match="connect"):
|
|
client._make_request("GET", "/x")
|
|
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# Login #
|
|
# --------------------------------------------------------------------------- #
|
|
|
|
|
|
class TestLogin:
|
|
def test_login_success(self, client):
|
|
client.session.post.return_value = _resp("Ok.", json_decodable=False)
|
|
assert client.login() is True
|
|
assert client._authenticated is True
|
|
|
|
def test_login_wrong_credentials(self, client):
|
|
client.session.post.return_value = _resp("Fails.", json_decodable=False)
|
|
with pytest.raises(QBittorrentAuthError):
|
|
client.login()
|
|
assert client._authenticated is False
|
|
|
|
def test_login_api_error_translated_to_auth_error(self, client):
|
|
client.session.post.return_value = _http_error(403)
|
|
with pytest.raises(QBittorrentAuthError):
|
|
client.login()
|
|
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# get_torrents (auto-login behavior) #
|
|
# --------------------------------------------------------------------------- #
|
|
|
|
|
|
class TestGetTorrents:
|
|
def test_auto_logs_in_then_fetches(self, client):
|
|
# Order: 1) login POST, 2) torrents/info GET
|
|
client.session.post.return_value = _resp("Ok.", json_decodable=False)
|
|
client.session.get.return_value = _resp(
|
|
[
|
|
{
|
|
"hash": "h1",
|
|
"name": "Foo",
|
|
"size": 100,
|
|
"progress": 0.5,
|
|
"state": "downloading",
|
|
"dlspeed": 1024,
|
|
"upspeed": 512,
|
|
"eta": 60,
|
|
"num_seeds": 5,
|
|
"num_leechs": 1,
|
|
"ratio": 0.1,
|
|
"category": "movies",
|
|
"save_path": "/dl",
|
|
}
|
|
]
|
|
)
|
|
torrents = client.get_torrents()
|
|
assert len(torrents) == 1
|
|
assert torrents[0].name == "Foo"
|
|
assert torrents[0].progress == 50.0 # 0.5 → 50%
|
|
assert client._authenticated is True
|
|
|
|
def test_non_list_returns_empty(self, client):
|
|
client._authenticated = True
|
|
client.session.get.return_value = _resp({"oops": "bad"})
|
|
assert client.get_torrents() == []
|
|
|
|
def test_filter_and_category_propagated(self, client):
|
|
client._authenticated = True
|
|
client.session.get.return_value = _resp([])
|
|
client.get_torrents(filter="completed", category="movies")
|
|
params = client.session.get.call_args.kwargs["params"]
|
|
assert params == {"filter": "completed", "category": "movies"}
|
|
|
|
def test_skips_unparseable_torrents(self, client):
|
|
client._authenticated = True
|
|
# _parse_torrent uses .get on every field with sensible defaults, so
|
|
# malformed dicts almost never raise — patch the parser to force it.
|
|
client.session.get.return_value = _resp([{"good": True}])
|
|
with patch.object(client, "_parse_torrent", side_effect=Exception("nope")):
|
|
assert client.get_torrents() == []
|
|
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# add_torrent #
|
|
# --------------------------------------------------------------------------- #
|
|
|
|
|
|
class TestAddTorrent:
|
|
def test_add_success(self, client):
|
|
client._authenticated = True
|
|
client.session.post.return_value = _resp("Ok.", json_decodable=False)
|
|
assert client.add_torrent("magnet:?xt=foo") is True
|
|
|
|
def test_add_unexpected_response(self, client):
|
|
client._authenticated = True
|
|
client.session.post.return_value = _resp("Fails.", json_decodable=False)
|
|
assert client.add_torrent("magnet:?xt=foo") is False
|
|
|
|
def test_add_payload(self, client):
|
|
client._authenticated = True
|
|
client.session.post.return_value = _resp("Ok.", json_decodable=False)
|
|
client.add_torrent(
|
|
"magnet:?xt=foo", category="movies", save_path="/dl", paused=True
|
|
)
|
|
payload = client.session.post.call_args.kwargs["data"]
|
|
assert payload["urls"] == "magnet:?xt=foo"
|
|
assert payload["paused"] == "true"
|
|
assert payload["category"] == "movies"
|
|
assert payload["savepath"] == "/dl"
|
|
|
|
def test_paused_false_serialized(self, client):
|
|
client._authenticated = True
|
|
client.session.post.return_value = _resp("Ok.", json_decodable=False)
|
|
client.add_torrent("magnet:?xt=foo")
|
|
payload = client.session.post.call_args.kwargs["data"]
|
|
assert payload["paused"] == "false"
|
|
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# Mutations (delete, pause, resume, recheck, set_location) #
|
|
# --------------------------------------------------------------------------- #
|
|
|
|
|
|
class TestMutations:
|
|
def _ok(self, client):
|
|
client._authenticated = True
|
|
client.session.post.return_value = _resp("Ok.", json_decodable=False)
|
|
|
|
def test_delete_success(self, client):
|
|
self._ok(client)
|
|
assert client.delete_torrent("hash1", delete_files=True) is True
|
|
payload = client.session.post.call_args.kwargs["data"]
|
|
assert payload["hashes"] == "hash1"
|
|
assert payload["deleteFiles"] == "true"
|
|
|
|
def test_delete_no_files_default(self, client):
|
|
self._ok(client)
|
|
client.delete_torrent("hash1")
|
|
assert client.session.post.call_args.kwargs["data"]["deleteFiles"] == "false"
|
|
|
|
def test_pause(self, client):
|
|
self._ok(client)
|
|
assert client.pause_torrent("hash1") is True
|
|
|
|
def test_resume(self, client):
|
|
self._ok(client)
|
|
assert client.resume_torrent("hash1") is True
|
|
|
|
def test_recheck(self, client):
|
|
self._ok(client)
|
|
assert client.recheck("hash1") is True
|
|
|
|
def test_set_location(self, client):
|
|
self._ok(client)
|
|
assert client.set_location("hash1", "/new/path") is True
|
|
payload = client.session.post.call_args.kwargs["data"]
|
|
assert payload == {"hashes": "hash1", "location": "/new/path"}
|
|
|
|
def test_mutation_propagates_api_error(self, client):
|
|
client._authenticated = True
|
|
client.session.post.return_value = _http_error(500)
|
|
with pytest.raises(QBittorrentAPIError):
|
|
client.delete_torrent("hash1")
|
|
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# find_by_name #
|
|
# --------------------------------------------------------------------------- #
|
|
|
|
|
|
def _torrent_dict(name, save_path=None):
|
|
return {
|
|
"hash": "h",
|
|
"name": name,
|
|
"size": 1,
|
|
"progress": 0.0,
|
|
"state": "x",
|
|
"dlspeed": 0,
|
|
"upspeed": 0,
|
|
"eta": 0,
|
|
"num_seeds": 0,
|
|
"num_leechs": 0,
|
|
"ratio": 0.0,
|
|
"save_path": save_path,
|
|
}
|
|
|
|
|
|
class TestFindByName:
|
|
def test_exact_match(self, client):
|
|
client._authenticated = True
|
|
client.session.get.return_value = _resp(
|
|
[_torrent_dict("Foundation.S01"), _torrent_dict("Other")]
|
|
)
|
|
result = client.find_by_name("Foundation.S01")
|
|
assert isinstance(result, TorrentInfo)
|
|
assert result.name == "Foundation.S01"
|
|
|
|
def test_case_insensitive_match(self, client):
|
|
client._authenticated = True
|
|
client.session.get.return_value = _resp([_torrent_dict("foundation.s01")])
|
|
result = client.find_by_name("Foundation.S01")
|
|
assert result is not None
|
|
assert result.name == "foundation.s01"
|
|
|
|
def test_save_path_fallback(self, client):
|
|
client._authenticated = True
|
|
client.session.get.return_value = _resp(
|
|
[_torrent_dict("Different", save_path="/dl/Foundation.S01")]
|
|
)
|
|
result = client.find_by_name("Foundation.S01")
|
|
assert result is not None
|
|
assert result.save_path.endswith("Foundation.S01")
|
|
|
|
def test_no_match_returns_none(self, client):
|
|
client._authenticated = True
|
|
client.session.get.return_value = _resp([_torrent_dict("nope")])
|
|
assert client.find_by_name("Foundation.S01") is None
|
|
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# _parse_torrent #
|
|
# --------------------------------------------------------------------------- #
|
|
|
|
|
|
class TestParseTorrent:
|
|
def test_defaults(self, client):
|
|
t = client._parse_torrent({})
|
|
assert t.hash == ""
|
|
assert t.name == "Unknown"
|
|
assert t.progress == 0.0
|
|
assert t.state == "unknown"
|
|
|
|
def test_progress_converted_to_percentage(self, client):
|
|
t = client._parse_torrent({"progress": 0.75})
|
|
assert t.progress == 75.0
|
|
|
|
def test_full_payload(self, client):
|
|
t = client._parse_torrent(
|
|
{
|
|
"hash": "h",
|
|
"name": "n",
|
|
"size": 1024,
|
|
"progress": 1.0,
|
|
"state": "uploading",
|
|
"dlspeed": 100,
|
|
"upspeed": 50,
|
|
"eta": 0,
|
|
"num_seeds": 10,
|
|
"num_leechs": 2,
|
|
"ratio": 2.5,
|
|
"category": "movies",
|
|
"save_path": "/dl",
|
|
}
|
|
)
|
|
assert t.progress == 100.0
|
|
assert t.ratio == 2.5
|
|
assert t.category == "movies"
|
|
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# logout #
|
|
# --------------------------------------------------------------------------- #
|
|
|
|
|
|
class TestLogout:
|
|
def test_logout_success(self, client):
|
|
client._authenticated = True
|
|
client.session.post.return_value = _resp("", json_decodable=False)
|
|
assert client.logout() is True
|
|
assert client._authenticated is False
|
|
|
|
def test_logout_swallows_errors(self, client):
|
|
client._authenticated = True
|
|
client.session.post.side_effect = RuntimeError("boom")
|
|
# Per implementation, logout returns False instead of raising.
|
|
assert client.logout() is False
|
|
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# get_torrent_properties #
|
|
# --------------------------------------------------------------------------- #
|
|
|
|
|
|
class TestGetTorrentProperties:
|
|
def test_properties_returned(self, client):
|
|
client._authenticated = True
|
|
client.session.get.return_value = _resp({"piece_size": 16384})
|
|
assert client.get_torrent_properties("h")["piece_size"] == 16384
|