handler_test.py

import os

from ipinfo.cache.default import DefaultCache
from ipinfo.details import Details
from ipinfo.handler import Handler
from ipinfo import handler_utils
from ipinfo.error import APIError
from ipinfo.exceptions import RequestQuotaExceededError
import ipinfo
import pytest
import requests


def test_init():
    token = "mytesttoken"
    handler = Handler(token)
    assert handler.access_token == token
    assert isinstance(handler.cache, DefaultCache)
    assert "US" in handler.countries


def test_headers():
    token = "mytesttoken"
    handler = Handler(token, headers={"custom_field": "yes"})
    headers = handler_utils.get_headers(token, handler.headers)

    assert "user-agent" in headers
    assert "accept" in headers
    assert "authorization" in headers
    assert "custom_field" in headers


def test_get_details():
    token = os.environ.get("IPINFO_TOKEN", "")
    handler = Handler(token)
    details = handler.getDetails("8.8.8.8")
    assert isinstance(details, Details)
    assert details.ip == "8.8.8.8"
    assert details.hostname == "dns.google"
    assert details.city == "Mountain View"
    assert details.region == "California"
    assert details.country == "US"
    assert details.country_name == "United States"
    assert details.isEU == False
    country_flag = details.country_flag
    assert country_flag["emoji"] == "🇺🇸"
    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_currency = details.country_currency
    assert country_currency["code"] == "USD"
    assert country_currency["symbol"] == "$"
    continent = details.continent
    assert continent["code"] == "NA"
    assert continent["name"] == "North America"
    assert details.loc is not None
    assert details.latitude is not None
    assert details.longitude is not None
    assert details.postal == "94043"
    assert details.timezone == "America/Los_Angeles"
    if token:
        asn = details.asn
        assert asn["asn"] == "AS15169"
        assert asn["name"] == "Google LLC"
        assert asn["domain"] == "google.com"
        assert asn["route"] == "8.8.8.0/24"
        assert asn["type"] == "hosting"

        company = details.company
        assert company["name"] == "Google LLC"
        assert company["domain"] == "google.com"
        assert company["type"] == "hosting"

        privacy = details.privacy
        assert privacy["vpn"] == False
        assert privacy["proxy"] == False
        assert privacy["tor"] == False
        assert privacy["relay"] == False
        assert privacy["hosting"] == True
        assert privacy["service"] == ""

        abuse = details.abuse
        assert (
            abuse["address"]
            == "US, CA, Mountain View, 1600 Amphitheatre Parkway, 94043"
        )
        assert abuse["country"] == "US"
        assert abuse["email"] == "network-abuse@google.com"
        assert abuse["name"] == "Abuse"
        assert abuse["network"] == "8.8.8.0/24"
        assert abuse["phone"] == "+1-650-253-0000"

        domains = details.domains
        assert domains["ip"] == "8.8.8.8"
        # NOTE: actual number changes too much
        assert "total" in domains
        assert len(domains["domains"]) == 5

@pytest.mark.parametrize(
    ("mock_resp_status_code", "mock_resp_headers", "mock_resp_error_msg", "expected_error_json"),
    [
        pytest.param(503, {"Content-Type": "text/plain"}, b"Service Unavailable", {"error": "Service Unavailable"}, id="5xx_not_json"),
        pytest.param(403, {"Content-Type": "application/json"}, b'{"message": "missing token"}', {"message": "missing token"}, id="4xx_json"),
        pytest.param(400, {"Content-Type": "application/json"}, b'{"message": "missing field"}', {"message": "missing field"}, id="400"),
    ]
)
def test_get_details_error(monkeypatch, mock_resp_status_code, mock_resp_headers, mock_resp_error_msg, expected_error_json):
    def mock_get(*args, **kwargs):
        response = requests.Response()
        response.status_code = mock_resp_status_code
        response.headers = mock_resp_headers
        response._content = mock_resp_error_msg
        return response

    monkeypatch.setattr(requests, 'get', mock_get)
    token = os.environ.get("IPINFO_TOKEN", "")
    handler = Handler(token)

    with pytest.raises(APIError) as exc_info:
        handler.getDetails("8.8.8.8")
    assert exc_info.value.error_code == mock_resp_status_code
    assert exc_info.value.error_json == expected_error_json

