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

from __future__ import annotations

import json
from datetime import date

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"
        payload = json.loads(request.content)
        assert payload["query"].strip()
        assert payload["xxx"] == "xxxx"
        assert payload["start_date"] == ""
        assert payload["end_date"] == ""
        assert payload["top_k"] == ""
        return httpx.Response(
            200,
            json={
                "results": [
                    {
                        "recUuid": "record/123",
                        "content": "test content",
                        "title": "test 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(
            "  test query  ",
            settings,
            transport=httpx.MockTransport(handler),
        )
    )

    assert len(requests) == 1
    assert settings.payload == {"query": "configured value", "xxx": "xxxx"}
    assert result == {
        "query": "test query",
        "total_results": 2,
        "results": [
            {
                "recUuid": "record/123",
                "content": "test content",
                "title": "test 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(
    ("query", "expected_start", "expected_end"),
    [
        ("近期政策", "2026-05-13", "2026-08-13"),
        ("近几天政策", "2026-08-06", "2026-08-13"),
        ("查询 2026年8月10日 到 2026-08-13 的政策", "2026-08-10", "2026-08-13"),
    ],
)
async def test_configurable_search_adds_date_payload(
    query: str,
    expected_start: str,
    expected_end: str,
) -> None:
    async def handler(request: httpx.Request) -> httpx.Response:
        payload = json.loads(request.content)
        assert payload["start_date"] == expected_start
        assert payload["end_date"] == expected_end
        assert payload["top_k"] == ""
        return httpx.Response(200, json={"results": []})

    await execute_search(
        query,
        _settings(),
        today=date(2026, 8, 13),
        transport=httpx.MockTransport(handler),
    )


@pytest.mark.asyncio
async def test_configurable_search_explicit_date_args_and_top_k_win() -> None:
    async def handler(request: httpx.Request) -> httpx.Response:
        payload = json.loads(request.content)
        assert payload["start_date"] == "2026-08-10"
        assert payload["end_date"] == "2026-08-13"
        assert payload["top_k"] == "15"
        return httpx.Response(200, json={"results": []})

    await execute_search(
        "近期政策",
        _settings(),
        start_date="2026/8/10",
        end_date="2026年8月13日",
        top_k=15,
        today=date(2026, 8, 13),
        transport=httpx.MockTransport(handler),
    )


@pytest.mark.asyncio
async def test_configurable_search_infers_top_k_from_query() -> None:
    async def handler(request: httpx.Request) -> httpx.Response:
        payload = json.loads(request.content)
        assert payload["top_k"] == "15"
        return httpx.Response(200, json={"results": []})

    await execute_search(
        "近期政策 返回15条",
        _settings(),
        today=date(2026, 8, 13),
        transport=httpx.MockTransport(handler),
    )


@pytest.mark.asyncio
async def test_configurable_search_accepts_root_array_results() -> None:
    async def handler(_: httpx.Request) -> httpx.Response:
        return httpx.Response(
            200,
            json=[
                {
                    "recUuid": "root-array-record",
                    "content": "direct array content",
                    "title": "direct array title",
                    "url": "https://original.example.test/root-array",
                }
            ],
        )

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

    assert result == {
        "query": "array response",
        "total_results": 1,
        "results": [
            {
                "recUuid": "root-array-record",
                "content": "direct array content",
                "title": "direct array title",
                "url": "https://original.example.test/root-array",
            }
        ],
    }


@pytest.mark.asyncio
async def test_configurable_search_normalizes_metadata_wrapped_results() -> None:
    async def handler(_: httpx.Request) -> httpx.Response:
        return httpx.Response(
            200,
            json=[
                {
                    "page_content": "outer content",
                    "metadata": {
                        "m_title": "metadata title",
                        "content": "metadata content",
                        "source1": "https://source.example.test/article/1",
                        "m_publish": "2026-08-13",
                    },
                }
            ],
        )

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

    assert result["results"] == [
        {
            "recUuid": "",
            "content": "outer content",
            "title": "metadata title",
            "source": "https://source.example.test/article/1",
            "time": "2026-08-13",
            "publish_time": "2026-08-13",
            "url": "https://source.example.test/article/1",
        }
    ]


@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)