diff --git a/tests/handler_async_test.py b/tests/handler_async_test.py index 2520932..0e1a903 100644 --- a/tests/handler_async_test.py +++ b/tests/handler_async_test.py @@ -2,15 +2,15 @@ import os import sys +import aiohttp +import ipinfo +import pytest +from ipinfo import handler_utils from ipinfo.cache.default import DefaultCache from ipinfo.details import Details -from ipinfo.handler_async import AsyncHandler -from ipinfo import handler_utils from ipinfo.error import APIError from ipinfo.exceptions import RequestQuotaExceededError -import ipinfo -import pytest -import aiohttp +from ipinfo.handler_async import AsyncHandler skip_if_python_3_11_or_later = sys.version_info >= (3, 11) @@ -78,8 +78,7 @@ async def test_get_details(): assert country_flag["unicode"] == "U+1F1FA U+1F1F8" country_flag_url = details.country_flag_url assert ( - country_flag_url - == "https://cdn.ipinfo.io/static/images/countries-flags/US.svg" + country_flag_url == "https://cdn.ipinfo.io/static/images/countries-flags/US.svg" ) country_currency = details.country_currency assert country_currency["code"] == "USD" @@ -132,21 +131,59 @@ async def test_get_details(): await handler.deinit() + @pytest.mark.parametrize( - ("mock_resp_status_code", "mock_resp_headers", "mock_resp_error_msg", "expected_error_json"), + ( + "mock_resp_status_code", + "mock_resp_headers", + "mock_resp_error_msg", + "expected_error_json", + ), [ - pytest.param(503, {"Content-Type": "text/plain"}, "Service Unavailable", {"error": "Service Unavailable"}, id="5xx_not_json"), - pytest.param(403, {"Content-Type": "application/json"}, '{"message": "missing token"}', {"message": "missing token"}, id="4xx_json"), - pytest.param(400, {"Content-Type": "application/json"}, '{"message": "missing field"}', {"message": "missing field"}, id="400"), - ] + pytest.param( + 503, + {"Content-Type": "text/plain"}, + "Service Unavailable", + {"error": "Service Unavailable"}, + id="5xx_not_json", + ), + pytest.param( + 403, + {"Content-Type": "application/json"}, + '{"message": "missing token"}', + {"message": "missing token"}, + id="4xx_json", + ), + pytest.param( + 400, + {"Content-Type": "application/json"}, + '{"message": "missing field"}', + {"message": "missing field"}, + id="400", + ), + ], ) @pytest.mark.asyncio -async def test_get_details_error(monkeypatch, mock_resp_status_code, mock_resp_headers, mock_resp_error_msg, expected_error_json): +async def test_get_details_error( + monkeypatch, + mock_resp_status_code, + mock_resp_headers, + mock_resp_error_msg, + expected_error_json, +): async def mock_get(*args, **kwargs): - response = MockResponse(status=mock_resp_status_code, text=mock_resp_error_msg, headers=mock_resp_headers) + response = MockResponse( + status=mock_resp_status_code, + text=mock_resp_error_msg, + headers=mock_resp_headers, + ) return response - monkeypatch.setattr(aiohttp.ClientSession, 'get', lambda *args, **kwargs: aiohttp.client._RequestContextManager(mock_get())) + monkeypatch.setattr( + aiohttp.ClientSession, + "get", + lambda *args, **kwargs: aiohttp.client._RequestContextManager(mock_get()), + ) token = os.environ.get("IPINFO_TOKEN", "") handler = AsyncHandler(token) with pytest.raises(APIError) as exc_info: @@ -154,18 +191,24 @@ async def mock_get(*args, **kwargs): assert exc_info.value.error_code == mock_resp_status_code assert exc_info.value.error_json == expected_error_json + @pytest.mark.asyncio async def test_get_details_quota_error(monkeypatch): async def mock_get(*args, **kwargs): response = MockResponse(status=429, text="Quota exceeded", headers={}) return response - monkeypatch.setattr(aiohttp.ClientSession, 'get', lambda *args, **kwargs: aiohttp.client._RequestContextManager(mock_get())) + monkeypatch.setattr( + aiohttp.ClientSession, + "get", + lambda *args, **kwargs: aiohttp.client._RequestContextManager(mock_get()), + ) token = os.environ.get("IPINFO_TOKEN", "") handler = AsyncHandler(token) with pytest.raises(RequestQuotaExceededError): await handler.getDetails("8.8.8.8") + ############# # BATCH TESTS ############# @@ -198,7 +241,9 @@ def _check_batch_details(ips, details, token): assert "domains" in d -@pytest.mark.skipif(skip_if_python_3_11_or_later, reason="Requires Python 3.10 or earlier") +@pytest.mark.skipif( + skip_if_python_3_11_or_later, reason="Requires Python 3.10 or earlier" +) @pytest.mark.parametrize("batch_size", [None, 1, 2, 3]) @pytest.mark.asyncio async def test_get_batch_details(batch_size): @@ -229,15 +274,15 @@ async def test_get_iterative_batch_details(batch_size): _check_iterative_batch_details(ips, details, token) -@pytest.mark.skipif(skip_if_python_3_11_or_later, reason="Requires Python 3.10 or earlier") +@pytest.mark.skipif( + skip_if_python_3_11_or_later, reason="Requires Python 3.10 or earlier" +) @pytest.mark.parametrize("batch_size", [None, 1, 2, 3]) @pytest.mark.asyncio async def test_get_batch_details_total_timeout(batch_size): handler, token, ips = _prepare_batch_test() with pytest.raises(ipinfo.exceptions.TimeoutExceededError): - await handler.getBatchDetails( - ips, batch_size=batch_size, timeout_total=0.001 - ) + await handler.getBatchDetails(ips, batch_size=batch_size, timeout_total=0.001) await handler.deinit() @@ -260,30 +305,65 @@ async def test_bogon_details(): @pytest.mark.asyncio -async def test_get_resproxy(): - token = os.environ.get("IPINFO_TOKEN", "") - if not token: - pytest.skip("token required for resproxy tests") - handler = AsyncHandler(token) - # Use an IP known to be a residential proxy (from API documentation) +async def test_get_resproxy(monkeypatch): + mock_response = MockResponse( + json.dumps( + { + "ip": "175.107.211.204", + "last_seen": "2025-01-20", + "percent_days_seen": 0.85, + "service": "example_service", + } + ), + 200, + {"Content-Type": "application/json"}, + ) + + def mock_get(*args, **kwargs): + return mock_response + + handler = AsyncHandler("test_token") + handler._ensure_aiohttp_ready() + monkeypatch.setattr(handler.httpsess, "get", mock_get) + details = await handler.getResproxy("175.107.211.204") assert isinstance(details, Details) assert details.ip == "175.107.211.204" - assert details.last_seen is not None - assert details.percent_days_seen is not None - assert details.service is not None + assert details.last_seen == "2025-01-20" + assert details.percent_days_seen == 0.85 + assert details.service == "example_service" await handler.deinit() @pytest.mark.asyncio -async def test_get_resproxy_caching(): - token = os.environ.get("IPINFO_TOKEN", "") - if not token: - pytest.skip("token required for resproxy tests") - handler = AsyncHandler(token) +async def test_get_resproxy_caching(monkeypatch): + call_count = 0 + + def mock_get(*args, **kwargs): + nonlocal call_count + call_count += 1 + return MockResponse( + json.dumps( + { + "ip": "175.107.211.204", + "last_seen": "2025-01-20", + "percent_days_seen": 0.85, + "service": "example_service", + } + ), + 200, + {"Content-Type": "application/json"}, + ) + + handler = AsyncHandler("test_token") + handler._ensure_aiohttp_ready() + monkeypatch.setattr(handler.httpsess, "get", mock_get) + # First call should hit the API details1 = await handler.getResproxy("175.107.211.204") # Second call should hit the cache details2 = await handler.getResproxy("175.107.211.204") assert details1.ip == details2.ip - await handler.deinit() \ No newline at end of file + # Verify only one API call was made (second was cached) + assert call_count == 1 + await handler.deinit() diff --git a/tests/handler_test.py b/tests/handler_test.py index c16cdbe..8fe21ae 100644 --- a/tests/handler_test.py +++ b/tests/handler_test.py @@ -243,27 +243,60 @@ def test_iterative_bogon_details(): ################# -def test_get_resproxy(): - token = os.environ.get("IPINFO_TOKEN", "") - if not token: - pytest.skip("token required for resproxy tests") +def test_get_resproxy(monkeypatch): + def mock_get(*args, **kwargs): + response = requests.Response() + response.status_code = 200 + response.headers = {"Content-Type": "application/json"} + response._content = b'{"ip": "175.107.211.204", "last_seen": "2025-01-20", "percent_days_seen": 0.85, "service": "example_service"}' + return response + + monkeypatch.setattr(requests, "get", mock_get) + token = "test_token" handler = Handler(token) - # Use an IP known to be a residential proxy (from API documentation) details = handler.getResproxy("175.107.211.204") assert isinstance(details, Details) assert details.ip == "175.107.211.204" - assert details.last_seen is not None - assert details.percent_days_seen is not None - assert details.service is not None + assert details.last_seen == "2025-01-20" + assert details.percent_days_seen == 0.85 + assert details.service == "example_service" -def test_get_resproxy_caching(): - token = os.environ.get("IPINFO_TOKEN", "") - if not token: - pytest.skip("token required for resproxy tests") +def test_get_resproxy_caching(monkeypatch): + call_count = 0 + + def mock_get(*args, **kwargs): + nonlocal call_count + call_count += 1 + response = requests.Response() + response.status_code = 200 + response.headers = {"Content-Type": "application/json"} + response._content = b'{"ip": "175.107.211.204", "last_seen": "2025-01-20", "percent_days_seen": 0.85, "service": "example_service"}' + return response + + monkeypatch.setattr(requests, "get", mock_get) + token = "test_token" handler = Handler(token) # First call should hit the API details1 = handler.getResproxy("175.107.211.204") # Second call should hit the cache details2 = handler.getResproxy("175.107.211.204") - assert details1.ip == details2.ip \ No newline at end of file + assert details1.ip == details2.ip + # Verify only one API call was made (second was cached) + assert call_count == 1 + + +def test_get_resproxy_empty(monkeypatch): + def mock_get(*args, **kwargs): + response = requests.Response() + response.status_code = 200 + response.headers = {"Content-Type": "application/json"} + response._content = b"{}" + return response + + monkeypatch.setattr(requests, "get", mock_get) + token = "test_token" + handler = Handler(token) + details = handler.getResproxy("8.8.8.8") + assert isinstance(details, Details) + assert details.all == {}