Function test_threaded_backpressure

tests/test_threaded_parse.py:271–287

Test that backpressure works with max_concurrent_results=1.

()

def test_threaded_backpressure():
    """Test that backpressure works with max_concurrent_results=1."""
    filename = LARGE_SAMPLE_PDF

    parser = DoclingThreadedPdfParser(
        parser_config=ThreadedPdfParserConfig(
            loglevel="fatal",
            threads=2,
            max_concurrent_results=1,
            boundary_type=PdfPageBoundaryType.CROP_BOX,
        ),
        decode_config=_make_decode_config(),
    )

    key = parser.load(filename)
    count = sum(1 for result in parser.iterate_results() if result.success)
    assert count == parser.page_count(key)


def test_threaded_single_thread():