" for number in range(12)])
+ )
+ body = _paragraph("".join(_image(number=number) for number in range(12)))
+
+ result = converter.convert(_docx(body), _INFO)
+
+ assert result.markdown == "\n\n".join(f"Image {number}" for number in range(12))
+
+
+def test_returned_image_and_literal_text_use_normal_html_rendering() -> None:
+ render = Mock(
+ return_value=(
+ "
" for number in range(12)])
- )
- body = _paragraph("".join(_image(number=number) for number in range(12)))
-
- result = converter.convert(_docx(body), _INFO)
-
- assert result.markdown == "\n\n".join(f"Image {number}" for number in range(12))
-
-
-def test_returned_image_and_literal_text_use_normal_html_rendering() -> None:
- render = Mock(
- return_value=(
- "
+
+"""
+ path = tmp_path / "strike.html"
+ path.write_text(html, encoding="utf-8")
+ markdown = MarkItDown().convert(str(path)).markdown
+
+ assert markdown == "\n\n".join(
+ [
+ # , and the obsolete all mean strikethrough
+ "Plain ~~s element~~ after.",
+ "Plain ~~del element~~ after.",
+ "Plain ~~strike element~~ after.",
+ # Surrounding whitespace stays outside of the markup ...
+ "Spaces A ~~B~~ C.",
+ # ... and runs of it collapse to a single space
+ "Runs D ~~E~~ F.",
+ # An empty element contributes nothing
+ "Empty GH.",
+ # A line break inside the element is kept, and the markup
+ # survives it because strikethrough may span a single newline
+ "Newline I~~J\nK~~L.",
+ "Break M~~N\nO~~P.",
+ ]
+ )
+
+
+def test_deeply_nested_html_fallback() -> None:
+ """Large, deeply nested HTML should fall back to plain-text extraction
+ instead of silently returning unconverted HTML (issue #1636).
+
+ Note: This test uses sys.setrecursionlimit to guarantee a RecursionError
+ regardless of the host environment's default limit, making it deterministic
+ across different platforms and CI configurations.
+ """
+ import sys
+ import warnings
+
+ markitdown = MarkItDown()
+
+ # Use a small recursion limit so the test is environment-independent.
+ # We restore the original limit in a finally block to avoid side-effects.
+ original_limit = sys.getrecursionlimit()
+ low_limit = 200 # well below markdownify's traversal depth for depth=500
+
+ # Build HTML with nesting deep enough to trigger RecursionError
+ depth = 500
+ html = ""
+ for _ in range(depth):
+ html += '
'
+ html += "
Deep content with bold text
"
+ for _ in range(depth):
+ html += "
"
+ html += ""
+
+ try:
+ sys.setrecursionlimit(low_limit)
+ with warnings.catch_warnings(record=True) as w:
+ warnings.simplefilter("always")
+ result = markitdown.convert_stream(
+ io.BytesIO(html.encode("utf-8")),
+ file_extension=".html",
+ )
+
+ # Should have emitted a warning about the fallback
+ recursion_warnings = [x for x in w if "deeply nested" in str(x.message)]
+ assert len(recursion_warnings) > 0
+
+ finally:
+ sys.setrecursionlimit(original_limit)
+
+ # The output should contain the text content, not raw HTML
+ assert "Deep content" in result.markdown
+ assert "bold text" in result.markdown
+ assert "
-
-"""
- path = tmp_path / "strike.html"
- path.write_text(html, encoding="utf-8")
- markdown = MarkItDown().convert(str(path)).markdown
-
- assert markdown == "\n\n".join(
- [
- # , and the obsolete all mean strikethrough
- "Plain ~~s element~~ after.",
- "Plain ~~del element~~ after.",
- "Plain ~~strike element~~ after.",
- # Surrounding whitespace stays outside of the markup ...
- "Spaces A ~~B~~ C.",
- # ... and runs of it collapse to a single space
- "Runs D ~~E~~ F.",
- # An empty element contributes nothing
- "Empty GH.",
- # A line break inside the element is kept, and the markup
- # survives it because strikethrough may span a single newline
- "Newline I~~J\nK~~L.",
- "Break M~~N\nO~~P.",
- ]
- )
-
-
-def test_docx_equations() -> None:
- markitdown = MarkItDown()
- docx_file = os.path.join(TEST_FILES_DIR, "equations.docx")
- result = markitdown.convert(docx_file)
-
- # Check for inline equation m=1 (wrapped with single $) is present
- assert "$m=1$" in result.text_content, "Inline equation $m=1$ not found"
-
- # Find block equations wrapped with double $$ and check if they are present
- block_equations = re.findall(r"\$\$(.+?)\$\$", result.text_content)
- assert block_equations, "No block equations found in the document."
-
-
-def test_docx_zip_filename_casing_mismatch() -> None:
- """Test that DOCX files with inconsistent ZIP filename casing are handled.
-
- Some document generators produce .docx files where the central directory
- records one casing (e.g. 'word/document.xml') but the local file headers
- record another (e.g. 'Word/Document.XML'). Python's zipfile module raises
- BadZipFile when reading such files. This test verifies that MarkItDown
- handles this gracefully.
-
- See: https://github.com/microsoft/markitdown/issues/1812
- """
- import struct
-
- markitdown = MarkItDown()
- docx_file = os.path.join(TEST_FILES_DIR, "test.docx")
-
- # Read the original docx and get its expected content
- original_result = markitdown.convert(docx_file)
- assert original_result.markdown.strip(), "Original DOCX should have content"
-
- # Read raw bytes and corrupt the local file header filenames
- with open(docx_file, "rb") as f:
- raw = bytearray(f.read())
-
- # Find all local file headers and uppercase their filenames
- corrupted = bytearray(raw)
- offset = 0
- patched_count = 0
- while offset + 30 <= len(corrupted):
- if corrupted[offset : offset + 4] != b"PK\x03\x04":
- break
- fname_len = struct.unpack_from(" len(corrupted):
- break
- # Uppercase the filename in the local header
- old_name = corrupted[offset + 30 : offset + 30 + fname_len]
- new_name = old_name.upper()
- if old_name != new_name:
- corrupted[offset + 30 : offset + 30 + fname_len] = new_name
- patched_count += 1
- comp_size = struct.unpack_from(" 0, "Should have patched at least one local file header"
-
- # Verify the corrupted file would fail with plain zipfile
- with pytest.raises(zipfile.BadZipFile):
- with zipfile.ZipFile(io.BytesIO(bytes(corrupted)), "r") as zf:
- for name in zf.namelist():
- zf.read(name)
-
- # Verify MarkItDown can still convert it
- corrupted_result = markitdown.convert_stream(
- io.BytesIO(bytes(corrupted)),
- file_extension=".docx",
- )
- assert (
- corrupted_result.markdown.strip()
- ), "Corrupted DOCX should still produce content"
- # Content should be equivalent to the original
- assert (
- original_result.markdown.strip() == corrupted_result.markdown.strip()
- ), "Corrupted DOCX should produce the same output as original"
-
-
-def test_docx_zip_filename_non_casing_mismatch_still_rejected() -> None:
- """Only case-only local/central filename disagreements are repaired.
-
- Equal encoded length does not imply two names differ only in case, so the
- repair must not be used to wave through archives whose local and central
- directory names genuinely disagree -- zipfile rejects those for good
- reason, and honouring that keeps the workaround scoped to issue #1812.
- """
- import struct
-
- from markitdown.converter_utils.docx.pre_process import (
- _fix_zip_filename_casing,
- )
-
- def build_zip(first_name: str) -> bytes:
- buf = io.BytesIO()
- with zipfile.ZipFile(buf, "w") as zf:
- zf.writestr(first_name, b"payload-a")
- zf.writestr("word/document.xml", b"payload-b")
- return buf.getvalue()
-
- def replace_first_local_name(data: bytes, new_name: bytes) -> bytes:
- """Rewrite only the first local file header's name, leaving the central
- directory -- the authoritative copy -- untouched."""
- raw = bytearray(data)
- fname_len = struct.unpack_from(" bool:
- try:
- with zipfile.ZipFile(io.BytesIO(data), "r") as zf:
- for name in zf.namelist():
- zf.read(name)
- except zipfile.BadZipFile:
- return False
- return True
-
- original = "[Content_Types].xml"
-
- # A case-only mismatch is repaired ...
- case_only = replace_first_local_name(build_zip(original), original.upper().encode())
- assert not reads_cleanly(case_only)
- repaired = _fix_zip_filename_casing(io.BytesIO(case_only))
- assert reads_cleanly(repaired.read())
-
- # ... but an equal-length mismatch that is not case-only is left alone,
- # so zipfile still refuses the archive.
- for mutated_name in (
- b"ZBnoudou^Uxqdr\\/ylm", # unrelated, same length
- b"[Content_Types].xmy", # a single differing character
- b"[content_types]/xml", # differs only outside the cased characters
- ):
- data = replace_first_local_name(build_zip(original), mutated_name)
- assert not reads_cleanly(data), f"{mutated_name!r} should not be readable"
- result = _fix_zip_filename_casing(io.BytesIO(data))
- assert not reads_cleanly(
- result.read()
- ), f"{mutated_name!r} must not be silently repaired"
-
-
-def test_docx_malformed_equations() -> None:
- """Malformed equations should not crash the converter (issue #1979)."""
- import zipfile
- from io import BytesIO
-
- docx_xml = """
-
-
-
- Normal text
-
-
-
- x+1
-
-
-
- After good equation
-
-
-
-
-
- After empty oMathPara
-
-
-
-
-
- After empty inline oMath
-
-
-
- y+2
-
-
- After unnamespaced oMathPara
-
- z+3
-
- After unnamespaced oMath
-
-
-"""
-
- buf = BytesIO()
- with zipfile.ZipFile(buf, "w") as z:
- z.writestr("word/document.xml", docx_xml.encode("utf-8"))
- z.writestr(
- "[Content_Types].xml",
- """
-
-
-
-
-""",
- )
- z.writestr(
- "word/_rels/document.xml.rels",
- """
-
-""",
- )
- z.writestr(
- "_rels/.rels",
- """
-
-
-""",
- )
-
- buf.seek(0)
- markitdown = MarkItDown()
- result = markitdown.convert(buf)
- assert "Normal text" in result.markdown
- # A malformed equation must not abort the math pre-processing step, which
- # would silently drop the LaTeX conversion of every other equation too
- assert "$x+1$" in result.markdown or "$$x+1$$" in result.markdown
- assert "After empty oMathPara" in result.markdown
- assert "After empty inline oMath" in result.markdown
- assert "After unnamespaced oMathPara" in result.markdown
- assert "After unnamespaced oMath" in result.markdown
-
-
-def test_xlsx_legacy_show_zeroes_sheetview(tmp_path) -> None:
- from openpyxl import Workbook
-
- base_path = tmp_path / "base.xlsx"
- xlsx_path = tmp_path / "legacy_show_zeroes.xlsx"
-
- workbook = Workbook()
- sheet = workbook.active
- sheet.title = "Data"
- sheet["A1"] = "hello"
- sheet["B1"] = "world"
- # A cell whose text is byte-identical to the malformed attribute. openpyxl stores it
- # as an inline string, so it lands in the very worksheet part that gets repaired.
- sheet["A2"] = ' showZeroes="0" '
- workbook.save(base_path)
-
- with zipfile.ZipFile(base_path) as source:
- with zipfile.ZipFile(xlsx_path, "w", zipfile.ZIP_DEFLATED) as target:
- for item in source.infolist():
- data = source.read(item.filename)
- if item.filename == "xl/worksheets/sheet1.xml":
- data = data.replace(
- b" None:
- from markitdown.converters._xlsx_converter import _rename_show_zeroes_attribute
-
- worksheet_xml = (
- b""
- b''
- b''
- b""
- # openpyxl ignores custom sheet views entirely, so they need no repair
- b''
- b''
- b' showZeroes="0" '
- b""
- b'IF(A1=" showZeroes=","x","y")y'
- b""
- )
-
- repaired = _rename_show_zeroes_attribute(worksheet_xml)
-
- # Every start tag is repaired, whatever separates its attributes ...
- assert repaired.count(b"showZeros=") == 2
- assert b'' in repaired
- assert b'' in repaired
-
- # ... and nothing outside those start tags is rewritten.
- assert b' showZeroes="0" ' in repaired
- assert b'IF(A1=" showZeroes=","x","y")' in repaired
- assert b'' in repaired
-
-
def test_input_as_strings() -> None:
markitdown = MarkItDown()
@@ -895,263 +322,10 @@ def test_get_content_disposition_filename_decodes_rfc5987() -> None:
)
-def test_pptx_chart_multi_series_conversion() -> None:
- """Charts with multiple series and many categories must convert correctly.
-
- Regression test for the slow path in PptxConverter._convert_chart_to_markdown,
- where ``series.values[idx]`` was evaluated inside the (category x series) loop.
- In python-pptx each ``series.values`` access rescans the cached points via
- XPath (O(n) per lookup), so the old code was O(n^2) per series (and rebuilt
- the whole tuple for every category), making large charts extremely slow.
-
- The values are now materialized once per series. This test builds a chart
- with enough categories that the regressed code path would be pathologically
- slow, and verifies the resulting Markdown table is correct across series.
- """
- pptx = pytest.importorskip("pptx")
- from pptx.util import Inches
- from pptx.chart.data import CategoryChartData
- from pptx.enum.chart import XL_CHART_TYPE
-
- n_categories = 200
- categories = [f"C{i}" for i in range(n_categories)]
- series_a = [float(i) for i in range(n_categories)]
- series_b = [float(i * 2) for i in range(n_categories)]
-
- presentation = pptx.Presentation()
- slide = presentation.slides.add_slide(presentation.slide_layouts[5])
- chart_data = CategoryChartData()
- chart_data.categories = categories
- chart_data.add_series("Series A", series_a)
- chart_data.add_series("Series B", series_b)
- slide.shapes.add_chart(
- XL_CHART_TYPE.COLUMN_CLUSTERED,
- Inches(1),
- Inches(1),
- Inches(8),
- Inches(5),
- chart_data,
- )
-
- buffer = io.BytesIO()
- presentation.save(buffer)
- buffer.seek(0)
-
- result = MarkItDown().convert_stream(buffer, file_extension=".pptx")
- md = result.markdown
-
- # Both series headers are present
- assert "Series A" in md
- assert "Series B" in md
- # First and last categories are present (nothing truncated)
- assert "| C0 |" in md
- assert f"| C{n_categories - 1} |" in md
- # A representative row carries the correct value for each series
- assert "| C10 | 10.0 | 20.0 |" in md
-
-
-def test_deeply_nested_html_fallback() -> None:
- """Large, deeply nested HTML should fall back to plain-text extraction
- instead of silently returning unconverted HTML (issue #1636).
-
- Note: This test uses sys.setrecursionlimit to guarantee a RecursionError
- regardless of the host environment's default limit, making it deterministic
- across different platforms and CI configurations.
- """
- import sys
- import warnings
-
- markitdown = MarkItDown()
-
- # Use a small recursion limit so the test is environment-independent.
- # We restore the original limit in a finally block to avoid side-effects.
- original_limit = sys.getrecursionlimit()
- low_limit = 200 # well below markdownify's traversal depth for depth=500
-
- # Build HTML with nesting deep enough to trigger RecursionError
- depth = 500
- html = ""
- for _ in range(depth):
- html += '
'
- html += "
Deep content with bold text
"
- for _ in range(depth):
- html += "
"
- html += ""
-
- try:
- sys.setrecursionlimit(low_limit)
- with warnings.catch_warnings(record=True) as w:
- warnings.simplefilter("always")
- result = markitdown.convert_stream(
- io.BytesIO(html.encode("utf-8")),
- file_extension=".html",
- )
-
- # Should have emitted a warning about the fallback
- recursion_warnings = [x for x in w if "deeply nested" in str(x.message)]
- assert len(recursion_warnings) > 0
-
- finally:
- sys.setrecursionlimit(original_limit)
-
- # The output should contain the text content, not raw HTML
- assert "Deep content" in result.markdown
- assert "bold text" in result.markdown
- assert "
" not in result.markdown
-
-
-def test_deeply_nested_rss_item_fallback() -> None:
- """Deeply nested HTML inside an RSS item should fall back to plain-text
- extraction instead of silently embedding raw unconverted HTML in the
- markdown output (same failure class as the HTML converter fix in #1644).
-
- Note: This test uses sys.setrecursionlimit to guarantee a RecursionError
- regardless of the host environment's default limit, making it deterministic
- across different platforms and CI configurations.
- """
- import sys
- import warnings
-
- markitdown = MarkItDown()
-
- # Use a small recursion limit so the test is environment-independent.
- # We restore the original limit in a finally block to avoid side-effects.
- original_limit = sys.getrecursionlimit()
- low_limit = 200 # well below markdownify's traversal depth for depth=500
-
- # Build an RSS item whose content is deeply nested HTML
- depth = 500
- item_html = ""
- for _ in range(depth):
- item_html += '
'
- item_html += "
Deep feed content with bold text
"
- for _ in range(depth):
- item_html += "
"
-
- rss = (
- ''
- ''
- ""
- "Test Feed"
- "A test feed"
- ""
- "Deep Item"
- f""
- ""
- ""
- ""
- )
- atom = (
- ''
- ''
- "Test Feed"
- ""
- "Deep Entry"
- f''
- ""
- ""
- )
-
- try:
- sys.setrecursionlimit(low_limit)
- with warnings.catch_warnings(record=True) as w:
- warnings.simplefilter("always")
- result = markitdown.convert_stream(
- io.BytesIO(rss.encode("utf-8")),
- file_extension=".rss",
- )
-
- # Should have emitted a warning about the fallback
- recursion_warnings = [x for x in w if "deeply nested" in str(x.message)]
- assert len(recursion_warnings) > 0
-
- # strict=True should expose the conversion failure rather than applying
- # the plain-text fallback.
- with pytest.raises(RecursionError):
- RssConverter().convert(
- io.BytesIO(rss.encode("utf-8")),
- StreamInfo(extension=".rss"),
- strict=True,
- )
- with pytest.raises(RecursionError):
- RssConverter().convert(
- io.BytesIO(atom.encode("utf-8")),
- StreamInfo(extension=".atom"),
- strict=True,
- )
- finally:
- sys.setrecursionlimit(original_limit)
-
- # The output should contain the text content, not raw HTML
- assert "Deep feed content" in result.markdown
- assert "bold text" in result.markdown
- assert "
" not in result.markdown
-
-
-@pytest.mark.skipif(
- os.name == "nt",
- reason="The DOCX fixture embeds a POSIX file:///tmp/test_rlink.txt target.",
-)
-def test_doc_rlink() -> None:
- # Test for: CVE-2025-11849
- markitdown = MarkItDown()
-
- # Document with rlink
- docx_file = os.path.join(TEST_FILES_DIR, "rlink.docx")
-
- # Directory containing the target rlink file
- rlink_tmp_dir = os.path.abspath(os.sep + "tmp")
-
- # Ensure the tmp directory exists
- if not os.path.exists(rlink_tmp_dir):
- pytest.skip(f"Skipping rlink test; {rlink_tmp_dir} directory does not exist.")
- return
-
- rlink_file_path = os.path.join(rlink_tmp_dir, "test_rlink.txt")
- rlink_content = "de658225-569e-4e3d-9ed2-cfb6abf927fc"
- b64_prefix = (
- "ZGU2NTgyMjUtNTY5ZS00ZTNkLTllZDItY2ZiNmFiZjk" # base64 prefix of rlink_content
- )
-
- if os.path.exists(rlink_file_path):
- with open(rlink_file_path, "r", encoding="utf-8") as f:
- existing_content = f.read()
- if existing_content != rlink_content:
- raise ValueError(
- f"Existing {rlink_file_path} content does not match expected content."
- )
- else:
- with open(rlink_file_path, "w", encoding="utf-8") as f:
- f.write(rlink_content)
-
- try:
- result = markitdown.convert(docx_file, keep_data_uris=True).text_content
- assert (
- b64_prefix not in result
- ) # Make sure the target file was NOT embedded in the output
- finally:
- os.remove(rlink_file_path)
-
-
-@pytest.mark.skipif(
- skip_remote,
- reason="do not run tests that query external urls",
-)
-def test_markitdown_remote() -> None:
- markitdown = MarkItDown()
-
- # By URL
- result = markitdown.convert(PDF_TEST_URL)
- for test_string in PDF_TEST_STRINGS:
- assert test_string in result.text_content
-
- # Youtube
- # result = markitdown.convert(YOUTUBE_TEST_URL)
- # for test_string in YOUTUBE_TEST_STRINGS:
- # assert test_string in result.text_content
+# Youtube
+# result = markitdown.convert(YOUTUBE_TEST_URL)
+# for test_string in YOUTUBE_TEST_STRINGS:
+# assert test_string in result.text_content
@pytest.mark.skipif(
@@ -1189,61 +363,6 @@ def test_exceptions() -> None:
assert type(exc_info.value.attempts[0].converter).__name__ == "PptxConverter"
-def test_pptx_converter_treats_none_llm_caption_as_empty(monkeypatch) -> None:
- from markitdown.converters import _pptx_converter
-
- calls = 0
-
- def none_caption(*args, **kwargs):
- nonlocal calls
- calls += 1
- return None
-
- monkeypatch.setattr(_pptx_converter, "llm_caption", none_caption)
-
- result = MarkItDown().convert(
- os.path.join(TEST_FILES_DIR, "test.pptx"),
- llm_client=MagicMock(),
- llm_model="test-model",
- )
-
- assert calls > 0
- assert (
- ""
- in result.markdown
- )
-
-
-@pytest.mark.skipif(
- skip_exiftool,
- reason="do not run if exiftool is not installed",
-)
-def test_markitdown_exiftool() -> None:
- which_exiftool = shutil.which("exiftool")
- assert which_exiftool is not None
-
- # Test explicitly setting the location of exiftool
- markitdown = MarkItDown(exiftool_path=which_exiftool)
- result = markitdown.convert(os.path.join(TEST_FILES_DIR, "test.jpg"))
- for key in JPG_TEST_EXIFTOOL:
- target = f"{key}: {JPG_TEST_EXIFTOOL[key]}"
- assert target in result.text_content
-
- # Test setting the exiftool path through an environment variable
- os.environ["EXIFTOOL_PATH"] = which_exiftool
- markitdown = MarkItDown()
- result = markitdown.convert(os.path.join(TEST_FILES_DIR, "test.jpg"))
- for key in JPG_TEST_EXIFTOOL:
- target = f"{key}: {JPG_TEST_EXIFTOOL[key]}"
- assert target in result.text_content
-
- # Test some other media types
- result = markitdown.convert(os.path.join(TEST_FILES_DIR, "test.mp3"))
- for key in MP3_TEST_EXIFTOOL:
- target = f"{key}: {MP3_TEST_EXIFTOOL[key]}"
- assert target in result.text_content
-
-
def test_markitdown_llm_parameters() -> None:
"""Test that LLM parameters are correctly passed to the client."""
mock_client = MagicMock()
@@ -1314,137 +433,6 @@ def test_markitdown_llm() -> None:
validate_strings(result, PPTX_TEST_STRINGS)
-def test_pptx_chart_no_title_text_frame() -> None:
- from markitdown.converters._pptx_converter import PptxConverter
-
- # python-pptx's ChartTitle.text_frame is destructive -- it creates a text
- # frame if one isn't already present, so it never returns None.
- # has_text_frame is the property that actually reflects presence/absence,
- # which is what the has_title=True-but-no-text-frame case looks like
- # against the real library.
- mock_chart = MagicMock()
- mock_chart.has_title = True
- mock_chart.chart_title.has_text_frame = False
-
- mock_category = MagicMock()
- mock_category.label = "Cat 1"
- mock_chart.plots = [MagicMock(categories=[mock_category])]
-
- mock_series = MagicMock()
- mock_series.name = "Series 1"
- mock_series.values = [10.0]
- mock_chart.series = [mock_series]
-
- converter = PptxConverter()
- result = converter._convert_chart_to_markdown(mock_chart)
-
- assert "### Chart" in result
- assert "Cat 1" in result
- assert "Series 1" in result
- assert ":" not in result
-
-
-def test_pptx_chart_with_title_text_frame() -> None:
- from markitdown.converters._pptx_converter import PptxConverter
-
- mock_chart = MagicMock()
- mock_chart.has_title = True
- mock_chart.chart_title.has_text_frame = True
- mock_chart.chart_title.text_frame.text = "Revenue"
-
- mock_category = MagicMock()
- mock_category.label = "Cat 1"
- mock_chart.plots = [MagicMock(categories=[mock_category])]
-
- mock_series = MagicMock()
- mock_series.name = "Series 1"
- mock_series.values = [10.0]
- mock_chart.series = [mock_series]
-
- converter = PptxConverter()
- result = converter._convert_chart_to_markdown(mock_chart)
-
- assert "### Chart: Revenue" in result
-
-
-def test_youtube_converter_missing_title_metadata() -> None:
- """Missing titles fall back to HTML when no video content is extracted."""
- from unittest.mock import patch
- from markitdown.converters._youtube_converter import YouTubeConverter
-
- converter = YouTubeConverter()
- stream_info = StreamInfo(
- mimetype="text/html",
- extension=".html",
- url="https://www.youtube.com/watch?v=12345",
- )
-
- with patch(
- "markitdown.converters._youtube_converter.IS_YOUTUBE_TRANSCRIPT_CAPABLE",
- False,
- ):
- # Case 1: Stream with no title metadata or title tag
- html_content_no_title = b"Video Content"
- stream_no_title = io.BytesIO(html_content_no_title)
- result_no_title = converter.convert(stream_no_title, stream_info)
- assert result_no_title.title is None
- assert result_no_title.markdown == "Video Content"
-
- # Case 2: Stream with an empty tag
- html_content_empty_title = (
- b"Video Content"
- )
- stream_empty_title = io.BytesIO(html_content_empty_title)
- result_empty_title = converter.convert(stream_empty_title, stream_info)
- assert result_empty_title.title is None
- assert result_empty_title.markdown == "Video Content"
-
- # Case 3: Stream whose title is only available from the tag
- html_content_title_tag = b"Fallback TitleVideo Content"
- stream_title_tag = io.BytesIO(html_content_title_tag)
- result_title_tag = converter.convert(stream_title_tag, stream_info)
- assert result_title_tag.title == "Fallback Title"
- assert result_title_tag.markdown == "# YouTube\n\n## Fallback Title\n"
-
-
-def test_zip_duplicate_filenames_preserve_each_entry() -> None:
- """Same-named ZIP entries must retain their own content and archive order."""
- buf = io.BytesIO()
- with zipfile.ZipFile(buf, "w") as archive:
- archive.writestr("notes.txt", "First archived entry.")
- with pytest.warns(UserWarning, match="Duplicate name"):
- archive.writestr("notes.txt", "Second archived entry.")
- buf.seek(0)
-
- result = MarkItDown().convert_stream(
- buf, stream_info=StreamInfo(extension=".zip", filename="duplicate.zip")
- )
-
- assert result.markdown == (
- "Content from the zip file `duplicate.zip`:\n\n"
- "## File: notes.txt\n\nFirst archived entry.\n\n"
- "## File: notes.txt\n\nSecond archived entry."
- )
-
-
-def test_zip_stream_no_filename_header() -> None:
- """Regression test: ZipConverter must not render the literal string 'None'
- in the output header when the stream has no associated URL, local path, or
- filename (e.g. when called via convert_stream() without stream_info)."""
- markitdown = MarkItDown()
-
- buf = io.BytesIO()
- with zipfile.ZipFile(buf, "w") as zf:
- zf.writestr("hello.txt", "Hello world")
- buf.seek(0)
-
- result = markitdown.convert_stream(
- buf, stream_info=StreamInfo(mimetype="application/zip")
- )
- assert result.markdown.startswith("Content from the zip file `(unknown)`:\n\n")
- assert "Hello world" in result.markdown
-
-
def test_ipynb_heading_title_preserves_leading_hash() -> None:
"""Heading marker removal must not eat a leading '#' from the title text.
@@ -1563,367 +551,320 @@ def test_ipynb_accepts_non_ascii() -> None:
assert result is True
-def test_epub_metadata_nodevalue():
- from defusedxml.minidom import parseString
- from markitdown.converters._epub_converter import EpubConverter
-
- xml_data = (
- ''
- "Structured Title"
- "Author 1"
- "Author 2"
- ""
- ""
- ""
- )
- dom = parseString(xml_data)
- converter = EpubConverter()
+# File URI validation
+
+UNC_URI_PATHS = [
+ "//server.example/share/x.txt",
+ "///server.example/share/x.txt",
+ "////server.example/share/x.txt",
+ r"/\server.example/share/x.txt",
+ r"\/server.example/share/x.txt",
+ r"\\server.example\share\x.txt",
+ "/%2Fserver.example/share/x.txt",
+ "/%2fserver.example/share/x.txt",
+ "/%5Cserver.example/share/x.txt",
+ "/%5cserver.example/share/x.txt",
+ "%2F%2Fserver.example/share/x.txt",
+ "%5C%5Cserver.example%5Cshare%5Cx.txt",
+ "/%5C%3F%5CUNC%5Cserver.example%5Cshare%5Cx.txt",
+ "/%5C.%5CUNC%5Cserver.example%5Cshare%5Cx.txt",
+ "/%5C%3F%5CC:%5CTemp%5Cx.txt",
+ "/%5C.%5CPhysicalDrive0",
+]
- title = converter._get_text_from_node(dom, "dc:title")
- assert title == "Structured Title"
+BLOCKED_URIS = (
+ [
+ f"file://{authority}{path}"
+ for authority in ("", "localhost")
+ for path in UNC_URI_PATHS
+ # An authority must be followed by a slash to introduce its path.
+ if path.startswith("/")
+ ]
+ + [f"file:{path}" for path in UNC_URI_PATHS if not path.startswith("/")]
+ + [
+ "FILE:////server.example/share/x.txt",
+ "file:////C:/Temp/x.txt",
+ ]
+)
- creators = converter._get_all_texts_from_nodes(dom, "dc:creator")
- assert creators == ["Author 1", "Author 2"]
- publisher = converter._get_text_from_node(dom, "dc:publisher")
- assert publisher is None
+@pytest.mark.parametrize("uri", BLOCKED_URIS)
+def test_file_uri_rejects_unc_or_device_path(uri: str) -> None:
+ with pytest.raises(ValueError, match="UNC"):
+ file_uri_to_path(uri)
- missing = converter._get_text_from_node(dom, "dc:date")
- assert missing is None
+class _NoFileAccessMarkItDown(MarkItDown):
+ def convert_local(self, *args, **kwargs):
+ pytest.fail("Rejected file URIs must not reach convert_local()")
-_EPUB_CONTAINER = (
- ''
- ''
- ''
-)
-_EPUB_OPF = (
- ''
- ''
- ''
- "Example book"
- ''
- ''
-)
+@pytest.fixture(scope="module")
+def no_file_access_markitdown() -> MarkItDown:
+ return _NoFileAccessMarkItDown(enable_builtins=False)
-_EPUB_CHAPTER = (
- ''
- "
Chapter text.
"
- ''
- ""
-)
+@pytest.mark.parametrize("method", ["convert", "convert_uri"])
+@pytest.mark.parametrize("uri", BLOCKED_URIS + ["file://server.example/share/x.txt"])
+def test_convert_rejects_remote_file_uri_before_file_access(
+ no_file_access_markitdown: MarkItDown, method: str, uri: str
+) -> None:
+ with pytest.raises(ValueError, match="Unsupported file URI"):
+ getattr(no_file_access_markitdown, method)(uri)
-def _build_epub() -> bytes:
- buf = io.BytesIO()
- with zipfile.ZipFile(buf, "w") as zf:
- zf.writestr("mimetype", "application/epub+zip")
- zf.writestr("META-INF/container.xml", _EPUB_CONTAINER)
- zf.writestr("OEBPS/content.opf", _EPUB_OPF)
- zf.writestr("OEBPS/ch1.xhtml", _EPUB_CHAPTER)
- return buf.getvalue()
+@pytest.mark.parametrize(
+ "path, blocked",
+ [
+ (r"\\server\share\x.txt", True),
+ ("//server/share/x.txt", True),
+ (r"/\server/share/x.txt", True),
+ (r"\/server/share/x.txt", True),
+ (r"\\?\UNC\server\share\x.txt", True),
+ (r"\\?\C:\Temp\x.txt", True),
+ (r"\\.\PhysicalDrive0", True),
+ (r"C:\Temp\x.txt", False),
+ ("C:/Temp/x.txt", False),
+ ("/home/user/x.txt", False),
+ ("relative.txt", False),
+ ],
+)
+def test_resolved_path_validation(path: str, blocked: bool) -> None:
+ assert _is_unc_or_device_path(path) is blocked
-def test_epub_honors_keep_data_uris() -> None:
- """EPUB chapters must be converted with the options the caller passed."""
- result = MarkItDown().convert_stream(
- io.BytesIO(_build_epub()),
- stream_info=StreamInfo(extension=".epub"),
- keep_data_uris=True,
- )
- assert (
- "" in result.markdown
+def test_file_uri_rejects_unc_after_conversion(monkeypatch: pytest.MonkeyPatch) -> None:
+ # Check the final guard independently of the URI prefix check. Replace only
+ # the helper's converter binding, leaving the stdlib functions unchanged.
+ monkeypatch.setattr(
+ uri_utils, "url2pathname", lambda _: "//server.example/share/x.txt"
)
+ with pytest.raises(ValueError, match="UNC"):
+ file_uri_to_path("file:///local.txt")
-def test_epub_truncates_data_uris_by_default() -> None:
- """Without the option, the default truncation must still apply."""
- result = MarkItDown().convert_stream(
- io.BytesIO(_build_epub()), stream_info=StreamInfo(extension=".epub")
- )
- assert "" in result.markdown
- assert "iVBORw0KGgo" not in result.markdown
+@pytest.mark.parametrize("filename", ["notes.txt", "notes café.txt", "%5C%5Cx.txt"])
+@pytest.mark.parametrize("authority", ["", "localhost"])
+def test_local_file_uri_conversion(
+ tmp_path: Path, filename: str, authority: str
+) -> None:
+ local_file = tmp_path / filename
+ local_file.write_text("Local file contents", encoding="utf-8")
+ uri = local_file.as_uri().replace("file://", f"file://{authority}", 1)
+ netloc, path = file_uri_to_path(uri)
-def test_epub_metadata_and_text_are_unchanged() -> None:
- """The rest of the conversion must not move."""
- result = MarkItDown().convert_stream(
- io.BytesIO(_build_epub()), stream_info=StreamInfo(extension=".epub")
- )
+ assert netloc == (authority or None)
+ assert path == str(local_file)
+ markitdown = MarkItDown()
+ assert markitdown.convert(uri).markdown == "Local file contents"
+ assert markitdown.convert(str(local_file)).markdown == "Local file contents"
- assert result.title == "Example book"
- assert "**Title:** Example book" in result.markdown
- assert "Chapter text." in result.markdown
+@pytest.mark.skipif(os.name != "nt", reason="Requires native Windows path conversion")
+@pytest.mark.parametrize("authority", ["", "localhost"])
+@pytest.mark.parametrize(
+ "uri_path, expected",
+ [
+ ("/C:/Temp/notes.txt", r"C:\Temp\notes.txt"),
+ ("/C%3A/Temp/notes.txt", r"C:\Temp\notes.txt"),
+ ("/C%3a/Temp/notes.txt", r"C:\Temp\notes.txt"),
+ ("/C:/Temp/notes%20caf%C3%A9.txt", "C:\\Temp\\notes café.txt"),
+ ("/C:/Temp/%255C%255Cx.txt", r"C:\Temp\%5C%5Cx.txt"),
+ ],
+)
+def test_windows_local_drive_uri(authority: str, uri_path: str, expected: str) -> None:
+ netloc, path = file_uri_to_path(f"file://{authority}{uri_path}")
-def test_json_with_late_non_ascii_character(tmp_path) -> None:
- payload = {
- "record": {
- "title": "Example record",
- "abstract": "This is sample test. " * 500,
- },
- "notes": "non-ASCII character: è",
- }
- json_path = tmp_path / "input.json"
- json_path.write_text(
- json.dumps(payload, ensure_ascii=False, indent=2), encoding="utf-8"
- )
+ assert netloc == (authority or None)
+ assert path == expected
- result = MarkItDown().convert(str(json_path))
- assert "non-ASCII character: è" in result.text_content
+# Charset sampling
+_SAMPLE_SIZE = 65536
-###############################################################################
-# CSV converter.
-###############################################################################
+_SPLIT_CHARACTERS = [
+ (character, split)
+ for character in ("\u00e9", "\u65e5", "\U0001f600")
+ for split in range(1, len(character.encode("utf-8")))
+]
-def _convert_csv(data: bytes, charset: str | None = None) -> str:
- stream_info = StreamInfo(extension=".csv", charset=charset)
+def _split_utf8_json(character: str, split: int) -> bytes:
+ prefix = '{"name":"r\u00e9sum\u00e9","notes":"'.encode("utf-8")
return (
- MarkItDown()
- .convert_stream(io.BytesIO(data), stream_info=stream_info)
- .text_content
+ prefix
+ + b"a" * (_SAMPLE_SIZE - split - len(prefix))
+ + (character + '"}').encode("utf-8")
)
-def test_csv_utf8_bom_is_stripped_from_the_header() -> None:
- result = _convert_csv(b"\xef\xbb\xbfname,age\nAlice,30\n")
-
- assert "\ufeff" not in result
- assert result.startswith("| name | age |")
- assert "| Alice | 30 |" in result
-
-
-def test_csv_utf8_bom_is_stripped_when_charset_is_known() -> None:
- result = _convert_csv(b"\xef\xbb\xbfname,age\nAlice,30\n", charset="utf-8")
-
- assert "\ufeff" not in result
- assert result.startswith("| name | age |")
-
-
-def test_csv_leading_blank_line_does_not_destroy_the_table() -> None:
- result = _convert_csv(b"\nname,age\nAlice,30\n")
-
- assert result == "| name | age |\n| --- | --- |\n| Alice | 30 |"
+@pytest.fixture(scope="module")
+def markitdown() -> MarkItDown:
+ return MarkItDown()
-def test_csv_trailing_blank_lines_are_skipped() -> None:
- result = _convert_csv(b"name,age\nAlice,30\n\n\n")
+@pytest.mark.parametrize("character,split", _SPLIT_CHARACTERS)
+def test_charset_sample_completes_only_the_split_character(
+ character: str, split: int
+) -> None:
+ data = _split_utf8_json(character, split)
+ stream = io.BytesIO(data)
+ expected_size = _SAMPLE_SIZE + len(character.encode("utf-8")) - split
- assert result == "| name | age |\n| --- | --- |\n| Alice | 30 |"
+ sample = _read_charset_sample(stream)
+ assert sample == data[:expected_size]
+ assert stream.tell() == expected_size
+ assert expected_size <= _SAMPLE_SIZE + 3
+ sample.decode("utf-8")
-def test_csv_blank_lines_between_rows_are_kept() -> None:
- result = _convert_csv(b"name,age\n\nAlice,30\n\nBob,40\n")
- assert (
- result == "| name | age |\n| --- | --- |\n| Alice | 30 |\n| | |\n| Bob | 40 |"
- )
+@pytest.mark.parametrize("character,split", _SPLIT_CHARACTERS)
+def test_split_utf8_json_preserves_content(
+ markitdown: MarkItDown, character: str, split: int
+) -> None:
+ data = _split_utf8_json(character, split)
+ stream = io.BytesIO(data)
+ guesses = markitdown._get_stream_info_guesses(stream, StreamInfo())
+ assert guesses[0].charset == "utf-8"
+ assert stream.tell() == 0
-def test_csv_all_blank_input_returns_empty_markdown() -> None:
- result = _convert_csv(b"\n\n\n")
+ result = markitdown.convert_stream(stream)
- assert result == ""
+ assert result.markdown == data.decode("utf-8")
@pytest.mark.parametrize(
- "data,expected",
+ "sample,tail",
[
- (
- b"banner\nname,age,city\nAlice,30,Seattle\n",
- "| banner | | |\n| --- | --- | --- |\n"
- "| name | age | city |\n| Alice | 30 | Seattle |",
- ),
- (
- b"name,age\nAlice\n\nBob,40,Seattle\nCarol,25\n",
- "| name | age | |\n| --- | --- | --- |\n"
- "| Alice | | |\n| | | |\n"
- "| Bob | 40 | Seattle |\n| Carol | 25 | |",
- ),
- (
- b"name\nAlice,,\n",
- "| name | | |\n| --- | --- | --- |\n| Alice | | |",
- ),
- (
- b"name,age,city\nAlice,30\nBob\n",
- "| name | age | city |\n| --- | --- | --- |\n"
- "| Alice | 30 | |\n| Bob | | |",
- ),
+ (b"", b""),
+ (b"ordinary text", b""),
+ (b"short incomplete \xc3", b""),
+ (b"a" * _SAMPLE_SIZE, b"\xc3\xa9"),
+ (b"a" * (_SAMPLE_SIZE - 2) + b"\xc3\xa9", b"\xf0\x9f\x98\x80"),
+ (b"a" * (_SAMPLE_SIZE - 1) + b"\xc3", b""),
+ (b"a" * (_SAMPLE_SIZE - 1) + b"\xf0", b"\x9f"),
+ (b"a" * (_SAMPLE_SIZE - 1) + b"\xc3", b"x"),
+ (b"a" * (_SAMPLE_SIZE - 1) + b"\xe0", b"\x80\x80"),
+ (b"a" * (_SAMPLE_SIZE - 1) + b"\xed", b"\xa0\x80"),
+ (b"\xff" + b"a" * (_SAMPLE_SIZE - 2) + b"\xc3", b"\xa9"),
+ ],
+ ids=[
+ "empty",
+ "short",
+ "short-incomplete",
+ "ascii",
+ "complete-utf8",
+ "incomplete-at-eof",
+ "incomplete-after-lookahead",
+ "invalid-continuation",
+ "overlong",
+ "surrogate",
+ "non-utf8-prefix",
],
- ids=["preamble", "widest-row-late", "trailing-empty-fields", "widest-header"],
)
-def test_csv_table_matches_widest_row(data: bytes, expected: str) -> None:
- assert _convert_csv(data) == expected
-
-
-def test_csv_pipe_in_cell_is_escaped() -> None:
- result = _convert_csv(b'name,description\nWidget,"cheap | fast"\n')
-
- # The pipe must be escaped rather than emitted raw, or it splits the row.
- assert "| Widget | cheap \\| fast |" in result
-
-
-def test_csv_pipe_preceded_by_backslash_is_still_escaped() -> None:
- # `left\|right` must not become `left\\|right`: the doubled backslash is a
- # literal backslash, which would leave the pipe acting as a delimiter and
- # let a renderer drop `right`. The run of backslashes is doubled instead,
- # so the escaping `\|` survives.
- result = _convert_csv(b'name,description\nWidget,"left\\|right"\n')
-
- assert r"| Widget | left\\\|right |" in result
-
-
-def test_csv_pipe_preceded_by_two_backslashes_is_still_escaped() -> None:
- # An even-length run is just as dangerous once the naive `\|` is appended,
- # so check a longer run as well.
- result = _convert_csv(b'name,description\nWidget,"left\\\\|right"\n')
-
- assert r"| Widget | left\\\\\|right |" in result
-
-
-def test_csv_newline_in_quoted_cell_does_not_split_the_row() -> None:
- result = _convert_csv(b'name,notes\nWidget,"line one\nline two"\n')
-
- # The table must stay on one line per record; the embedded break collapses
- # to a space.
- assert len(result.splitlines()) == 3
- assert "| Widget | line one line two |" in result
-
-
-def test_csv_carriage_returns_in_quoted_cell_collapse_to_spaces() -> None:
- result = _convert_csv(b'name,notes\nWidget,"line one\r\nline two\rline three"\n')
+def test_other_charset_samples_are_unchanged(sample: bytes, tail: bytes) -> None:
+ stream = io.BytesIO(sample + tail)
- assert len(result.splitlines()) == 3
- assert "| Widget | line one line two line three |" in result
+ assert _read_charset_sample(stream) == sample
+ assert stream.tell() <= _SAMPLE_SIZE + 3
-def test_csv_pipe_in_header_is_escaped() -> None:
- result = _convert_csv(b'"a | b",c\n1,2\n')
+def test_charset_guesses_restore_nonzero_stream_position(
+ markitdown: MarkItDown,
+) -> None:
+ stream = io.BytesIO(b"prefix" + _split_utf8_json("\U0001f600", 1))
+ stream.seek(len(b"prefix"))
- assert "| a \\| b | c |" in result
+ markitdown._get_stream_info_guesses(stream, StreamInfo())
+ assert stream.tell() == len(b"prefix")
-def test_csv_plain_values_are_unchanged() -> None:
- # Guards against over-escaping ordinary content.
- result = _convert_csv(b"name,description\nWidget,cheap and fast\n")
- assert "| Widget | cheap and fast |" in result
- assert "\\" not in result
+def test_explicit_charset_still_takes_precedence(markitdown: MarkItDown) -> None:
+ data = _split_utf8_json("\u00e9", 1)
+ result = markitdown.convert_stream(
+ io.BytesIO(data),
+ stream_info=StreamInfo(extension=".json", charset="cp1252"),
+ )
-def test_csv_backslash_without_a_pipe_is_left_alone() -> None:
- # Only backslashes that guard a pipe are doubled; a Windows path stays
- # readable.
- result = _convert_csv(b"name,path\nWidget,C:\\temp\\file.txt\n")
+ assert result.markdown == data.decode("cp1252")
- assert r"| Widget | C:\temp\file.txt |" in result
+# Undetectable charsets
-@pytest.mark.parametrize(
- "suffix,escaped_suffix",
- [("", ""), ("|", r"\|"), ("x|", r"x\|")],
-)
-def test_csv_long_backslash_runs(suffix: str, escaped_suffix: str) -> None:
- backslashes = "\\" * 65_536
- value = backslashes + suffix
- expected = backslashes * (2 if suffix == "|" else 1) + escaped_suffix
+# Tests for bytes no charset decodes.
+#
+# ``charset_normalizer.from_bytes(data).best()`` returns ``None`` when nothing
+# decodes the bytes, and ``str(None)`` is the four-character string ``"None"``,
+# so a binary file handed over with a text extension became a document whose
+# entire content was the word "None" -- four characters that were never in the
+# file.
- result = _convert_csv(f"{value}\n{value}\n".encode("utf-8"), charset="utf-8")
+# Random bytes: no charset claims them, which is what makes `best()` answer None.
+# Fixed rather than generated, so the test does not depend on chance.
+UNDECODABLE = bytes((7 * i * i + 251 * i + 193) % 256 for i in range(4096))
- assert result == f"| {expected} |\n| --- |\n| {expected} |"
+def test_the_fixture_really_is_undecodable() -> None:
+ """Guards the guard: if some charset claimed these bytes the rest would pass
+ for the wrong reason."""
+ assert from_bytes(UNDECODABLE).best() is None
-# ---------------------------------------------------------------------------
-# Regression test for issue #1960:
-# exiftool_path pointing to a nonexistent binary used to leak a raw
-# FileNotFoundError. It should now be wrapped in RuntimeError with a
-# message that includes the path.
-# ---------------------------------------------------------------------------
-
-def test_exiftool_metadata_with_nonexistent_binary():
- """#1960: nonexistent exiftool_path raises RuntimeError, not FileNotFoundError."""
- from markitdown.converters._exiftool import exiftool_metadata
-
- exiftool_path = "/this/does/not/exist/exiftool"
- with pytest.raises(
- RuntimeError,
- match=re.escape(f"Failed to invoke exiftool at {exiftool_path}"),
- ):
- exiftool_metadata(io.BytesIO(b""), exiftool_path=exiftool_path)
+def test_binary_with_a_text_extension_is_not_the_word_none() -> None:
+ markitdown = MarkItDown()
+ for extension in (".txt", ".md"):
+ result = markitdown.convert_stream(
+ io.BytesIO(UNDECODABLE), file_extension=extension
+ )
+ assert result.markdown != "None"
-def test_exiftool_metadata_with_no_path():
- """Sanity check: exiftool_path=None still returns {} (early return)."""
- from markitdown.converters._exiftool import exiftool_metadata
+def test_binary_with_a_csv_extension_is_not_a_table_of_none() -> None:
+ markitdown = MarkItDown()
+ result = markitdown.convert_stream(io.BytesIO(UNDECODABLE), file_extension=".csv")
+ assert result.markdown != "| None |\n| --- |"
- assert exiftool_metadata(io.BytesIO(b""), exiftool_path=None) == {}
+def test_ordinary_text_is_unchanged() -> None:
+ """The fallback only runs when detection fails; everything else is as it was."""
+ markitdown = MarkItDown()
-def test_exiftool_metadata_invocation_oserror():
- """#1960: an OSError while running exiftool is wrapped, and the stream is restored."""
- from unittest.mock import patch
+ text = markitdown.convert_stream(
+ io.BytesIO(b"hello, world\n"), file_extension=".txt"
+ )
+ assert text.markdown == "hello, world\n"
- from markitdown.converters._exiftool import exiftool_metadata
+ table = markitdown.convert_stream(io.BytesIO(b"a,b\n1,2\n"), file_extension=".csv")
+ assert table.markdown == "| a | b |\n| --- | --- |\n| 1 | 2 |"
- def fake_run(args, **kwargs):
- # The version check succeeds ...
- if "-ver" in args:
- return SimpleNamespace(stdout="12.24\n")
- # ... but the actual metadata invocation fails.
- raise OSError("exiftool vanished")
- exiftool_path = "/usr/bin/exiftool"
- file_stream = io.BytesIO(b"0123456789")
- file_stream.seek(4)
+def test_text_containing_the_word_none_survives() -> None:
+ """The bug was a whole document reading "None", not the word appearing in
+ one."""
+ markitdown = MarkItDown()
+ result = markitdown.convert_stream(
+ io.BytesIO(b"value,note\nNone,missing\n"), file_extension=".csv"
+ )
+ assert result.markdown == "| value | note |\n| --- | --- |\n| None | missing |"
- with patch("markitdown.converters._exiftool.subprocess.run", side_effect=fake_run):
- with pytest.raises(
- RuntimeError,
- match=re.escape(f"Failed to invoke exiftool at {exiftool_path}"),
- ):
- exiftool_metadata(file_stream, exiftool_path=exiftool_path)
- assert file_stream.tell() == 4
+def test_a_declared_charset_still_wins() -> None:
+ """`stream_info.charset` short-circuits detection and is untouched."""
+ markitdown = MarkItDown()
+ body = "名称,代码\n浦发银行,600000\n".encode("gb18030")
+ result = markitdown.convert_stream(
+ io.BytesIO(body),
+ stream_info=StreamInfo(extension=".csv", charset="gb18030"),
+ )
+ assert "浦发银行" in result.markdown
if __name__ == "__main__":
- """Runs this file's tests from the command line."""
- for test in [
- test_stream_info_operations,
- test_data_uris,
- test_file_uris,
- test_docx_comments,
- test_docx_zip_filename_casing_mismatch,
- test_docx_zip_filename_non_casing_mismatch_still_rejected,
- test_zip_duplicate_filenames_preserve_each_entry,
- test_input_as_strings,
- test_markitdown_remote,
- test_speech_transcription,
- test_exceptions,
- test_doc_rlink,
- test_markitdown_exiftool,
- test_markitdown_llm_parameters,
- test_markitdown_llm,
- test_pptx_chart_no_title_text_frame,
- test_pptx_chart_with_title_text_frame,
- test_ipynb_accepts_non_ascii,
- test_epub_metadata_nodevalue,
- test_exiftool_metadata_with_nonexistent_binary,
- test_exiftool_metadata_with_no_path,
- test_exiftool_metadata_invocation_oserror,
- ]:
- print(f"Running {test.__name__}...", end="")
- test()
- print("OK")
- print("All tests passed!")
+ raise SystemExit(pytest.main([__file__]))
diff --git a/packages/markitdown/tests/test_module_vectors.py b/packages/markitdown/tests/test_module_vectors.py
index 3f51550a83..d6d9d680c6 100644
--- a/packages/markitdown/tests/test_module_vectors.py
+++ b/packages/markitdown/tests/test_module_vectors.py
@@ -3,8 +3,6 @@
import time
import pytest
import base64
-import re
-import zipfile
from pathlib import Path
@@ -201,37 +199,6 @@ def test_convert_stream_keep_data_uris(test_vector):
assert string not in result.markdown
-def test_convert_docx_with_style_missing_type(tmp_path):
- """DOCX conversion should not fail when a style entry is missing w:type."""
- source_path = os.path.join(TEST_FILES_DIR, "test.docx")
- malformed_path = tmp_path / "missing_style_type.docx"
-
- with zipfile.ZipFile(source_path, mode="r") as zip_input:
- with zipfile.ZipFile(malformed_path, mode="w") as zip_output:
- for item in zip_input.infolist():
- content = zip_input.read(item.filename)
- if item.filename == "word/styles.xml":
- styles_xml = content.decode("utf-8")
- styles_xml, count = re.subn(
- r' None:
if __name__ == "__main__":
- pytest.main([__file__, "-v"])
+ raise SystemExit(pytest.main([__file__]))
diff --git a/packages/markitdown/tests/test_pdf_tables.py b/packages/markitdown/tests/test_pdf.py
similarity index 66%
rename from packages/markitdown/tests/test_pdf_tables.py
rename to packages/markitdown/tests/test_pdf.py
index d26de0cb98..8ae2ec464a 100644
--- a/packages/markitdown/tests/test_pdf_tables.py
+++ b/packages/markitdown/tests/test_pdf.py
@@ -1,11 +1,20 @@
-#!/usr/bin/env python3 -m pytest
-"""Tests for PDF table extraction functionality."""
+"""PDF conversion, table extraction, numbering, and page cleanup."""
+import io
import os
import re
+from unittest.mock import MagicMock, patch
+
import pytest
from markitdown import MarkItDown
+from markitdown.converters._pdf_converter import (
+ PARTIAL_NUMBERING_PATTERN,
+ _merge_partial_numbering_lines,
+)
+
+
+# Table extraction
TEST_FILES_DIR = os.path.join(os.path.dirname(__file__), "test_files")
@@ -51,8 +60,8 @@ def extract_markdown_tables(text_content):
for line in lines:
line = line.strip()
if line.startswith("|") and line.endswith("|"):
- # Skip separator rows (contain only dashes and pipes)
- if re.match(r"^\|[\s\-|]+\|$", line):
+ # A separator needs dashes in every cell; retain blank data rows.
+ if re.match(r"^\|(?:\s*-+\s*\|)+$", line):
continue
# Parse cells from the row
cells = [cell.strip() for cell in line.split("|")[1:-1]]
@@ -271,7 +280,7 @@ def test_borderless_table_extraction(self, markitdown):
), "Second table should come before Recommendations"
def test_borderless_table_no_duplication(self, markitdown):
- """Test that borderless table content is not duplicated excessively."""
+ """Each SKU reference appears once, without missing or duplicated content."""
pdf_path = os.path.join(
TEST_FILES_DIR, "SPARSE-2024-INV-1234_borderless_table.pdf"
)
@@ -282,13 +291,9 @@ def test_borderless_table_no_duplication(self, markitdown):
result = markitdown.convert(pdf_path)
text_content = result.text_content
- # Count occurrences of unique table data - should not be excessively duplicated
- # SKU-8847 appears in both tables, plus possibly once in summary text
+ # SKU-8847 appears once in each of two tables and once in the analysis.
sku_count = text_content.count("SKU-8847")
- # Should appear at most 4 times (2 tables + minor text references), not more
- assert (
- sku_count <= 4
- ), f"SKU-8847 appears too many times ({sku_count}), suggests duplication issue"
+ assert sku_count == 3, f"Expected 3 SKU-8847 references, found {sku_count}"
def test_borderless_table_correct_position(self, markitdown):
"""Test that tables appear in correct positions relative to text."""
@@ -626,31 +631,6 @@ def test_academic_pdf_extraction(self, markitdown):
"multiagentconversations" not in text_content.lower()
), "Text should have proper spacing between words"
- def test_scanned_pdf_handling(self, markitdown):
- """Test handling of scanned/image-based PDF (no text layer).
-
- Expected output: Empty - scanned PDFs without OCR have no text layer.
- """
- pdf_path = os.path.join(
- TEST_FILES_DIR, "MEDRPT-2024-PAT-3847_medical_report_scan.pdf"
- )
-
- if not os.path.exists(pdf_path):
- pytest.skip(f"Test file not found: {pdf_path}")
-
- result = markitdown.convert(pdf_path)
-
- # Scanned PDFs without OCR have no text layer, so extraction should be empty
- assert (
- result is not None
- ), "Converter should return a result even for scanned PDFs"
- assert result.text_content is not None, "text_content should not be None"
-
- # Verify extraction is empty (no text layer in scanned PDF)
- assert (
- result.text_content.strip() == ""
- ), f"Scanned PDF should have empty extraction, got: '{result.text_content[:100]}...'"
-
def test_movie_theater_booking_pdf_extraction(self, markitdown):
"""Test extraction of movie theater booking PDF with complex tables.
@@ -746,35 +726,7 @@ def test_movie_theater_full_output(self, markitdown):
with open(expected_path, "r", encoding="utf-8") as f:
expected_output = f.read()
- # Compare outputs
- actual_lines = [line.rstrip() for line in actual_output.split("\n")]
- expected_lines = [line.rstrip() for line in expected_output.split("\n")]
-
- # Check line count
- assert abs(len(actual_lines) - len(expected_lines)) <= 2, (
- f"Line count mismatch: actual={len(actual_lines)}, "
- f"expected={len(expected_lines)}"
- )
-
- # Check structural elements
- assert actual_output.count("|") > 80, "Should have many pipe separators"
- assert actual_output.count("---") > 8, "Should have table separators"
-
- # Validate critical sections
- for section in [
- "BOOKING ORDER",
- "STARLIGHT CINEMAS",
- "2024-12-5678",
- "Holiday Spectacular",
- "$12,500.00",
- ]:
- assert section in actual_output, f"Missing section: {section}"
-
- # Check table structure
- table_rows = [line for line in actual_lines if line.startswith("|")]
- assert (
- len(table_rows) > 15
- ), f"Should have >15 table rows, got {len(table_rows)}"
+ assert actual_output == expected_output
def test_sparse_borderless_table_full_output(self, markitdown):
"""Test complete output for SPARSE borderless table PDF."""
@@ -799,28 +751,7 @@ def test_sparse_borderless_table_full_output(self, markitdown):
with open(expected_path, "r", encoding="utf-8") as f:
expected_output = f.read()
- # Compare outputs
- actual_lines = [line.rstrip() for line in actual_output.split("\n")]
- expected_lines = [line.rstrip() for line in expected_output.split("\n")]
-
- # Check line count is close
- assert abs(len(actual_lines) - len(expected_lines)) <= 2, (
- f"Line count mismatch: actual={len(actual_lines)}, "
- f"expected={len(expected_lines)}"
- )
-
- # Check structural elements
- assert actual_output.count("|") > 50, "Should have many pipe separators"
-
- # Validate critical sections
- for section in [
- "INVENTORY RECONCILIATION REPORT",
- "SPARSE-2024-INV-1234",
- "SKU-8847",
- "SKU-9201",
- "Variance Analysis",
- ]:
- assert section in actual_output, f"Missing section: {section}"
+ assert actual_output == expected_output
def test_repair_multipage_full_output(self, markitdown):
"""Test complete output for REPAIR multipage invoice PDF."""
@@ -841,28 +772,7 @@ def test_repair_multipage_full_output(self, markitdown):
with open(expected_path, "r", encoding="utf-8") as f:
expected_output = f.read()
- # Compare outputs
- actual_lines = [line.rstrip() for line in actual_output.split("\n")]
- expected_lines = [line.rstrip() for line in expected_output.split("\n")]
-
- # Check line count is close
- assert abs(len(actual_lines) - len(expected_lines)) <= 2, (
- f"Line count mismatch: actual={len(actual_lines)}, "
- f"expected={len(expected_lines)}"
- )
-
- # Check structural elements
- assert actual_output.count("|") > 40, "Should have many pipe separators"
-
- # Validate critical sections
- for section in [
- "ZAVA AUTO REPAIR",
- "Gabriel Diaz",
- "Jeep",
- "Grand Cherokee",
- "GRAND TOTAL",
- ]:
- assert section in actual_output, f"Missing section: {section}"
+ assert actual_output == expected_output
def test_receipt_full_output(self, markitdown):
"""Test complete output for RECEIPT retail purchase PDF."""
@@ -887,25 +797,7 @@ def test_receipt_full_output(self, markitdown):
with open(expected_path, "r", encoding="utf-8") as f:
expected_output = f.read()
- # Compare outputs
- actual_lines = [line.rstrip() for line in actual_output.split("\n")]
- expected_lines = [line.rstrip() for line in expected_output.split("\n")]
-
- # Check line count is close
- assert abs(len(actual_lines) - len(expected_lines)) <= 2, (
- f"Line count mismatch: actual={len(actual_lines)}, "
- f"expected={len(expected_lines)}"
- )
-
- # Validate critical sections
- for section in [
- "TECHMART ELECTRONICS",
- "TXN-98765-2024",
- "Sarah Mitchell",
- "$821.14",
- "RETURN POLICY",
- ]:
- assert section in actual_output, f"Missing section: {section}"
+ assert actual_output == expected_output
def test_academic_paper_full_output(self, markitdown):
"""Test complete output for academic paper PDF."""
@@ -924,32 +816,10 @@ def test_academic_paper_full_output(self, markitdown):
with open(expected_path, "r", encoding="utf-8") as f:
expected_output = f.read()
- # Compare outputs
- actual_lines = [line.rstrip() for line in actual_output.split("\n")]
- expected_lines = [line.rstrip() for line in expected_output.split("\n")]
-
- # Check line count is close
- assert abs(len(actual_lines) - len(expected_lines)) <= 2, (
- f"Line count mismatch: actual={len(actual_lines)}, "
- f"expected={len(expected_lines)}"
- )
-
- # Academic paper should not have pipe separators
- assert (
- actual_output.count("|") == 0
- ), "Academic paper should not have pipe separators"
-
- # Validate critical sections
- for section in [
- "Introduction",
- "Large language models",
- "agents",
- "multi-agent",
- ]:
- assert section in actual_output, f"Missing section: {section}"
+ assert actual_output == expected_output
def test_medical_scan_full_output(self, markitdown):
- """Test complete output for medical report scan PDF (empty, no text layer)."""
+ """A scanned PDF without OCR produces empty output and no Markdown tables."""
pdf_path = os.path.join(
TEST_FILES_DIR, "MEDRPT-2024-PAT-3847_medical_report_scan.pdf"
)
@@ -976,6 +846,9 @@ def test_medical_scan_full_output(self, markitdown):
assert (
expected_output.strip() == ""
), "Expected output should be empty for scanned PDF"
+ assert (
+ extract_markdown_tables(actual_output) == []
+ ), "Scanned PDF should have no extracted tables"
class TestPdfTableMarkdownFormat:
@@ -1112,64 +985,38 @@ def test_receipt_has_no_tables(self, markitdown):
total_table_rows < 5
), f"Receipt should not have significant tables, found {total_table_rows} rows"
- def test_scanned_pdf_no_tables(self, markitdown):
- """Test that scanned PDF has empty extraction and no tables."""
- pdf_path = os.path.join(
- TEST_FILES_DIR, "MEDRPT-2024-PAT-3847_medical_report_scan.pdf"
- )
-
- if not os.path.exists(pdf_path):
- pytest.skip(f"Test file not found: {pdf_path}")
-
- result = markitdown.convert(pdf_path)
-
- # Scanned PDF with no text layer should have empty extraction
- assert (
- result.text_content.strip() == ""
- ), "Scanned PDF should have empty extraction"
-
- tables = extract_markdown_tables(result.text_content)
-
- # Scanned PDF with no text layer should have no tables
- assert len(tables) == 0, "Scanned PDF should have no extracted tables"
-
- def test_all_pdfs_table_rows_consistent(self, markitdown):
+ @pytest.mark.parametrize(
+ "pdf_file, expected_table_count",
+ [
+ ("SPARSE-2024-INV-1234_borderless_table.pdf", 2),
+ ("REPAIR-2022-INV-001_multipage.pdf", 6),
+ ("RECEIPT-2024-TXN-98765_retail_purchase.pdf", 0),
+ ("test.pdf", 0),
+ ],
+ )
+ def test_all_pdfs_table_rows_consistent(
+ self, markitdown, pdf_file, expected_table_count
+ ):
"""Test that all PDF tables have rows with pipe-separated content.
Note: With gap-based column detection, rows may have different column counts
depending on how content is spaced in the PDF. What's important is that each
row has pipe separators and the content is readable.
"""
- pdf_files = [
- "SPARSE-2024-INV-1234_borderless_table.pdf",
- "REPAIR-2022-INV-001_multipage.pdf",
- "RECEIPT-2024-TXN-98765_retail_purchase.pdf",
- "test.pdf",
- ]
-
- for pdf_file in pdf_files:
- pdf_path = os.path.join(TEST_FILES_DIR, pdf_file)
- if not os.path.exists(pdf_path):
- continue
-
- result = markitdown.convert(pdf_path)
- tables = extract_markdown_tables(result.text_content)
-
- for table_idx, table in enumerate(tables):
- if not table:
- continue
+ pdf_path = os.path.join(TEST_FILES_DIR, pdf_file)
+ result = markitdown.convert(pdf_path)
+ assert result.text_content.strip(), f"{pdf_file}: No text extracted"
- # Verify each row has at least one column (pipe-separated content)
- for row_idx, row in enumerate(table):
- assert (
- len(row) >= 1
- ), f"{pdf_file}: Table {table_idx}, row {row_idx} has no columns"
+ tables = extract_markdown_tables(result.text_content)
+ assert (
+ len(tables) == expected_table_count
+ ), f"{pdf_file}: Expected {expected_table_count} tables, found {len(tables)}"
- # Verify the row has non-empty content
- row_content = " ".join(cell.strip() for cell in row)
- assert (
- len(row_content.strip()) > 0
- ), f"{pdf_file}: Table {table_idx}, row {row_idx} is empty"
+ for table_idx, table in enumerate(tables):
+ for row_idx, row in enumerate(table):
+ assert any(
+ cell.strip() for cell in row
+ ), f"{pdf_file}: Table {table_idx}, row {row_idx} is empty"
def test_borderless_table_data_integrity(self, markitdown):
"""Test that borderless table extraction preserves data integrity."""
@@ -1196,3 +1043,464 @@ def test_borderless_table_data_integrity(self, markitdown):
table_text = str(second_table)
assert "Electronics" in table_text, "Second table should contain Electronics"
assert "Hardware" in table_text, "Second table should contain Hardware"
+
+
+# MasterFormat numbering
+
+
+class TestMasterFormatPartialNumbering:
+ """Test handling of MasterFormat-style partial numbering (.1, .2, etc.)."""
+
+ def test_partial_numbering_pattern_regex(self):
+ """Test that the partial numbering regex pattern correctly matches."""
+
+ # Should match partial numbering patterns
+ assert PARTIAL_NUMBERING_PATTERN.match(".1") is not None
+ assert PARTIAL_NUMBERING_PATTERN.match(".2") is not None
+ assert PARTIAL_NUMBERING_PATTERN.match(".10") is not None
+ assert PARTIAL_NUMBERING_PATTERN.match(".99") is not None
+
+ # Should NOT match other patterns
+ assert PARTIAL_NUMBERING_PATTERN.match("1.") is None
+ assert PARTIAL_NUMBERING_PATTERN.match("1.2") is None
+ assert PARTIAL_NUMBERING_PATTERN.match(".1.2") is None
+ assert PARTIAL_NUMBERING_PATTERN.match("text") is None
+ assert PARTIAL_NUMBERING_PATTERN.match(".a") is None
+ assert PARTIAL_NUMBERING_PATTERN.match("") is None
+
+ def test_masterformat_partial_numbering_not_split(self):
+ """Test that MasterFormat partial numbering stays with associated text.
+
+ MasterFormat documents use partial numbering like:
+ .1 The intent of this Request for Proposal...
+ .2 Available information relative to...
+
+ These should NOT be split into separate table columns, but kept
+ as coherent text lines with the number followed by its description.
+ """
+ pdf_path = os.path.join(TEST_FILES_DIR, "masterformat_partial_numbering.pdf")
+
+ markitdown = MarkItDown()
+ result = markitdown.convert(pdf_path)
+ text_content = result.text_content
+
+ # Partial numberings should NOT appear isolated on their own lines
+ # If they're isolated, it means the parser incorrectly split them from their text
+ lines = text_content.split("\n")
+ isolated_numberings = []
+ for line in lines:
+ stripped = line.strip()
+ # Check if line contains ONLY a partial numbering (with possible whitespace/pipes)
+ cleaned = stripped.replace("|", "").strip()
+ if cleaned in [".1", ".2", ".3", ".4", ".5", ".6", ".7", ".8", ".9", ".10"]:
+ isolated_numberings.append(stripped)
+
+ assert len(isolated_numberings) == 0, (
+ f"Partial numberings should not be isolated from their text. "
+ f"Found isolated: {isolated_numberings}"
+ )
+
+ # Verify that partial numberings appear WITH following text on the same line
+ # Look for patterns like ".1 The intent" or ".1 Some text"
+ partial_with_text = re.findall(r"\.\d+\s+\w+", text_content)
+ assert (
+ len(partial_with_text) > 0
+ ), "Expected to find partial numberings followed by text on the same line"
+
+ def test_masterformat_content_preserved(self):
+ """Test that MasterFormat document content is fully preserved."""
+ pdf_path = os.path.join(TEST_FILES_DIR, "masterformat_partial_numbering.pdf")
+
+ markitdown = MarkItDown()
+ result = markitdown.convert(pdf_path)
+ text_content = result.text_content
+
+ # Verify key content from the MasterFormat document is preserved
+ expected_content = [
+ "RFP for Construction Management Services",
+ "Section 00 00 43",
+ "Instructions to Respondents",
+ "Ken Sargent House",
+ "INTENT",
+ "Request for Proposal",
+ "KEN SARGENT HOUSE",
+ "GRANDE PRAIRIE, ALBERTA",
+ "Section 00 00 45",
+ ]
+
+ for content in expected_content:
+ assert (
+ content in text_content
+ ), f"Expected content '{content}' not found in extracted text"
+
+ # Verify partial numbering is followed by text on the same line
+ # .1 should be followed by "The intent" on the same line
+ assert re.search(
+ r"\.1\s+The intent", text_content
+ ), "Partial numbering .1 should be followed by 'The intent' text"
+
+ # .2 should be followed by "Available information" on the same line
+ assert re.search(
+ r"\.2\s+Available information", text_content
+ ), "Partial numbering .2 should be followed by 'Available information' text"
+
+ # Ensure text content is not empty and has reasonable length
+ assert (
+ len(text_content.strip()) > 100
+ ), "MasterFormat document should have substantial text content"
+
+ @pytest.mark.parametrize(
+ "blank_lines",
+ ["", "\n\n", " \n\t"],
+ ids=["one-blank-line", "multiple-blank-lines", "whitespace-only-lines"],
+ )
+ def test_merge_partial_numbering_with_empty_lines_between(self, blank_lines):
+ """Merge across blank lines while preserving surrounding text and paragraphs."""
+ text = (
+ "Heading\n\n"
+ f".1\n{blank_lines}\nThe intent of this Request...\n"
+ "Continuation.\n\n"
+ f".10\n{blank_lines}\nAvailable information."
+ )
+ expected = (
+ "Heading\n\n"
+ ".1 The intent of this Request...\n"
+ "Continuation.\n\n"
+ ".10 Available information."
+ )
+
+ assert _merge_partial_numbering_lines(text) == expected
+
+ def test_multiple_partial_numberings_all_merged(self):
+ """Test that all partial numberings in a document are properly merged."""
+ pdf_path = os.path.join(TEST_FILES_DIR, "masterformat_partial_numbering.pdf")
+
+ markitdown = MarkItDown()
+ result = markitdown.convert(pdf_path)
+ text_content = result.text_content
+
+ # Count occurrences of merged partial numberings (number followed by text)
+ merged_count = len(re.findall(r"\.\d+\s+[A-Za-z]", text_content))
+
+ # Count isolated partial numberings (number alone on a line)
+ isolated_count = 0
+ for line in text_content.split("\n"):
+ stripped = line.strip()
+ if re.match(r"^\.\d+$", stripped):
+ isolated_count += 1
+
+ assert (
+ merged_count >= 2
+ ), f"Expected at least 2 merged partial numberings, found {merged_count}"
+ assert (
+ isolated_count == 0
+ ), f"Found {isolated_count} isolated partial numberings that weren't merged"
+
+
+# Page cleanup and extraction fallback
+
+# Tests for PDF converter memory optimization.
+#
+# Verifies that:
+# - page.close() is called after processing each page (frees cached data)
+# - Plain-text PDFs fall back to pdfminer when no form pages are found
+# - Mixed PDFs use form extraction only on form-style pages
+
+
+def _make_form_page():
+ """Create a mock page with 3-column table-like word positions."""
+ page = MagicMock()
+ page.width = 612
+ page.close = MagicMock()
+ page.extract_words.return_value = [
+ {"text": "Name", "x0": 50, "x1": 100, "top": 10, "bottom": 20},
+ {"text": "Value", "x0": 250, "x1": 300, "top": 10, "bottom": 20},
+ {"text": "Unit", "x0": 450, "x1": 500, "top": 10, "bottom": 20},
+ {"text": "Alpha", "x0": 50, "x1": 100, "top": 30, "bottom": 40},
+ {"text": "100", "x0": 250, "x1": 280, "top": 30, "bottom": 40},
+ {"text": "kg", "x0": 450, "x1": 470, "top": 30, "bottom": 40},
+ {"text": "Beta", "x0": 50, "x1": 100, "top": 50, "bottom": 60},
+ {"text": "200", "x0": 250, "x1": 280, "top": 50, "bottom": 60},
+ {"text": "lb", "x0": 450, "x1": 470, "top": 50, "bottom": 60},
+ ]
+ return page
+
+
+def _make_plain_page():
+ """Create a mock page with single-line paragraph (no table structure)."""
+ page = MagicMock()
+ page.width = 612
+ page.close = MagicMock()
+ page.extract_words.return_value = [
+ {
+ "text": "This is a long paragraph of plain text.",
+ "x0": 50,
+ "x1": 550,
+ "top": 10,
+ "bottom": 20,
+ },
+ ]
+ page.extract_text.return_value = "This is a long paragraph of plain text."
+ return page
+
+
+def _mock_pdfplumber_open(pages):
+ """Return a mock pdfplumber.open that yields the given pages."""
+
+ def mock_open(stream):
+ mock_pdf = MagicMock()
+ mock_pdf.pages = pages
+ mock_pdf.__enter__ = MagicMock(return_value=mock_pdf)
+ mock_pdf.__exit__ = MagicMock(return_value=False)
+ return mock_pdf
+
+ return mock_open
+
+
+class TestPdfMemoryOptimization:
+ """Test that PDF conversion cleans up per-page caches to limit memory."""
+
+ def test_page_close_called_on_every_page(self):
+ """Verify page.close() is called on every page during conversion.
+
+ This ensures cached word/layout data is freed after each page,
+ preventing O(n) memory growth with page count.
+ """
+ num_pages = 20
+ pages = [_make_form_page() for _ in range(num_pages)]
+
+ with patch(
+ "markitdown.converters._pdf_converter.pdfplumber"
+ ) as mock_pdfplumber:
+ mock_pdfplumber.open.side_effect = _mock_pdfplumber_open(pages)
+
+ md = MarkItDown()
+ buf = io.BytesIO(b"fake pdf content")
+ from markitdown import StreamInfo
+
+ md.convert_stream(
+ buf,
+ stream_info=StreamInfo(extension=".pdf", mimetype="application/pdf"),
+ )
+
+ # page.close() must be called on ALL pages
+ for i, page in enumerate(pages):
+ assert page.close.called, (
+ f"page.close() was NOT called on page {i} — "
+ "this would cause memory to accumulate"
+ )
+
+ def test_plain_text_pdf_falls_back_to_pdfminer(self):
+ """Verify all-plain-text PDFs fall back to pdfminer.
+
+ When no page has form-style content, the converter should discard
+ pdfplumber results and use pdfminer for the whole document (better
+ text spacing for prose).
+ """
+ num_pages = 50
+ pages = [_make_plain_page() for _ in range(num_pages)]
+
+ with patch(
+ "markitdown.converters._pdf_converter.pdfplumber"
+ ) as mock_pdfplumber, patch(
+ "markitdown.converters._pdf_converter.pdfminer"
+ ) as mock_pdfminer:
+ mock_pdfplumber.open.side_effect = _mock_pdfplumber_open(pages)
+ mock_pdfminer.high_level.extract_text.return_value = "Plain text content"
+
+ md = MarkItDown()
+ buf = io.BytesIO(b"fake pdf content")
+ from markitdown import StreamInfo
+
+ result = md.convert_stream(
+ buf,
+ stream_info=StreamInfo(extension=".pdf", mimetype="application/pdf"),
+ )
+
+ # pdfminer should be used for the final text extraction
+ assert mock_pdfminer.high_level.extract_text.called, (
+ "pdfminer.high_level.extract_text was not called — "
+ "plain-text PDFs should fall back to pdfminer"
+ )
+ assert result.text_content is not None
+
+ def test_plain_text_pdf_still_closes_all_pages(self):
+ """Even for plain-text PDFs, page.close() must be called on every page."""
+ num_pages = 30
+ pages = [_make_plain_page() for _ in range(num_pages)]
+
+ with patch(
+ "markitdown.converters._pdf_converter.pdfplumber"
+ ) as mock_pdfplumber, patch(
+ "markitdown.converters._pdf_converter.pdfminer"
+ ) as mock_pdfminer:
+ mock_pdfplumber.open.side_effect = _mock_pdfplumber_open(pages)
+ mock_pdfminer.high_level.extract_text.return_value = "text"
+
+ md = MarkItDown()
+ buf = io.BytesIO(b"fake pdf content")
+ from markitdown import StreamInfo
+
+ md.convert_stream(
+ buf,
+ stream_info=StreamInfo(extension=".pdf", mimetype="application/pdf"),
+ )
+
+ for i, page in enumerate(pages):
+ assert (
+ page.close.called
+ ), f"page.close() was NOT called on plain-text page {i}"
+
+ def test_mixed_pdf_uses_form_extraction_per_page(self):
+ """In a mixed PDF, form pages get table extraction while plain pages don't.
+
+ Ensures we don't miss form-style pages and don't waste work
+ running form extraction on plain-text pages.
+ """
+ # Pages 0,2,4 are form-style; pages 1,3 are plain text
+ pages = [
+ _make_form_page(), # 0 - form
+ _make_plain_page(), # 1 - plain
+ _make_form_page(), # 2 - form
+ _make_plain_page(), # 3 - plain
+ _make_form_page(), # 4 - form
+ ]
+
+ with patch(
+ "markitdown.converters._pdf_converter.pdfplumber"
+ ) as mock_pdfplumber:
+ mock_pdfplumber.open.side_effect = _mock_pdfplumber_open(pages)
+
+ md = MarkItDown()
+ buf = io.BytesIO(b"fake pdf content")
+ from markitdown import StreamInfo
+
+ result = md.convert_stream(
+ buf,
+ stream_info=StreamInfo(extension=".pdf", mimetype="application/pdf"),
+ )
+
+ # All pages should have close() called
+ for i, page in enumerate(pages):
+ assert page.close.called, f"page.close() not called on page {i}"
+
+ # Form pages (0,2,4) should have extract_words called
+ for i in [0, 2, 4]:
+ assert pages[
+ i
+ ].extract_words.called, f"extract_words not called on form page {i}"
+
+ # Result should contain table content from form pages
+ assert result.text_content is not None
+ assert (
+ "|" in result.text_content
+ ), "Expected markdown table pipes in output from form-style pages"
+
+ def test_only_one_pdfplumber_open_call(self):
+ """Verify pdfplumber.open is called exactly once (single pass)."""
+ pages = [_make_form_page() for _ in range(10)]
+
+ with patch(
+ "markitdown.converters._pdf_converter.pdfplumber"
+ ) as mock_pdfplumber:
+ mock_pdfplumber.open.side_effect = _mock_pdfplumber_open(pages)
+
+ md = MarkItDown()
+ buf = io.BytesIO(b"fake pdf content")
+ from markitdown import StreamInfo
+
+ md.convert_stream(
+ buf,
+ stream_info=StreamInfo(extension=".pdf", mimetype="application/pdf"),
+ )
+
+ assert mock_pdfplumber.open.call_count == 1, (
+ f"Expected 1 pdfplumber.open call (single pass), "
+ f"got {mock_pdfplumber.open.call_count}"
+ )
+
+ @pytest.mark.skipif(
+ not os.path.exists(
+ os.path.join(TEST_FILES_DIR, "REPAIR-2022-INV-001_multipage.pdf")
+ ),
+ reason="Multipage invoice PDF not available",
+ )
+ def test_real_pdf_page_cleanup(self):
+ """Close each real page before the next page or document cleanup begins."""
+ import pdfplumber
+
+ events = []
+ page_numbers = []
+ original_extract_words = pdfplumber.page.Page.extract_words
+ original_page_close = pdfplumber.page.Page.close
+ original_pdf_close = pdfplumber.PDF.close
+
+ def tracking_extract_words(page, *args, **kwargs):
+ events.append(("extract", page.page_number))
+ return original_extract_words(page, *args, **kwargs)
+
+ def tracking_page_close(page):
+ events.append(("close", page.page_number))
+ return original_page_close(page)
+
+ def tracking_pdf_close(pdf):
+ page_numbers.extend(page.page_number for page in pdf.pages)
+ events.append(("document_close", None))
+ return original_pdf_close(pdf)
+
+ with patch.object(
+ pdfplumber.page.Page, "extract_words", tracking_extract_words
+ ), patch.object(
+ pdfplumber.page.Page, "close", tracking_page_close
+ ), patch.object(
+ pdfplumber.PDF, "close", tracking_pdf_close
+ ):
+ pdf_path = os.path.join(TEST_FILES_DIR, "REPAIR-2022-INV-001_multipage.pdf")
+ MarkItDown().convert(pdf_path)
+
+ # Assert outside the spies: conversion catches pdfplumber exceptions and
+ # falls back to pdfminer, which could otherwise hide assertion failures.
+ assert len(page_numbers) > 1, "The fixture must exercise multiple pages"
+ document_close = events.index(("document_close", None))
+ for index, page_number in enumerate(page_numbers):
+ extracted = events.index(("extract", page_number))
+ closed = events.index(("close", page_number))
+ next_extraction = (
+ events.index(("extract", page_numbers[index + 1]))
+ if index + 1 < len(page_numbers)
+ else document_close
+ )
+ assert extracted < closed < next_extraction, (
+ f"Page {page_number} must be closed after extraction and before "
+ f"the next page or document cleanup: {events}"
+ )
+
+
+# Conversion regressions
+
+skip_remote = (
+ True if os.environ.get("GITHUB_ACTIONS") else False
+) # Don't run these tests in CI
+
+PDF_TEST_URL = "https://arxiv.org/pdf/2308.08155v2.pdf"
+
+PDF_TEST_STRINGS = [
+ "While there is contemporaneous exploration of multi-agent approaches"
+]
+
+
+@pytest.mark.skipif(
+ skip_remote,
+ reason="do not run tests that query external urls",
+)
+def test_markitdown_remote() -> None:
+ markitdown = MarkItDown()
+
+ # By URL
+ result = markitdown.convert(PDF_TEST_URL)
+ for test_string in PDF_TEST_STRINGS:
+ assert test_string in result.text_content
+
+
+if __name__ == "__main__":
+ raise SystemExit(pytest.main([__file__]))
diff --git a/packages/markitdown/tests/test_pdf_masterformat.py b/packages/markitdown/tests/test_pdf_masterformat.py
deleted file mode 100644
index 8d3eb07398..0000000000
--- a/packages/markitdown/tests/test_pdf_masterformat.py
+++ /dev/null
@@ -1,171 +0,0 @@
-#!/usr/bin/env python3 -m pytest
-"""Tests for MasterFormat-style partial numbering in PDF conversion."""
-
-import os
-import re
-import pytest
-
-from markitdown import MarkItDown
-from markitdown.converters._pdf_converter import PARTIAL_NUMBERING_PATTERN
-
-TEST_FILES_DIR = os.path.join(os.path.dirname(__file__), "test_files")
-
-
-class TestMasterFormatPartialNumbering:
- """Test handling of MasterFormat-style partial numbering (.1, .2, etc.)."""
-
- def test_partial_numbering_pattern_regex(self):
- """Test that the partial numbering regex pattern correctly matches."""
-
- # Should match partial numbering patterns
- assert PARTIAL_NUMBERING_PATTERN.match(".1") is not None
- assert PARTIAL_NUMBERING_PATTERN.match(".2") is not None
- assert PARTIAL_NUMBERING_PATTERN.match(".10") is not None
- assert PARTIAL_NUMBERING_PATTERN.match(".99") is not None
-
- # Should NOT match other patterns
- assert PARTIAL_NUMBERING_PATTERN.match("1.") is None
- assert PARTIAL_NUMBERING_PATTERN.match("1.2") is None
- assert PARTIAL_NUMBERING_PATTERN.match(".1.2") is None
- assert PARTIAL_NUMBERING_PATTERN.match("text") is None
- assert PARTIAL_NUMBERING_PATTERN.match(".a") is None
- assert PARTIAL_NUMBERING_PATTERN.match("") is None
-
- def test_masterformat_partial_numbering_not_split(self):
- """Test that MasterFormat partial numbering stays with associated text.
-
- MasterFormat documents use partial numbering like:
- .1 The intent of this Request for Proposal...
- .2 Available information relative to...
-
- These should NOT be split into separate table columns, but kept
- as coherent text lines with the number followed by its description.
- """
- pdf_path = os.path.join(TEST_FILES_DIR, "masterformat_partial_numbering.pdf")
-
- markitdown = MarkItDown()
- result = markitdown.convert(pdf_path)
- text_content = result.text_content
-
- # Partial numberings should NOT appear isolated on their own lines
- # If they're isolated, it means the parser incorrectly split them from their text
- lines = text_content.split("\n")
- isolated_numberings = []
- for line in lines:
- stripped = line.strip()
- # Check if line contains ONLY a partial numbering (with possible whitespace/pipes)
- cleaned = stripped.replace("|", "").strip()
- if cleaned in [".1", ".2", ".3", ".4", ".5", ".6", ".7", ".8", ".9", ".10"]:
- isolated_numberings.append(stripped)
-
- assert len(isolated_numberings) == 0, (
- f"Partial numberings should not be isolated from their text. "
- f"Found isolated: {isolated_numberings}"
- )
-
- # Verify that partial numberings appear WITH following text on the same line
- # Look for patterns like ".1 The intent" or ".1 Some text"
- partial_with_text = re.findall(r"\.\d+\s+\w+", text_content)
- assert (
- len(partial_with_text) > 0
- ), "Expected to find partial numberings followed by text on the same line"
-
- def test_masterformat_content_preserved(self):
- """Test that MasterFormat document content is fully preserved."""
- pdf_path = os.path.join(TEST_FILES_DIR, "masterformat_partial_numbering.pdf")
-
- markitdown = MarkItDown()
- result = markitdown.convert(pdf_path)
- text_content = result.text_content
-
- # Verify key content from the MasterFormat document is preserved
- expected_content = [
- "RFP for Construction Management Services",
- "Section 00 00 43",
- "Instructions to Respondents",
- "Ken Sargent House",
- "INTENT",
- "Request for Proposal",
- "KEN SARGENT HOUSE",
- "GRANDE PRAIRIE, ALBERTA",
- "Section 00 00 45",
- ]
-
- for content in expected_content:
- assert (
- content in text_content
- ), f"Expected content '{content}' not found in extracted text"
-
- # Verify partial numbering is followed by text on the same line
- # .1 should be followed by "The intent" on the same line
- assert re.search(
- r"\.1\s+The intent", text_content
- ), "Partial numbering .1 should be followed by 'The intent' text"
-
- # .2 should be followed by "Available information" on the same line
- assert re.search(
- r"\.2\s+Available information", text_content
- ), "Partial numbering .2 should be followed by 'Available information' text"
-
- # Ensure text content is not empty and has reasonable length
- assert (
- len(text_content.strip()) > 100
- ), "MasterFormat document should have substantial text content"
-
- def test_merge_partial_numbering_with_empty_lines_between(self):
- """Test that partial numberings merge correctly even with empty lines between.
-
- When PDF extractors produce output like:
- .1
-
- The intent of this Request...
-
- The merge logic should still combine them properly.
- """
- pdf_path = os.path.join(TEST_FILES_DIR, "masterformat_partial_numbering.pdf")
-
- markitdown = MarkItDown()
- result = markitdown.convert(pdf_path)
- text_content = result.text_content
-
- # The merged result should have .1 and .2 followed by text
- # Check that we don't have patterns like ".1\n\nThe intent" (unmerged)
- lines = text_content.split("\n")
-
- for i, line in enumerate(lines):
- stripped = line.strip()
- # If we find an isolated partial numbering, the merge failed
- if stripped in [".1", ".2", ".3", ".4", ".5", ".6", ".7", ".8"]:
- # Check if next non-empty line exists and wasn't merged
- for j in range(i + 1, min(i + 3, len(lines))):
- if lines[j].strip():
- pytest.fail(
- f"Partial numbering '{stripped}' on line {i} was not "
- f"merged with following text '{lines[j].strip()[:30]}...'"
- )
- break
-
- def test_multiple_partial_numberings_all_merged(self):
- """Test that all partial numberings in a document are properly merged."""
- pdf_path = os.path.join(TEST_FILES_DIR, "masterformat_partial_numbering.pdf")
-
- markitdown = MarkItDown()
- result = markitdown.convert(pdf_path)
- text_content = result.text_content
-
- # Count occurrences of merged partial numberings (number followed by text)
- merged_count = len(re.findall(r"\.\d+\s+[A-Za-z]", text_content))
-
- # Count isolated partial numberings (number alone on a line)
- isolated_count = 0
- for line in text_content.split("\n"):
- stripped = line.strip()
- if re.match(r"^\.\d+$", stripped):
- isolated_count += 1
-
- assert (
- merged_count >= 2
- ), f"Expected at least 2 merged partial numberings, found {merged_count}"
- assert (
- isolated_count == 0
- ), f"Found {isolated_count} isolated partial numberings that weren't merged"
diff --git a/packages/markitdown/tests/test_pdf_memory.py b/packages/markitdown/tests/test_pdf_memory.py
deleted file mode 100644
index 1731dd63e6..0000000000
--- a/packages/markitdown/tests/test_pdf_memory.py
+++ /dev/null
@@ -1,364 +0,0 @@
-#!/usr/bin/env python3 -m pytest
-"""Tests for PDF converter memory optimization.
-
-Verifies that:
-- page.close() is called after processing each page (frees cached data)
-- Plain-text PDFs fall back to pdfminer when no form pages are found
-- Mixed PDFs use form extraction only on form-style pages
-- Memory stays constant regardless of page count
-"""
-
-import gc
-import io
-import os
-import tracemalloc
-
-import pytest
-from unittest.mock import patch, MagicMock
-
-from markitdown import MarkItDown
-
-TEST_FILES_DIR = os.path.join(os.path.dirname(__file__), "test_files")
-
-
-def _has_fpdf2() -> bool:
- try:
- import fpdf # noqa: F401
-
- return True
- except ImportError:
- return False
-
-
-def _make_form_page():
- """Create a mock page with 3-column table-like word positions."""
- page = MagicMock()
- page.width = 612
- page.close = MagicMock()
- page.extract_words.return_value = [
- {"text": "Name", "x0": 50, "x1": 100, "top": 10, "bottom": 20},
- {"text": "Value", "x0": 250, "x1": 300, "top": 10, "bottom": 20},
- {"text": "Unit", "x0": 450, "x1": 500, "top": 10, "bottom": 20},
- {"text": "Alpha", "x0": 50, "x1": 100, "top": 30, "bottom": 40},
- {"text": "100", "x0": 250, "x1": 280, "top": 30, "bottom": 40},
- {"text": "kg", "x0": 450, "x1": 470, "top": 30, "bottom": 40},
- {"text": "Beta", "x0": 50, "x1": 100, "top": 50, "bottom": 60},
- {"text": "200", "x0": 250, "x1": 280, "top": 50, "bottom": 60},
- {"text": "lb", "x0": 450, "x1": 470, "top": 50, "bottom": 60},
- ]
- return page
-
-
-def _make_plain_page():
- """Create a mock page with single-line paragraph (no table structure)."""
- page = MagicMock()
- page.width = 612
- page.close = MagicMock()
- page.extract_words.return_value = [
- {
- "text": "This is a long paragraph of plain text.",
- "x0": 50,
- "x1": 550,
- "top": 10,
- "bottom": 20,
- },
- ]
- page.extract_text.return_value = "This is a long paragraph of plain text."
- return page
-
-
-def _mock_pdfplumber_open(pages):
- """Return a mock pdfplumber.open that yields the given pages."""
-
- def mock_open(stream):
- mock_pdf = MagicMock()
- mock_pdf.pages = pages
- mock_pdf.__enter__ = MagicMock(return_value=mock_pdf)
- mock_pdf.__exit__ = MagicMock(return_value=False)
- return mock_pdf
-
- return mock_open
-
-
-class TestPdfMemoryOptimization:
- """Test that PDF conversion cleans up per-page caches to limit memory."""
-
- def test_page_close_called_on_every_page(self):
- """Verify page.close() is called on every page during conversion.
-
- This ensures cached word/layout data is freed after each page,
- preventing O(n) memory growth with page count.
- """
- num_pages = 20
- pages = [_make_form_page() for _ in range(num_pages)]
-
- with patch(
- "markitdown.converters._pdf_converter.pdfplumber"
- ) as mock_pdfplumber:
- mock_pdfplumber.open.side_effect = _mock_pdfplumber_open(pages)
-
- md = MarkItDown()
- buf = io.BytesIO(b"fake pdf content")
- from markitdown import StreamInfo
-
- md.convert_stream(
- buf,
- stream_info=StreamInfo(extension=".pdf", mimetype="application/pdf"),
- )
-
- # page.close() must be called on ALL pages
- for i, page in enumerate(pages):
- assert page.close.called, (
- f"page.close() was NOT called on page {i} — "
- "this would cause memory to accumulate"
- )
-
- def test_plain_text_pdf_falls_back_to_pdfminer(self):
- """Verify all-plain-text PDFs fall back to pdfminer.
-
- When no page has form-style content, the converter should discard
- pdfplumber results and use pdfminer for the whole document (better
- text spacing for prose).
- """
- num_pages = 50
- pages = [_make_plain_page() for _ in range(num_pages)]
-
- with patch(
- "markitdown.converters._pdf_converter.pdfplumber"
- ) as mock_pdfplumber, patch(
- "markitdown.converters._pdf_converter.pdfminer"
- ) as mock_pdfminer:
- mock_pdfplumber.open.side_effect = _mock_pdfplumber_open(pages)
- mock_pdfminer.high_level.extract_text.return_value = "Plain text content"
-
- md = MarkItDown()
- buf = io.BytesIO(b"fake pdf content")
- from markitdown import StreamInfo
-
- result = md.convert_stream(
- buf,
- stream_info=StreamInfo(extension=".pdf", mimetype="application/pdf"),
- )
-
- # pdfminer should be used for the final text extraction
- assert mock_pdfminer.high_level.extract_text.called, (
- "pdfminer.high_level.extract_text was not called — "
- "plain-text PDFs should fall back to pdfminer"
- )
- assert result.text_content is not None
-
- def test_plain_text_pdf_still_closes_all_pages(self):
- """Even for plain-text PDFs, page.close() must be called on every page."""
- num_pages = 30
- pages = [_make_plain_page() for _ in range(num_pages)]
-
- with patch(
- "markitdown.converters._pdf_converter.pdfplumber"
- ) as mock_pdfplumber, patch(
- "markitdown.converters._pdf_converter.pdfminer"
- ) as mock_pdfminer:
- mock_pdfplumber.open.side_effect = _mock_pdfplumber_open(pages)
- mock_pdfminer.high_level.extract_text.return_value = "text"
-
- md = MarkItDown()
- buf = io.BytesIO(b"fake pdf content")
- from markitdown import StreamInfo
-
- md.convert_stream(
- buf,
- stream_info=StreamInfo(extension=".pdf", mimetype="application/pdf"),
- )
-
- for i, page in enumerate(pages):
- assert (
- page.close.called
- ), f"page.close() was NOT called on plain-text page {i}"
-
- def test_mixed_pdf_uses_form_extraction_per_page(self):
- """In a mixed PDF, form pages get table extraction while plain pages don't.
-
- Ensures we don't miss form-style pages and don't waste work
- running form extraction on plain-text pages.
- """
- # Pages 0,2,4 are form-style; pages 1,3 are plain text
- pages = [
- _make_form_page(), # 0 - form
- _make_plain_page(), # 1 - plain
- _make_form_page(), # 2 - form
- _make_plain_page(), # 3 - plain
- _make_form_page(), # 4 - form
- ]
-
- with patch(
- "markitdown.converters._pdf_converter.pdfplumber"
- ) as mock_pdfplumber:
- mock_pdfplumber.open.side_effect = _mock_pdfplumber_open(pages)
-
- md = MarkItDown()
- buf = io.BytesIO(b"fake pdf content")
- from markitdown import StreamInfo
-
- result = md.convert_stream(
- buf,
- stream_info=StreamInfo(extension=".pdf", mimetype="application/pdf"),
- )
-
- # All pages should have close() called
- for i, page in enumerate(pages):
- assert page.close.called, f"page.close() not called on page {i}"
-
- # Form pages (0,2,4) should have extract_words called
- for i in [0, 2, 4]:
- assert pages[
- i
- ].extract_words.called, f"extract_words not called on form page {i}"
-
- # Result should contain table content from form pages
- assert result.text_content is not None
- assert (
- "|" in result.text_content
- ), "Expected markdown table pipes in output from form-style pages"
-
- def test_only_one_pdfplumber_open_call(self):
- """Verify pdfplumber.open is called exactly once (single pass)."""
- pages = [_make_form_page() for _ in range(10)]
-
- with patch(
- "markitdown.converters._pdf_converter.pdfplumber"
- ) as mock_pdfplumber:
- mock_pdfplumber.open.side_effect = _mock_pdfplumber_open(pages)
-
- md = MarkItDown()
- buf = io.BytesIO(b"fake pdf content")
- from markitdown import StreamInfo
-
- md.convert_stream(
- buf,
- stream_info=StreamInfo(extension=".pdf", mimetype="application/pdf"),
- )
-
- assert mock_pdfplumber.open.call_count == 1, (
- f"Expected 1 pdfplumber.open call (single pass), "
- f"got {mock_pdfplumber.open.call_count}"
- )
-
- @pytest.mark.skipif(
- not os.path.exists(os.path.join(TEST_FILES_DIR, "test.pdf")),
- reason="test.pdf not available",
- )
- def test_real_pdf_page_cleanup(self):
- """Integration test: verify page.close() is called with a real PDF."""
- import pdfplumber
-
- close_call_count = 0
- original_close = pdfplumber.page.Page.close
-
- def tracking_close(self):
- nonlocal close_call_count
- close_call_count += 1
- original_close(self)
-
- with patch.object(pdfplumber.page.Page, "close", tracking_close):
- md = MarkItDown()
- pdf_path = os.path.join(TEST_FILES_DIR, "test.pdf")
- md.convert(pdf_path)
-
- assert (
- close_call_count > 0
- ), "page.close() was never called during PDF conversion"
-
-
-def _generate_table_pdf(num_pages: int) -> bytes:
- """Generate a PDF with table-like content on every page."""
- from fpdf import FPDF
-
- pdf = FPDF()
- pdf.set_auto_page_break(auto=False)
- for page_num in range(num_pages):
- pdf.add_page()
- pdf.set_font("Helvetica", size=10)
- pdf.set_xy(10, 10)
- pdf.cell(60, 8, "Parameter", border=1)
- pdf.cell(60, 8, "Value", border=1)
- pdf.cell(60, 8, "Unit", border=1)
- pdf.ln()
- for row in range(20):
- y = 18 + row * 8
- if y > 270:
- break
- pdf.set_xy(10, y)
- pdf.cell(60, 8, f"Param_{page_num}_{row}", border=1)
- pdf.cell(60, 8, f"{(page_num * 100 + row) * 1.23:.2f}", border=1)
- pdf.cell(60, 8, "kg/m2", border=1)
- return pdf.output()
-
-
-@pytest.mark.skipif(
- not _has_fpdf2(),
- reason="fpdf2 not installed",
-)
-class TestPdfMemoryBenchmark:
- """Benchmark: verify memory stays constant with page.close() fix."""
-
- def test_memory_does_not_grow_linearly(self):
- """Peak memory for 200 pages should be far less than without the fix.
-
- Without page.close(), 200 pages uses ~225 MiB (linear growth).
- With the fix, peak memory should stay under 30 MiB.
- """
- from markitdown import StreamInfo
-
- num_pages = 200
- pdf_bytes = _generate_table_pdf(num_pages)
-
- gc.collect()
- tracemalloc.start()
-
- md = MarkItDown()
- buf = io.BytesIO(pdf_bytes)
- md.convert_stream(buf, stream_info=StreamInfo(extension=".pdf"))
-
- _, peak = tracemalloc.get_traced_memory()
- tracemalloc.stop()
-
- peak_mib = peak / 1024 / 1024
- # Without the fix this would be ~225 MiB. With the fix it should
- # be well under 30 MiB. Use a generous threshold to avoid flaky
- # failures on different machines.
- assert peak_mib < 30, (
- f"Peak memory {peak_mib:.1f} MiB for {num_pages} pages is too high. "
- f"Expected < 30 MiB with page.close() fix."
- )
-
- def test_memory_constant_across_page_counts(self):
- """Peak memory should not scale linearly with page count.
-
- Converts 50-page and 200-page PDFs and asserts the peak memory
- ratio is much less than the 4x page count ratio.
- """
- from markitdown import StreamInfo
-
- results = {}
- for num_pages in [50, 200]:
- pdf_bytes = _generate_table_pdf(num_pages)
-
- gc.collect()
- tracemalloc.start()
-
- md = MarkItDown()
- buf = io.BytesIO(pdf_bytes)
- md.convert_stream(buf, stream_info=StreamInfo(extension=".pdf"))
-
- _, peak = tracemalloc.get_traced_memory()
- tracemalloc.stop()
- results[num_pages] = peak
-
- ratio = results[200] / results[50]
- # With O(n) memory growth the ratio would be ~4x.
- # With the fix it should be close to 1x (well under 2x).
- assert ratio < 2.0, (
- f"Memory ratio 200p/50p = {ratio:.2f}x — "
- f"expected < 2.0x (constant memory). "
- f"50p={results[50] / 1024 / 1024:.1f} MiB, "
- f"200p={results[200] / 1024 / 1024:.1f} MiB"
- )
diff --git a/packages/markitdown/tests/test_pptx_images.py b/packages/markitdown/tests/test_pptx.py
similarity index 55%
rename from packages/markitdown/tests/test_pptx_images.py
rename to packages/markitdown/tests/test_pptx.py
index fa6d579aef..8d2b9eb19c 100644
--- a/packages/markitdown/tests/test_pptx_images.py
+++ b/packages/markitdown/tests/test_pptx.py
@@ -1,12 +1,17 @@
-"""PPTX image hooks share native slide traversal and HTML rendering."""
+"""PowerPoint conversion, titles, notes, charts, and image hooks."""
import base64
import inspect
import io
+import os
from pathlib import Path
from typing import Any, BinaryIO, Callable, Optional
-from unittest.mock import Mock
+from unittest.mock import MagicMock, Mock
+import pptx
+import pptx.shapes.autoshape
+import pptx.slide
+import pytest
from bs4 import BeautifulSoup
from lxml import etree
from pptx import Presentation
@@ -14,7 +19,6 @@
from pptx.enum.chart import XL_CHART_TYPE
from pptx.enum.shapes import MSO_SHAPE_TYPE
from pptx.util import Inches
-import pytest
from markitdown import (
FileConversionException,
@@ -22,16 +26,299 @@
MissingDependencyException,
StreamInfo,
)
-from markitdown.converters import HtmlConverter, PptxConverter
-from markitdown.converters import _pptx_converter
+from markitdown.converters import HtmlConverter, PptxConverter, _pptx_converter
+
+
+# Slide titles
+
+BODY_TEXT = "Some body text on the slide."
+
+
+def _build_pptx_with_title(title: str | None) -> io.BytesIO:
+ """Build a one-slide "Title and Content" deck, optionally filling the title.
+
+ Passing None leaves the title placeholder as PowerPoint creates it: present
+ on the slide, showing "Click to add title", and carrying no text.
+ """
+ prs = pptx.Presentation()
+ slide = prs.slides.add_slide(prs.slide_layouts[1])
+ if title is not None:
+ slide.shapes.title.text = title
+ slide.placeholders[1].text = BODY_TEXT
+ buf = io.BytesIO()
+ prs.save(buf)
+ buf.seek(0)
+ return buf
+
+
+def _convert_title_slide(title: str | None) -> str:
+ return (
+ MarkItDown()
+ .convert_stream(
+ _build_pptx_with_title(title), stream_info=StreamInfo(extension=".pptx")
+ )
+ .markdown
+ )
+
+
+def _heading_lines(markdown: str) -> list[str]:
+ return [line for line in markdown.splitlines() if line.startswith("#")]
+
+
+def test_untouched_title_placeholder_produces_no_heading() -> None:
+ markdown = _convert_title_slide(None)
+
+ assert BODY_TEXT in markdown
+ assert _heading_lines(markdown) == []
+
+
+def test_empty_title_produces_no_heading() -> None:
+ markdown = _convert_title_slide("")
+
+ assert BODY_TEXT in markdown
+ assert _heading_lines(markdown) == []
+
+
+def test_whitespace_only_title_produces_no_heading() -> None:
+ markdown = _convert_title_slide(" ")
+
+ assert BODY_TEXT in markdown
+ assert _heading_lines(markdown) == []
+
+
+def test_title_with_text_is_still_emitted() -> None:
+ markdown = _convert_title_slide("Quarterly Results")
+
+ assert "# Quarterly Results" in markdown
+ assert BODY_TEXT in markdown
+
+
+# Speaker notes
+
+
+def _build_pptx_with_optional_notes(notes: str | None) -> io.BytesIO:
+ """Build a one-slide deck in memory, optionally attaching a notes slide."""
+ prs = pptx.Presentation()
+ slide = prs.slides.add_slide(prs.slide_layouts[5])
+ slide.shapes.title.text = "Slide title"
+ if notes is not None:
+ # Touching notes_slide creates the part, which is what PowerPoint does
+ # for a deck whose notes pane has been opened.
+ slide.notes_slide.notes_text_frame.text = notes
+ buf = io.BytesIO()
+ prs.save(buf)
+ buf.seek(0)
+ return buf
+
+
+def _convert_notes_slide(notes: str | None) -> str:
+ return (
+ MarkItDown()
+ .convert_stream(
+ _build_pptx_with_optional_notes(notes),
+ stream_info=StreamInfo(extension=".pptx"),
+ )
+ .markdown
+ )
+
+
+def test_empty_notes_slide_produces_no_notes_heading() -> None:
+ markdown = _convert_notes_slide("")
+
+ assert "Slide title" in markdown
+ assert "### Notes:" not in markdown
+
+
+def test_whitespace_only_notes_produce_no_notes_heading() -> None:
+ markdown = _convert_notes_slide(" \n\n ")
+
+ assert "Slide title" in markdown
+ assert "### Notes:" not in markdown
+
+
+def test_slide_without_a_notes_slide_produces_no_notes_heading() -> None:
+ markdown = _convert_notes_slide(None)
+
+ assert "Slide title" in markdown
+ assert "### Notes:" not in markdown
+
+
+def test_notes_with_text_are_still_emitted() -> None:
+ markdown = _convert_notes_slide("Remember to mention the budget.")
+
+ assert "### Notes:\nRemember to mention the budget." in markdown
+
+
+# Missing text
+
+# Regression test for #1808: PptxConverter must not crash when ``shape.text``
+# or ``notes_text_frame.text`` returns ``None`` (e.g. an ```` run with no
+# ```` child, or certain third-party-generated decks).
+
+TEST_FILES_DIR = os.path.join(os.path.dirname(__file__), "test_files")
+
+PPTX_FIXTURE = os.path.join(TEST_FILES_DIR, "test.pptx")
+
+
+def _build_pptx_with_notes() -> io.BytesIO:
+ """Build a minimal pptx with a notes slide attached, in memory."""
+ prs = pptx.Presentation()
+ blank = prs.slide_layouts[6]
+ slide = prs.slides.add_slide(blank)
+ slide.notes_slide.notes_text_frame.text = "some notes"
+ buf = io.BytesIO()
+ prs.save(buf)
+ buf.seek(0)
+ return buf
+
+
+def test_pptx_converter_handles_none_shape_text(monkeypatch):
+ """``shape.text`` returning None must not blow up the title / body paths."""
+ monkeypatch.setattr(
+ pptx.shapes.autoshape.Shape,
+ "text",
+ property(lambda self: None),
+ raising=True,
+ )
+
+ result = MarkItDown().convert(PPTX_FIXTURE)
+ assert result.markdown is not None
+
+
+def test_pptx_converter_handles_none_notes_text(monkeypatch):
+ """``notes_text_frame.text`` returning None must not blow up the notes path."""
+ real_notes = pptx.slide.NotesSlide.notes_text_frame
+
+ class _NoneText:
+ text = None
+
+ def fake_notes(self):
+ frame = real_notes.fget(self)
+ if frame is None:
+ return None
+ return _NoneText()
+
+ monkeypatch.setattr(
+ pptx.slide.NotesSlide,
+ "notes_text_frame",
+ property(fake_notes),
+ raising=True,
+ )
+
+ buf = _build_pptx_with_notes()
+ result = MarkItDown().convert_stream(buf, file_extension=".pptx")
+ assert result.markdown is not None
+
+
+# SVG pictures
+
+# Tests for PPTX SVG images that lack a rasterized fallback.
+#
+# PowerPoint stores an SVG picture as an ```` whose ``r:embed`` points to
+# a rasterized PNG fallback, plus an ```` extension that points to
+# the SVG. When a picture has no raster fallback the ```` has no
+# ``r:embed`` at all, so python-pptx's ``shape.image`` raises
+# ``ValueError("no embedded image")``. The converter must handle this gracefully
+# (resolving the SVG blip directly) instead of failing the whole conversion.
+# A tiny synthetic PPTX whose only picture is an SVG without a rasterized
+# fallback: the has no r:embed, only an extension
+# pointing to an embedded SVG. Its alt text is "Red square SVG".
+SVG_NO_FALLBACK_PPTX = os.path.join(TEST_FILES_DIR, "test_svg_no_fallback.pptx")
+
+_SVG_NS = "http://schemas.microsoft.com/office/drawing/2016/SVG/main"
+
+
+def _first_picture_shape(pptx_path):
+ presentation = Presentation(pptx_path)
+ converter = PptxConverter()
+ for slide in presentation.slides:
+ for shape in slide.shapes:
+ if converter._is_picture(shape):
+ return converter, shape
+ raise AssertionError(f"No picture shape found in {pptx_path}")
+
+
+def test_pptx_svg_without_raster_fallback() -> None:
+ md = MarkItDown()
+
+ # Default conversion should not raise and should emit the image alt text.
+ result = md.convert(SVG_NO_FALLBACK_PPTX)
+ assert "Red square SVG" in result.markdown
+
+ # keep_data_uris used to crash with ValueError("no embedded image"). It
+ # should now embed the SVG as a data URI.
+ result = md.convert(SVG_NO_FALLBACK_PPTX, keep_data_uris=True)
+ assert "data:image/svg+xml;base64," in result.markdown
+
+
+def test_get_image_info_resolves_svg_blip_without_fallback() -> None:
+ # Unit-level check: _get_image_info must resolve the raw SVG blob directly
+ # from the extension when shape.image raises because there is
+ # no rasterized fallback (no r:embed on the ).
+ converter, shape = _first_picture_shape(SVG_NO_FALLBACK_PPTX)
+
+ blob, content_type, _filename = converter._get_image_info(shape)
+
+ assert blob is not None and len(blob) > 0
+ assert content_type == "image/svg+xml"
+ assert b"