229 lines
8.4 KiB
Python
229 lines
8.4 KiB
Python
"""Tests for ``alfred.infrastructure.api.knaben.client.KnabenClient``.
|
|
|
|
- ``TestInit`` — explicit args override settings; no API key required.
|
|
- ``TestMakeRequest`` — error translation: timeout, 404, 429 (rate limit),
|
|
generic 5xx, and ``RequestException``.
|
|
- ``TestSearch`` — query validation, success path, empty hits, request
|
|
parameter wiring (search_field/order_by/etc.), 404 → empty list,
|
|
per-result parse failures are swallowed (best-effort parsing).
|
|
- ``TestParseTorrent`` — coverage of optional/missing fields and
|
|
``int(... or 0)`` coercion for null seeders/leechers.
|
|
|
|
All HTTP is mocked at ``alfred.infrastructure.api.knaben.client.requests``.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from unittest.mock import MagicMock, patch
|
|
|
|
import pytest
|
|
from requests.exceptions import HTTPError, RequestException, Timeout
|
|
|
|
from alfred.infrastructure.api_TO_CHECK.knaben.client import KnabenClient
|
|
from alfred.infrastructure.api_TO_CHECK.knaben.exceptions import (
|
|
KnabenAPIError,
|
|
KnabenNotFoundError,
|
|
)
|
|
|
|
|
|
def _ok_response(json_body):
|
|
r = MagicMock()
|
|
r.status_code = 200
|
|
r.json.return_value = json_body
|
|
r.raise_for_status.return_value = None
|
|
return r
|
|
|
|
|
|
def _http_error_response(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():
|
|
return KnabenClient(base_url="https://api.knaben.test/v1", timeout=5)
|
|
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# Init #
|
|
# --------------------------------------------------------------------------- #
|
|
|
|
|
|
class TestInit:
|
|
def test_default_base_url(self):
|
|
c = KnabenClient()
|
|
assert c.base_url == "https://api.knaben.org/v1"
|
|
|
|
def test_explicit_override(self):
|
|
c = KnabenClient(base_url="https://x", timeout=99)
|
|
assert c.base_url == "https://x"
|
|
assert c.timeout == 99
|
|
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# _make_request #
|
|
# --------------------------------------------------------------------------- #
|
|
|
|
|
|
class TestMakeRequest:
|
|
@patch("alfred.infrastructure.api.knaben.client.requests.post")
|
|
def test_timeout(self, mock_post, client):
|
|
mock_post.side_effect = Timeout("slow")
|
|
with pytest.raises(KnabenAPIError, match="timeout"):
|
|
client._make_request({"q": "x"})
|
|
|
|
@patch("alfred.infrastructure.api.knaben.client.requests.post")
|
|
def test_http_404(self, mock_post, client):
|
|
mock_post.return_value = _http_error_response(404)
|
|
with pytest.raises(KnabenNotFoundError):
|
|
client._make_request({"q": "x"})
|
|
|
|
@patch("alfred.infrastructure.api.knaben.client.requests.post")
|
|
def test_http_429_rate_limit(self, mock_post, client):
|
|
mock_post.return_value = _http_error_response(429)
|
|
with pytest.raises(KnabenAPIError, match="Rate limit"):
|
|
client._make_request({"q": "x"})
|
|
|
|
@patch("alfred.infrastructure.api.knaben.client.requests.post")
|
|
def test_http_500(self, mock_post, client):
|
|
mock_post.return_value = _http_error_response(500)
|
|
with pytest.raises(KnabenAPIError, match="500"):
|
|
client._make_request({"q": "x"})
|
|
|
|
@patch("alfred.infrastructure.api.knaben.client.requests.post")
|
|
def test_request_exception(self, mock_post, client):
|
|
mock_post.side_effect = RequestException("net")
|
|
with pytest.raises(KnabenAPIError, match="connect"):
|
|
client._make_request({"q": "x"})
|
|
|
|
@patch("alfred.infrastructure.api.knaben.client.requests.post")
|
|
def test_posts_json_body(self, mock_post, client):
|
|
mock_post.return_value = _ok_response({"hits": []})
|
|
client._make_request({"q": "x"})
|
|
call = mock_post.call_args
|
|
# KnabenClient sends params as JSON body, not query string
|
|
assert call.kwargs["json"] == {"q": "x"}
|
|
assert call.kwargs["timeout"] == 5
|
|
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# search #
|
|
# --------------------------------------------------------------------------- #
|
|
|
|
|
|
class TestSearch:
|
|
@pytest.mark.parametrize("bad", ["", None, 42])
|
|
def test_invalid_query(self, client, bad):
|
|
with pytest.raises(ValueError):
|
|
client.search(bad)
|
|
|
|
def test_query_too_long(self, client):
|
|
with pytest.raises(ValueError, match="too long"):
|
|
client.search("a" * 501)
|
|
|
|
@patch("alfred.infrastructure.api.knaben.client.requests.post")
|
|
def test_success(self, mock_post, client):
|
|
mock_post.return_value = _ok_response(
|
|
{
|
|
"hits": [
|
|
{
|
|
"title": "Inception.2010.1080p",
|
|
"size": "10 GB",
|
|
"seeders": 500,
|
|
"leechers": 50,
|
|
"magnetUrl": "magnet:?xt=...",
|
|
"hash": "abc",
|
|
"tracker": "rarbg",
|
|
"date": "2020-01-01",
|
|
"category": "movie",
|
|
}
|
|
]
|
|
}
|
|
)
|
|
results = client.search("Inception")
|
|
assert len(results) == 1
|
|
r = results[0]
|
|
assert r.title == "Inception.2010.1080p"
|
|
assert r.seeders == 500
|
|
assert r.magnet.startswith("magnet:")
|
|
assert r.info_hash == "abc"
|
|
|
|
@patch("alfred.infrastructure.api.knaben.client.requests.post")
|
|
def test_empty_hits_returns_empty_list(self, mock_post, client):
|
|
mock_post.return_value = _ok_response({"hits": []})
|
|
assert client.search("nothing") == []
|
|
|
|
@patch("alfred.infrastructure.api.knaben.client.requests.post")
|
|
def test_404_returns_empty_list(self, mock_post, client):
|
|
mock_post.return_value = _http_error_response(404)
|
|
assert client.search("nothing") == []
|
|
|
|
@patch("alfred.infrastructure.api.knaben.client.requests.post")
|
|
def test_request_parameters(self, mock_post, client):
|
|
mock_post.return_value = _ok_response({"hits": []})
|
|
client.search("Inception", limit=25)
|
|
params = mock_post.call_args.kwargs["json"]
|
|
assert params["query"] == "Inception"
|
|
assert params["search_field"] == "title"
|
|
assert params["order_by"] == "peers"
|
|
assert params["order_direction"] == "desc"
|
|
assert params["size"] == 25
|
|
assert params["hide_unsafe"] is True
|
|
assert params["hide_xxx"] is True
|
|
|
|
@patch("alfred.infrastructure.api.knaben.client.requests.post")
|
|
def test_default_limit(self, mock_post, client):
|
|
mock_post.return_value = _ok_response({"hits": []})
|
|
client.search("x")
|
|
assert mock_post.call_args.kwargs["json"]["size"] == 10
|
|
|
|
@patch("alfred.infrastructure.api.knaben.client.requests.post")
|
|
def test_unexpected_exception_propagates(self, mock_post, client):
|
|
# Anything other than KnabenNotFoundError bubbles up.
|
|
mock_post.side_effect = RuntimeError("boom")
|
|
with pytest.raises(RuntimeError):
|
|
client.search("x")
|
|
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# _parse_torrent #
|
|
# --------------------------------------------------------------------------- #
|
|
|
|
|
|
class TestParseTorrent:
|
|
def test_minimal(self, client):
|
|
r = client._parse_torrent({})
|
|
assert r.title == "Unknown"
|
|
assert r.size == "Unknown"
|
|
assert r.seeders == 0
|
|
assert r.leechers == 0
|
|
assert r.magnet == ""
|
|
|
|
def test_null_seeders_coerced_to_zero(self, client):
|
|
r = client._parse_torrent({"seeders": None, "leechers": None})
|
|
assert r.seeders == 0
|
|
assert r.leechers == 0
|
|
|
|
def test_optional_fields_propagated(self, client):
|
|
r = client._parse_torrent(
|
|
{
|
|
"title": "X",
|
|
"size": "1 GB",
|
|
"seeders": 10,
|
|
"leechers": 2,
|
|
"magnetUrl": "magnet:?",
|
|
"hash": "h",
|
|
"tracker": "t",
|
|
"date": "d",
|
|
"category": "c",
|
|
}
|
|
)
|
|
assert r.info_hash == "h"
|
|
assert r.tracker == "t"
|
|
assert r.upload_date == "d"
|
|
assert r.category == "c"
|