searxng-mcp/test_server.py

import os
from urllib.parse import parse_qs, urlparse

import httpx
import pytest
import respx

# Set env before importing server so the default URL is predictable
os.environ["SEARXNG_URL"] = "http://test-searxng:8080"

import server  # noqa: E402


# --- build_search_url tests ---


def test_build_search_url_basic():
    url = server.build_search_url("hello world")
    parsed = urlparse(url)
    params = parse_qs(parsed.query)
    assert parsed.scheme == "http"
    assert parsed.netloc == "test-searxng:8080"
    assert parsed.path == "/search"
    assert params["q"] == ["hello world"]
    assert params["format"] == ["json"]
    # No optional params
    assert "categories" not in params
    assert "language" not in params
    assert "pageno" not in params


def test_build_search_url_all_params():
    url = server.build_search_url(
        "test query",
        categories="news,science",
        language="de",
        pageno=3,
        time_range="month",
        safesearch=2,
    )
    params = parse_qs(urlparse(url).query)
    assert params["q"] == ["test query"]
    assert params["format"] == ["json"]
    assert params["categories"] == ["news,science"]
    assert params["language"] == ["de"]
    assert params["pageno"] == ["3"]
    assert params["time_range"] == ["month"]
    assert params["safesearch"] == ["2"]


# --- format_results tests ---


SAMPLE_RESULTS = {
    "results": [
        {
            "title": "Example Page",
            "url": "https://example.com",
            "content": "This is a sample result.",
            "engines": ["google", "bing"],
            "publishedDate": "2025-01-15",
        },
        {
            "title": "Another Page",
            "url": "https://another.com",
            "content": "Another result snippet.",
            "engines": ["duckduckgo"],
        },
    ],
    "total_results": 100,
    "time_taken": 0.5,
}


def test_format_results_basic():
    output = server.format_results(SAMPLE_RESULTS)
    assert "1. Example Page" in output
    assert "URL: https://example.com" in output
    assert "This is a sample result." in output
    assert "Source: google, bing" in output
    assert "Published: 2025-01-15" in output
    assert "2. Another Page" in output
    assert "URL: https://another.com" in output
    assert "(100 total results in 0.5s)" in output


def test_format_results_with_answers():
    data = {
        "results": [{"title": "A", "url": "https://a.com", "content": "a"}],
        "answers": ["42 is the answer"],
        "total_results": 1,
        "time_taken": 0.1,
    }
    output = server.format_results(data)
    assert "Direct answers:" in output
    assert "42 is the answer" in output
    # Answers should come before results
    assert output.index("Direct answers:") < output.index("1. A")


def test_format_results_empty():
    output = server.format_results({"results": []})
    assert output == "No results found."


def test_format_results_empty_no_key():
    output = server.format_results({})
    assert output == "No results found."


def test_format_results_max_results():
    data = {
        "results": [
            {"title": f"Result {i}", "url": f"https://r{i}.com", "content": f"Content {i}"}
            for i in range(20)
        ],
        "total_results": 20,
        "time_taken": 0.3,
    }
    output = server.format_results(data, max_results=5)
    assert "5. Result 4" in output
    assert "6." not in output


# --- web_search integration tests (mocked HTTP) ---


@pytest.mark.asyncio
@respx.mock
async def test_web_search_success():
    respx.get("http://test-searxng:8080/search").mock(
        return_value=httpx.Response(200, json=SAMPLE_RESULTS)
    )
    result = await server.web_search("hello")
    assert "Example Page" in result
    assert "Another Page" in result


@pytest.mark.asyncio
@respx.mock
async def test_web_search_network_error():
    respx.get("http://test-searxng:8080/search").mock(
        side_effect=httpx.ConnectError("Connection refused")
    )
    result = await server.web_search("hello")
    assert "Connection error" in result
    assert "test-searxng:8080" in result


@pytest.mark.asyncio
@respx.mock
async def test_web_search_non_200():
    respx.get("http://test-searxng:8080/search").mock(
        return_value=httpx.Response(500, text="Internal Server Error")
    )
    result = await server.web_search("hello")
    assert "HTTP 500" in result


@pytest.mark.asyncio
@respx.mock
async def test_web_search_timeout():
    respx.get("http://test-searxng:8080/search").mock(
        side_effect=httpx.ReadTimeout("timed out")
    )
    result = await server.web_search("hello")
    assert "Timeout" in result


@pytest.mark.asyncio
@respx.mock
async def test_web_search_max_results_clamped():
    data = {
        "results": [
            {"title": f"R{i}", "url": f"https://r{i}.com", "content": f"C{i}"}
            for i in range(10)
        ],
        "total_results": 10,
        "time_taken": 0.1,
    }
    respx.get("http://test-searxng:8080/search").mock(
        return_value=httpx.Response(200, json=data)
    )
    result = await server.web_search("hello", max_results=3)
    assert "3. R2" in result
    assert "4." not in result


def test_searxng_url_from_env():
    assert server.SEARXNG_URL == "http://test-searxng:8080"