forked from retoor/devplacepy
retoor
535e9c5dc1
The degradation guard in crawl() wrapped only chromium.launch(), while
the driver start sat outside it in the async context manager. A failure
to start the driver therefore propagated out of crawl() and failed the
whole DeepSearch job, contradicting the warning it logs on that path
("pages will use httpx only").
Start the driver inside the guarded block and stop it in the finally, so
an unavailable driver degrades to httpx-only fetching as intended. The
crawl loop body is unchanged apart from indentation.
138 lines
4.7 KiB
Python
138 lines
4.7 KiB
Python
# retoor <retoor@molodetz.nl>
|
|
|
|
from tests.conftest import run_async
|
|
|
|
from devplacepy.services.jobs.deepsearch import crawl as crawl_module
|
|
from devplacepy.services.jobs.deepsearch.crawl import (
|
|
CrawledPage,
|
|
_interleave,
|
|
_is_hostile,
|
|
_snippet_page,
|
|
crawl,
|
|
)
|
|
|
|
|
|
async def _no_stop() -> bool:
|
|
return False
|
|
|
|
|
|
def test_interleave_round_robins_and_dedupes():
|
|
buckets = [
|
|
[{"url": "a"}, {"url": "b"}],
|
|
[{"url": "c"}, {"url": "a"}],
|
|
[{"url": "d"}],
|
|
]
|
|
assert [item["url"] for item in _interleave(buckets)] == ["a", "c", "d", "b"]
|
|
|
|
|
|
def test_is_hostile_matches_social_domains():
|
|
assert _is_hostile("https://x.com/user/status/1")
|
|
assert _is_hostile("https://www.youtube.com/watch?v=abc")
|
|
assert _is_hostile("https://old.reddit.com/r/x")
|
|
assert not _is_hostile("https://blog.example.com/post")
|
|
|
|
|
|
def test_snippet_page_prefers_content_over_description():
|
|
candidate = {
|
|
"url": "https://x.com/a/status/1",
|
|
"title": "Tweet",
|
|
"description": "short",
|
|
"content": "This is the full rsearch content, deliberately written long enough to clear the snippet minimum length floor so that it is kept as a real source easily.",
|
|
}
|
|
page = _snippet_page(candidate, 0)
|
|
assert page is not None
|
|
assert page.source == "search"
|
|
assert "full rsearch content" in page.text
|
|
|
|
|
|
def test_snippet_page_none_when_too_thin():
|
|
assert _snippet_page({"url": "https://x.com/a", "content": "tiny"}, 0) is None
|
|
|
|
|
|
def test_crawl_uses_snippet_for_hostile_and_skips_fetch(monkeypatch):
|
|
fetched = []
|
|
|
|
async def fake_fetch(url, depth, browser=None):
|
|
fetched.append(url)
|
|
return CrawledPage(url=url, title="T", text="x " * 200, source="httpx", status=200, depth=depth)
|
|
|
|
monkeypatch.setattr(crawl_module, "fetch_page", fake_fetch)
|
|
candidates = [
|
|
{
|
|
"url": "https://x.com/ThePrimeagen/status/1",
|
|
"title": "Prime",
|
|
"description": "",
|
|
"content": "The real tweet text returned by rsearch for this social post, well past the snippet minimum length threshold so it survives as a usable source here.",
|
|
}
|
|
]
|
|
outcome = run_async(
|
|
crawl(candidates, 6, lambda f: None, lambda u: False, _no_stop, query="q", depth=1)
|
|
)
|
|
assert fetched == []
|
|
assert len(outcome.pages) == 1
|
|
assert outcome.pages[0].source == "search"
|
|
|
|
|
|
def test_crawl_prefers_richer_crawl_over_snippet(monkeypatch):
|
|
async def fake_fetch(url, depth, browser=None):
|
|
return CrawledPage(url=url, title="Article", text="Deep article body. " * 60, source="httpx", status=200, depth=depth)
|
|
|
|
monkeypatch.setattr(crawl_module, "fetch_page", fake_fetch)
|
|
candidates = [
|
|
{
|
|
"url": "https://blog.example.com/a",
|
|
"title": "Blog",
|
|
"description": "",
|
|
"content": "A short snippet that is longer than the floor but shorter than the crawled article body itself.",
|
|
}
|
|
]
|
|
outcome = run_async(
|
|
crawl(candidates, 6, lambda f: None, lambda u: False, _no_stop, query="q", depth=1)
|
|
)
|
|
assert outcome.pages[0].source == "httpx"
|
|
|
|
|
|
def test_crawl_falls_back_to_snippet_when_fetch_empty(monkeypatch):
|
|
async def fake_fetch(url, depth, browser=None):
|
|
return None
|
|
|
|
monkeypatch.setattr(crawl_module, "fetch_page", fake_fetch)
|
|
candidates = [
|
|
{
|
|
"url": "https://walled.example.com/a",
|
|
"title": "Wall",
|
|
"description": "",
|
|
"content": "The search snippet holds the real content that the login-walled page refused to serve to the bot today, and it is comfortably past the snippet minimum length floor.",
|
|
}
|
|
]
|
|
outcome = run_async(
|
|
crawl(candidates, 6, lambda f: None, lambda u: False, _no_stop, query="q", depth=1)
|
|
)
|
|
assert len(outcome.pages) == 1
|
|
assert outcome.pages[0].source == "search"
|
|
|
|
|
|
def test_crawl_degrades_to_httpx_when_the_playwright_driver_cannot_start(monkeypatch):
|
|
import playwright.async_api as playwright_api
|
|
|
|
class FailingDriver:
|
|
async def start(self):
|
|
raise RuntimeError("driver missing")
|
|
|
|
monkeypatch.setattr(playwright_api, "async_playwright", lambda: FailingDriver())
|
|
|
|
async def fake_fetch(url, depth, browser=None):
|
|
return CrawledPage(
|
|
url=url, title="T", text="body " * 100, source="httpx", status=200, depth=depth
|
|
)
|
|
|
|
monkeypatch.setattr(crawl_module, "fetch_page", fake_fetch)
|
|
candidates = [
|
|
{"url": "https://example.com/a", "title": "A", "description": "", "content": ""}
|
|
]
|
|
outcome = run_async(
|
|
crawl(candidates, 6, lambda f: None, lambda u: False, _no_stop, query="q", depth=1)
|
|
)
|
|
assert len(outcome.pages) == 1
|
|
assert outcome.pages[0].source == "httpx"
|