"""Mis-decoded bodies never become snapshots, entities or events (2026-09-12 Brotli incident)."""
from __future__ import annotations
import gzip
import brotli
from companyatlas.fetch import decode_text, looks_binary, text_quality
from companyatlas.sdk.models import ExtractedJob, ExtractedNewsItem, ExtractedPerson, Extraction
from companyatlas.services.pipeline import _drop_corrupt_entities
HTML = "Société Générale — résultatsÉconomie 日本語 Zürich".encode()
def test_binary_bodies_are_flagged() -> None:
assert looks_binary(brotli.compress(HTML), "text/html; charset=utf-8")
assert looks_binary(gzip.compress(HTML), "text/html")
assert not looks_binary(HTML, "text/html")
assert not looks_binary(b"%PDF-1.7 binary", "application/pdf") # non-textual types are left to their connectors
def test_decode_prefers_clean_codec() -> None:
assert text_quality(decode_text(HTML, "text/html")) == 1.0
assert "bénéfices" in decode_text("Groupe Lactalis – bénéfices élevés".encode("cp1252"), "text/html")
sjis = "トヨタ自動車".encode("shift_jis")
assert "トヨタ" in decode_text(sjis, "text/html")
garbage = brotli.compress(HTML).decode("utf-8", errors="replace")
assert text_quality(garbage) < 0.99
def test_corrupt_entities_are_dropped() -> None:
ex = Extraction(text="ok", blocks=[], title="News ��", jobs=[ExtractedJob(title="ML Engineer"), ExtractedJob(title="��7�c")],
people=[ExtractedPerson(name="Jane Doe"), ExtractedPerson(name="1234")],
news=[ExtractedNewsItem(title="Quarterly results", url="https://x.com/a"), ExtractedNewsItem(title="ng��t1]", url="https://x.com/b")])
_drop_corrupt_entities(ex)
assert [j.title for j in ex.jobs] == ["ML Engineer"]
assert [p.name for p in ex.people] == ["Jane Doe"]
assert [n.title for n in ex.news] == ["Quarterly results"]