Oleksandr Kozachuk
2caa18a17f
search_games used `where name ~ "<q>"*`: prefix-only, diacritic- and word-order-sensitive -- "MARVEL Tōkon" found nothing though IGDB has it. Switch to IGDB's native `search` clause (relevance-ranked, diacritic-insensitive). fetch_url and image downloads read bodies with content.read(n), which returns only the first buffered chunk (~7 KB): pages collapsed to their <title>. httpread.read_capped collects chunks up to the byte cap; image downloads read limit+1 so over-limit files are still rejected instead of cached truncated (IMG-10). Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
278 lines
11 KiB
Python
278 lines
11 KiB
Python
import unittest
|
|
from unittest.mock import Mock, patch
|
|
|
|
import requests
|
|
|
|
from fjerkroa_bot.igdblib import IGDBQuery
|
|
|
|
|
|
class TestIGDBQuery(unittest.TestCase):
|
|
def setUp(self):
|
|
self.client_id = "test_client_id"
|
|
self.api_key = "test_api_key"
|
|
self.igdb = IGDBQuery(self.client_id, self.api_key)
|
|
|
|
def test_init(self):
|
|
"""Test IGDBQuery initialization."""
|
|
self.assertEqual(self.igdb.client_id, self.client_id)
|
|
self.assertEqual(self.igdb.igdb_api_key, self.api_key)
|
|
|
|
@patch("fjerkroa_bot.igdblib.requests.post")
|
|
def test_send_igdb_request_success(self, mock_post):
|
|
"""Test successful IGDB API request."""
|
|
mock_response = Mock()
|
|
mock_response.json.return_value = {"id": 1, "name": "Test Game"}
|
|
mock_response.raise_for_status.return_value = None
|
|
mock_post.return_value = mock_response
|
|
|
|
result = self.igdb.send_igdb_request("games", "fields name; limit 1;")
|
|
|
|
self.assertEqual(result, {"id": 1, "name": "Test Game"})
|
|
mock_post.assert_called_once_with(
|
|
"https://api.igdb.com/v4/games",
|
|
headers={"Client-ID": self.client_id, "Authorization": f"Bearer {self.api_key}"},
|
|
data="fields name; limit 1;",
|
|
)
|
|
|
|
@patch("fjerkroa_bot.igdblib.requests.post")
|
|
def test_send_igdb_request_failure(self, mock_post):
|
|
"""Test IGDB API request failure."""
|
|
mock_post.side_effect = requests.RequestException("API Error")
|
|
|
|
result = self.igdb.send_igdb_request("games", "fields name; limit 1;")
|
|
|
|
self.assertIsNone(result)
|
|
|
|
def test_build_query_basic(self):
|
|
"""Test building basic query."""
|
|
query = IGDBQuery.build_query(["name", "summary"])
|
|
expected = "fields name,summary; limit 10;"
|
|
self.assertEqual(query, expected)
|
|
|
|
def test_build_query_with_limit(self):
|
|
"""Test building query with custom limit."""
|
|
query = IGDBQuery.build_query(["name"], limit=5)
|
|
expected = "fields name; limit 5;"
|
|
self.assertEqual(query, expected)
|
|
|
|
def test_build_query_with_offset(self):
|
|
"""Test building query with offset."""
|
|
query = IGDBQuery.build_query(["name"], offset=10)
|
|
expected = "fields name; limit 10; offset 10;"
|
|
self.assertEqual(query, expected)
|
|
|
|
def test_build_query_with_filters(self):
|
|
"""Test building query with filters."""
|
|
filters = {"name": "Mario", "platform": "Nintendo"}
|
|
query = IGDBQuery.build_query(["name"], filters=filters)
|
|
expected = "fields name; limit 10; where name Mario & platform Nintendo;"
|
|
self.assertEqual(query, expected)
|
|
|
|
def test_build_query_empty_fields(self):
|
|
"""Test building query with empty fields."""
|
|
query = IGDBQuery.build_query([])
|
|
expected = "fields *; limit 10;"
|
|
self.assertEqual(query, expected)
|
|
|
|
def test_build_query_none_fields(self):
|
|
"""Test building query with None fields."""
|
|
query = IGDBQuery.build_query(None)
|
|
expected = "fields *; limit 10;"
|
|
self.assertEqual(query, expected)
|
|
|
|
@patch.object(IGDBQuery, "send_igdb_request")
|
|
def test_generalized_igdb_query(self, mock_send):
|
|
"""Test generalized IGDB query method."""
|
|
mock_send.return_value = [{"id": 1, "name": "Test Game"}]
|
|
|
|
params = {"name": "Mario"}
|
|
result = self.igdb.generalized_igdb_query(params, "games", ["name"], limit=5)
|
|
|
|
expected_query = 'fields name; limit 5; where name ~ "Mario"*;'
|
|
|
|
mock_send.assert_called_once_with("games", expected_query)
|
|
self.assertEqual(result, [{"id": 1, "name": "Test Game"}])
|
|
|
|
@patch.object(IGDBQuery, "send_igdb_request")
|
|
def test_generalized_igdb_query_with_additional_filters(self, mock_send):
|
|
"""Test generalized query with additional filters."""
|
|
mock_send.return_value = [{"id": 1, "name": "Test Game"}]
|
|
|
|
params = {"name": "Mario"}
|
|
additional_filters = {"platform": "= 1"}
|
|
self.igdb.generalized_igdb_query(params, "games", ["name"], additional_filters, limit=5)
|
|
|
|
expected_query = 'fields name; limit 5; where name ~ "Mario"* & platform = 1;'
|
|
mock_send.assert_called_once_with("games", expected_query)
|
|
|
|
def test_create_query_function(self):
|
|
"""Test creating a query function."""
|
|
func_def = self.igdb.create_query_function("test_func", "Test function", {"name": {"type": "string"}}, "games", ["name"], limit=5)
|
|
|
|
self.assertEqual(func_def["name"], "test_func")
|
|
self.assertEqual(func_def["description"], "Test function")
|
|
self.assertEqual(func_def["parameters"]["type"], "object")
|
|
self.assertIn("function", func_def)
|
|
|
|
@patch.object(IGDBQuery, "generalized_igdb_query")
|
|
def test_platform_families(self, mock_query):
|
|
"""Test platform families caching."""
|
|
mock_query.return_value = [{"id": 1, "name": "PlayStation"}, {"id": 2, "name": "Nintendo"}]
|
|
|
|
# First call
|
|
result1 = self.igdb.platform_families()
|
|
expected = {1: "PlayStation", 2: "Nintendo"}
|
|
self.assertEqual(result1, expected)
|
|
|
|
# Second call should use cache
|
|
result2 = self.igdb.platform_families()
|
|
self.assertEqual(result2, expected)
|
|
|
|
# Should only call the API once due to caching
|
|
mock_query.assert_called_once_with({}, "platform_families", ["id", "name"], limit=500)
|
|
|
|
@patch.object(IGDBQuery, "generalized_igdb_query")
|
|
@patch.object(IGDBQuery, "platform_families")
|
|
def test_platforms(self, mock_families, mock_query):
|
|
"""Test platforms method."""
|
|
mock_families.return_value = {1: "PlayStation"}
|
|
mock_query.return_value = [
|
|
{"id": 1, "name": "PlayStation 5", "alternative_name": "PS5", "abbreviation": "PS5", "platform_family": 1},
|
|
{"id": 2, "name": "Nintendo Switch"},
|
|
]
|
|
|
|
self.igdb.platforms()
|
|
|
|
# Test passes if no exception is raised
|
|
|
|
mock_query.assert_called_once_with(
|
|
{}, "platforms", ["id", "name", "alternative_name", "abbreviation", "platform_family"], limit=500
|
|
)
|
|
|
|
@patch.object(IGDBQuery, "generalized_igdb_query")
|
|
def test_game_info(self, mock_query):
|
|
"""Test game info method."""
|
|
mock_query.return_value = [{"id": 1, "name": "Super Mario Bros"}]
|
|
|
|
result = self.igdb.game_info("Mario")
|
|
|
|
expected_fields = [
|
|
"id",
|
|
"name",
|
|
"alternative_names",
|
|
"game_type",
|
|
"release_dates",
|
|
"franchise",
|
|
"language_supports",
|
|
"keywords",
|
|
"platforms",
|
|
"rating",
|
|
"summary",
|
|
]
|
|
|
|
mock_query.assert_called_once_with({"name": "Mario"}, "games", expected_fields, limit=100)
|
|
self.assertEqual(result, [{"id": 1, "name": "Super Mario Bros"}])
|
|
|
|
|
|
class TestIGDBNativeSearch(unittest.TestCase):
|
|
def test_build_query_with_search_term(self):
|
|
"""search_games uses IGDB full-text search, not a name prefix filter."""
|
|
query = IGDBQuery.build_query(["name"], {"game_type": "= 0"}, limit=5, search_term="Marvel Tōkon")
|
|
self.assertEqual(query, 'search "Marvel Tōkon"; fields name; limit 5; where game_type = 0;')
|
|
|
|
def test_search_term_escapes_quotes_and_backslashes(self):
|
|
query = IGDBQuery.build_query(["name"], search_term='say "hi" \\ bye')
|
|
self.assertIn('search "say \\"hi\\" \\\\ bye";', query)
|
|
|
|
@patch.object(IGDBQuery, "generalized_igdb_query")
|
|
def test_search_games_passes_search_term(self, mock_query):
|
|
mock_query.return_value = []
|
|
IGDBQuery("cid", "token").search_games("Elden Ring", limit=3)
|
|
_, kwargs = mock_query.call_args
|
|
self.assertEqual(kwargs["search_term"], "Elden Ring")
|
|
self.assertEqual(mock_query.call_args.args[0], {})
|
|
|
|
|
|
class TestIGDBTokenRefresh(unittest.TestCase):
|
|
@staticmethod
|
|
def _oauth_response(token="fresh_token", expires_in=5_000_000):
|
|
response = Mock()
|
|
response.json.return_value = {"access_token": token, "expires_in": expires_in}
|
|
response.raise_for_status.return_value = None
|
|
return response
|
|
|
|
@staticmethod
|
|
def _api_response(payload, status_code=200):
|
|
response = Mock()
|
|
response.status_code = status_code
|
|
response.json.return_value = payload
|
|
response.raise_for_status.return_value = None
|
|
return response
|
|
|
|
@patch("fjerkroa_bot.igdblib.requests.post")
|
|
def test_fetches_token_when_only_secret_configured(self, mock_post):
|
|
"""Without a static token, the first request fetches one via Twitch OAuth."""
|
|
mock_post.side_effect = [self._oauth_response(), self._api_response([{"id": 1}])]
|
|
igdb = IGDBQuery("cid", client_secret="secret")
|
|
|
|
result = igdb.send_igdb_request("games", "fields name; limit 1;")
|
|
|
|
self.assertEqual(result, [{"id": 1}])
|
|
oauth_call, api_call = mock_post.call_args_list
|
|
self.assertEqual(oauth_call.args[0], "https://id.twitch.tv/oauth2/token")
|
|
self.assertEqual(
|
|
oauth_call.kwargs["params"],
|
|
{"client_id": "cid", "client_secret": "secret", "grant_type": "client_credentials"},
|
|
)
|
|
self.assertEqual(api_call.kwargs["headers"]["Authorization"], "Bearer fresh_token")
|
|
|
|
@patch("fjerkroa_bot.igdblib.requests.post")
|
|
def test_refreshes_and_retries_on_401(self, mock_post):
|
|
"""A 401 with a configured secret triggers one refresh and retry."""
|
|
mock_post.side_effect = [
|
|
self._api_response(None, status_code=401),
|
|
self._oauth_response(),
|
|
self._api_response([{"id": 2}]),
|
|
]
|
|
igdb = IGDBQuery("cid", "expired_token", client_secret="secret")
|
|
|
|
result = igdb.send_igdb_request("games", "fields name; limit 1;")
|
|
|
|
self.assertEqual(result, [{"id": 2}])
|
|
self.assertEqual(igdb.igdb_api_key, "fresh_token")
|
|
self.assertEqual(mock_post.call_args_list[2].kwargs["headers"]["Authorization"], "Bearer fresh_token")
|
|
|
|
@patch("fjerkroa_bot.igdblib.time.time")
|
|
@patch("fjerkroa_bot.igdblib.requests.post")
|
|
def test_proactive_refresh_before_expiry(self, mock_post, mock_time):
|
|
"""An expired self-fetched token is refreshed before the request."""
|
|
mock_time.return_value = 1_000_000.0
|
|
mock_post.side_effect = [self._oauth_response("token_a", expires_in=5_000_000), self._api_response([])]
|
|
igdb = IGDBQuery("cid", client_secret="secret")
|
|
igdb.send_igdb_request("games", "fields name;")
|
|
|
|
# jump past the token expiry -> next request refreshes first
|
|
mock_time.return_value = 1_000_000.0 + 5_000_000
|
|
mock_post.side_effect = [self._oauth_response("token_b"), self._api_response([])]
|
|
igdb.send_igdb_request("games", "fields name;")
|
|
|
|
self.assertEqual(igdb.igdb_api_key, "token_b")
|
|
|
|
@patch("fjerkroa_bot.igdblib.requests.post")
|
|
def test_no_refresh_without_secret(self, mock_post):
|
|
"""Static-token setups keep the old behavior: no OAuth calls, error -> None."""
|
|
response = Mock()
|
|
response.status_code = 401
|
|
response.raise_for_status.side_effect = requests.RequestException("401 Client Error")
|
|
mock_post.return_value = response
|
|
igdb = IGDBQuery("cid", "expired_token")
|
|
|
|
result = igdb.send_igdb_request("games", "fields name; limit 1;")
|
|
|
|
self.assertIsNone(result)
|
|
mock_post.assert_called_once()
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|