82 lines
2.7 KiB
Python
82 lines
2.7 KiB
Python
from __future__ import annotations
|
|
|
|
import json
|
|
import unittest
|
|
from types import SimpleNamespace
|
|
from unittest.mock import AsyncMock, patch
|
|
|
|
from app.web_tools import WebPageTools, _ensure_public_url
|
|
|
|
|
|
class WebPageToolsTests(unittest.IsolatedAsyncioTestCase):
|
|
async def test_private_network_urls_are_rejected(self) -> None:
|
|
for url in (
|
|
"http://127.0.0.1/admin",
|
|
"http://10.0.0.1/",
|
|
"http://169.254.169.254/latest/meta-data/",
|
|
"http://[::1]/",
|
|
):
|
|
with self.subTest(url=url):
|
|
with self.assertRaisesRegex(ValueError, "private network"):
|
|
await _ensure_public_url(url)
|
|
|
|
async def test_reads_page_as_bounded_markdown(self) -> None:
|
|
crawl_result = SimpleNamespace(
|
|
success=True,
|
|
url="https://example.com/article",
|
|
redirected_url=None,
|
|
status_code=200,
|
|
metadata={"title": "Example"},
|
|
markdown="abcdefghij",
|
|
)
|
|
with (
|
|
patch(
|
|
"app.web_tools._ensure_public_url",
|
|
AsyncMock(side_effect=lambda value: value),
|
|
) as ensure_public,
|
|
patch(
|
|
"app.web_tools._crawl_with_playwright",
|
|
AsyncMock(return_value=crawl_result),
|
|
) as crawl,
|
|
):
|
|
payload = json.loads(
|
|
await WebPageTools(timeout_seconds=12, max_chars=5).read_web_page(
|
|
"https://example.com/article"
|
|
)
|
|
)
|
|
|
|
self.assertEqual(payload["markdown"], "abcde")
|
|
self.assertTrue(payload["truncated"])
|
|
self.assertEqual(payload["title"], "Example")
|
|
self.assertEqual(ensure_public.await_count, 2)
|
|
crawl.assert_awaited_once_with("https://example.com/article", 12)
|
|
|
|
async def test_crawl_failure_is_reported_to_agent(self) -> None:
|
|
crawl_result = SimpleNamespace(
|
|
success=False,
|
|
url="https://example.com",
|
|
redirected_url=None,
|
|
error_message="blocked",
|
|
)
|
|
with (
|
|
patch(
|
|
"app.web_tools._ensure_public_url",
|
|
AsyncMock(side_effect=lambda value: value),
|
|
),
|
|
patch(
|
|
"app.web_tools._crawl_with_playwright",
|
|
AsyncMock(return_value=crawl_result),
|
|
),
|
|
):
|
|
with self.assertRaisesRegex(RuntimeError, "blocked"):
|
|
await WebPageTools().read_web_page("https://example.com")
|
|
|
|
def test_tool_is_registered_for_all_agent_sessions(self) -> None:
|
|
tools = WebPageTools().tools()
|
|
|
|
self.assertEqual([tool.name for tool in tools], ["read_web_page"])
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|