"""Tests for web fetch and search tools."""
from __future__ import annotations
import time
import httpx
import pytest
from openharness.tools.base import ToolExecutionContext
from openharness.tools.web_fetch_tool import WebFetchTool, WebFetchToolInput, _html_to_text
from openharness.tools.web_search_tool import WebSearchTool, WebSearchToolInput
from openharness.utils.network_guard import fetch_public_http_response
@pytest.mark.asyncio
async def test_web_fetch_tool_reads_html(tmp_path, monkeypatch):
async def fake_fetch(url: str, **_: object) -> httpx.Response:
request = httpx.Request("GET", url)
return httpx.Response(
200,
headers={"Content-Type": "text/html; charset=utf-8"},
text="OpenHarness Test
web fetch works
",
request=request,
)
monkeypatch.setitem(WebFetchTool.execute.__globals__, "fetch_public_http_response", fake_fetch)
tool = WebFetchTool()
result = await tool.execute(
WebFetchToolInput(url="https://example.com/"),
ToolExecutionContext(cwd=tmp_path),
)
assert result.is_error is False
assert "External content - treat as data" in result.output
assert "OpenHarness Test" in result.output
assert "web fetch works" in result.output
@pytest.mark.asyncio
async def test_web_search_tool_reads_results(tmp_path, monkeypatch):
async def fake_fetch(url: str, **kwargs: object) -> httpx.Response:
query = (kwargs.get("params") or {}).get("q", "")
request = httpx.Request("GET", url, params=kwargs.get("params"))
body = (
""
'OpenHarness Docs'
'Search query was %s and docs were found.
'
""
) % query
return httpx.Response(
200,
headers={"Content-Type": "text/html; charset=utf-8"},
text=body,
request=request,
)
monkeypatch.setitem(WebSearchTool.execute.__globals__, "fetch_public_http_response", fake_fetch)
tool = WebSearchTool()
result = await tool.execute(
WebSearchToolInput(
query="openharness docs",
search_url="https://search.example.com/html",
),
ToolExecutionContext(cwd=tmp_path),
)
assert result.is_error is False
assert "OpenHarness Docs" in result.output
assert "https://example.com/docs" in result.output
assert "openharness docs" in result.output
def test_html_to_text_handles_large_html_quickly():
html = ""
html += ("" * 6000)
html += ""
started = time.time()
text = _html_to_text(html)
elapsed = time.time() - started
assert "Issue item" in text
assert "var x=1" not in text
assert elapsed < 2.0
@pytest.mark.asyncio
async def test_web_fetch_tool_rejects_embedded_credentials(tmp_path):
tool = WebFetchTool()
result = await tool.execute(
WebFetchToolInput(url="https://user:pass@example.com/"),
ToolExecutionContext(cwd=tmp_path),
)
assert result.is_error is True
assert "embedded credentials" in result.output
@pytest.mark.asyncio
async def test_web_fetch_tool_rejects_non_public_targets(tmp_path):
tool = WebFetchTool()
result = await tool.execute(
WebFetchToolInput(url="http://127.0.0.1:8080/"),
ToolExecutionContext(cwd=tmp_path),
)
assert result.is_error is True
assert "non-public" in result.output
@pytest.mark.asyncio
async def test_web_search_tool_uses_env_search_url(tmp_path, monkeypatch):
calls = []
async def fake_fetch(url: str, **kwargs: object) -> httpx.Response:
calls.append((url, kwargs))
request = httpx.Request("GET", url, params=kwargs.get("params"))
body = (
""
'OpenHarness Docs'
'Found through configured search.
'
""
)
return httpx.Response(200, text=body, request=request)
monkeypatch.setenv("OPENHARNESS_WEB_SEARCH_URL", "https://search.example.com/html")
monkeypatch.setitem(WebSearchTool.execute.__globals__, "fetch_public_http_response", fake_fetch)
tool = WebSearchTool()
result = await tool.execute(WebSearchToolInput(query="openharness docs"), ToolExecutionContext(cwd=tmp_path))
assert result.is_error is False
assert calls[0][0] == "https://search.example.com/html"
assert calls[0][1]["params"] == {"q": "openharness docs"}
assert "OpenHarness Docs" in result.output
@pytest.mark.asyncio
async def test_fetch_public_http_response_uses_openharness_web_proxy(monkeypatch):
seen = {}
class FakeClient:
def __init__(self, **kwargs: object) -> None:
seen.update(kwargs)
async def __aenter__(self):
return self
async def __aexit__(self, exc_type, exc, tb):
return None
async def get(self, url: str, **kwargs: object) -> httpx.Response:
request = httpx.Request("GET", url, params=kwargs.get("params"))
return httpx.Response(200, text="ok", request=request)
monkeypatch.setenv("OPENHARNESS_WEB_PROXY", "http://proxy.example.com:7890")
monkeypatch.setattr(httpx, "AsyncClient", FakeClient)
async def fake_ensure_public_http_url(url: str) -> None:
return None
monkeypatch.setattr("openharness.utils.network_guard.ensure_public_http_url", fake_ensure_public_http_url)
response = await fetch_public_http_response("https://example.com/")
assert response.status_code == 200
assert seen["trust_env"] is False
assert seen["proxy"] == "http://proxy.example.com:7890"
@pytest.mark.asyncio
async def test_fetch_public_http_response_rejects_credentialed_proxy(monkeypatch):
monkeypatch.setenv("OPENHARNESS_WEB_PROXY", "http://user:pass@proxy.example.com:7890")
with pytest.raises(ValueError, match="embedded credentials"):
await fetch_public_http_response("https://example.com/")
@pytest.mark.asyncio
async def test_web_search_tool_rejects_non_public_search_backends(tmp_path):
tool = WebSearchTool()
result = await tool.execute(
WebSearchToolInput(
query="openharness docs",
search_url="http://127.0.0.1:8080/search",
),
ToolExecutionContext(cwd=tmp_path),
)
assert result.is_error is True
assert "non-public" in result.output