4be6afb541
- Add APITimeoutError exception and retry logic to db_get - Add timeout handling to db_post, db_put, db_patch, db_delete - Fix get_team_by_owner to prefer non-gauntlet team (PostgreSQL migration fix) - Code formatting cleanup (black)
222 lines
8.6 KiB
Python
222 lines
8.6 KiB
Python
import asyncio
|
|
import pytest
|
|
import aiohttp
|
|
from unittest.mock import Mock, patch, AsyncMock, MagicMock
|
|
from exceptions import DatabaseError, APITimeoutError
|
|
import api_calls
|
|
|
|
|
|
class TestUtilityFunctions:
|
|
"""Test utility functions in api_calls."""
|
|
|
|
def test_param_char_with_params(self):
|
|
"""Test param_char returns & when other_params is truthy."""
|
|
assert api_calls.param_char(True) == "&"
|
|
assert api_calls.param_char(["param1"]) == "&"
|
|
assert api_calls.param_char({"key": "value"}) == "&"
|
|
assert api_calls.param_char("some_param") == "&"
|
|
|
|
def test_param_char_without_params(self):
|
|
"""Test param_char returns ? when other_params is falsy."""
|
|
assert api_calls.param_char(False) == "?"
|
|
assert api_calls.param_char(None) == "?"
|
|
assert api_calls.param_char([]) == "?"
|
|
assert api_calls.param_char({}) == "?"
|
|
assert api_calls.param_char("") == "?"
|
|
assert api_calls.param_char(0) == "?"
|
|
|
|
@patch("api_calls.DB_URL", "https://test.example.com/api")
|
|
def test_get_req_url_basic(self):
|
|
"""Test basic URL generation without object_id or params."""
|
|
result = api_calls.get_req_url("teams")
|
|
expected = "https://test.example.com/api/v2/teams"
|
|
assert result == expected
|
|
|
|
@patch("api_calls.DB_URL", "https://test.example.com/api")
|
|
def test_get_req_url_with_version(self):
|
|
"""Test URL generation with custom API version."""
|
|
result = api_calls.get_req_url("teams", api_ver=1)
|
|
expected = "https://test.example.com/api/v1/teams"
|
|
assert result == expected
|
|
|
|
@patch("api_calls.DB_URL", "https://test.example.com/api")
|
|
def test_get_req_url_with_object_id(self):
|
|
"""Test URL generation with object_id."""
|
|
result = api_calls.get_req_url("teams", object_id=123)
|
|
expected = "https://test.example.com/api/v2/teams/123"
|
|
assert result == expected
|
|
|
|
@patch("api_calls.DB_URL", "https://test.example.com/api")
|
|
def test_get_req_url_with_params(self):
|
|
"""Test URL generation with parameters."""
|
|
params = [("season", "7"), ("active", "true")]
|
|
result = api_calls.get_req_url("teams", params=params)
|
|
expected = "https://test.example.com/api/v2/teams?season=7&active=true"
|
|
assert result == expected
|
|
|
|
@patch("api_calls.DB_URL", "https://test.example.com/api")
|
|
def test_get_req_url_complete(self):
|
|
"""Test URL generation with all parameters."""
|
|
params = [("season", "7"), ("limit", "10")]
|
|
result = api_calls.get_req_url("games", api_ver=1, object_id=456, params=params)
|
|
expected = "https://test.example.com/api/v1/games/456?season=7&limit=10"
|
|
assert result == expected
|
|
|
|
@patch("api_calls.logger")
|
|
def test_log_return_value_short_string(self, mock_logger):
|
|
"""Test logging short return values."""
|
|
api_calls.log_return_value("Short log message")
|
|
mock_logger.info.assert_called_once_with("\n\nreturn: Short log message")
|
|
|
|
@patch("api_calls.logger")
|
|
def test_log_return_value_long_string(self, mock_logger):
|
|
"""Test logging long return values that get chunked."""
|
|
long_string = "A" * 5000 # 5000 character string
|
|
api_calls.log_return_value(long_string)
|
|
|
|
# Should have been called twice (first chunk + second chunk)
|
|
assert mock_logger.info.call_count == 2
|
|
# First call should include the "return:" prefix
|
|
assert "\n\nreturn: " in mock_logger.info.call_args_list[0][0][0]
|
|
|
|
@patch("api_calls.logger")
|
|
def test_log_return_value_extremely_long_string(self, mock_logger):
|
|
"""Test logging extremely long return values that get snipped."""
|
|
extremely_long_string = (
|
|
"B" * 400000
|
|
) # 400k character string (exceeds 300k limit)
|
|
api_calls.log_return_value(extremely_long_string)
|
|
|
|
# Should warn about snipping
|
|
mock_logger.warning.assert_called_with("[ S N I P P E D ]")
|
|
|
|
def test_team_hash(self):
|
|
"""Test team hash generation."""
|
|
mock_team = {"sname": "TestTeam", "gmid": 1234567}
|
|
|
|
result = api_calls.team_hash(mock_team)
|
|
# Expected format: last char + gmid/6950123 + second-to-last char + gmid/42069123
|
|
expected = f"m{1234567 / 6950123:.0f}a{1234567 / 42069123:.0f}"
|
|
assert result == expected
|
|
|
|
|
|
# Note: Async database function tests are complex due to aiohttp mocking
|
|
# For now, focusing on utility functions which provide significant coverage improvement
|
|
|
|
|
|
class TestSpecificFunctions:
|
|
"""Test specific API wrapper functions."""
|
|
|
|
@pytest.mark.asyncio
|
|
@patch("api_calls.db_get")
|
|
async def test_get_team_by_abbrev_found(self, mock_db_get):
|
|
"""Test get_team_by_abbrev function when team is found."""
|
|
mock_db_get.return_value = {
|
|
"count": 1,
|
|
"teams": [{"id": 123, "abbrev": "TEST", "name": "Test Team"}],
|
|
}
|
|
|
|
result = await api_calls.get_team_by_abbrev("TEST")
|
|
|
|
assert result == {"id": 123, "abbrev": "TEST", "name": "Test Team"}
|
|
mock_db_get.assert_called_once_with("teams", params=[("abbrev", "TEST")])
|
|
|
|
@pytest.mark.asyncio
|
|
@patch("api_calls.db_get")
|
|
async def test_get_team_by_abbrev_not_found(self, mock_db_get):
|
|
"""Test get_team_by_abbrev function when team is not found."""
|
|
mock_db_get.return_value = {"count": 0, "teams": []}
|
|
|
|
result = await api_calls.get_team_by_abbrev("NONEXISTENT")
|
|
|
|
assert result is None
|
|
mock_db_get.assert_called_once_with("teams", params=[("abbrev", "NONEXISTENT")])
|
|
|
|
@pytest.mark.asyncio
|
|
@patch("api_calls.db_post")
|
|
async def test_post_to_dex(self, mock_db_post):
|
|
"""Test post_to_dex function."""
|
|
mock_db_post.return_value = {"id": 456, "posted": True}
|
|
|
|
mock_player = {"id": 123}
|
|
mock_team = {"id": 456}
|
|
|
|
result = await api_calls.post_to_dex(mock_player, mock_team)
|
|
|
|
assert result == {"id": 456, "posted": True}
|
|
mock_db_post.assert_called_once_with(
|
|
"paperdex", payload={"player_id": 123, "team_id": 456}
|
|
)
|
|
|
|
|
|
class TestEnvironmentConfiguration:
|
|
"""Test environment-based configuration."""
|
|
|
|
def test_db_url_exists(self):
|
|
"""Test that DB_URL is configured."""
|
|
assert api_calls.DB_URL is not None
|
|
assert "manticorum.com" in api_calls.DB_URL
|
|
|
|
def test_auth_token_exists(self):
|
|
"""Test that AUTH_TOKEN is configured."""
|
|
assert api_calls.AUTH_TOKEN is not None
|
|
assert "Authorization" in api_calls.AUTH_TOKEN
|
|
|
|
|
|
class TestTimeoutAndRetry:
|
|
"""Test timeout and retry logic for API calls.
|
|
|
|
These tests verify that:
|
|
1. Default timeout values are correctly set
|
|
2. db_get has retry parameter, mutation methods do not
|
|
3. APITimeoutError exception exists and is a subclass of DatabaseError
|
|
"""
|
|
|
|
def test_default_timeout_values(self):
|
|
"""Test that default timeout values are set correctly.
|
|
|
|
Default should be 5 seconds for all functions.
|
|
db_get should have retries parameter, mutation methods should not.
|
|
"""
|
|
import inspect
|
|
|
|
# Check db_get signature - should have both timeout and retries
|
|
sig = inspect.signature(api_calls.db_get)
|
|
assert sig.parameters["timeout"].default == 5
|
|
assert sig.parameters["retries"].default == 3
|
|
|
|
# Check mutation functions - should have timeout but no retries param
|
|
for func_name in ["db_post", "db_patch", "db_put", "db_delete"]:
|
|
func = getattr(api_calls, func_name)
|
|
sig = inspect.signature(func)
|
|
assert sig.parameters["timeout"].default == 5, (
|
|
f"{func_name} should have default timeout=5"
|
|
)
|
|
assert "retries" not in sig.parameters, (
|
|
f"{func_name} should not have retries parameter"
|
|
)
|
|
|
|
def test_api_timeout_error_exists(self):
|
|
"""Test that APITimeoutError exception is properly defined.
|
|
|
|
APITimeoutError should be a subclass of DatabaseError so existing
|
|
error handlers that catch DatabaseError will also catch timeouts.
|
|
"""
|
|
assert issubclass(APITimeoutError, DatabaseError)
|
|
assert issubclass(APITimeoutError, Exception)
|
|
|
|
# Test that it can be instantiated with a message
|
|
error = APITimeoutError("Test timeout message")
|
|
assert "Test timeout message" in str(error)
|
|
|
|
def test_client_timeout_import(self):
|
|
"""Test that ClientTimeout is properly imported from aiohttp.
|
|
|
|
This verifies the timeout functionality can be used.
|
|
"""
|
|
from aiohttp import ClientTimeout
|
|
|
|
# Create a timeout object to verify it works
|
|
timeout = ClientTimeout(total=5)
|
|
assert timeout.total == 5
|