"""web_view must open up to three URLs at once and never lose the pages it got.
Pages take a second to several, so a batch is worth it only if it actually runs
concurrently — and only if the shared text budget and the per-URL deadlines keep
one slow or broken page from costing the others. The loader lives in
``conftest.py`` next to this file.
"""
import asyncio
import time
def _page(text: str = "hello") -> dict:
return {"success": True, "output": f"Title: t\n\n{text}", "error": None,
"metadata": {}}
def _failed(error: str = "invalid_url") -> dict:
return {"success": False, "output": "URL must start with http:// or https://",
"error": error, "metadata": {}}
class TestSingleUrl:
async def test_one_url_keeps_the_plain_output(
self, navi_web_browse, monkeypatch
):
seen: dict = {}
async def view_one(url, screenshot=False, wait_until="networkidle"):
seen["url"] = url
seen["screenshot"] = screenshot
return _page()
monkeypatch.setattr(navi_web_browse, "_view_one", view_one)
result = await navi_web_browse.web_view("https://example.com", screenshot=True)
assert result == _page()
assert seen == {"url": "https://example.com", "screenshot": True}
async def test_an_invalid_url_is_reported_without_launching(
self, navi_web_browse
):
result = await navi_web_browse.web_view("ftp://example.com")
assert result["success"] is False
assert result["error"] == "invalid_url"
class TestBatch:
async def test_urls_open_concurrently(self, navi_web_browse, monkeypatch):
opened: list[str] = []
async def view_one(url, screenshot=False, wait_until="networkidle"):
opened.append(url)
await asyncio.sleep(0.1)
return _page(url)
monkeypatch.setattr(navi_web_browse, "_view_one", view_one)
began = time.monotonic()
result = await navi_web_browse.web_view(["https://a", "https://b", "https://c"])
elapsed = time.monotonic() - began
assert result["success"] is True
assert opened == ["https://a", "https://b", "https://c"]
assert elapsed < 0.25, f"three 0.1s pages took {elapsed:.2f}s — not concurrent"
assert result["output"].count("### ") == 3
async def test_blocks_follow_the_request_order(self, navi_web_browse, monkeypatch):
async def view_one(url, screenshot=False, wait_until="networkidle"):
await asyncio.sleep(0.1 if url == "https://a" else 0)
return _page(url)
monkeypatch.setattr(navi_web_browse, "_view_one", view_one)
result = await navi_web_browse.web_view(["https://a", "https://b"])
assert result["output"].index("### https://a") < result["output"].index(
"### https://b"
)
async def test_more_than_three_are_capped_with_a_note(
self, navi_web_browse, monkeypatch
):
opened: list[str] = []
async def view_one(url, screenshot=False, wait_until="networkidle"):
opened.append(url)
return _page()
monkeypatch.setattr(navi_web_browse, "_view_one", view_one)
result = await navi_web_browse.web_view(
[f"https://{i}" for i in range(5)]
)
assert len(opened) == 3
assert "Only the first 3 of 5 URLs were opened" in result["output"]
async def test_duplicates_collapse(self, navi_web_browse, monkeypatch):
opened: list[str] = []
async def view_one(url, screenshot=False, wait_until="networkidle"):
opened.append(url)
return _page()
monkeypatch.setattr(navi_web_browse, "_view_one", view_one)
await navi_web_browse.web_view(["https://a", " https://a ", "https://b"])
assert opened == ["https://a", "https://b"]
class TestFailures:
async def test_one_failed_page_does_not_erase_the_others(
self, navi_web_browse, monkeypatch
):
async def view_one(url, screenshot=False, wait_until="networkidle"):
return _failed() if url == "https://b" else _page()
monkeypatch.setattr(navi_web_browse, "_view_one", view_one)
result = await navi_web_browse.web_view(["https://a", "https://b", "https://c"])
assert result["success"] is True
assert "### https://b -> failed:" in result["output"]
assert [item["success"] for item in result["metadata"]["urls"]] == [
True, False, True,
]
async def test_a_page_timing_out_does_not_erase_the_others(
self, navi_web_browse, monkeypatch
):
async def view_one(url, screenshot=False, wait_until="networkidle"):
if url == "https://b":
await asyncio.sleep(10)
return _page()
monkeypatch.setattr(navi_web_browse, "_view_one", view_one)
monkeypatch.setattr(navi_web_browse, "_BATCH_DEADLINE", 0.05)
result = await navi_web_browse.web_view(["https://a", "https://b"])
assert result["success"] is True
assert "### https://b -> failed:" in result["output"]
async def test_a_batch_with_no_pages_fails(self, navi_web_browse, monkeypatch):
async def view_one(url, screenshot=False, wait_until="networkidle"):
return _failed()
monkeypatch.setattr(navi_web_browse, "_view_one", view_one)
result = await navi_web_browse.web_view(["https://a", "https://b"])
assert result["success"] is False
assert result["error"] == "all_urls_failed"
async def test_no_url_is_an_error(self, navi_web_browse):
result = await navi_web_browse.web_view([" ", ""])
assert result["success"] is False
assert result["error"] == "empty_url"
async def test_a_screenshot_batch_is_refused_up_front(
self, navi_web_browse, monkeypatch
):
"""`screenshot` returns image metadata for one page; with three there is
nowhere to put them, so say so instead of silently dropping two."""
opened: list[str] = []
async def view_one(url, screenshot=False, wait_until="networkidle"):
opened.append(url)
return _page()
monkeypatch.setattr(navi_web_browse, "_view_one", view_one)
result = await navi_web_browse.web_view(
["https://a", "https://b"], screenshot=True
)
assert result["success"] is False
assert result["error"] == "screenshot_batch"
assert opened == []
class TestTextBudget:
async def test_the_batch_stops_at_its_shared_text_budget(
self, navi_web_browse, monkeypatch
):
"""Three pages at the single-page cap would be 60k chars of context; the
batch stops at 40k and says where it stopped."""
big = "x" * (navi_web_browse._MAX_TEXT)
async def view_one(url, screenshot=False, wait_until="networkidle"):
return _page(big)
monkeypatch.setattr(navi_web_browse, "_view_one", view_one)
result = await navi_web_browse.web_view(["https://a", "https://b", "https://c"])
assert len(result["output"]) < navi_web_browse._BATCH_TEXT + 2_000
assert "40k-char" in result["output"]
assert result["metadata"]["urls"][-1]["truncated"] is True
async def test_a_single_page_keeps_its_own_cap(self, navi_web_browse, monkeypatch):
async def view_one(url, screenshot=False, wait_until="networkidle"):
return _page("x" * navi_web_browse._MAX_TEXT)
monkeypatch.setattr(navi_web_browse, "_view_one", view_one)
result = await navi_web_browse.web_view("https://a")
assert "40k-char" not in result["output"]
class TestToolLayer:
"""A call that opened nothing must reach the journal as an error.
``view_tool`` used to return the JSON with ``"success": false`` and no
``isError`` — which is how a server whose browser could not launch at all
looked like a working one, with a 1400-char payload I mistook for page text.
"""
async def test_no_page_opened_raises(self, navi_web_server, monkeypatch):
async def failed(url, screenshot=False, wait_until="networkidle"):
return {"success": False, "output": "Browser error: no browser",
"error": "all_urls_failed", "metadata": {}}
monkeypatch.setattr(navi_web_server, "web_view", failed)
try:
await navi_web_server.view_tool(["https://a", "https://b"])
except RuntimeError as exc:
assert "Browser error" in str(exc)
else:
raise AssertionError("a failed web_view did not raise")
async def test_a_partial_batch_still_returns(
self, navi_web_server, monkeypatch
):
async def partly(url, screenshot=False, wait_until="networkidle"):
return {"success": True, "output": "### https://a\nok", "error": None,
"metadata": {"urls": [{"url": "https://a", "success": True}]}}
monkeypatch.setattr(navi_web_server, "web_view", partly)
body = await navi_web_server.view_tool(["https://a", "https://b"])
assert '"success": true' in body
