engineering/arborist
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions
arborist/tests/crawler/test_bridge.py
russell@unturf.com 8d6961fcc1
aborist/arborist 
modified:   .gitlab-ci.yml
	modified:   bench/qa_questions.txt
	modified:   bench/qa_sweep.py
	modified:   bench/run.sh
	modified:   docs/TICKETS.md
	modified:   docs/_source/README.md
	modified:   docs/_source/_ext/makefile_targets.py
	modified:   docs/_source/api/cli.rst
	modified:   docs/_source/api/distill.rst
	modified:   docs/_source/api/mesh.rst
	modified:   docs/_source/api/qa.rst
	modified:   docs/_source/api/retrieval.rst
	modified:   docs/_source/api/storage.rst
	modified:   docs/_source/api/substrate.rst
	modified:   docs/_source/concepts.rst
	modified:   docs/_source/conf.py
	modified:   docs/_source/cookbook.rst
	modified:   docs/_source/index.rst
	modified:   docs/_source/license.rst
	modified:   docs/_source/quickstart.rst
	modified:   docs/bench-maxing.md
	modified:   docs/benchmarks.md
	modified:   docs/cti-architecture.md
	modified:   docs/diagrams/aborist-modules.dot
	modified:   docs/diagrams/aborist-modules.svg
	modified:   docs/diagrams/mesh-data-flow.dot
	modified:   docs/diagrams/mesh-epoch-lifecycle.dot
	modified:   docs/diagrams/mesh-epoch-lifecycle.svg
	modified:   docs/diagrams/mesh-group-decisions.dot
	modified:   docs/diagrams/mesh-group-decisions.svg
	modified:   docs/diagrams/mesh-identity-stack.dot
	modified:   docs/diagrams/mesh-secret-envelope.dot
	modified:   docs/mesh.md
	modified:   docs/qa-modes-bench.md
	modified:   docs/seven-point-program.md
	modified:   docs/tickets/ticket-000001-retrieval-keywords-audit-gap.md
	modified:   docs/tickets/ticket-000002-reference-frame-polarity-contract.md
	modified:   docs/tickets/ticket-000003-anchor-class-warrant.md
	modified:   docs/tickets/ticket-000005-label-ladder-migration.md
	modified:   docs/tickets/ticket-000006-bench-emergent-findings.md
	modified:   docs/tickets/ticket-000007-query-layer-hyphen-fold.md
	modified:   docs/tickets/ticket-000008-broad-quantifier-preflight-guard.md
	modified:   docs/tickets/ticket-000009-quantifier-preflight-dag-binding.md
	modified:   docs/tickets/ticket-000010-metacognition-preflight-guard.md
	modified:   docs/tickets/ticket-000011-soft-preflight-hint-sidecar.md
	modified:   scripts/backfill_concepts.py
	modified:   scripts/bench_emergent.py
	modified:   tests/crawler/test_async_web_fetcher.py
	modified:   tests/crawler/test_bridge.py
	modified:   tests/crawler/test_web_fetch.py
	modified:   tests/test_bench_qa_sweep.py
	modified:   tests/test_burn.py
	modified:   tests/test_burn_doc.py
	modified:   tests/test_claim_lattice.py
	modified:   tests/test_cli_render.py
	modified:   tests/test_compress.py
	modified:   tests/test_concepts.py
	modified:   tests/test_dag.py
	modified:   tests/test_directives.py
	modified:   tests/test_distill.py
	modified:   tests/test_distill_recursive.py
	modified:   tests/test_evict.py
	modified:   tests/test_frame.py
	modified:   tests/test_grok_source.py
	modified:   tests/test_html_source.py
	modified:   tests/test_ingest.py
	modified:   tests/test_inspect.py
	modified:   tests/test_journal.py
	modified:   tests/test_keys.py
	modified:   tests/test_llm_context_base.py
	modified:   tests/test_merkle.py
	modified:   tests/test_mesh.py
	modified:   tests/test_mesh_aead.py
	modified:   tests/test_mesh_chain.py
	modified:   tests/test_mesh_cli.py
	modified:   tests/test_mesh_cli_pull.py
	modified:   tests/test_mesh_wire.py
	modified:   tests/test_mesh_wire_e2e.py
	modified:   tests/test_metacognition.py
	modified:   tests/test_migration_audit_mode.py
	modified:   tests/test_providence_source.py
	modified:   tests/test_qa.py
	modified:   tests/test_qa_quality_live.py
	modified:   tests/test_quantifier_caps.py
	modified:   tests/test_quantifier_classifier.py
	modified:   tests/test_quantifier_phase4.py
	modified:   tests/test_quantifier_reminder.py
	modified:   tests/test_query.py
	modified:   tests/test_reclassify.py
	modified:   tests/test_repair.py
	modified:   tests/test_resume.py
	modified:   tests/test_snapshot.py
	modified:   tests/test_soft_preflight.py
	modified:   tests/test_tfidf.py
	modified:   tests/test_vcs_source.py
	modified:   tests/test_verify.py
	modified:   tests/test_verify_json.py
	modified:   tests/test_versioned_ingest.py
	modified:   tests/test_warrant.py
	modified:   tests/test_wikipedia_old.py
	modified:   tests/test_wikipedia_xml.py
	modified:   tests/test_wikitext.py
