"""Door B extraction registry: file bytes -> text, per file type (Phase 2 step 1). Core stdlib extractors (md/txt/csv/json/html) plus the fail-fast gates: unknown extension, the [extract]-gated binary types when the extra is absent, corrupt (non-UTF-8) bytes, and an empty CSV. All file-type->text extraction lives HERE (the guard is text-only); this step adds zero runtime dependency and no guard call — it is pure, deterministic plumbing. """ from __future__ import annotations import pytest from llm_ingestion_okf import ExtractionError, extract_text # --- core extractors: happy path (a fixture per type) --- def test_md_passthrough() -> None: assert extract_text("note.md", b"# Title\n\nBody\n") == "# Title\n\nBody\n" def test_txt_passthrough() -> None: assert extract_text("note.txt", b"plain text") == "plain text" def test_utf8_sig_bom_is_stripped() -> None: # utf-8-sig so a BOM never leaks into the first character (baseline parity # with Door A's read_csv). assert extract_text("note.txt", b"\xef\xbb\xbfhello") == "hello" def test_csv_renders_the_phase_1_markdown_table() -> None: out = extract_text("data.csv", b"name,age\nAda,36\n") assert out == "| name | age |\n| --- | --- |\n| Ada | 36 |\n" def test_json_is_verbatim_inside_a_fenced_block() -> None: out = extract_text("cfg.json", b'{"k": 1}\n') assert out == '```\n{"k": 1}\n```\n' def test_html_text_via_htmlparser() -> None: # Block boundaries separate words; tags themselves contribute no text. out = extract_text("page.html", b"

Title

Hello world

") assert out == "Title Hello world" def test_html_skips_script_and_style() -> None: html = b"

Keep

" assert extract_text("page.html", html) == "Keep" def test_htm_is_an_html_alias() -> None: assert extract_text("page.htm", b"

hi

") == "hi" def test_extension_dispatch_is_case_insensitive() -> None: assert extract_text("NOTE.MD", b"x") == "x" # --- fail-fast: unknown extension --- def test_unknown_extension_fails_fast() -> None: with pytest.raises(ExtractionError) as excinfo: extract_text("archive.zip", b"PK\x03\x04") assert excinfo.value.code == "extractor_unknown" def test_missing_extension_fails_fast() -> None: with pytest.raises(ExtractionError) as excinfo: extract_text("README", b"x") assert excinfo.value.code == "extractor_unknown" # --- fail-fast: [extract]-gated binary types without the extra --- @pytest.mark.parametrize("name", ["doc.pdf", "doc.docx", "sheet.xlsx"]) def test_optional_type_without_extra_fails_fast(name: str) -> None: with pytest.raises(ExtractionError) as excinfo: extract_text(name, b"binary") assert excinfo.value.code == "extractor_extra_missing" # The error names the extra so the operator knows the remedy — never a # silent skip, never a bundled parser in core. assert "extract" in str(excinfo.value) # --- fail-fast: corrupt (non-UTF-8) bytes on a text type --- def test_invalid_utf8_fails_fast_typed() -> None: # Never a leaked UnicodeDecodeError — the "always typed" doctrine holds for # Door B too (cf. Door A wrapping path ValueError as SourceError). with pytest.raises(ExtractionError) as excinfo: extract_text("note.txt", b"\xffbad") assert excinfo.value.code == "extractor_decode_error" # --- fail-fast: empty CSV (no header row) --- def test_empty_csv_fails_fast() -> None: with pytest.raises(ExtractionError) as excinfo: extract_text("empty.csv", b"") assert excinfo.value.code == "extractor_empty_csv"