136 lines
4.8 KiB
Python
136 lines
4.8 KiB
Python
"""Unit tests for the bounded site crawler (#92): crawl_site (WRITE + confirmation)."""
|
|
|
|
from typing import Any
|
|
|
|
import pytest
|
|
|
|
import backend.tools.crawler as crawler
|
|
from backend.tools.api import ToolConfirmationRequired, ToolContext, ToolError, call_tool
|
|
|
|
|
|
class FakeResponse:
|
|
def __init__(self, content: bytes = b"", status_code: int = 200,
|
|
headers: dict | None = None):
|
|
self.content = content
|
|
self.status_code = status_code
|
|
self.headers = headers or {"content-type": "text/html; charset=utf-8"}
|
|
self.encoding = "utf-8"
|
|
|
|
def raise_for_status(self):
|
|
pass
|
|
|
|
|
|
def _ctx() -> ToolContext:
|
|
return ToolContext(
|
|
user={"username": "tester", "role": "admin", "vaults": ["*"]},
|
|
audit_enabled=False,
|
|
)
|
|
|
|
|
|
@pytest.fixture
|
|
def vault(tmp_path, monkeypatch):
|
|
vault_dir = tmp_path / "Vault"
|
|
vault_dir.mkdir()
|
|
monkeypatch.setitem(__import__("backend.indexer", fromlist=["index"]).index,
|
|
"Vault", {"name": "Vault", "path": str(vault_dir), "config": {}})
|
|
return vault_dir
|
|
|
|
|
|
PAGE_A = (
|
|
b"<html><head><title>Docs</title></head><body>"
|
|
b"<p>Bienvenue sur la documentation.</p>"
|
|
b'<a href="/page-b">Suite</a><a href="https://other.dev/x">ext</a>'
|
|
b"</body></html>"
|
|
)
|
|
PAGE_B = (
|
|
b"<html><head><title>Page B</title></head><body><p>Details ici.</p></body></html>"
|
|
)
|
|
|
|
|
|
@pytest.fixture
|
|
def local_urls(monkeypatch):
|
|
"""Skip the DNS-based SSRF guard: test hosts are fake, HTTP is mocked."""
|
|
monkeypatch.setattr(crawler, "_assert_public_http_url", lambda url: url)
|
|
|
|
|
|
@pytest.fixture
|
|
def two_pages(monkeypatch, local_urls):
|
|
def fake_get(url, **kw):
|
|
url = str(url)
|
|
if url.endswith("/page-b"):
|
|
return FakeResponse(content=PAGE_B)
|
|
return FakeResponse(content=PAGE_A)
|
|
|
|
monkeypatch.setattr(crawler.httpx, "get", fake_get)
|
|
|
|
|
|
class TestConfirmation:
|
|
def test_requires_confirmation(self, vault, two_pages):
|
|
with pytest.raises(ToolConfirmationRequired):
|
|
call_tool("crawl_site", _ctx(), {
|
|
"url": "https://docs.example.dev/start",
|
|
"vault": "Vault", "path": "Crawls/docs.md",
|
|
})
|
|
|
|
|
|
class TestCrawl:
|
|
def test_saves_same_host_pages(self, vault, two_pages):
|
|
out = call_tool("crawl_site", _ctx(), {
|
|
"url": "https://docs.example.dev/start",
|
|
"vault": "Vault", "path": "Crawls/docs.md",
|
|
}, confirm=True)
|
|
assert out.ok and out.data["pages"] == 2
|
|
digest = (vault / "Crawls" / "docs.md").read_text(encoding="utf-8")
|
|
assert "# Crawl de docs.example.dev" in digest
|
|
assert "Bienvenue sur la documentation." in digest
|
|
assert "Details ici." in digest
|
|
assert "other.dev" not in digest
|
|
|
|
def test_max_pages_bound(self, vault, monkeypatch, local_urls):
|
|
# A link farm: every page links to a new page — cap at max_pages.
|
|
def fake_get(url, **kw):
|
|
url = str(url)
|
|
n = int(url.rsplit("/", 1)[-1] or 0)
|
|
return FakeResponse(
|
|
content=f"<html><head><title>P{n}</title></head><body>"
|
|
f"<p>page {n}</p><a href=\"/{n + 1}\">next</a></body></html>".encode())
|
|
|
|
monkeypatch.setattr(crawler.httpx, "get", fake_get)
|
|
out = call_tool("crawl_site", _ctx(), {
|
|
"url": "https://farm.example.dev/0",
|
|
"vault": "Vault", "path": "farm.md", "max_pages": 3,
|
|
}, confirm=True)
|
|
assert out.ok and out.data["pages"] == 3
|
|
|
|
def test_no_pages_recovered(self, vault, monkeypatch, local_urls):
|
|
def dead_get(*a, **kw):
|
|
raise crawler.httpx.ConnectError("down")
|
|
|
|
monkeypatch.setattr(crawler.httpx, "get", dead_get)
|
|
with pytest.raises(ToolError) as ei:
|
|
call_tool("crawl_site", _ctx(), {
|
|
"url": "https://dead.example.dev/", "vault": "Vault", "path": "x.md",
|
|
}, confirm=True)
|
|
assert ei.value.code == "crawl_failed"
|
|
|
|
def test_internal_url_rejected(self, vault):
|
|
with pytest.raises(ToolError) as ei:
|
|
call_tool("crawl_site", _ctx(), {
|
|
"url": "http://127.0.0.1:8080/", "vault": "Vault", "path": "x.md",
|
|
}, confirm=True)
|
|
assert ei.value.code in ("ssrf_blocked", "dns_error")
|
|
|
|
def test_binary_content_skipped(self, vault, monkeypatch, local_urls):
|
|
def fake_get(url, **kw):
|
|
url = str(url)
|
|
if url.endswith("/x.pdf"):
|
|
return FakeResponse(content=b"%PDF-1.4", headers={"content-type": "application/pdf"})
|
|
return FakeResponse(content=PAGE_A)
|
|
|
|
monkeypatch.setattr(crawler.httpx, "get", fake_get)
|
|
out = call_tool("crawl_site", _ctx(), {
|
|
"url": "https://docs.example.dev/start",
|
|
"vault": "Vault", "path": "docs.md", "max_pages": 5,
|
|
}, confirm=True)
|
|
assert out.ok and out.data["pages"] >= 1
|