"""
Tests for HTTP helper functions (fetch_url, fetch_json).
Uses aioresponses to properly mock aiohttp ClientSession, which creates
its own session internally rather than accepting an injected one.
"""
import pytest
from aiohttp import ClientResponseError
from aioresponses import aioresponses
from threat_intel_mcp.server import fetch_url, fetch_json
class TestFetchURL:
"""Test fetch_url function."""
@pytest.mark.asyncio
async def test_fetch_url_success(self):
"""Should successfully fetch URL content."""
expected_text = "Test content"
with aioresponses() as mocked:
mocked.get("http://example.com/test", body=expected_text)
result = await fetch_url("http://example.com/test")
assert result == expected_text
@pytest.mark.asyncio
async def test_fetch_url_with_headers(self):
"""Should pass custom headers to request."""
expected_text = "Test content"
custom_headers = {"Authorization": "Bearer token123"}
with aioresponses() as mocked:
mocked.get("http://example.com/test", body=expected_text)
result = await fetch_url(
"http://example.com/test", headers=custom_headers
)
assert result == expected_text
@pytest.mark.asyncio
async def test_fetch_url_with_timeout(self):
"""Should use custom timeout without error."""
expected_text = "Test content"
with aioresponses() as mocked:
mocked.get("http://example.com/test", body=expected_text)
result = await fetch_url(
"http://example.com/test", timeout=60
)
assert result == expected_text
@pytest.mark.asyncio
async def test_fetch_url_http_error(self):
"""Should raise exception on HTTP error."""
with aioresponses() as mocked:
mocked.get("http://example.com/notfound", status=404)
with pytest.raises(ClientResponseError):
await fetch_url("http://example.com/notfound")
class TestFetchJSON:
"""Test fetch_json function."""
@pytest.mark.asyncio
async def test_fetch_json_success(self):
"""Should successfully fetch and parse JSON."""
expected_data = {"key": "value", "number": 42}
with aioresponses() as mocked:
mocked.get(
"http://example.com/api/test",
payload=expected_data,
)
result = await fetch_json("http://example.com/api/test")
assert result == expected_data
@pytest.mark.asyncio
async def test_fetch_json_with_headers(self):
"""Should pass custom headers to JSON request."""
expected_data = {"result": "success"}
custom_headers = {"x-api-key": "test-key"}
with aioresponses() as mocked:
mocked.get(
"http://example.com/api/test",
payload=expected_data,
)
result = await fetch_json(
"http://example.com/api/test", headers=custom_headers
)
assert result == expected_data
@pytest.mark.asyncio
async def test_fetch_json_array(self):
"""Should handle JSON arrays."""
expected_data = [{"id": 1}, {"id": 2}]
with aioresponses() as mocked:
mocked.get(
"http://example.com/api/items",
payload=expected_data,
)
result = await fetch_json("http://example.com/api/items")
assert result == expected_data
assert isinstance(result, list)
@pytest.mark.asyncio
async def test_fetch_json_http_error(self):
"""Should raise exception on HTTP error."""
with aioresponses() as mocked:
mocked.get("http://example.com/api/error", status=500)
with pytest.raises(ClientResponseError):
await fetch_json("http://example.com/api/error")
