118 lines
3.5 KiB
Python
118 lines
3.5 KiB
Python
"""fetch_url executor 单元测试。
|
||
|
||
mock 边界：
|
||
- ``respx`` 模拟 httpx.AsyncClient 的 GET（不真实联网）。
|
||
- executor 直接调用，不经过 ToolRuntime（runtime 集成路径由 test_runtime 覆盖）。
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
from collections.abc import AsyncIterator
|
||
from typing import Any
|
||
|
||
import httpx
|
||
import pytest
|
||
import respx
|
||
|
||
from app.application.tool_runtime.web_tools import _fetch_url_executor
|
||
|
||
|
||
@pytest.fixture
|
||
async def http_client() -> AsyncIterator[httpx.AsyncClient]:
|
||
async with httpx.AsyncClient() as client:
|
||
yield client
|
||
|
||
|
||
def _ctx(http_client: httpx.AsyncClient, **extra: Any) -> dict[str, Any]:
|
||
"""构造 executor 上下文。"""
|
||
base: dict[str, Any] = {"http_client": http_client, "trace_id": "trace_test"}
|
||
base.update(extra)
|
||
return base
|
||
|
||
|
||
@respx.mock
|
||
async def test_fetch_url_converts_html_to_markdown(
|
||
http_client: httpx.AsyncClient,
|
||
) -> None:
|
||
"""fetch_url 把 HTML 转成 markdown 并截断。"""
|
||
html = (
|
||
"<html><head><title>T</title></head>"
|
||
"<body><h1>标题</h1><p>这是一段正文。</p>"
|
||
'<a href="https://example.com">链接</a></body></html>'
|
||
)
|
||
respx.get("https://example.com/page").mock(return_value=httpx.Response(200, text=html))
|
||
|
||
result = await _fetch_url_executor({"url": "https://example.com/page"}, _ctx(http_client))
|
||
|
||
assert result.ok
|
||
assert "标题" in result.stdout
|
||
assert "这是一段正文。" in result.stdout
|
||
# html2text 默认保留链接
|
||
assert "example.com" in result.stdout
|
||
|
||
|
||
@respx.mock
|
||
async def test_fetch_url_truncates_to_max_chars(
|
||
http_client: httpx.AsyncClient,
|
||
) -> None:
|
||
"""max_chars 截断生效。"""
|
||
long_text = "A" * 500
|
||
html = f"<html><body><p>{long_text}</p></body></html>"
|
||
respx.get("https://example.com/big").mock(return_value=httpx.Response(200, text=html))
|
||
|
||
result = await _fetch_url_executor(
|
||
{"url": "https://example.com/big", "max_chars": 50}, _ctx(http_client)
|
||
)
|
||
|
||
assert result.ok
|
||
assert "已截断" in result.stdout
|
||
assert len(result.stdout) < 500
|
||
|
||
|
||
@respx.mock
|
||
async def test_fetch_url_http_error_returns_error(
|
||
http_client: httpx.AsyncClient,
|
||
) -> None:
|
||
"""HTTP 错误返回 ok=False。"""
|
||
respx.get("https://example.com/missing").mock(
|
||
return_value=httpx.Response(404, text="not found")
|
||
)
|
||
|
||
result = await _fetch_url_executor({"url": "https://example.com/missing"}, _ctx(http_client))
|
||
|
||
assert not result.ok
|
||
assert "404" in result.stderr
|
||
|
||
|
||
async def test_fetch_url_missing_url_returns_error(
|
||
http_client: httpx.AsyncClient,
|
||
) -> None:
|
||
"""缺 url 输入时返回 ok=False。"""
|
||
result = await _fetch_url_executor({}, _ctx(http_client))
|
||
|
||
assert not result.ok
|
||
assert "url" in result.stderr
|
||
|
||
|
||
async def test_fetch_url_missing_http_client_returns_error() -> None:
|
||
"""缺注入的 http_client 时返回 ok=False。"""
|
||
result = await _fetch_url_executor({"url": "https://example.com"}, None)
|
||
|
||
assert not result.ok
|
||
assert "http_client" in result.stderr
|
||
|
||
|
||
@respx.mock
|
||
async def test_fetch_url_network_error_returns_error(
|
||
http_client: httpx.AsyncClient,
|
||
) -> None:
|
||
"""网络错误（HTTPError）返回 ok=False 而不抛异常。"""
|
||
respx.get("https://example.com/neterr").mock(
|
||
side_effect=httpx.ConnectError("connection refused")
|
||
)
|
||
|
||
result = await _fetch_url_executor({"url": "https://example.com/neterr"}, _ctx(http_client))
|
||
|
||
assert not result.ok
|
||
assert "请求失败" in result.stderr
|