siftlode/backend/tests/test_search_scrape.py

"""youtube/search_scrape — the zero-quota InnerTube search (the module was 20% covered).

Everything here breaks silently when YouTube moves: a parser that stops finding videos, a cursor that
stops paging, a fallback key that outlives its usefulness. The network is an httpx MockTransport.
"""

import json

import httpx
import pytest

from app.youtube import search_scrape

RESULTS_HTML = (
    '<script>{"INNERTUBE_API_KEY":"live-key","INNERTUBE_CLIENT_VERSION":"2.20261001"}</script>'
)


def _video(vid: str, title: str, *, simple: bool = False) -> dict:
    return {
        "videoRenderer": {
            "videoId": vid,
            "title": {"simpleText": title}
            if simple
            else {"runs": [{"text": title[:3]}, {"text": title[3:]}]},
            "ownerText": {
                "runs": [
                    {
                        "text": "Chan",
                        "navigationEndpoint": {"browseEndpoint": {"browseId": "UCchan"}},
                    }
                ]
            },
            "thumbnail": {"thumbnails": [{"url": "small.jpg"}, {"url": "big.jpg"}]},
        }
    }


PAGE = {
    "contents": {
        "sectionListRenderer": {
            "contents": [
                {
                    "itemSectionRenderer": {
                        "contents": [
                            _video("v1", "First video"),
                            {"channelRenderer": {"channelId": "UCx"}},  # not a video: skipped
                            {
                                "reelShelfRenderer": {
                                    "items": [{"reelItemRenderer": {"videoId": "s1"}}]
                                }
                            },
                            _video("v2", "Second", simple=True),
                            _video("v1", "First video"),  # a repeat in another shelf: once
                            {"videoRenderer": {"title": {"simpleText": "no id"}}},
                        ]
                    }
                },
                {
                    "continuationItemRenderer": {
                        "continuationEndpoint": {"continuationCommand": {"token": "NEXT"}}
                    }
                },
            ]
        }
    }
}


@pytest.fixture
def net(monkeypatch):
    """Route every httpx.Client the module opens through a handler the test sets."""
    calls: list[httpx.Request] = []
    state: dict = {"handler": None}

    def transport(request: httpx.Request) -> httpx.Response:
        calls.append(request)
        return state["handler"](request)

    real = httpx.Client
    monkeypatch.setattr(
        search_scrape.httpx,
        "Client",
        lambda **kw: real(transport=httpx.MockTransport(transport), **kw),
    )
    monkeypatch.setattr(search_scrape, "_config", None)
    state["calls"] = calls
    return state


def _serve(results_html=RESULTS_HTML, search=PAGE, status=200):
    def handler(request: httpx.Request) -> httpx.Response:
        if request.url.path == "/results":
            if isinstance(results_html, Exception):
                raise results_html
            return httpx.Response(200, text=results_html)
        if isinstance(search, str):
            return httpx.Response(status, text=search)
        return httpx.Response(status, json=search)

    return handler


def test_a_page_flattens_its_videos_in_order_and_hands_back_the_cursor(net):
    net["handler"] = _serve()
    page = search_scrape.search("cats")
    assert page["next_page_token"] == "NEXT"
    assert [i["id"] for i in page["items"]] == ["v1", "v2"]
    assert page["items"][0] == {
        "id": "v1",
        "channel_id": "UCchan",
        "channel_title": "Chan",
        "title": "First video",
        "published_at": None,
        "thumbnail_url": "big.jpg",
        "live_broadcast": "none",
    }
    assert page["items"][1]["title"] == "Second"
    post = net["calls"][-1]
    assert post.url.params["key"] == "live-key"
    body = json.loads(post.content)
    assert body["query"] == "cats" and body["params"] == search_scrape._TYPE_VIDEO
    assert body["context"]["client"]["clientVersion"] == "2.20261001"


def test_the_next_page_sends_the_cursor_not_the_query_and_the_end_has_none(net):
    net["handler"] = _serve(search={"onResponseReceivedCommands": [_video("v9", "Last")]})
    page = search_scrape.search("cats", continuation="NEXT")
    body = json.loads(net["calls"][-1].content)
    assert body["continuation"] == "NEXT" and "query" not in body
    assert [i["id"] for i in page["items"]] == ["v9"]
    assert page["next_page_token"] is None


@pytest.mark.parametrize(
    ("search", "status"),
    [
        ("<html>consent</html>", 200),
        ({"error": "blocked"}, 429),
        ({"responseContext": {}, "alerts": ["unusual traffic"]}, 200),
        ([], 200),
    ],
    ids=["not-json", "http-error", "unknown-shape", "not-an-object"],
)
def test_a_blocked_or_garbled_answer_is_an_error_not_an_empty_page(net, search, status):
    net["handler"] = _serve(search=search, status=status)
    with pytest.raises(search_scrape.ScrapeError):
        search_scrape.search("cats")


def test_a_search_with_no_results_is_an_empty_page_not_an_error(net):
    net["handler"] = _serve(search={"contents": {"sectionListRenderer": {"contents": []}}})
    assert search_scrape.search("zzzzqqq") == {"items": [], "next_page_token": None}


def test_the_live_key_is_fetched_once_and_cached(net):
    net["handler"] = _serve()
    search_scrape.search("a")
    search_scrape.search("b")
    assert [c.url.path for c in net["calls"]].count("/results") == 1


@pytest.mark.parametrize(
    "results_html",
    [httpx.ConnectError("down"), "<html>a consent page, no keys</html>"],
    ids=["fetch-failed", "no-key-on-page"],
)
def test_a_failed_key_fetch_falls_back_for_now_and_tries_again_next_time(net, results_html):
    """The fallback pair used to be CACHED for the life of the process: one failed fetch pinned
    every later search to a hard-coded key, and once that went stale search returned nothing,
    with no error anywhere."""
    net["handler"] = _serve(results_html=results_html)
    search_scrape.search("a")
    assert net["calls"][-1].url.params["key"] == search_scrape._FALLBACK_KEY

    net["handler"] = _serve()
    search_scrape.search("b")
    assert net["calls"][-1].url.params["key"] == "live-key"