hkuds--openharness
tjb-tech
5dd8b952ec
A lightweight open-source Python implementation of the Agent Harness architecture. 44x lighter than Claude Code (11K vs 512K lines), 98% core tool coverage. - 43 tools with Pydantic validation and parallel execution - Skills system compatible with anthropics/skills (17+ tested) - Plugin system compatible with claude-code/plugins (12+ tested) - API retry with exponential backoff - Multi-level permissions with path rules - React/Ink TUI with "Oh my Harness!" branding - 114 unit tests + 6 E2E test suites - MIT License
87 行
2.9 KiB
Python
87 行
2.9 KiB
Python
"""Tests for web fetch and search tools."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import contextlib
|
|
import threading
|
|
from http.server import BaseHTTPRequestHandler, HTTPServer
|
|
|
|
import pytest
|
|
|
|
from urllib.parse import parse_qs, urlparse
|
|
|
|
from openharness.tools.base import ToolExecutionContext
|
|
from openharness.tools.web_fetch_tool import WebFetchTool, WebFetchToolInput
|
|
from openharness.tools.web_search_tool import WebSearchTool, WebSearchToolInput
|
|
|
|
|
|
class _Handler(BaseHTTPRequestHandler):
|
|
def do_GET(self) -> None: # noqa: N802
|
|
query = parse_qs(urlparse(self.path).query).get("q", [""])[0]
|
|
if query:
|
|
body = (
|
|
"<html><body>"
|
|
'<a class="result__a" href="https://example.com/docs">OpenHarness Docs</a>'
|
|
'<div class="result__snippet">Search query was %s and docs were found.</div>'
|
|
"</body></html>"
|
|
) % query
|
|
else:
|
|
body = "<html><body><h1>OpenHarness Test</h1><p>web fetch works</p></body></html>"
|
|
encoded = body.encode("utf-8")
|
|
self.send_response(200)
|
|
self.send_header("Content-Type", "text/html; charset=utf-8")
|
|
self.send_header("Content-Length", str(len(encoded)))
|
|
self.end_headers()
|
|
self.wfile.write(encoded)
|
|
|
|
def log_message(self, format: str, *args) -> None: # noqa: A003
|
|
del format, args
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_web_fetch_tool_reads_html(tmp_path):
|
|
server = HTTPServer(("127.0.0.1", 0), _Handler)
|
|
thread = threading.Thread(target=server.serve_forever, daemon=True)
|
|
thread.start()
|
|
try:
|
|
tool = WebFetchTool()
|
|
result = await tool.execute(
|
|
WebFetchToolInput(url=f"http://127.0.0.1:{server.server_port}/"),
|
|
ToolExecutionContext(cwd=tmp_path),
|
|
)
|
|
finally:
|
|
server.shutdown()
|
|
with contextlib.suppress(Exception):
|
|
server.server_close()
|
|
thread.join(timeout=1)
|
|
|
|
assert result.is_error is False
|
|
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):
|
|
server = HTTPServer(("127.0.0.1", 0), _Handler)
|
|
thread = threading.Thread(target=server.serve_forever, daemon=True)
|
|
thread.start()
|
|
try:
|
|
tool = WebSearchTool()
|
|
result = await tool.execute(
|
|
WebSearchToolInput(
|
|
query="openharness docs",
|
|
search_url=f"http://127.0.0.1:{server.server_port}/search",
|
|
),
|
|
ToolExecutionContext(cwd=tmp_path),
|
|
)
|
|
finally:
|
|
server.shutdown()
|
|
with contextlib.suppress(Exception):
|
|
server.server_close()
|
|
thread.join(timeout=1)
|
|
|
|
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
|