def test_get_details_quota_error(monkeypatch):
    def mock_get(*args, **kwargs):
        response = requests.Response()
        response.status_code = 429
        return response

    monkeypatch.setattr(requests, 'get', mock_get)
    token = os.environ.get("IPINFO_TOKEN", "")
    handler = Handler(token)

    with pytest.raises(RequestQuotaExceededError):
        handler.getDetails("8.8.8.8")

#############
# BATCH TESTS
#############

_batch_ip_addrs = ["1.1.1.1", "8.8.8.8", "9.9.9.9"]


def _prepare_batch_test():
    """Helper for preparing batch test cases."""
    token = os.environ.get("IPINFO_TOKEN", "")
    if not token:
        pytest.skip("token required for batch tests")
    handler = Handler(token)
    return handler, token, _batch_ip_addrs


def _check_batch_details(ips, details, token):
    """Helper for batch tests."""
    for ip in ips:
        assert ip in details
        d = details[ip]
        assert d["ip"] == ip
        assert "country" in d
        assert "country_name" in d
        if token:
            assert "asn" in d
            assert "company" in d
            assert "privacy" in d
            assert "abuse" in d
            assert "domains" in d


def _check_iterative_batch_details(details, token):
    """Helper for iterative batch tests."""
    assert "ip" in details, "Key 'ip' not found in details"
    assert "country" in details, "Key 'country' not found in details"
    assert "city" in details, "Key 'city' not found in details"
    if token:
        assert "asn" in details, "Key 'asn' not found in details"
        assert "company" in details, "Key 'company' not found in details"
        assert "privacy" in details, "Key 'privacy' not found in details"
        assert "abuse" in details, "Key 'abuse' not found in details"
        assert "domains" in details, "Key 'domains' not found in details"


@pytest.mark.parametrize("batch_size", [None, 1, 2, 3])
def test_get_batch_details(batch_size):
    handler, token, ips = _prepare_batch_test()
    details = handler.getBatchDetails(ips, batch_size=batch_size)
    _check_batch_details(ips, details, token)


@pytest.mark.parametrize("batch_size", [1, 2])
def test_get_batch_details_total_timeout(batch_size):
    handler, token, ips = _prepare_batch_test()
    with pytest.raises(ipinfo.exceptions.TimeoutExceededError):
        handler.getBatchDetails(
            ips, batch_size=batch_size, timeout_total=0.001
        )


@pytest.mark.parametrize("batch_size", [None, 1, 2, 3])
def test_get_iterative_batch_details(batch_size):
    handler, token, ips = _prepare_batch_test()
    details_iterator = handler.getBatchDetailsIter(ips, batch_size=batch_size)
    for details in details_iterator:
        _check_iterative_batch_details(details, token)


#############
# MAP TESTS
#############

# Disabled temporarily
#
# def test_get_map():
#     handler = Handler()
#     mapUrl = handler.getMap(open("tests/map-ips.txt").read().splitlines())
#     print(f"got URL={mapUrl}")


#############
# BOGON TESTS
#############


def test_bogon_details():
    token = os.environ.get("IPINFO_TOKEN", "")
    handler = Handler(token)
    details = handler.getDetails("127.0.0.1")
    assert isinstance(details, Details)
    assert details.all == {"bogon": True, "ip": "127.0.0.1"}


def test_iterative_bogon_details():
    token = os.environ.get("IPINFO_TOKEN", "")
    handler = Handler(token)
    details = next(handler.getBatchDetailsIter(["127.0.0.1"]))
    assert details.all == {"bogon": True, "ip": "127.0.0.1"}


#################
# RESPROXY TESTS
#################


def test_get_resproxy():
    token = os.environ.get("IPINFO_TOKEN", "")
    if not token:
        pytest.skip("token required for resproxy tests")
    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


def test_get_resproxy_caching():
    token = os.environ.get("IPINFO_TOKEN", "")
    if not token:
        pytest.skip("token required for resproxy tests")
    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