|
# 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):
|
|
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):
|
|
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):
|
|
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"
|