2026-05-07 09:31:49 -04:00

549 lines
18 KiB
Python
Raw Blame History

"""Bridge tests: crawler discovery + ingest + recrawl-check.
These tests stub the network. The verbatim crawler tests in this same
directory exercise the underlying AsyncWebFetcher; here we exercise
the arborist bridge logic on top of it (BFS bounds, same-domain filter,
ETag capture, conditional HEAD classification).
"""
from __future__ import annotations
import json
import time
from pathlib import Path
from unittest.mock import patch
import httpx
import pytest
from arborist.sources.crawler.bridge import (
_normalize,
_same_domain,
crawl_seed,
ingest_crawled,
recrawl_check,
)
from arborist.store import connect
# ---------------------------------------------------------------------------
# URL helpers (pure)
# ---------------------------------------------------------------------------
def test_same_domain_exact_hostname():
assert _same_domain("https://x.com/a", "https://x.com/b") is True
def test_same_domain_rejects_subdomain():
assert _same_domain("https://x.com/a", "https://docs.x.com/b") is False
def test_same_domain_rejects_other_host():
assert _same_domain("https://x.com/a", "https://y.com/a") is False
def test_normalize_strips_fragment():
assert _normalize("https://x.com/a#section") == "https://x.com/a"
def test_normalize_preserves_query():
assert _normalize("https://x.com/a?q=1") == "https://x.com/a?q=1"
# ---------------------------------------------------------------------------
# crawl_seed: BFS discovery (no real network — mock the fetcher)
# ---------------------------------------------------------------------------
class _StubFetcher:
"""Stand-in for AsyncWebFetcher that returns pre-canned (html, links)
based on a URL→links map. Enough to drive the bridge's BFS.
"""
def __init__(self, page_links: dict[str, list[str]]):
self.page_links = page_links
async def fetch_webpage(self, url: str, extract_links: bool = False, **kw):
return "<html></html>", self.page_links.get(url, [])
def _run_crawl(page_links, seed, depth, max_pages):
"""Helper: drive _crawl_seed_async with a stub fetcher."""
import asyncio
from arborist.sources.crawler import bridge
return asyncio.run(
bridge._crawl_seed_async(
seed, max_depth=depth, max_pages=max_pages, fetcher=_StubFetcher(page_links)
)
)
def test_crawl_seed_breadth_first_within_limit():
page_links = {
"https://x.com/": ["/a", "/b"],
"https://x.com/a": ["/c"],
"https://x.com/b": [],
"https://x.com/c": [],
}
out = _run_crawl(page_links, "https://x.com/", depth=2, max_pages=10)
# Seed first, then both depth-1 pages, then the depth-2 page.
assert out[0] == "https://x.com/"
assert "https://x.com/a" in out
assert "https://x.com/b" in out
assert "https://x.com/c" in out
def test_crawl_seed_respects_max_pages():
page_links = {f"https://x.com/p{i}": [f"/p{i+1}"] for i in range(50)}
page_links["https://x.com/"] = ["/p0"]
out = _run_crawl(page_links, "https://x.com/", depth=10, max_pages=5)
assert len(out) == 5
def test_crawl_seed_max_pages_zero_means_no_cap():
"""Default `--max-pages 0` lets depth be the only bound. Pin so a
future "let's set a sensible default" PR can't silently re-introduce
a cap."""
# Linear chain of 30 pages within depth 30; max_pages=0 should
# discover all of them.
page_links = {f"https://x.com/p{i}": [f"/p{i+1}"] for i in range(30)}
page_links["https://x.com/"] = ["/p0"]
out = _run_crawl(page_links, "https://x.com/", depth=30, max_pages=0)
assert len(out) >= 30 # seed + at least 30 chained pages
def test_crawl_seed_respects_max_depth():
page_links = {
"https://x.com/": ["/a"],
"https://x.com/a": ["/b"],
"https://x.com/b": ["/c"],
"https://x.com/c": [],
}
out = _run_crawl(page_links, "https://x.com/", depth=1, max_pages=10)
# Seed (depth 0) + /a (depth 1) only. /b is depth 2, skipped.
assert "https://x.com/" in out
assert "https://x.com/a" in out
assert "https://x.com/b" not in out
def test_crawl_seed_drops_off_domain_links():
page_links = {
"https://x.com/": ["https://other.com/foo", "/local"],
"https://x.com/local": [],
}
out = _run_crawl(page_links, "https://x.com/", depth=2, max_pages=10)
assert "https://other.com/foo" not in out
assert "https://x.com/local" in out
def test_crawl_seed_dedups():
page_links = {
"https://x.com/": ["/a", "/a", "/b"],
"https://x.com/a": ["/b"],
"https://x.com/b": ["/a"],
}
out = _run_crawl(page_links, "https://x.com/", depth=3, max_pages=10)
# Each URL appears at most once even with cycles.
assert len(out) == len(set(out))
def test_crawl_seed_fast_flag_creates_fast_mode_fetcher(monkeypatch):
"""When fast=True and no fetcher is supplied, the bridge constructs an
AsyncWebFetcher with fast_mode=True. Pin so the kwarg can't silently
drop on the floor."""
captured: dict = {}
class _CapturingFetcher:
def __init__(self, **kwargs):
captured.update(kwargs)
async def fetch_webpage(self, url, extract_links=False, **kw):
return None, []
from arborist.sources.crawler import bridge
monkeypatch.setattr(bridge, "AsyncWebFetcher", _CapturingFetcher)
bridge.crawl_seed("https://x.com/", max_depth=1, max_pages=1, fast=True)
assert captured.get("fast_mode") is True
def test_crawl_seed_default_creates_polite_fetcher(monkeypatch):
"""fast=False (default) → fast_mode=False on the underlying fetcher."""
captured: dict = {}
class _CapturingFetcher:
def __init__(self, **kwargs):
captured.update(kwargs)
async def fetch_webpage(self, url, extract_links=False, **kw):
return None, []
from arborist.sources.crawler import bridge
monkeypatch.setattr(bridge, "AsyncWebFetcher", _CapturingFetcher)
bridge.crawl_seed("https://x.com/", max_depth=1, max_pages=1)
assert captured.get("fast_mode") is False
def test_crawl_seed_passes_extract_links_true():
"""Regression: AsyncWebFetcher.fetch_webpage defaults extract_links=False,
so without the explicit kwarg BFS terminates after the seed (only one
URL ever lands in `discovered`). Pin the kwarg by recording the calls
on a stub fetcher."""
class _RecordingFetcher:
def __init__(self):
self.calls: list[dict] = []
async def fetch_webpage(self, url, extract_links=False, **kw):
self.calls.append({"url": url, "extract_links": extract_links})
return "<html></html>", []
import asyncio
from arborist.sources.crawler import bridge
fetcher = _RecordingFetcher()
asyncio.run(
bridge._crawl_seed_async(
"https://x.com/", max_depth=2, max_pages=5, fetcher=fetcher
)
)
assert fetcher.calls, "fetcher was never called"
for call in fetcher.calls:
assert call["extract_links"] is True, (
f"BFS must request link extraction; got {call!r}"
)
# ---------------------------------------------------------------------------
# ingest_crawled: stores ETag + Last-Modified per document_root
# ---------------------------------------------------------------------------
@pytest.fixture
def http_meta_db(tmp_path):
db = tmp_path / "meta.db"
# Touch the DB so the schema is initialised, then close — we'll
# reopen as the test needs.
conn = connect(db)
try:
pass
finally:
conn.close()
return db
def _patched_httpx_get(url_to_response: dict[str, dict]):
"""Return a context-manager class that fakes httpx.Client.get."""
class _FakeClient:
def __init__(self, *a, **kw):
self._responses = url_to_response
def __enter__(self):
return self
def __exit__(self, *a):
pass
def get(self, url):
spec = self._responses.get(url)
if spec is None:
raise httpx.HTTPError(f"no stub for {url}")
req = httpx.Request("GET", url)
return httpx.Response(
spec.get("status", 200),
headers=spec.get("headers", {"content-type": "text/html"}),
content=spec.get("body", "<html><body>hi</body></html>").encode(),
request=req,
)
def head(self, url, headers=None):
spec = self._responses.get(url)
if spec is None:
raise httpx.HTTPError(f"no stub for {url}")
head_status = spec.get("head_status", 200)
req = httpx.Request("HEAD", url, headers=headers or {})
return httpx.Response(
head_status,
headers=spec.get("head_headers", {}),
request=req,
)
return _FakeClient
def test_ingest_crawled_writes_http_meta(http_meta_db):
url = "https://x.com/page"
responses = {
url: {
"status": 200,
"headers": {
"content-type": "text/html",
"etag": '"abc123"',
"last-modified": "Tue, 28 Apr 2026 00:00:00 GMT",
},
"body": "<html><body><h1>Title</h1>Hello world.</body></html>",
}
}
with patch("arborist.sources.crawler.bridge.httpx.Client", _patched_httpx_get(responses)):
conn = connect(http_meta_db)
try:
result = ingest_crawled(conn, [url])
finally:
conn.close()
assert result["http_meta_written"] == 1
assert len(result["documents"]) == 1
document_root = result["documents"][0]["document_root"]
conn = connect(http_meta_db)
try:
row = conn.execute(
"SELECT etag, last_modified, last_status, last_fetched_at, last_checked_at "
"FROM document_http_meta WHERE document_root = ?",
(document_root,),
).fetchone()
finally:
conn.close()
assert row["etag"] == '"abc123"'
assert row["last_modified"] == "Tue, 28 Apr 2026 00:00:00 GMT"
assert row["last_status"] == 200
assert row["last_fetched_at"] > 0
def test_ingest_crawled_skips_non_html(http_meta_db):
url = "https://x.com/file.pdf"
responses = {
url: {
"status": 200,
"headers": {"content-type": "application/pdf"},
"body": "%PDF-1.4 binary",
}
}
with patch("arborist.sources.crawler.bridge.httpx.Client", _patched_httpx_get(responses)):
conn = connect(http_meta_db)
try:
result = ingest_crawled(conn, [url])
finally:
conn.close()
assert result["http_meta_written"] == 0
assert result["inserted"] == 0
@pytest.mark.parametrize(
"feed_url",
[
"https://x.com/feeds/all.atom.xml",
"https://x.com/feed/",
"https://x.com/atom",
"https://x.com/rss",
"https://x.com/sitemap.xml",
"https://x.com/sitemaps/posts.xml",
"https://x.com/wp-rss2.xml",
"https://x.com/index.atom",
],
)
def test_ingest_crawled_skips_feed_urls_by_path(http_meta_db, feed_url):
"""Feed/sitemap URLs are crawl infrastructure, not knowledge. Skip
even before the fetch — pin so a future "but XML can be useful"
rewrite can't silently re-introduce feed pollution."""
# Response stub never gets used because the path filter is pre-fetch,
# but we provide one in case the filter regresses.
responses = {feed_url: {"status": 200, "headers": {"content-type": "text/html"}}}
with patch("arborist.sources.crawler.bridge.httpx.Client", _patched_httpx_get(responses)):
conn = connect(http_meta_db)
try:
result = ingest_crawled(conn, [feed_url])
finally:
conn.close()
assert result["http_meta_written"] == 0
assert result["inserted"] == 0
@pytest.mark.parametrize(
"ct",
[
"application/atom+xml",
"application/rss+xml",
"application/rdf+xml",
"application/xml",
"text/xml",
"application/atom+xml; charset=utf-8",
],
)
def test_ingest_crawled_skips_feed_content_types(http_meta_db, ct):
"""A feed served at a non-feed path still gets rejected by
Content-Type. xhtml is allowed (treated as HTML elsewhere)."""
url = "https://x.com/innocent-looking-path" # no feed-pattern in URL
responses = {
url: {
"status": 200,
"headers": {"content-type": ct},
"body": "<?xml version='1.0'?><feed></feed>",
}
}
with patch("arborist.sources.crawler.bridge.httpx.Client", _patched_httpx_get(responses)):
conn = connect(http_meta_db)
try:
result = ingest_crawled(conn, [url])
finally:
conn.close()
assert result["http_meta_written"] == 0
assert result["inserted"] == 0
def test_ingest_crawled_keeps_xhtml(http_meta_db):
"""xhtml IS HTML — keep it, even though the content type ends in +xml."""
url = "https://x.com/page"
responses = {
url: {
"status": 200,
"headers": {"content-type": "application/xhtml+xml"},
"body": "<html><body><h1>T</h1>Real content here.</body></html>",
}
}
with patch("arborist.sources.crawler.bridge.httpx.Client", _patched_httpx_get(responses)):
conn = connect(http_meta_db)
try:
result = ingest_crawled(conn, [url])
finally:
conn.close()
assert result["inserted"] == 1
assert result["http_meta_written"] == 1
# ---------------------------------------------------------------------------
# recrawl_check: conditional HEAD classification
# ---------------------------------------------------------------------------
def _seed_meta(conn, *, document_root: str, document_uri: str, etag: str | None = None):
"""Insert a documents row + http_meta row for recheck tests."""
from arborist.store import transaction
with transaction(conn):
conn.execute(
"INSERT INTO documents "
"(document_root, document_uri, source_type, kind, compression_depth, "
" title, chunking_version, canonicalization_version, schema_version, "
" ingest_ts, hit_count) "
"VALUES (?, ?, 'html', 'surface', 0, NULL, 'tok-512-v1', 'norm-v1', 'v9.8.0', ?, 0)",
(document_root, document_uri, int(time.time())),
)
conn.execute(
"INSERT INTO document_http_meta "
"(document_root, etag, last_modified, last_fetched_at, last_status, last_checked_at) "
"VALUES (?, ?, NULL, ?, 200, NULL)",
(document_root, etag, int(time.time())),
)
def test_recrawl_check_classifies_304_as_fresh(tmp_path):
db = tmp_path / "rc.db"
conn = connect(db)
try:
_seed_meta(conn, document_root="aa" * 32, document_uri="https://x.com/p", etag='"v1"')
finally:
conn.close()
responses = {"https://x.com/p": {"head_status": 304}}
with patch("arborist.sources.crawler.bridge.httpx.Client", _patched_httpx_get(responses)):
conn = connect(db)
try:
result = recrawl_check(conn, limit=10)
finally:
conn.close()
assert result["fresh"] == 1
assert result["stale"] == 0
assert result["items"][0]["verdict"] == "fresh"
def test_recrawl_check_classifies_200_as_stale(tmp_path):
db = tmp_path / "rc.db"
conn = connect(db)
try:
_seed_meta(conn, document_root="bb" * 32, document_uri="https://x.com/q", etag='"v1"')
finally:
conn.close()
responses = {"https://x.com/q": {"head_status": 200}}
with patch("arborist.sources.crawler.bridge.httpx.Client", _patched_httpx_get(responses)):
conn = connect(db)
try:
result = recrawl_check(conn, limit=10)
finally:
conn.close()
assert result["stale"] == 1
assert result["items"][0]["verdict"] == "stale"
def test_recrawl_check_classifies_404_as_gone(tmp_path):
db = tmp_path / "rc.db"
conn = connect(db)
try:
_seed_meta(conn, document_root="cc" * 32, document_uri="https://x.com/r", etag='"v1"')
finally:
conn.close()
responses = {"https://x.com/r": {"head_status": 404}}
with patch("arborist.sources.crawler.bridge.httpx.Client", _patched_httpx_get(responses)):
conn = connect(db)
try:
result = recrawl_check(conn, limit=10)
finally:
conn.close()
assert result["gone"] == 1
def test_recrawl_check_domain_filter(tmp_path):
db = tmp_path / "rc.db"
conn = connect(db)
try:
_seed_meta(conn, document_root="11" * 32, document_uri="https://keep.com/a", etag='"v"')
_seed_meta(conn, document_root="22" * 32, document_uri="https://drop.com/a", etag='"v"')
finally:
conn.close()
responses = {
"https://keep.com/a": {"head_status": 304},
# drop.com would error if the filter is broken; we expect it to be skipped.
}
with patch("arborist.sources.crawler.bridge.httpx.Client", _patched_httpx_get(responses)):
conn = connect(db)
try:
result = recrawl_check(conn, domain="keep.com", limit=10)
finally:
conn.close()
assert result["examined"] == 1
assert result["items"][0]["document_uri"] == "https://keep.com/a"
def test_recrawl_check_updates_last_checked_at(tmp_path):
db = tmp_path / "rc.db"
conn = connect(db)
try:
_seed_meta(conn, document_root="33" * 32, document_uri="https://x.com/s", etag='"v1"')
finally:
conn.close()
responses = {"https://x.com/s": {"head_status": 304}}
with patch("arborist.sources.crawler.bridge.httpx.Client", _patched_httpx_get(responses)):
conn = connect(db)
try:
recrawl_check(conn, limit=10)
finally:
conn.close()
conn = connect(db)
try:
row = conn.execute(
"SELECT last_status, last_checked_at FROM document_http_meta WHERE document_root='33' || '33' || '33' || '33' || '33' || '33' || '33' || '33' || '33' || '33' || '33' || '33' || '33' || '33' || '33' || '33' || '33' || '33' || '33' || '33' || '33' || '33' || '33' || '33' || '33' || '33' || '33' || '33' || '33' || '33' || '33' || '33'"
).fetchone()
finally:
conn.close()
assert row["last_status"] == 304
assert row["last_checked_at"] is not None
Reference in a new issue View git blame Copy permalink