backend/tests/test_websearch.py

"""No-network guards for the Tavily image search.

The feature is best-effort by contract: every failure path must return an
empty list rather than raise, because an image lookup must never be able to
break or delay an answer.
"""
import httpx
import pytest

from app import websearch
from app.config import settings
from app.websearch import _parse, search_images, search_subject

PAYLOAD = {
    "images": [
        {"url": "https://a.example/1.jpg", "description": "a photo"},
        {"url": "https://a.example/1.jpg", "description": "dupe"},
        {"url": "", "description": "no url"},
        {"url": "ftp://a.example/3.jpg", "description": "wrong scheme"},
        # plain http would be blocked as mixed content on the TLS deployment
        {"url": "http://a.example/5.jpg", "description": "insecure"},
        {"url": "https://a.example/2.jpg", "description": "x" * 500},
        "https://a.example/4.jpg",  # bare string (no image descriptions)
    ],
}


def test_parse_filters_and_dedupes():
    out = _parse(PAYLOAD, limit=10)
    assert [i.url for i in out] == [
        "https://a.example/1.jpg",
        "https://a.example/2.jpg",
        "https://a.example/4.jpg",
    ]
    assert len(out[1].description) == 200      # truncated
    assert out[2].description == ""            # bare string -> no description


def test_parse_respects_limit():
    assert len(_parse(PAYLOAD, limit=2)) == 2


def test_parse_tolerates_missing_key():
    assert _parse({}, limit=3) == []


@pytest.mark.parametrize("query,expected", [
    # scaffolding stripped
    ("show me photos of the new iPhone 17", "the new iPhone 17"),
    ("what does the newest iphone model look like", "the newest iphone model"),
    ("What do red pandas look like?", "red pandas"),
    ("pictures of mount fuji", "mount fuji"),
    # a leading article gets absorbed with the scaffolding; harmless for search
    ("can you show me the tesla roadster", "tesla roadster"),
    ("i wanna see pics of the eiffel tower", "the eiffel tower"),
    # already a clean subject — left alone
    ("who is the current prime minister of japan",
     "who is the current prime minister of japan"),
    ("best places to visit in kyoto right now",
     "best places to visit in kyoto right now"),
    # all scaffolding, no subject -> fall back to the original
    ("show me photos", "show me photos"),
])
def test_search_subject(query, expected):
    assert search_subject(query) == expected


@pytest.fixture
def enabled(monkeypatch):
    monkeypatch.setattr(settings, "tavily_api_key", "tvly-test")
    monkeypatch.setattr(settings, "image_search_enabled", True)


def _stub(monkeypatch, handler):
    """Point websearch's module client at a mock transport."""
    client = httpx.AsyncClient(transport=httpx.MockTransport(handler))
    monkeypatch.setattr(websearch, "_client", client)


@pytest.mark.asyncio
async def test_search_images_happy_path(monkeypatch, enabled):
    _stub(monkeypatch, lambda req: httpx.Response(200, json=PAYLOAD))
    out = await search_images("mount fuji", limit=3)
    assert len(out) == 3
    assert out[0].url == "https://a.example/1.jpg"


@pytest.mark.asyncio
async def test_search_images_swallows_http_error(monkeypatch, enabled):
    _stub(monkeypatch, lambda req: httpx.Response(500, text="boom"))
    assert await search_images("mount fuji") == []


@pytest.mark.asyncio
async def test_search_images_swallows_timeout(monkeypatch, enabled):
    def handler(req):
        raise httpx.ReadTimeout("too slow", request=req)

    _stub(monkeypatch, handler)
    assert await search_images("mount fuji") == []


@pytest.mark.asyncio
async def test_search_images_swallows_bad_json(monkeypatch, enabled):
    _stub(monkeypatch, lambda req: httpx.Response(200, text="not json"))
    assert await search_images("mount fuji") == []


@pytest.mark.asyncio
async def test_disabled_without_key(monkeypatch):
    monkeypatch.setattr(settings, "tavily_api_key", "")
    # No transport stub: a real call would fail the test by hanging/erroring
    assert await search_images("mount fuji") == []
    assert not websearch.enabled()


@pytest.mark.asyncio
async def test_empty_query_short_circuits(monkeypatch, enabled):
    assert await search_images("   ") == []