SPB Git forge

spb/trawls

Public
3commits 1branches 0releases
456.0 KBsize
maindefault branch
18 days agolast push
Python 76.2% JavaScript 11.3% CSS 6.3% HTML 5.9%
5.3 KB · 130 lines python
Raw Blame History
1import pytest23from trawls.core.antibot.detect import detect4from trawls.core.scheduler.dedup import ContentDeduper, hamming, normalize_url, same_site, simhash5from trawls.core.security import SsrfError, check_url6from trawls.extract.css import extract_css7from trawls.models import ChunkOptions, CssField, ErrorCode8from trawls.processors.chunker import chunk9from trawls.processors.structured.metadata import extract_metadata101112def test_normalize_url() -> None:13    assert normalize_url("HTTP://Ex.COM:80/a/b/?utm_source=x&b=2&a=1#frag") == "http://ex.com/a/b?a=1&b=2"14    assert normalize_url("https://ex.com") == "https://ex.com/"15    assert normalize_url("/rel", "https://ex.com/dir/page") == "https://ex.com/rel"16    assert normalize_url("javascript:void(0)") is None17    assert normalize_url("mailto:a@b.c") is None18    assert normalize_url("https://ex.com/a?fbclid=1&gclid=2") == "https://ex.com/a"192021def test_same_site() -> None:22    assert same_site("https://ex.com/a", "https://ex.com/b", False)23    assert not same_site("https://ex.com/a", "https://blog.ex.com/b", False)24    assert same_site("https://ex.com/a", "https://blog.ex.com/b", True)25    assert not same_site("https://ex.com/a", "https://other.com/b", True)262728def test_simhash_dedup() -> None:29    a = "Le chalut est un filet remorqué par un navire. " * 2030    b = a.replace("navire", "bateau", 1)31    c = "Texte complètement différent sur les oiseaux migrateurs et leurs routes. " * 2032    assert hamming(simhash(a), simhash(b)) <= 333    assert hamming(simhash(a), simhash(c)) > 1034    d = ContentDeduper()35    assert d.is_duplicate(a, "u1") is None36    assert d.is_duplicate(b, "u2") == "u1"37    assert d.is_duplicate(c, "u3") is None383940@pytest.mark.asyncio41async def test_ssrf() -> None:42    for bad in (43        "http://127.0.0.1/",44        "http://localhost:8080/x",45        "http://169.254.169.254/latest",46        "http://10.0.0.1/",47        "http://192.168.2.1/",48        "ftp://ex.com/",49        "http://[::1]/",50        "http://0.0.0.0/",51    ):52        with pytest.raises(SsrfError):53            await check_url(bad)54    assert await check_url("https://1.1.1.1/") == "https://1.1.1.1/"555657def test_detect() -> None:58    assert (59        detect(60            "<html><title>Just a moment...</title><div id='cf-browser-verification'></div></html>", 50361        ).kind62        == "cloudflare"63    )64    assert (65        detect('<html><body><div id="root"></div><script src="/app.js"></script></body></html>', 200).kind66        == "spa"67    )68    ok = detect("<html><body><p>" + "Du contenu normal et long. " * 50 + "</p></body></html>", 200)69    assert ok.kind is None70    assert detect("<html><body><h1>Access Denied</h1></body></html>", 403).blocked717273def test_chunker() -> None:74    md = (75        "# Titre\n\nIntro courte.\n\n## Section A\n\n"76        + ("Phrase de la section A. " * 80)77        + "\n\n## Section B\n\nPetite.\n"78    )79    cs = chunk(80        md,81        "https://ex.com",82        ChunkOptions(strategy="by_heading", max_tokens=200, min_tokens=20, overlap_tokens=20),83    )84    assert len(cs) >= 385    assert all(c.token_count <= 260 for c in cs)86    assert any("Section A" in c.heading_path for c in cs)87    assert cs[0].index == 0 and cs[-1].index == len(cs) - 188    ct = chunk("Mot. " * 2000, "u", ChunkOptions(strategy="by_tokens", size_tokens=100, overlap_tokens=10))89    assert len(ct) > 10 and all(c.token_count <= 130 for c in ct)909192def test_css_extract() -> None:93    html = "<html><body><h1> Produit  X </h1><span class='price'>1 234,50 $</span><a class='more' href='/p/1'>+</a><ul><li>a</li><li>b</li></ul><time datetime='2026-09-05'>5 sept</time></body></html>"94    data, errs = extract_css(95        html,96        {97            "title": CssField(selector="h1"),98            "price": CssField(selector=".price", type="float"),99            "url": CssField(selector="a.more", attr="href", type="url"),100            "items": CssField(selector="li", multiple=True),101            "date": CssField(selector="time", attr="datetime", type="date"),102            "missing": CssField(selector=".nope", type="int"),103        },104        "https://ex.com/x",105    )106    assert data["title"] == "Produit X"107    assert data["price"] == 1234.5108    assert data["url"] == "https://ex.com/p/1"109    assert data["items"] == ["a", "b"]110    assert data["date"].startswith("2026-09-05")111    assert data["missing"] is None and "missing" in errs112113114def test_metadata() -> None:115    html = """<html lang="fr-CA"><head><title>T</title><meta property="og:title" content="OG T"><meta name="description" content="D">116    <link rel="canonical" href="/canon"><script type="application/ld+json">{"@type":"Article","headline":"H","author":{"name":"Simon"},"datePublished":"2026-01-02"}</script></head><body></body></html>"""117    m = extract_metadata(html, "https://ex.com/p")118    assert m.title == "OG T" and m.description == "D" and m.language == "fr"119    assert m.canonical_url == "https://ex.com/canon"120    assert m.author == "Simon" and m.published_at == "2026-01-02"121    assert m.jsonld and m.jsonld[0]["@type"] == "Article"122123124def test_error_codes_retryable() -> None:125    from trawls.models import ErrorInfo126127    assert ErrorInfo.make(ErrorCode.HTTP_5XX, "x").retryable128    assert not ErrorInfo.make(ErrorCode.BLOCKED, "x").retryable129    assert not ErrorInfo.make(ErrorCode.ROBOTS_DISALLOWED, "x").retryable130