import pytest
from trawls.core.antibot.detect import detect
from trawls.core.scheduler.dedup import ContentDeduper, hamming, normalize_url, same_site, simhash
from trawls.core.security import SsrfError, check_url
from trawls.extract.css import extract_css
from trawls.models import ChunkOptions, CssField, ErrorCode
from trawls.processors.chunker import chunk
from trawls.processors.structured.metadata import extract_metadata
def test_normalize_url() -> None:
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"
assert normalize_url("https://ex.com") == "https://ex.com/"
assert normalize_url("/rel", "https://ex.com/dir/page") == "https://ex.com/rel"
assert normalize_url("javascript:void(0)") is None
assert normalize_url("mailto:a@b.c") is None
assert normalize_url("https://ex.com/a?fbclid=1&gclid=2") == "https://ex.com/a"
def test_same_site() -> None:
assert same_site("https://ex.com/a", "https://ex.com/b", False)
assert not same_site("https://ex.com/a", "https://blog.ex.com/b", False)
assert same_site("https://ex.com/a", "https://blog.ex.com/b", True)
assert not same_site("https://ex.com/a", "https://other.com/b", True)
def test_simhash_dedup() -> None:
a = "Le chalut est un filet remorqué par un navire. " * 20
b = a.replace("navire", "bateau", 1)
c = "Texte complètement différent sur les oiseaux migrateurs et leurs routes. " * 20
assert hamming(simhash(a), simhash(b)) <= 3
assert hamming(simhash(a), simhash(c)) > 10
d = ContentDeduper()
assert d.is_duplicate(a, "u1") is None
assert d.is_duplicate(b, "u2") == "u1"
assert d.is_duplicate(c, "u3") is None
@pytest.mark.asyncio
async def test_ssrf() -> None:
for bad in (
"http://127.0.0.1/",
"http://localhost:8080/x",
"http://169.254.169.254/latest",
"http://10.0.0.1/",
"http://192.168.2.1/",
"ftp://ex.com/",
"http://[::1]/",
"http://0.0.0.0/",
):
with pytest.raises(SsrfError):
await check_url(bad)
assert await check_url("https://1.1.1.1/") == "https://1.1.1.1/"
def test_detect() -> None:
assert (
detect(
"Just a moment...", 503
).kind
== "cloudflare"
)
assert (
detect('', 200).kind
== "spa"
)
ok = detect("" + "Du contenu normal et long. " * 50 + "
", 200)
assert ok.kind is None
assert detect("Access Denied
", 403).blocked
def test_chunker() -> None:
md = (
"# Titre\n\nIntro courte.\n\n## Section A\n\n"
+ ("Phrase de la section A. " * 80)
+ "\n\n## Section B\n\nPetite.\n"
)
cs = chunk(
md,
"https://ex.com",
ChunkOptions(strategy="by_heading", max_tokens=200, min_tokens=20, overlap_tokens=20),
)
assert len(cs) >= 3
assert all(c.token_count <= 260 for c in cs)
assert any("Section A" in c.heading_path for c in cs)
assert cs[0].index == 0 and cs[-1].index == len(cs) - 1
ct = chunk("Mot. " * 2000, "u", ChunkOptions(strategy="by_tokens", size_tokens=100, overlap_tokens=10))
assert len(ct) > 10 and all(c.token_count <= 130 for c in ct)
def test_css_extract() -> None:
html = " Produit X
1 234,50 $+"
data, errs = extract_css(
html,
{
"title": CssField(selector="h1"),
"price": CssField(selector=".price", type="float"),
"url": CssField(selector="a.more", attr="href", type="url"),
"items": CssField(selector="li", multiple=True),
"date": CssField(selector="time", attr="datetime", type="date"),
"missing": CssField(selector=".nope", type="int"),
},
"https://ex.com/x",
)
assert data["title"] == "Produit X"
assert data["price"] == 1234.5
assert data["url"] == "https://ex.com/p/1"
assert data["items"] == ["a", "b"]
assert data["date"].startswith("2026-09-05")
assert data["missing"] is None and "missing" in errs
def test_metadata() -> None:
html = """T
"""
m = extract_metadata(html, "https://ex.com/p")
assert m.title == "OG T" and m.description == "D" and m.language == "fr"
assert m.canonical_url == "https://ex.com/canon"
assert m.author == "Simon" and m.published_at == "2026-01-02"
assert m.jsonld and m.jsonld[0]["@type"] == "Article"
def test_error_codes_retryable() -> None:
from trawls.models import ErrorInfo
assert ErrorInfo.make(ErrorCode.HTTP_5XX, "x").retryable
assert not ErrorInfo.make(ErrorCode.BLOCKED, "x").retryable
assert not ErrorInfo.make(ErrorCode.ROBOTS_DISALLOWED, "x").retryable