"""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.knaben.client import KnabenClient from alfred.infrastructure.api.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"