Michael
417 lines
17 KiB
Python
417 lines
17 KiB
Python
"""Tests for ResearchTools: web_search and fetch_url."""
|
|
|
|
import json
|
|
import httpx
|
|
from unittest.mock import MagicMock, patch, PropertyMock
|
|
|
|
import pytest
|
|
from gitea.tools.research_tools import ResearchTools, _MAX_CONTENT_CHARS
|
|
|
|
|
|
@pytest.fixture
|
|
def tools() -> ResearchTools:
|
|
return ResearchTools()
|
|
|
|
|
|
# ------------------------------------------------------------------ #
|
|
# _smart_truncate
|
|
# ------------------------------------------------------------------ #
|
|
|
|
|
|
class TestSmartTruncate:
|
|
def test_no_truncation_if_short(self, tools: ResearchTools) -> None:
|
|
text = "hello world"
|
|
assert tools._smart_truncate(text, max_chars=100) == text
|
|
|
|
def test_cuts_at_paragraph_boundary(self, tools: ResearchTools) -> None:
|
|
# Two paragraphs; the boundary falls after the 70% mark of max_chars
|
|
para1 = "A" * 80
|
|
para2 = "B" * 80
|
|
text = para1 + "\n\n" + para2
|
|
result = tools._smart_truncate(text, max_chars=100)
|
|
# Should cut at the \n\n, not mid-word
|
|
assert "truncated" in result
|
|
assert result.startswith(para1)
|
|
|
|
def test_hard_cut_when_no_good_boundary(self, tools: ResearchTools) -> None:
|
|
# Single block — no paragraph boundary available
|
|
text = "x" * 200
|
|
result = tools._smart_truncate(text, max_chars=100)
|
|
assert "truncated" in result
|
|
assert result.startswith("x" * 100)
|
|
|
|
def test_exact_length_not_truncated(self, tools: ResearchTools) -> None:
|
|
text = "a" * 100
|
|
assert tools._smart_truncate(text, max_chars=100) == text
|
|
|
|
|
|
# ------------------------------------------------------------------ #
|
|
# _html_to_markdown — regex fallback (no optional deps needed)
|
|
# ------------------------------------------------------------------ #
|
|
|
|
|
|
class TestHtmlToMarkdown:
|
|
def _patch_imports(self, tools: ResearchTools) -> str:
|
|
"""Return result when optional deps are unavailable."""
|
|
# Force all optional imports to fail → regex fallback
|
|
import builtins
|
|
real_import = builtins.__import__
|
|
|
|
def mock_import(name: str, *args, **kwargs): # type: ignore[no-untyped-def]
|
|
if name in ("trafilatura", "readability", "markdownify"):
|
|
raise ImportError(f"mocked missing: {name}")
|
|
return real_import(name, *args, **kwargs)
|
|
|
|
with patch("builtins.__import__", side_effect=mock_import):
|
|
return tools._html_to_markdown("<p>Hello <b>world</b></p>")
|
|
|
|
def test_regex_fallback_removes_tags(self, tools: ResearchTools) -> None:
|
|
result = self._patch_imports(tools)
|
|
assert "Hello" in result
|
|
assert "world" in result
|
|
assert "<p>" not in result
|
|
assert "<b>" not in result
|
|
|
|
def test_regex_fallback_removes_script(self, tools: ResearchTools) -> None:
|
|
import builtins
|
|
real_import = builtins.__import__
|
|
|
|
def mock_import(name: str, *args, **kwargs): # type: ignore[no-untyped-def]
|
|
if name in ("trafilatura", "readability", "markdownify"):
|
|
raise ImportError
|
|
return real_import(name, *args, **kwargs)
|
|
|
|
with patch("builtins.__import__", side_effect=mock_import):
|
|
result = tools._html_to_markdown("<script>evil()</script>visible")
|
|
assert "evil" not in result
|
|
assert "visible" in result
|
|
|
|
def test_regex_fallback_decodes_entities(self, tools: ResearchTools) -> None:
|
|
import builtins
|
|
real_import = builtins.__import__
|
|
|
|
def mock_import(name: str, *args, **kwargs): # type: ignore[no-untyped-def]
|
|
if name in ("trafilatura", "readability", "markdownify"):
|
|
raise ImportError
|
|
return real_import(name, *args, **kwargs)
|
|
|
|
with patch("builtins.__import__", side_effect=mock_import):
|
|
result = tools._html_to_markdown("Tom & Jerry <3>")
|
|
assert "Tom & Jerry <3>" in result
|
|
|
|
|
|
# ------------------------------------------------------------------ #
|
|
# _format_results
|
|
# ------------------------------------------------------------------ #
|
|
|
|
|
|
class TestFormatResults:
|
|
def test_numbered_list(self, tools: ResearchTools) -> None:
|
|
results = [
|
|
{"title": "Title A", "href": "https://a.com", "body": "Snippet A"},
|
|
{"title": "Title B", "href": "https://b.com", "body": "Snippet B"},
|
|
]
|
|
output = tools._format_results(results, "test query")
|
|
assert "[1]" in output
|
|
assert "[2]" in output
|
|
assert "https://a.com" in output
|
|
assert "Snippet A" in output
|
|
|
|
def test_includes_date_when_present(self, tools: ResearchTools) -> None:
|
|
results = [{"title": "T", "href": "https://x.com", "body": "S", "published_date": "2024-01"}]
|
|
output = tools._format_results(results, "q")
|
|
assert "Date: 2024-01" in output
|
|
|
|
def test_no_date_field_when_absent(self, tools: ResearchTools) -> None:
|
|
results = [{"title": "T", "href": "https://x.com", "body": "S"}]
|
|
output = tools._format_results(results, "q")
|
|
assert "Date:" not in output
|
|
|
|
def test_snippet_truncated_to_250_chars(self, tools: ResearchTools) -> None:
|
|
long_body = "x" * 500
|
|
results = [{"title": "T", "href": "u", "body": long_body}]
|
|
output = tools._format_results(results, "q")
|
|
assert "x" * 251 not in output # body was truncated before formatting
|
|
|
|
|
|
# ------------------------------------------------------------------ #
|
|
# _search_searxng
|
|
# ------------------------------------------------------------------ #
|
|
|
|
|
|
class TestSearchSearxng:
|
|
def _make_searxng_response(self, results: list[dict]) -> MagicMock:
|
|
mock_resp = MagicMock()
|
|
mock_resp.json.return_value = {"results": results}
|
|
mock_resp.raise_for_status = MagicMock()
|
|
return mock_resp
|
|
|
|
@patch("gitea.tools.research_tools.httpx.Client")
|
|
def test_returns_formatted_results_on_success(
|
|
self, mock_cls: MagicMock, tools: ResearchTools
|
|
) -> None:
|
|
results = [
|
|
{"title": "SearXNG Result", "url": "https://example.com", "content": "snippet"},
|
|
]
|
|
mock_cls.return_value.__enter__.return_value.get.return_value = (
|
|
self._make_searxng_response(results)
|
|
)
|
|
output = tools._search_searxng("test query", num_results=5)
|
|
assert output is not None
|
|
assert "[1]" in output
|
|
assert "example.com" in output
|
|
|
|
@patch("gitea.tools.research_tools.httpx.Client")
|
|
def test_returns_none_when_empty_results(
|
|
self, mock_cls: MagicMock, tools: ResearchTools
|
|
) -> None:
|
|
mock_cls.return_value.__enter__.return_value.get.return_value = (
|
|
self._make_searxng_response([])
|
|
)
|
|
assert tools._search_searxng("nothing", num_results=5) is None
|
|
|
|
@patch("gitea.tools.research_tools.httpx.Client")
|
|
def test_returns_none_on_connection_error(
|
|
self, mock_cls: MagicMock, tools: ResearchTools
|
|
) -> None:
|
|
mock_cls.return_value.__enter__.return_value.get.side_effect = (
|
|
httpx.ConnectError("refused")
|
|
)
|
|
assert tools._search_searxng("query", num_results=5) is None
|
|
|
|
@patch("gitea.tools.research_tools.httpx.Client")
|
|
def test_respects_time_range_param(
|
|
self, mock_cls: MagicMock, tools: ResearchTools
|
|
) -> None:
|
|
mock_cls.return_value.__enter__.return_value.get.return_value = (
|
|
self._make_searxng_response([])
|
|
)
|
|
tools._search_searxng("q", num_results=5, time_range="month")
|
|
call_kwargs = mock_cls.return_value.__enter__.return_value.get.call_args
|
|
params = call_kwargs[1].get("params", call_kwargs[0][1] if len(call_kwargs[0]) > 1 else {})
|
|
assert params.get("time_range") == "month"
|
|
|
|
@patch("gitea.tools.research_tools.httpx.Client")
|
|
def test_web_search_uses_searxng_first(
|
|
self, mock_cls: MagicMock, tools: ResearchTools
|
|
) -> None:
|
|
"""web_search should return SearXNG results without touching DDGS."""
|
|
results = [{"title": "From SearXNG", "url": "https://sx.com", "content": "content"}]
|
|
mock_cls.return_value.__enter__.return_value.get.return_value = (
|
|
self._make_searxng_response(results)
|
|
)
|
|
output = tools.web_search("python typing")
|
|
assert "From SearXNG" in output or "[1]" in output
|
|
|
|
|
|
# ------------------------------------------------------------------ #
|
|
# web_search — with DDGS mocked
|
|
# ------------------------------------------------------------------ #
|
|
|
|
|
|
class TestWebSearch:
|
|
def _make_ddgs_result(self) -> list[dict[str, str]]:
|
|
return [
|
|
{"title": "Foo Docs", "href": "https://foo.com/docs", "body": "Learn about Foo."},
|
|
{"title": "Bar Guide", "href": "https://bar.com", "body": "A guide to Bar."},
|
|
]
|
|
|
|
def _mock_ddgs(self, results: list[dict[str, str]]) -> MagicMock:
|
|
mock_ddgs_instance = MagicMock()
|
|
mock_ddgs_instance.__enter__ = MagicMock(return_value=mock_ddgs_instance)
|
|
mock_ddgs_instance.__exit__ = MagicMock(return_value=False)
|
|
mock_ddgs_instance.text.return_value = iter(results)
|
|
return mock_ddgs_instance
|
|
|
|
@patch("gitea.tools.research_tools.time.sleep")
|
|
def test_returns_numbered_results(
|
|
self, _mock_sleep: MagicMock, tools: ResearchTools
|
|
) -> None:
|
|
mock_cls = MagicMock(return_value=self._mock_ddgs(self._make_ddgs_result()))
|
|
with patch("gitea.tools.research_tools.DDGS", mock_cls):
|
|
result = tools.web_search("python httpx")
|
|
assert "[1]" in result
|
|
assert "foo.com" in result
|
|
|
|
@patch("gitea.tools.research_tools.time.sleep")
|
|
def test_no_results_returns_helpful_message(
|
|
self, _mock_sleep: MagicMock, tools: ResearchTools
|
|
) -> None:
|
|
mock_cls = MagicMock(return_value=self._mock_ddgs([]))
|
|
with patch("gitea.tools.research_tools.DDGS", mock_cls):
|
|
result = tools.web_search("xyzzy-not-real")
|
|
assert "No results" in result
|
|
|
|
def test_clamps_num_results_max(self, tools: ResearchTools) -> None:
|
|
ddgs_mock = self._mock_ddgs(self._make_ddgs_result())
|
|
mock_cls = MagicMock(return_value=ddgs_mock)
|
|
with patch("gitea.tools.research_tools.DDGS", mock_cls):
|
|
tools.web_search("q", num_results=999)
|
|
# DDGS.text should be called with max_results clamped to 20
|
|
ddgs_mock.text.assert_called_once()
|
|
_, kwargs = ddgs_mock.text.call_args
|
|
assert kwargs.get("max_results", 0) <= 20
|
|
|
|
def test_clamps_num_results_min(self, tools: ResearchTools) -> None:
|
|
ddgs_mock = self._mock_ddgs(self._make_ddgs_result())
|
|
mock_cls = MagicMock(return_value=ddgs_mock)
|
|
with patch("gitea.tools.research_tools.DDGS", mock_cls):
|
|
tools.web_search("q", num_results=0)
|
|
_, kwargs = ddgs_mock.text.call_args
|
|
assert kwargs.get("max_results", 0) >= 1
|
|
|
|
@patch("gitea.tools.research_tools.time.sleep")
|
|
def test_retries_on_rate_limit(
|
|
self, mock_sleep: MagicMock, tools: ResearchTools
|
|
) -> None:
|
|
"""Should retry up to 3 times with exponential backoff on RatelimitException."""
|
|
rate_exc = Exception("rate limit")
|
|
ddgs_mock = MagicMock()
|
|
ddgs_mock.__enter__ = MagicMock(return_value=ddgs_mock)
|
|
ddgs_mock.__exit__ = MagicMock(return_value=False)
|
|
ddgs_mock.text.side_effect = rate_exc
|
|
|
|
with patch("gitea.tools.research_tools.DDGS", return_value=ddgs_mock), \
|
|
patch("gitea.tools.research_tools.RatelimitException", type(rate_exc)), \
|
|
patch("gitea.tools.research_tools.DuckDuckGoSearchException", ValueError):
|
|
result = tools.web_search("q")
|
|
|
|
assert isinstance(result, str) # Returns error string, not raise
|
|
|
|
|
|
|
|
# ------------------------------------------------------------------ #
|
|
# fetch_url
|
|
# ------------------------------------------------------------------ #
|
|
|
|
|
|
class TestFetchUrl:
|
|
def _make_response(
|
|
self,
|
|
text: str,
|
|
content_type: str = "text/html; charset=utf-8",
|
|
status_code: int = 200,
|
|
) -> MagicMock:
|
|
mock_resp = MagicMock()
|
|
mock_resp.text = text
|
|
mock_resp.headers = {"content-type": content_type}
|
|
mock_resp.status_code = status_code
|
|
mock_resp.raise_for_status = MagicMock()
|
|
return mock_resp
|
|
|
|
def test_rejects_non_http_url(self, tools: ResearchTools) -> None:
|
|
result = tools.fetch_url("ftp://example.com/file")
|
|
assert "Invalid URL" in result
|
|
|
|
def test_rejects_no_scheme(self, tools: ResearchTools) -> None:
|
|
result = tools.fetch_url("example.com")
|
|
assert "Invalid URL" in result
|
|
|
|
@patch("gitea.tools.research_tools.httpx.Client")
|
|
def test_pretty_prints_json_response(
|
|
self, mock_cls: MagicMock, tools: ResearchTools
|
|
) -> None:
|
|
data = {"key": "value", "n": 42}
|
|
mock_resp = self._make_response(
|
|
json.dumps(data), content_type="application/json"
|
|
)
|
|
mock_resp.json = MagicMock(return_value=data)
|
|
mock_cls.return_value.__enter__.return_value.get.return_value = mock_resp
|
|
|
|
result = tools.fetch_url("https://api.example.com/data")
|
|
assert '"key": "value"' in result
|
|
|
|
@patch("gitea.tools.research_tools.httpx.Client")
|
|
def test_returns_plain_text_as_is(
|
|
self, mock_cls: MagicMock, tools: ResearchTools
|
|
) -> None:
|
|
mock_resp = self._make_response("plain text content", content_type="text/plain")
|
|
mock_cls.return_value.__enter__.return_value.get.return_value = mock_resp
|
|
|
|
result = tools.fetch_url("https://example.com/readme.txt")
|
|
assert "plain text content" in result
|
|
|
|
@patch("gitea.tools.research_tools.httpx.Client")
|
|
def test_returns_markdown_file_as_is(
|
|
self, mock_cls: MagicMock, tools: ResearchTools
|
|
) -> None:
|
|
mock_resp = self._make_response("# Heading\nContent", content_type="text/plain")
|
|
mock_cls.return_value.__enter__.return_value.get.return_value = mock_resp
|
|
|
|
result = tools.fetch_url("https://example.com/README.md")
|
|
assert "# Heading" in result
|
|
|
|
@patch("gitea.tools.research_tools.httpx.Client")
|
|
def test_raw_html_when_extract_false(
|
|
self, mock_cls: MagicMock, tools: ResearchTools
|
|
) -> None:
|
|
html = "<html><body><p>Content</p></body></html>"
|
|
mock_resp = self._make_response(html)
|
|
mock_cls.return_value.__enter__.return_value.get.return_value = mock_resp
|
|
|
|
result = tools.fetch_url("https://example.com", extract_text=False)
|
|
assert "<p>" in result
|
|
|
|
@patch("gitea.tools.research_tools.httpx.Client")
|
|
def test_truncates_at_max_chars(
|
|
self, mock_cls: MagicMock, tools: ResearchTools
|
|
) -> None:
|
|
mock_resp = self._make_response("word " * 10000, content_type="text/plain")
|
|
mock_cls.return_value.__enter__.return_value.get.return_value = mock_resp
|
|
|
|
result = tools.fetch_url("https://example.com/big", max_chars=100)
|
|
assert "truncated" in result
|
|
|
|
@patch("gitea.tools.research_tools.httpx.Client")
|
|
def test_handles_http_404(
|
|
self, mock_cls: MagicMock, tools: ResearchTools
|
|
) -> None:
|
|
import httpx as _httpx
|
|
|
|
mock_response = MagicMock()
|
|
mock_response.status_code = 404
|
|
mock_cls.return_value.__enter__.return_value.get.side_effect = (
|
|
_httpx.HTTPStatusError("not found", request=MagicMock(), response=mock_response)
|
|
)
|
|
result = tools.fetch_url("https://example.com/missing")
|
|
assert "404" in result or "Failed" in result
|
|
|
|
@patch("gitea.tools.research_tools.httpx.Client")
|
|
def test_handles_timeout(
|
|
self, mock_cls: MagicMock, tools: ResearchTools
|
|
) -> None:
|
|
import httpx as _httpx
|
|
|
|
mock_cls.return_value.__enter__.return_value.get.side_effect = (
|
|
_httpx.TimeoutException("timed out")
|
|
)
|
|
result = tools.fetch_url("https://slow.example.com")
|
|
assert "timed out" in result.lower() or "timeout" in result.lower()
|
|
|
|
@patch("gitea.tools.research_tools.httpx.Client")
|
|
def test_html_extraction_called_for_html_content(
|
|
self, mock_cls: MagicMock, tools: ResearchTools
|
|
) -> None:
|
|
html = "<html><body><p>Hello world from main content</p></body></html>"
|
|
mock_resp = self._make_response(html)
|
|
mock_cls.return_value.__enter__.return_value.get.return_value = mock_resp
|
|
|
|
with patch.object(tools, "_html_to_markdown", return_value="Hello world") as mock_extract:
|
|
result = tools.fetch_url("https://example.com")
|
|
mock_extract.assert_called_once_with(html)
|
|
assert "Hello world" in result
|
|
|
|
@patch("gitea.tools.research_tools.httpx.Client")
|
|
def test_default_max_chars_is_20k(
|
|
self, mock_cls: MagicMock, tools: ResearchTools
|
|
) -> None:
|
|
mock_resp = self._make_response("a" * 30000, content_type="text/plain")
|
|
mock_cls.return_value.__enter__.return_value.get.return_value = mock_resp
|
|
|
|
result = tools.fetch_url("https://example.com/long")
|
|
assert "truncated" in result
|
|
# Content before truncation note should be ~20k chars
|
|
content_before = result.split("[Content truncated")[0]
|
|
# Allow a small buffer for the trailing \n\n appended before the truncation note
|
|
assert len(content_before) <= _MAX_CONTENT_CHARS + 4
|