spb/trouve-ka Public
Trouve-KA — moteur de recherche web indépendant, Québec-first. Crawler distribué, index OpenSearch, ranking bilingue, galerie d'images. En prod : www.trouve-ka.com
Python 76.8%
TypeScript 15.7%
SQL 3.9%
Shell 1.4%
CSS 1.3%
Dockerfile 0.7%
1# Trouve-KA — tests du fetcher (web de fixtures via MockTransport)2# Author: Simon-Pierre Boucher3# Contact: contact@spboucher.ai45import functools67import httpx8import pytest910import trouveka.crawler.fetcher as fetcher_mod11from trouveka.config import Settings12from trouveka.crawler.fetcher import Fetcher13from trouveka.shared import is_safe_url14from trouveka.types import ErrorCode151617@pytest.fixture(autouse=True)18def no_dns(monkeypatch):19 # Les hôtes de fixtures ne résolvent pas en vrai DNS : on garde toute la20 # logique SSRF (schémas, hôtes bloqués, IP littérales) sans résolution.21 monkeypatch.setattr(fetcher_mod, "is_safe_url", functools.partial(is_safe_url, resolve=False))222324def make_fetcher(handler, **overrides) -> Fetcher:25 settings = Settings(max_response_bytes=1000, max_redirects=3, fetch_timeout=5, **overrides)26 client = httpx.AsyncClient(transport=httpx.MockTransport(handler))27 return Fetcher(client, settings)282930async def test_simple_fetch():31 def handler(request):32 return httpx.Response(200, html="<html><title>ok</title></html>",33 headers={"content-type": "text/html; charset=utf-8"})3435 result = await make_fetcher(handler).fetch("https://site.qc.ca/page")36 assert result.status_code == 20037 assert result.error_code == ErrorCode.OK38 assert b"ok" in result.body394041async def test_redirect_followed_with_ssrf_recheck():42 def handler(request):43 if request.url.path == "/a":44 return httpx.Response(301, headers={"location": "/b"})45 return httpx.Response(200, html="fini", headers={"content-type": "text/html"})4647 result = await make_fetcher(handler).fetch("https://site.qc.ca/a")48 assert result.status_code == 20049 assert result.final_url == "https://site.qc.ca/b"50 assert result.redirect_chain == ["https://site.qc.ca/b"]515253async def test_redirect_to_private_ip_blocked():54 def handler(request):55 return httpx.Response(302, headers={"location": "http://169.254.169.254/latest/meta-data/"})5657 result = await make_fetcher(handler).fetch("https://site.qc.ca/evil")58 assert result.error_code == ErrorCode.SSRF_BLOCKED596061async def test_too_many_redirects():62 def handler(request):63 n = int(request.url.path.strip("/") or 0)64 return httpx.Response(301, headers={"location": f"/{n + 1}"})6566 result = await make_fetcher(handler).fetch("https://site.qc.ca/0")67 assert result.error_code == ErrorCode.TOO_MANY_REDIRECTS686970async def test_body_size_limit():71 def handler(request):72 return httpx.Response(200, content=b"x" * 5000, headers={"content-type": "text/html"})7374 result = await make_fetcher(handler).fetch("https://site.qc.ca/gros")75 assert result.error_code == ErrorCode.TOO_LARGE767778async def test_unsupported_content_type():79 def handler(request):80 return httpx.Response(200, content=b"%PDF-1.4", headers={"content-type": "application/pdf"})8182 result = await make_fetcher(handler).fetch("https://site.qc.ca/doc.pdf")83 assert result.error_code == ErrorCode.UNSUPPORTED_CONTENT848586async def test_conditional_304():87 def handler(request):88 assert request.headers.get("if-none-match") == 'W/"abc"'89 return httpx.Response(304)9091 result = await make_fetcher(handler).fetch("https://site.qc.ca/page", etag='W/"abc"')92 assert result.status_code == 304939495async def test_4xx_and_5xx_classified():96 async def run(status):97 def handler(request):98 return httpx.Response(status)99 return await make_fetcher(handler).fetch("https://site.qc.ca/x")100101 assert (await run(404)).error_code == ErrorCode.HTTP_4XX102 assert (await run(503)).error_code == ErrorCode.HTTP_5XX103104105async def test_ssrf_blocked_upfront():106 def handler(request): # pragma: no cover — ne doit jamais être appelé107 raise AssertionError("fetch ne doit pas partir")108109 result = await make_fetcher(handler).fetch("http://127.0.0.1:8080/admin")110 assert result.error_code == ErrorCode.SSRF_BLOCKED111