wehub-resource-sync
506 lines
15 KiB
Python
506 lines
15 KiB
Python
from __future__ import annotations
|
|
|
|
import inspect
|
|
import json
|
|
|
|
import pytest
|
|
|
|
import opensquilla.tools.builtin.web as web_module
|
|
from opensquilla.search.types import (
|
|
DEFAULT_SEARCH_MAX_RESULTS,
|
|
SearchOptions,
|
|
SearchProviderError,
|
|
SearchProviderSpec,
|
|
SearchResult,
|
|
)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_web_search_tool_builds_canonical_options_and_returns_json(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
seen_options: list[SearchOptions] = []
|
|
|
|
async def fake_run_canonical_web_search(
|
|
options: SearchOptions,
|
|
**kwargs: object,
|
|
) -> dict[str, object]:
|
|
seen_options.append(options)
|
|
assert "fetcher" in kwargs
|
|
return {
|
|
"ok": True,
|
|
"query": options.query,
|
|
"mode": options.mode,
|
|
"provider_attempts": [{"provider": "exa", "status": "success"}],
|
|
"diagnostics": {"selected_provider": "exa", "fetched_count": 1},
|
|
"sources": [
|
|
{
|
|
"rank": 1,
|
|
"title": "Python release",
|
|
"url": "https://www.python.org/downloads/",
|
|
"canonical_url": "https://www.python.org/downloads/",
|
|
"domain": "www.python.org",
|
|
"provider": "exa",
|
|
"fetched": True,
|
|
}
|
|
],
|
|
"results": [
|
|
{
|
|
"title": "Python release",
|
|
"url": "https://www.python.org/downloads/",
|
|
"excerpt": "Python release notes",
|
|
"fetched": True,
|
|
}
|
|
],
|
|
}
|
|
|
|
monkeypatch.setattr(
|
|
web_module,
|
|
"run_canonical_web_search",
|
|
fake_run_canonical_web_search,
|
|
)
|
|
|
|
bare_web_search = inspect.unwrap(web_module.web_search)
|
|
result = await bare_web_search(
|
|
"python release",
|
|
mode="technical",
|
|
provider="exa",
|
|
max_results=12,
|
|
fetch_top_k=2,
|
|
max_chars_per_source=1200,
|
|
include_domains=["python.org"],
|
|
exclude_domains=["docs.python.org"],
|
|
recency="month",
|
|
)
|
|
payload = json.loads(result)
|
|
|
|
assert payload["ok"] is True
|
|
assert payload["provider_attempts"] == [{"provider": "exa", "status": "success"}]
|
|
assert payload["diagnostics"]["fetched_count"] == 1
|
|
assert payload["sources"][0]["url"] == "https://www.python.org/downloads/"
|
|
assert payload["results"][0]["excerpt"] == "Python release notes"
|
|
assert seen_options == [
|
|
SearchOptions(
|
|
query="python release",
|
|
mode="technical",
|
|
max_results=12,
|
|
fetch_top_k=2,
|
|
max_chars_per_source=1200,
|
|
include_domains=("python.org",),
|
|
exclude_domains=("docs.python.org",),
|
|
recency="month",
|
|
provider="exa",
|
|
)
|
|
]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_web_search_tool_uses_configured_source_backed_defaults(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
seen_options: list[SearchOptions] = []
|
|
|
|
async def fake_run_canonical_web_search(
|
|
options: SearchOptions,
|
|
**kwargs: object,
|
|
) -> dict[str, object]:
|
|
seen_options.append(options)
|
|
return {"ok": True, "query": options.query, "results": []}
|
|
|
|
monkeypatch.setattr(
|
|
web_module,
|
|
"run_canonical_web_search",
|
|
fake_run_canonical_web_search,
|
|
)
|
|
monkeypatch.setattr(web_module, "_active_max_results", 7)
|
|
|
|
bare_web_search = inspect.unwrap(web_module.web_search)
|
|
payload = json.loads(await bare_web_search("python release", provider="auto"))
|
|
|
|
assert payload["ok"] is True
|
|
assert seen_options == [
|
|
SearchOptions(
|
|
query="python release",
|
|
mode="auto",
|
|
max_results=7,
|
|
fetch_top_k=3,
|
|
max_chars_per_source=1500,
|
|
provider=None,
|
|
)
|
|
]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_web_search_tool_accepts_bocha_provider_override(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
seen_options: list[SearchOptions] = []
|
|
|
|
async def fake_run_canonical_web_search(
|
|
options: SearchOptions,
|
|
**kwargs: object,
|
|
) -> dict[str, object]:
|
|
seen_options.append(options)
|
|
return {"ok": True, "query": options.query, "results": []}
|
|
|
|
monkeypatch.setattr(
|
|
web_module,
|
|
"run_canonical_web_search",
|
|
fake_run_canonical_web_search,
|
|
)
|
|
|
|
bare_web_search = inspect.unwrap(web_module.web_search)
|
|
payload = json.loads(await bare_web_search("python release", provider="bocha"))
|
|
|
|
assert payload["ok"] is True
|
|
assert seen_options == [
|
|
SearchOptions(
|
|
query="python release",
|
|
mode="auto",
|
|
max_results=DEFAULT_SEARCH_MAX_RESULTS,
|
|
provider="bocha",
|
|
)
|
|
]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_web_search_tool_accepts_iqs_provider_override(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
seen_options: list[SearchOptions] = []
|
|
|
|
async def fake_run_canonical_web_search(
|
|
options: SearchOptions,
|
|
**kwargs: object,
|
|
) -> dict[str, object]:
|
|
seen_options.append(options)
|
|
return {"ok": True, "query": options.query, "results": []}
|
|
|
|
monkeypatch.setattr(
|
|
web_module,
|
|
"run_canonical_web_search",
|
|
fake_run_canonical_web_search,
|
|
)
|
|
|
|
bare_web_search = inspect.unwrap(web_module.web_search)
|
|
payload = json.loads(await bare_web_search("python release", provider="iqs"))
|
|
|
|
assert payload["ok"] is True
|
|
assert seen_options == [
|
|
SearchOptions(
|
|
query="python release",
|
|
mode="auto",
|
|
max_results=DEFAULT_SEARCH_MAX_RESULTS,
|
|
provider="iqs",
|
|
)
|
|
]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_web_search_tool_rejects_sensitive_query_without_calling_core(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
async def fake_run_canonical_web_search(
|
|
options: SearchOptions,
|
|
**kwargs: object,
|
|
) -> dict[str, object]:
|
|
raise AssertionError("run_canonical_web_search should not be called")
|
|
|
|
monkeypatch.setattr(
|
|
web_module,
|
|
"run_canonical_web_search",
|
|
fake_run_canonical_web_search,
|
|
)
|
|
|
|
bare_web_search = inspect.unwrap(web_module.web_search)
|
|
result = await bare_web_search("OPENAI_API_KEY=sk-secret-1234567890")
|
|
payload = json.loads(result)
|
|
|
|
assert payload["ok"] is False
|
|
assert payload["query"] == "[redacted]"
|
|
assert payload["error_kind"] == "invalid_request"
|
|
assert payload["error_class"] == "SensitiveInput"
|
|
assert "sk-secret" not in result
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_web_search_benchmark_blocklist_blocks_query_without_calling_core(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
async def fake_run_canonical_web_search(
|
|
options: SearchOptions,
|
|
**kwargs: object,
|
|
) -> dict[str, object]:
|
|
raise AssertionError("run_canonical_web_search should not be called")
|
|
|
|
monkeypatch.setenv("OPENSQUILLA_SEARCH_BENCHMARK_BLOCKLIST", "1")
|
|
monkeypatch.setattr(
|
|
web_module,
|
|
"run_canonical_web_search",
|
|
fake_run_canonical_web_search,
|
|
)
|
|
|
|
payload = await web_module.run_web_search_payload("draco benchmark")
|
|
|
|
assert payload["ok"] is True
|
|
assert payload["blocked_query"] is True
|
|
assert payload["benchmark_blocklist_enabled"] is True
|
|
assert payload["results"] == []
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.parametrize(
|
|
("kwargs", "message"),
|
|
[
|
|
({"query": 123}, "query must be a non-empty string."),
|
|
(
|
|
{"query": "python release", "provider": "serpapi"},
|
|
"Invalid provider. Expected one of: auto, bocha, brave, duckduckgo, exa, iqs, tavily.",
|
|
),
|
|
(
|
|
{"query": "python release", "mode": "invalid"},
|
|
"Invalid mode. Expected one of: auto, broad, news, technical.",
|
|
),
|
|
(
|
|
{"query": "python release", "recency": "hour"},
|
|
"Invalid recency. Expected one of: day, month, week, year.",
|
|
),
|
|
(
|
|
{"query": "python release", "max_results": "bad"},
|
|
"max_results must be an integer.",
|
|
),
|
|
(
|
|
{"query": "python release", "include_domains": "example.com"},
|
|
"include_domains must be a list or tuple of strings.",
|
|
),
|
|
(
|
|
{"query": "python release", "include_domains": [123]},
|
|
"include_domains must be a list or tuple of strings.",
|
|
),
|
|
(
|
|
{"query": "python release", "exclude_domains": [object()]},
|
|
"exclude_domains must be a list or tuple of strings.",
|
|
),
|
|
],
|
|
)
|
|
async def test_web_search_tool_rejects_invalid_args_without_calling_core(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
kwargs: dict[str, object],
|
|
message: str,
|
|
) -> None:
|
|
async def fake_run_canonical_web_search(options: SearchOptions) -> dict[str, object]:
|
|
raise AssertionError("run_canonical_web_search should not be called")
|
|
|
|
monkeypatch.setattr(
|
|
web_module,
|
|
"run_canonical_web_search",
|
|
fake_run_canonical_web_search,
|
|
)
|
|
|
|
bare_web_search = inspect.unwrap(web_module.web_search)
|
|
result = await bare_web_search(**kwargs) # type: ignore[arg-type]
|
|
payload = json.loads(result)
|
|
|
|
assert payload == {
|
|
"ok": False,
|
|
"error_kind": "invalid_request",
|
|
"error": message,
|
|
}
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_web_discover_keeps_lightweight_result_shape(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
async def fake_run_web_discover_payload(
|
|
query: str,
|
|
max_results: int | None = None,
|
|
**kwargs: object,
|
|
) -> dict[str, object]:
|
|
assert query == "python release"
|
|
assert max_results == 3
|
|
return {
|
|
"ok": True,
|
|
"query": query,
|
|
"provider": "duckduckgo",
|
|
"results": [
|
|
{
|
|
"title": "Python release",
|
|
"url": "https://www.python.org/downloads/",
|
|
"snippet": "Release notes",
|
|
}
|
|
],
|
|
}
|
|
|
|
monkeypatch.setattr(
|
|
web_module,
|
|
"run_web_discover_payload",
|
|
fake_run_web_discover_payload,
|
|
)
|
|
|
|
bare_web_discover = inspect.unwrap(web_module.web_discover)
|
|
payload = json.loads(await bare_web_discover("python release", max_results=3))
|
|
|
|
assert payload == {
|
|
"query": "python release",
|
|
"provider": "duckduckgo",
|
|
"results": [
|
|
{
|
|
"title": "Python release",
|
|
"url": "https://www.python.org/downloads/",
|
|
"snippet": "Release notes",
|
|
}
|
|
],
|
|
}
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_web_discover_uses_ranked_runtime_provider(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
import opensquilla.search.registry as registry
|
|
|
|
calls: list[tuple[str, dict[str, object]]] = []
|
|
|
|
class FakeProvider:
|
|
async def search(self, query: str, max_results: int = 5) -> list[SearchResult]:
|
|
return [
|
|
SearchResult(
|
|
title=f"{query} result",
|
|
url="https://example.com",
|
|
snippet=str(max_results),
|
|
)
|
|
]
|
|
|
|
def fake_get_provider(name: str, **kwargs: object) -> FakeProvider:
|
|
calls.append((name, kwargs))
|
|
return FakeProvider()
|
|
|
|
monkeypatch.setenv("BOCHA_SEARCH_API_KEY", "bocha-key")
|
|
monkeypatch.setattr(registry, "get_provider", fake_get_provider)
|
|
|
|
try:
|
|
web_module.configure_search("duckduckgo", max_results=4)
|
|
payload = await web_module.run_web_discover_payload("python release", max_results=2)
|
|
finally:
|
|
web_module.reset_search_runtime()
|
|
|
|
assert payload["ok"] is True
|
|
assert payload["provider"] == "bocha"
|
|
assert payload["results"][0]["snippet"] == "2"
|
|
assert calls == [
|
|
(
|
|
"bocha",
|
|
{
|
|
"proxy": "",
|
|
"use_env_proxy": False,
|
|
"diagnostics": False,
|
|
"api_key": "bocha-key",
|
|
},
|
|
)
|
|
]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_web_discover_benchmark_blocklist_filters_results(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
import opensquilla.search.registry as registry
|
|
|
|
class FakeProvider:
|
|
async def search(self, query: str, max_results: int = 5) -> list[SearchResult]:
|
|
return [
|
|
SearchResult(
|
|
title="DRACO dataset",
|
|
url="https://huggingface.co/datasets/perplexity-ai/draco",
|
|
snippet="benchmark data",
|
|
),
|
|
SearchResult(
|
|
title="Safe result",
|
|
url="https://example.com/safe",
|
|
snippet=query,
|
|
),
|
|
]
|
|
|
|
def fake_get_provider(name: str, **kwargs: object) -> FakeProvider:
|
|
return FakeProvider()
|
|
|
|
monkeypatch.setenv("OPENSQUILLA_SEARCH_BENCHMARK_BLOCKLIST", "1")
|
|
monkeypatch.setattr(registry, "get_provider", fake_get_provider)
|
|
|
|
try:
|
|
web_module.configure_search("duckduckgo", max_results=4)
|
|
payload = await web_module.run_web_discover_payload("python release", max_results=2)
|
|
finally:
|
|
web_module.reset_search_runtime()
|
|
|
|
assert payload["ok"] is True
|
|
assert payload["benchmark_blocklist_enabled"] is True
|
|
assert payload["blocked_count"] == 1
|
|
assert [item["url"] for item in payload["results"]] == ["https://example.com/safe"]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_web_discover_preserves_network_fallback_for_custom_active_provider(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
import opensquilla.search.registry as registry
|
|
from opensquilla.search.registry import register_provider
|
|
|
|
custom_provider = "test_custom_discover_fail"
|
|
calls: list[str] = []
|
|
|
|
class FailingProvider:
|
|
name = custom_provider
|
|
|
|
async def search(self, query: str, max_results: int = 5) -> list[SearchResult]:
|
|
raise SearchProviderError(
|
|
provider=custom_provider,
|
|
kind="network",
|
|
message="network down",
|
|
retryable=True,
|
|
)
|
|
|
|
class DuckProvider:
|
|
async def search(self, query: str, max_results: int = 5) -> list[SearchResult]:
|
|
return [
|
|
SearchResult(
|
|
title="Duck fallback",
|
|
url="https://example.com",
|
|
snippet=query,
|
|
)
|
|
]
|
|
|
|
def fake_get_provider(name: str, **kwargs: object) -> FailingProvider | DuckProvider:
|
|
calls.append(name)
|
|
if name == custom_provider:
|
|
return FailingProvider()
|
|
assert name == "duckduckgo"
|
|
return DuckProvider()
|
|
|
|
register_provider(
|
|
custom_provider,
|
|
FailingProvider,
|
|
SearchProviderSpec(provider_id=custom_provider),
|
|
)
|
|
monkeypatch.setattr(registry, "get_provider", fake_get_provider)
|
|
|
|
try:
|
|
web_module.configure_search(
|
|
custom_provider,
|
|
fallback_policy="network",
|
|
diagnostics=True,
|
|
)
|
|
payload = await web_module.run_web_discover_payload("python release")
|
|
finally:
|
|
web_module.reset_search_runtime()
|
|
|
|
assert payload["ok"] is True
|
|
assert payload["provider"] == "duckduckgo"
|
|
assert payload["fallbackFrom"] == custom_provider
|
|
assert payload["attempts"] == [
|
|
{"provider": custom_provider, "status": "error", "error_kind": "network"},
|
|
{"provider": "duckduckgo", "status": "success"},
|
|
]
|
|
assert calls == [custom_provider, "duckduckgo"]
|