deerflow-code/offline-backend-20260512/backend/packages/harness/tests/test_configurable_web_search.py

from __future__ import annotations

import json

import httpx
import pytest

from deerflow.community.configurable_search.tools import (
    ConfigurableSearchSettings,
    execute_search,
)
from deerflow.config.app_config import AppConfig
from deerflow.config.tool_config import ToolConfig
from deerflow.tools.tools import get_available_tools


def _settings(**overrides: object) -> ConfigurableSearchSettings:
    values = {
        "enabled": True,
        "endpoint": "https://search.example.test/searchName",
        "payload": {"query": "configured value", "xxx": "xxxx"},
        "verify_ssl": False,
        "timeout": 30.0,
        "max_results": 5,
        "result_url_template": "https://search.example.test/mindId/xxx?recUuid={recUuid}",
    }
    values.update(overrides)
    return ConfigurableSearchSettings(**values)  # type: ignore[arg-type]


@pytest.mark.asyncio
async def test_configurable_search_overrides_query_and_normalizes_nullable_results() -> None:
    requests: list[httpx.Request] = []

    async def handler(request: httpx.Request) -> httpx.Response:
        requests.append(request)
        assert request.method == "POST"
        assert json.loads(request.content) == {"query": "测试检索词", "xxx": "xxxx"}
        return httpx.Response(
            200,
            json={
                "results": [
                    {
                        "recUuid": "record/123",
                        "content": "检索内容",
                        "title": "标题",
                        "url": "https://original.example.test/one",
                    },
                    {
                        "recUuid": None,
                        "content": None,
                        "title": None,
                        "url": "https://original.example.test/two",
                    },
                    "invalid row",
                ]
            },
        )

    settings = _settings()
    result = json.loads(
        await execute_search(
            "  测试检索词  ",
            settings,
            transport=httpx.MockTransport(handler),
        )
    )

    assert len(requests) == 1
    assert settings.payload == {"query": "configured value", "xxx": "xxxx"}
    assert result == {
        "query": "测试检索词",
        "total_results": 2,
        "results": [
            {
                "recUuid": "record/123",
                "content": "检索内容",
                "title": "标题",
                "url": "https://search.example.test/mindId/xxx?recUuid=record%2F123",
            },
            {
                "recUuid": "",
                "content": "",
                "title": "",
                "url": "https://original.example.test/two",
            },
        ],
    }


@pytest.mark.asyncio
async def test_configurable_search_handles_empty_results() -> None:
    async def handler(_: httpx.Request) -> httpx.Response:
        return httpx.Response(200, json={"results": []})

    result = json.loads(
        await execute_search(
            "nothing",
            _settings(),
            transport=httpx.MockTransport(handler),
        )
    )

    assert result == {"query": "nothing", "total_results": 0, "results": []}


@pytest.mark.asyncio
@pytest.mark.parametrize(
    ("response_factory", "expected_error"),
    [
        (lambda request: httpx.Response(503, request=request), "Search API returned HTTP 503"),
        (
            lambda request: httpx.Response(200, content=b"not-json", request=request),
            "Search API request failed: JSONDecodeError",
        ),
        (
            lambda request: httpx.Response(200, json={"results": None}, request=request),
            "search API response field 'results' must be an array",
        ),
    ],
)
async def test_configurable_search_returns_safe_errors(
    response_factory,
    expected_error: str,
) -> None:
    async def handler(request: httpx.Request) -> httpx.Response:
        return response_factory(request)

    result = json.loads(
        await execute_search(
            "query",
            _settings(),
            transport=httpx.MockTransport(handler),
        )
    )

    assert result["results"] == []
    assert result["total_results"] == 0
    assert result["error"] == expected_error


@pytest.mark.asyncio
async def test_configurable_search_handles_timeout() -> None:
    async def handler(request: httpx.Request) -> httpx.Response:
        raise httpx.ReadTimeout("timed out", request=request)

    result = json.loads(
        await execute_search(
            "query",
            _settings(),
            transport=httpx.MockTransport(handler),
        )
    )

    assert result["error"] == "Search API request timed out"


@pytest.mark.asyncio
async def test_configurable_search_passes_verify_ssl_to_httpx(monkeypatch: pytest.MonkeyPatch) -> None:
    captured: dict[str, object] = {}

    class FakeAsyncClient:
        def __init__(self, **kwargs: object) -> None:
            captured.update(kwargs)

        async def __aenter__(self):
            return self

        async def __aexit__(self, *_: object) -> None:
            return None

        async def post(self, url: str, *, json: dict[str, object]) -> httpx.Response:
            request = httpx.Request("POST", url, json=json)
            return httpx.Response(200, json={"results": []}, request=request)

    monkeypatch.setattr(
        "deerflow.community.configurable_search.tools.httpx.AsyncClient",
        FakeAsyncClient,
    )

    await execute_search("query", _settings(verify_ssl=False))

    assert captured["verify"] is False
    assert captured["timeout"] == 30.0


def test_disabled_configured_tool_is_not_registered() -> None:
    config = AppConfig(
        sandbox={"use": "deerflow.sandbox.local:LocalSandboxProvider"},
        tools=[
            ToolConfig(
                name="web_search",
                group="web",
                enabled=False,
                use="deerflow.community.configurable_search.tools:web_search_tool",
                endpoint="https://search.example.test/searchName",
            )
        ],
    )

    available = get_available_tools(
        groups=["web"],
        include_mcp=False,
        app_config=config,
    )

    assert "web_search" not in {item.name for item in available}

    config.tools[0].enabled = True
    enabled_tools = get_available_tools(
        groups=["web"],
        include_mcp=False,
        app_config=config,
    )

    assert "web_search" in {item.name for item in enabled_tools}


def test_config_validation_requires_rec_uuid_url_placeholder() -> None:
    config = ToolConfig(
        name="web_search",
        group="web",
        use="deerflow.community.configurable_search.tools:web_search_tool",
        endpoint="https://search.example.test/searchName",
        result_url_template="https://search.example.test/mindId/xxx",
    )

    with pytest.raises(ValueError, match="must contain"):
        ConfigurableSearchSettings.from_tool_config(config)