386 lines
14 KiB
Python
386 lines
14 KiB
Python
"""
|
|
Unit tests for HTTP Client service.
|
|
|
|
Tests cover:
|
|
- Singleton pattern
|
|
- URL validation and security
|
|
- Cache key generation
|
|
- Cache entry management
|
|
- Cache-Control header parsing
|
|
- Rate limiting
|
|
- GET/POST request handling
|
|
"""
|
|
import sys
|
|
import unittest
|
|
import json
|
|
import hashlib
|
|
import time
|
|
import tempfile
|
|
import shutil
|
|
from pathlib import Path
|
|
from unittest.mock import MagicMock, patch, mock_open
|
|
|
|
# Add project root to path
|
|
project_root = Path(__file__).parent.parent.parent
|
|
if str(project_root) not in sys.path:
|
|
sys.path.insert(0, str(project_root))
|
|
|
|
# Mock requests before importing http_client
|
|
sys.modules['requests'] = MagicMock()
|
|
sys.modules['requests.adapters'] = MagicMock()
|
|
sys.modules['urllib3'] = MagicMock()
|
|
sys.modules['urllib3.util'] = MagicMock()
|
|
sys.modules['urllib3.util.retry'] = MagicMock()
|
|
|
|
from core.http_client import (
|
|
HTTPClient, get_http_client, HTTPClientError, URLSecurityError,
|
|
CacheEntry
|
|
)
|
|
|
|
|
|
class TestHTTPClientSingleton(unittest.TestCase):
|
|
"""Test HTTPClient singleton behavior."""
|
|
|
|
def tearDown(self):
|
|
"""Reset singleton after each test."""
|
|
HTTPClient._instance = None
|
|
global _http_client
|
|
import core.http_client
|
|
core.http_client._http_client = None
|
|
|
|
@patch('core.http_client.REQUESTS_AVAILABLE', True)
|
|
def test_singleton_instance(self):
|
|
"""Test that HTTPClient is a proper singleton."""
|
|
with patch.object(HTTPClient, '__init__', return_value=None):
|
|
HTTPClient._instance = None
|
|
client1 = HTTPClient.__new__(HTTPClient)
|
|
client1._initialized = True
|
|
client2 = HTTPClient.__new__(HTTPClient)
|
|
|
|
self.assertIs(client1, client2)
|
|
|
|
@patch('core.http_client.REQUESTS_AVAILABLE', True)
|
|
def test_get_http_client_singleton(self):
|
|
"""Test get_http_client returns singleton."""
|
|
import core.http_client
|
|
core.http_client._http_client = None
|
|
|
|
with patch.object(HTTPClient, '__init__', return_value=None):
|
|
client1 = get_http_client()
|
|
client2 = get_http_client()
|
|
|
|
self.assertIs(client1, client2)
|
|
|
|
|
|
class TestURLValidation(unittest.TestCase):
|
|
"""Test URL validation and security."""
|
|
|
|
def setUp(self):
|
|
"""Set up test client."""
|
|
HTTPClient._instance = None
|
|
with patch('core.http_client.REQUESTS_AVAILABLE', True):
|
|
with patch.object(HTTPClient, '_initialized', False):
|
|
self.client = MagicMock(spec=HTTPClient)
|
|
self.client._validate_url = HTTPClient._validate_url.__get__(self.client, HTTPClient)
|
|
|
|
def tearDown(self):
|
|
"""Reset singleton."""
|
|
HTTPClient._instance = None
|
|
|
|
def test_valid_http_url(self):
|
|
"""Test valid HTTP URL."""
|
|
result = self.client._validate_url("http://example.com/api")
|
|
self.assertEqual(result, "http://example.com/api")
|
|
|
|
def test_valid_https_url(self):
|
|
"""Test valid HTTPS URL."""
|
|
result = self.client._validate_url("https://api.example.com/v1/data")
|
|
self.assertEqual(result, "https://api.example.com/v1/data")
|
|
|
|
def test_empty_url(self):
|
|
"""Test empty URL."""
|
|
with self.assertRaises(URLSecurityError) as context:
|
|
self.client._validate_url("")
|
|
self.assertIn("cannot be empty", str(context.exception).lower())
|
|
|
|
def test_invalid_scheme(self):
|
|
"""Test URL with invalid scheme."""
|
|
with self.assertRaises(URLSecurityError) as context:
|
|
self.client._validate_url("ftp://example.com/file")
|
|
self.assertIn("scheme", str(context.exception).lower())
|
|
|
|
def test_path_traversal_in_url(self):
|
|
"""Test URL with path traversal."""
|
|
with self.assertRaises(URLSecurityError) as context:
|
|
self.client._validate_url("http://example.com/../../../etc/passwd")
|
|
self.assertIn("dangerous", str(context.exception).lower())
|
|
|
|
def test_null_byte_in_url(self):
|
|
"""Test URL with null byte."""
|
|
with self.assertRaises(URLSecurityError) as context:
|
|
self.client._validate_url("http://example.com/file\x00.txt")
|
|
self.assertIn("dangerous", str(context.exception).lower())
|
|
|
|
def test_at_symbol_in_url(self):
|
|
"""Test URL with @ symbol (credential injection)."""
|
|
with self.assertRaises(URLSecurityError) as context:
|
|
self.client._validate_url("http://user@evil.com@example.com")
|
|
self.assertIn("dangerous", str(context.exception).lower())
|
|
|
|
def test_backslash_in_url(self):
|
|
"""Test URL with backslash."""
|
|
with self.assertRaises(URLSecurityError) as context:
|
|
self.client._validate_url("http://example.com\\admin")
|
|
self.assertIn("dangerous", str(context.exception).lower())
|
|
|
|
def test_private_ip_blocked(self):
|
|
"""Test blocking private IP addresses."""
|
|
with self.assertRaises(URLSecurityError) as context:
|
|
self.client._validate_url("http://192.168.1.1/admin")
|
|
self.assertIn("restricted", str(context.exception).lower())
|
|
|
|
def test_loopback_ip_blocked(self):
|
|
"""Test blocking loopback IP addresses."""
|
|
with self.assertRaises(URLSecurityError) as context:
|
|
self.client._validate_url("http://127.0.0.1/secret")
|
|
self.assertIn("restricted", str(context.exception).lower())
|
|
|
|
|
|
class TestCacheKeyGeneration(unittest.TestCase):
|
|
"""Test cache key generation."""
|
|
|
|
def setUp(self):
|
|
"""Set up test client."""
|
|
HTTPClient._instance = None
|
|
with patch('core.http_client.REQUESTS_AVAILABLE', True):
|
|
self.client = MagicMock(spec=HTTPClient)
|
|
self.client._generate_cache_key = HTTPClient._generate_cache_key.__get__(self.client, HTTPClient)
|
|
|
|
def tearDown(self):
|
|
"""Reset singleton."""
|
|
HTTPClient._instance = None
|
|
|
|
def test_cache_key_without_params(self):
|
|
"""Test cache key generation without params."""
|
|
key = self.client._generate_cache_key("https://api.example.com/data")
|
|
|
|
self.assertIsInstance(key, str)
|
|
self.assertEqual(len(key), 64) # SHA256 hex length
|
|
|
|
def test_cache_key_with_params(self):
|
|
"""Test cache key generation with params."""
|
|
key1 = self.client._generate_cache_key(
|
|
"https://api.example.com/data",
|
|
{"key1": "value1", "key2": "value2"}
|
|
)
|
|
|
|
# Same params in different order should produce same key
|
|
key2 = self.client._generate_cache_key(
|
|
"https://api.example.com/data",
|
|
{"key2": "value2", "key1": "value1"}
|
|
)
|
|
|
|
self.assertEqual(key1, key2)
|
|
|
|
def test_cache_key_different_urls(self):
|
|
"""Test different URLs produce different keys."""
|
|
key1 = self.client._generate_cache_key("https://api.example.com/data1")
|
|
key2 = self.client._generate_cache_key("https://api.example.com/data2")
|
|
|
|
self.assertNotEqual(key1, key2)
|
|
|
|
|
|
class TestCachePathGeneration(unittest.TestCase):
|
|
"""Test cache path generation."""
|
|
|
|
def setUp(self):
|
|
"""Set up test client."""
|
|
HTTPClient._instance = None
|
|
with patch('core.http_client.REQUESTS_AVAILABLE', True):
|
|
with patch.object(HTTPClient, '__init__', return_value=None):
|
|
self.client = HTTPClient.__new__(HTTPClient)
|
|
self.client.cache_dir = Path("/cache")
|
|
|
|
def tearDown(self):
|
|
"""Reset singleton."""
|
|
HTTPClient._instance = None
|
|
|
|
def test_cache_path_structure(self):
|
|
"""Test cache path uses subdirectory structure."""
|
|
cache_key = "abcdef1234567890" * 4
|
|
path = self.client._get_cache_path(cache_key)
|
|
|
|
self.assertIn("ab", str(path))
|
|
self.assertIn("cd", str(path))
|
|
self.assertTrue(str(path).endswith(f"{cache_key}.json"))
|
|
|
|
|
|
class TestCacheControlParsing(unittest.TestCase):
|
|
"""Test Cache-Control header parsing."""
|
|
|
|
def setUp(self):
|
|
"""Set up test client."""
|
|
HTTPClient._instance = None
|
|
with patch('core.http_client.REQUESTS_AVAILABLE', True):
|
|
self.client = MagicMock(spec=HTTPClient)
|
|
self.client._parse_cache_control = HTTPClient._parse_cache_control.__get__(self.client, HTTPClient)
|
|
|
|
def tearDown(self):
|
|
"""Reset singleton."""
|
|
HTTPClient._instance = None
|
|
|
|
def test_parse_max_age(self):
|
|
"""Test parsing max-age directive."""
|
|
result = self.client._parse_cache_control("max-age=3600")
|
|
self.assertEqual(result, 3600)
|
|
|
|
def test_parse_no_cache(self):
|
|
"""Test parsing no-cache directive."""
|
|
result = self.client._parse_cache_control("no-cache")
|
|
self.assertEqual(result, 0)
|
|
|
|
def test_parse_no_store(self):
|
|
"""Test parsing no-store directive."""
|
|
result = self.client._parse_cache_control("no-store, max-age=0")
|
|
self.assertEqual(result, 0)
|
|
|
|
def test_parse_multiple_directives(self):
|
|
"""Test parsing multiple directives."""
|
|
result = self.client._parse_cache_control("public, max-age=7200, must-revalidate")
|
|
self.assertEqual(result, 7200)
|
|
|
|
def test_parse_empty_header(self):
|
|
"""Test parsing empty header."""
|
|
result = self.client._parse_cache_control("")
|
|
self.assertIsNone(result)
|
|
|
|
def test_parse_none_header(self):
|
|
"""Test parsing None header."""
|
|
result = self.client._parse_cache_control(None)
|
|
self.assertIsNone(result)
|
|
|
|
|
|
class TestCacheValidity(unittest.TestCase):
|
|
"""Test cache entry validity checking."""
|
|
|
|
def setUp(self):
|
|
"""Set up test client."""
|
|
HTTPClient._instance = None
|
|
with patch('core.http_client.REQUESTS_AVAILABLE', True):
|
|
with patch.object(HTTPClient, '__init__', return_value=None):
|
|
self.client = HTTPClient.__new__(HTTPClient)
|
|
|
|
def tearDown(self):
|
|
"""Reset singleton."""
|
|
HTTPClient._instance = None
|
|
|
|
def test_valid_cache_entry(self):
|
|
"""Test valid (non-expired) cache entry."""
|
|
entry = MagicMock(spec=CacheEntry)
|
|
entry.expires_at = time.time() + 3600 # 1 hour from now
|
|
|
|
result = self.client._is_cache_valid(entry)
|
|
self.assertTrue(result)
|
|
|
|
def test_expired_cache_entry(self):
|
|
"""Test expired cache entry."""
|
|
entry = MagicMock(spec=CacheEntry)
|
|
entry.expires_at = time.time() - 3600 # 1 hour ago
|
|
|
|
result = self.client._is_cache_valid(entry)
|
|
self.assertFalse(result)
|
|
|
|
|
|
class TestCacheEntryDataclass(unittest.TestCase):
|
|
"""Test CacheEntry dataclass."""
|
|
|
|
def test_cache_entry_creation(self):
|
|
"""Test creating a cache entry."""
|
|
entry = CacheEntry(
|
|
url="https://api.example.com/data",
|
|
status_code=200,
|
|
headers={"Content-Type": "application/json"},
|
|
content=b'{"key": "value"}',
|
|
cached_at=time.time(),
|
|
expires_at=time.time() + 3600
|
|
)
|
|
|
|
self.assertEqual(entry.url, "https://api.example.com/data")
|
|
self.assertEqual(entry.status_code, 200)
|
|
self.assertEqual(entry.content, b'{"key": "value"}')
|
|
|
|
def test_cache_entry_optional_fields(self):
|
|
"""Test cache entry with optional fields."""
|
|
entry = CacheEntry(
|
|
url="https://api.example.com/data",
|
|
status_code=200,
|
|
headers={},
|
|
content=b'{}',
|
|
cached_at=time.time(),
|
|
expires_at=time.time() + 3600,
|
|
cache_control="max-age=3600",
|
|
etag='"abc123"',
|
|
last_modified="Wed, 21 Oct 2024 07:28:00 GMT"
|
|
)
|
|
|
|
self.assertEqual(entry.etag, '"abc123"')
|
|
self.assertEqual(entry.last_modified, "Wed, 21 Oct 2024 07:28:00 GMT")
|
|
|
|
|
|
class TestHTTPClientInitialization(unittest.TestCase):
|
|
"""Test HTTPClient initialization."""
|
|
|
|
def tearDown(self):
|
|
"""Reset singleton."""
|
|
HTTPClient._instance = None
|
|
import core.http_client
|
|
core.http_client._http_client = None
|
|
|
|
@patch('core.http_client.REQUESTS_AVAILABLE', False)
|
|
def test_init_without_requests(self):
|
|
"""Test initialization when requests not available."""
|
|
HTTPClient._instance = None
|
|
with self.assertRaises(RuntimeError) as context:
|
|
HTTPClient()
|
|
self.assertIn("requests", str(context.exception).lower())
|
|
|
|
@patch('core.http_client.REQUESTS_AVAILABLE', True)
|
|
def test_init_creates_cache_dir(self):
|
|
"""Test initialization creates cache directory."""
|
|
HTTPClient._instance = None
|
|
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
|
cache_dir = Path(tmpdir) / "test_cache"
|
|
|
|
with patch.object(HTTPClient, '_initialized', False):
|
|
client = HTTPClient.__new__(HTTPClient)
|
|
client.cache_dir = cache_dir
|
|
client._initialized = False
|
|
client.__init__(cache_dir=str(cache_dir))
|
|
|
|
self.assertTrue(cache_dir.exists())
|
|
|
|
|
|
class TestHTTPExceptions(unittest.TestCase):
|
|
"""Test HTTP client exceptions."""
|
|
|
|
def test_http_client_error_is_exception(self):
|
|
"""Test HTTPClientError is an Exception."""
|
|
self.assertTrue(issubclass(HTTPClientError, Exception))
|
|
|
|
def test_url_security_error_is_http_client_error(self):
|
|
"""Test URLSecurityError is an HTTPClientError."""
|
|
self.assertTrue(issubclass(URLSecurityError, HTTPClientError))
|
|
|
|
def test_error_messages(self):
|
|
"""Test error message handling."""
|
|
error = HTTPClientError("Test error")
|
|
self.assertEqual(str(error), "Test error")
|
|
|
|
security_error = URLSecurityError("Security violation")
|
|
self.assertEqual(str(security_error), "Security violation")
|
|
|
|
|
|
if __name__ == '__main__':
|
|
unittest.main()
|