"""Web lookup service — scrape parsing, query building, and the SSRF guard. Nothing here touches the network: `_get` is monkeypatched, and the one test that needs a "public" hostname fakes the DNS answer. """ from __future__ import annotations import json import pytest from imptune.services import websearch # Captured before any monkeypatching so `TestFetchGuard` can put the real guard # back after the module-wide autouse fixture has stubbed it out. _REAL_ASSERT_FETCHABLE = websearch.assert_fetchable @pytest.fixture(autouse=True) def _allow_any_host(monkeypatch): """Neutralize the SSRF guard for parsing tests — it has its own tests below.""" monkeypatch.setattr(websearch, "assert_fetchable", lambda url: None) def _canned(monkeypatch, mapping: dict[str, bytes]): """Serve `_get` from a {url-substring: body} table.""" def fake_get(url, *, referer=None, max_bytes=0, extra_headers=None): for fragment, body in mapping.items(): if fragment in url: return body raise websearch.WebSearchError(f"unexpected url {url}") monkeypatch.setattr(websearch, "_get", fake_get) # -------------------------------------------------------------------------- # Image search # -------------------------------------------------------------------------- IMAGE_PAYLOAD = json.dumps( { "results": [ { "image": "https://cdn.example.com/m404.png", "thumbnail": "https://tn.example.com/m404.png", "title": "HP LaserJet M404", "width": 800, "height": 600, "url": "https://example.com/page", }, { "image": "data:image/png;base64,AAAA", "thumbnail": "https://tn.example.com/skip.png", "title": "inline", "width": 1, "height": 1, }, { "image": "https://cdn.example.com/second.jpg", "thumbnail": "", "title": "", "width": None, "height": None, }, ] } ).encode() def test_search_images_parses_results(monkeypatch): _canned( monkeypatch, {"duckduckgo.com/?q=": b'... vqd="4-12345678" ...', "i.js": IMAGE_PAYLOAD}, ) hits = websearch.search_images("hp laserjet m404") # The data: URI is dropped — only fetchable http(s) images survive. assert [h.url for h in hits] == [ "https://cdn.example.com/m404.png", "https://cdn.example.com/second.jpg", ] assert hits[0].width == 800 assert hits[0].source == "cdn.example.com" # A result with no thumbnail falls back to the full image. assert hits[1].thumbnail == "https://cdn.example.com/second.jpg" assert hits[1].width == 0 def test_search_images_honours_limit(monkeypatch): _canned( monkeypatch, {"duckduckgo.com/?q=": b'vqd="4-12345678"', "i.js": IMAGE_PAYLOAD}, ) assert len(websearch.search_images("x", limit=1)) == 1 def test_search_images_returns_empty_when_token_is_missing(monkeypatch): """A markup change upstream must degrade to "no results", not a 500.""" _canned(monkeypatch, {"duckduckgo.com/?q=": b"redesigned"}) assert websearch.search_images("anything") == [] def test_search_images_returns_empty_on_bad_json(monkeypatch): _canned( monkeypatch, {"duckduckgo.com/?q=": b'vqd="4-12345678"', "i.js": b""}, ) assert websearch.search_images("anything") == [] def test_search_images_skips_the_request_for_a_blank_query(monkeypatch): def explode(*args, **kwargs): raise AssertionError("should not hit the network") monkeypatch.setattr(websearch, "_get", explode) assert websearch.search_images(" ") == [] # -------------------------------------------------------------------------- # Page search # -------------------------------------------------------------------------- HTML_RESULTS = b"""
HP Universal Print Driver Download the UPD for Windows.
Direct link Second snippet
""" def test_search_pages_unwraps_the_redirect_and_strips_markup(monkeypatch): _canned(monkeypatch, {"html.duckduckgo.com": HTML_RESULTS}) hits = websearch.search_pages("hp upd") assert hits[0].url == "https://support.hp.com/upd" assert hits[0].title == "HP Universal Print Driver" assert hits[0].snippet == "Download the UPD for Windows." assert hits[0].host == "support.hp.com" assert hits[1].url == "https://direct.example.com/drivers" def test_search_pages_returns_empty_on_transport_failure(monkeypatch): def fail(*args, **kwargs): raise websearch.WebSearchError("boom") monkeypatch.setattr(websearch, "_get", fail) assert websearch.search_pages("hp upd") == [] # -------------------------------------------------------------------------- # SSRF guard # -------------------------------------------------------------------------- class TestFetchGuard: @pytest.fixture(autouse=True) def _real_guard(self, monkeypatch): monkeypatch.setattr(websearch, "assert_fetchable", _REAL_ASSERT_FETCHABLE) @pytest.mark.parametrize( "url", [ "file:///etc/passwd", "ftp://example.com/x", "gopher://example.com", ], ) def test_non_http_schemes_refused(self, url): with pytest.raises(websearch.WebSearchError): websearch.assert_fetchable(url) @pytest.mark.parametrize( "url", [ "http://127.0.0.1/admin", "http://localhost:8000/", "http://10.0.0.5/printer", "http://192.168.1.1/", "http://169.254.169.254/latest/meta-data/", "http://[::1]/", ], ) def test_private_targets_refused(self, url): """ImpTune sits on the printer LAN — this is the whole point of the guard.""" with pytest.raises(websearch.WebSearchError, match="private|resolve"): websearch.assert_fetchable(url) def test_public_target_allowed(self, monkeypatch): monkeypatch.setattr( websearch.socket, "getaddrinfo", lambda host, port: [(2, 1, 6, "", ("93.184.216.34", 0))], ) websearch.assert_fetchable("https://example.com/logo.png") def test_a_host_resolving_to_both_is_refused(self, monkeypatch): """One private answer in the set is enough to refuse the whole host.""" monkeypatch.setattr( websearch.socket, "getaddrinfo", lambda host, port: [ (2, 1, 6, "", ("93.184.216.34", 0)), (2, 1, 6, "", ("127.0.0.1", 0)), ], ) with pytest.raises(websearch.WebSearchError): websearch.assert_fetchable("https://rebind.example.com/x") # -------------------------------------------------------------------------- # Query builders and the kill switch # -------------------------------------------------------------------------- @pytest.mark.parametrize( ("typed", "expected"), [ ("HP LaserJet M404dn", "HP Universal Print Driver PCL6 download"), ("bizhub C300i (Konica Minolta)", "Konica Minolta Universal PCL Print Driver download"), ("Xerox VersaLink C405", "Xerox Global Print Driver download"), ], ) def test_generic_driver_query_uses_the_vendor_product_name(typed, expected): assert websearch.generic_driver_query(typed) == expected def test_generic_driver_query_falls_back_to_the_typed_model(): assert ( websearch.generic_driver_query("Acme 9000") == "Acme 9000 universal print driver download" ) def test_detect_brand_prefers_the_longest_match(): """"konica" and "konica minolta" both match — the specific one wins.""" assert websearch.detect_brand("Konica Minolta bizhub") == "konica minolta" assert websearch.detect_brand("Acme 9000") is None def test_image_query_appends_printer(): assert websearch.image_query("HP M404") == "HP M404 printer" assert websearch.image_query(" ") == "" def test_disabled_flag_blocks_every_lookup(monkeypatch): monkeypatch.setattr(websearch.cfg, "WEB_SEARCH", False) assert websearch.enabled() is False for call in ( lambda: websearch.search_images("x"), lambda: websearch.search_pages("x"), lambda: websearch.fetch_image("https://example.com/x.png"), ): with pytest.raises(websearch.WebSearchError, match="disabled"): call()