mirror of
https://github.com/tiennm99/litellm.git
synced 2026-08-11 16:26:07 +00:00
test: mock DNS resolution, hoist httpx import to module level
Greptile P1: six tests in test_url_utils.py performed real DNS lookups to example.com, violating the tests/test_litellm/ mock-only rule and risking offline CI failures. Add mock_dns_public and mock_dns_failure fixtures that monkeypatch socket.getaddrinfo on the url_utils module. Greptile P2: move 'import httpx' from inside _extract_redirect_url to module-level imports per CLAUDE.md style guide.
This commit is contained in:
@@ -14,6 +14,8 @@ from ipaddress import ip_address, ip_network
|
||||
from typing import Any, Tuple
|
||||
from urllib.parse import urlparse, urlunparse
|
||||
|
||||
import httpx
|
||||
|
||||
import litellm
|
||||
|
||||
_BLOCKED_NETWORKS = [
|
||||
@@ -142,8 +144,6 @@ _MAX_REDIRECTS = 10
|
||||
|
||||
def _extract_redirect_url(response: Any, request_url: str) -> str:
|
||||
"""Extract and resolve the redirect target from a response's Location header."""
|
||||
import httpx
|
||||
|
||||
location = response.headers.get("location")
|
||||
if not location:
|
||||
raise SSRFError("Redirect response has no Location header")
|
||||
|
||||
@@ -1,9 +1,34 @@
|
||||
import socket
|
||||
|
||||
import pytest
|
||||
|
||||
import litellm
|
||||
from litellm.litellm_core_utils import url_utils
|
||||
from litellm.litellm_core_utils.url_utils import SSRFError, _is_blocked_ip, validate_url
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def mock_dns_public(monkeypatch):
|
||||
"""Resolve any hostname to 93.184.216.34 (public)."""
|
||||
|
||||
def fake_getaddrinfo(host, port, *args, **kwargs):
|
||||
return [
|
||||
(socket.AF_INET, socket.SOCK_STREAM, 6, "", ("93.184.216.34", port or 80))
|
||||
]
|
||||
|
||||
monkeypatch.setattr(url_utils.socket, "getaddrinfo", fake_getaddrinfo)
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def mock_dns_failure(monkeypatch):
|
||||
"""Make every DNS lookup raise gaierror."""
|
||||
|
||||
def fake_getaddrinfo(host, port, *args, **kwargs):
|
||||
raise socket.gaierror("Name or service not known")
|
||||
|
||||
monkeypatch.setattr(url_utils.socket, "getaddrinfo", fake_getaddrinfo)
|
||||
|
||||
|
||||
class TestIsBlockedIp:
|
||||
def test_blocks_private(self):
|
||||
assert _is_blocked_ip("10.0.0.1") is True
|
||||
@@ -48,22 +73,22 @@ class TestValidateUrl:
|
||||
with pytest.raises(SSRFError):
|
||||
validate_url("http:///path")
|
||||
|
||||
def test_allows_public_https(self):
|
||||
def test_allows_public_https(self, mock_dns_public):
|
||||
rewritten, host = validate_url("https://example.com/image.png")
|
||||
assert host == "example.com"
|
||||
assert rewritten == "https://example.com/image.png"
|
||||
|
||||
def test_rewrites_public_http_to_ip(self):
|
||||
def test_rewrites_public_http_to_ip(self, mock_dns_public):
|
||||
rewritten, host = validate_url("http://example.com/image.png")
|
||||
assert host == "example.com"
|
||||
assert "example.com" not in rewritten
|
||||
|
||||
def test_preserves_path_and_query(self):
|
||||
def test_preserves_path_and_query(self, mock_dns_public):
|
||||
rewritten, host = validate_url("http://example.com/path?key=value")
|
||||
assert "/path" in rewritten
|
||||
assert "key=value" in rewritten
|
||||
|
||||
def test_dns_failure_raises(self):
|
||||
def test_dns_failure_raises(self, mock_dns_failure):
|
||||
with pytest.raises(SSRFError, match="DNS resolution failed"):
|
||||
validate_url("http://this-domain-does-not-exist-xyz123.invalid/test")
|
||||
|
||||
@@ -75,13 +100,17 @@ class TestValidateUrl:
|
||||
with pytest.raises(SSRFError):
|
||||
validate_url("http://[::1]/")
|
||||
|
||||
def test_https_rewrites_when_ssl_verify_disabled(self, monkeypatch):
|
||||
def test_https_rewrites_when_ssl_verify_disabled(
|
||||
self, monkeypatch, mock_dns_public
|
||||
):
|
||||
monkeypatch.setattr(litellm, "ssl_verify", False)
|
||||
rewritten, host = validate_url("https://example.com/image.png")
|
||||
assert host == "example.com"
|
||||
assert "example.com" not in rewritten # rewritten to IP
|
||||
|
||||
def test_https_not_rewritten_when_ssl_verify_enabled(self, monkeypatch):
|
||||
def test_https_not_rewritten_when_ssl_verify_enabled(
|
||||
self, monkeypatch, mock_dns_public
|
||||
):
|
||||
monkeypatch.setattr(litellm, "ssl_verify", True)
|
||||
rewritten, host = validate_url("https://example.com/image.png")
|
||||
assert rewritten == "https://example.com/image.png"
|
||||
|
||||
Reference in New Issue
Block a user