"""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 importlib.util import sys from pathlib import Path import pytest from llm_ingestion_okf import ExtractionError, ExtractionWarning, extract_text FIXTURES = Path(__file__).parent / "fixtures" # The parser-dependent tests below need the optional extra, so they are skipped # without it — an optional extra that forced its parser on every test run would # not be optional. The behaviour that must survive EITHER WAY is the rejection, # and that one is asserted unconditionally above via the import probe. requires_extract = pytest.mark.skipif( importlib.util.find_spec("pdfplumber") is None, reason="the optional [extract] extra is not installed", ) # --- 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"
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 the extra ships no parser for --- @pytest.mark.parametrize("name", ["doc.docx", "sheet.xlsx"]) def test_optional_type_without_a_parser_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) def test_pdf_without_the_extra_installed_keeps_the_same_rejection( monkeypatch: pytest.MonkeyPatch, ) -> None: """The gate became an import probe; the rejection did not change. Setting the module to None in sys.modules is what CPython treats as a failed import, so this exercises the uninstalled path REGARDLESS of whether the extra is installed in the running environment — a skip would have preserved nothing on the machine where the parser is present. """ monkeypatch.setitem(sys.modules, "pdfplumber", None) with pytest.raises(ExtractionError) as excinfo: extract_text("doc.pdf", b"%PDF-1.4") assert excinfo.value.code == "extractor_extra_missing" assert "extract" in str(excinfo.value) # --- pdf: the [extract] parser (pdfplumber) --- # The expected text is frozen against a committed fixture ON PURPOSE. Extraction # is deterministic within a parser version and NOT guaranteed across one # (pdfplumber pins pdfminer.six==20260107 exactly; pdfminer.six ships # date-stamped releases with no stability contract). This literal is what makes # a parser upgrade break something visible instead of drifting silently. KRAV_TEXT = "Krav til helning på utkilingen\n60 og 70 1:15" @requires_extract def test_pdf_extracts_text_with_label_and_value_on_one_line() -> None: data = (FIXTURES / "two-line-krav.pdf").read_bytes() with pytest.warns(ExtractionWarning): assert extract_text("krav.pdf", data) == KRAV_TEXT @requires_extract def test_pdf_extraction_is_byte_stable_across_calls() -> None: """Determinism is a promise this library already makes; hold it here too.""" data = (FIXTURES / "two-line-krav.pdf").read_bytes() with pytest.warns(ExtractionWarning): first = extract_text("krav.pdf", data) second = extract_text("krav.pdf", data) assert first == second @requires_extract def test_pdf_extraction_warns_that_drawn_content_is_not_recovered() -> None: """Figures are vector drawings: only the caption survives leg 2. Categorically true of text extraction, so it is stated as a warning on every PDF rather than guessed at per document — detecting "is there a figure here" would be exactly the layout heuristic this order declined. """ data = (FIXTURES / "two-line-krav.pdf").read_bytes() with pytest.warns(ExtractionWarning, match="figures"): extract_text("krav.pdf", data) @requires_extract def test_pdf_with_no_text_layer_fails_fast() -> None: """A scanned PDF yields nothing; an empty concept would be a silent skip.""" data = (FIXTURES / "no-text-layer.pdf").read_bytes() with pytest.raises(ExtractionError) as excinfo: extract_text("scan.pdf", data) assert excinfo.value.code == "extractor_empty_pdf" @requires_extract def test_corrupt_pdf_fails_fast_typed() -> None: """Never a leaked pdfminer exception — the "always typed" doctrine holds.""" with pytest.raises(ExtractionError) as excinfo: extract_text("broken.pdf", b"not a pdf at all") assert excinfo.value.code == "extractor_pdf_error" # --- 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"