116 lines
3.4 KiB
Python
116 lines
3.4 KiB
Python
import asyncio
|
|
import importlib.util
|
|
from pathlib import Path
|
|
|
|
import httpx
|
|
import pytest
|
|
|
|
|
|
def _load_web_search_module():
|
|
module_path = (
|
|
Path(__file__).resolve().parents[2] / "app" / "services" / "web_search.py"
|
|
)
|
|
spec = importlib.util.spec_from_file_location("tests_web_search_under_test", module_path)
|
|
module = importlib.util.module_from_spec(spec)
|
|
assert spec and spec.loader
|
|
spec.loader.exec_module(module)
|
|
return module
|
|
|
|
|
|
web_search = _load_web_search_module()
|
|
|
|
|
|
class FakeClient:
|
|
def __init__(self, response):
|
|
self.response = response
|
|
self.calls = []
|
|
|
|
async def post(self, url, *, headers, json):
|
|
self.calls.append({"url": url, "headers": headers, "json": json})
|
|
return self.response
|
|
|
|
|
|
def test_search_bocha_web_posts_expected_payload(monkeypatch):
|
|
monkeypatch.setattr(web_search.settings, "BOCHA_API_KEY", "sk-test")
|
|
monkeypatch.setattr(
|
|
web_search.settings,
|
|
"BOCHA_WEB_SEARCH_URL",
|
|
"https://api.bochaai.com/v1/web-search",
|
|
)
|
|
response = httpx.Response(
|
|
200,
|
|
json={"data": {"webPages": {"value": []}}},
|
|
request=httpx.Request("POST", "https://api.bochaai.com/v1/web-search"),
|
|
)
|
|
client = FakeClient(response)
|
|
|
|
result = asyncio.run(
|
|
web_search.search_bocha_web(
|
|
web_search.WebSearchRequest(
|
|
query="天津水务",
|
|
freshness="oneWeek",
|
|
summary=True,
|
|
count=5,
|
|
include=["example.com", "news.example.com"],
|
|
exclude=["spam.example.com"],
|
|
),
|
|
client=client,
|
|
)
|
|
)
|
|
|
|
assert result == {"data": {"webPages": {"value": []}}}
|
|
assert client.calls == [
|
|
{
|
|
"url": "https://api.bochaai.com/v1/web-search",
|
|
"headers": {
|
|
"Authorization": "Bearer sk-test",
|
|
"Content-Type": "application/json",
|
|
},
|
|
"json": {
|
|
"query": "天津水务",
|
|
"freshness": "oneWeek",
|
|
"summary": True,
|
|
"count": 5,
|
|
"include": "example.com,news.example.com",
|
|
"exclude": "spam.example.com",
|
|
},
|
|
}
|
|
]
|
|
|
|
|
|
def test_search_bocha_web_requires_api_key(monkeypatch):
|
|
monkeypatch.setattr(web_search.settings, "BOCHA_API_KEY", "")
|
|
|
|
with pytest.raises(web_search.BochaSearchConfigError):
|
|
asyncio.run(
|
|
web_search.search_bocha_web(
|
|
web_search.WebSearchRequest(query="天津水务"),
|
|
client=FakeClient(httpx.Response(200, json={})),
|
|
)
|
|
)
|
|
|
|
|
|
def test_search_bocha_web_surfaces_upstream_error(monkeypatch):
|
|
monkeypatch.setattr(web_search.settings, "BOCHA_API_KEY", "sk-test")
|
|
response = httpx.Response(
|
|
401,
|
|
json={"error": "invalid api key"},
|
|
request=httpx.Request("POST", "https://api.bochaai.com/v1/web-search"),
|
|
)
|
|
|
|
with pytest.raises(web_search.BochaSearchAPIError) as exc_info:
|
|
asyncio.run(
|
|
web_search.search_bocha_web(
|
|
web_search.WebSearchRequest(query="天津水务"),
|
|
client=FakeClient(response),
|
|
)
|
|
)
|
|
|
|
assert exc_info.value.status_code == 401
|
|
assert exc_info.value.detail == {"error": "invalid api key"}
|
|
|
|
|
|
def test_web_search_request_validates_count_range():
|
|
with pytest.raises(ValueError):
|
|
web_search.WebSearchRequest(query="天津水务", count=51)
|