devplacepy/tests/unit/services/jobs/deepsearch/crawl.py

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