Test threaded parsing with a single document.
()
| 176 | |
| 177 | |
| 178 | def test_threaded_single_document(): |
| 179 | """Test threaded parsing with a single document.""" |
| 180 | filename = SAMPLE_PDF |
| 181 | |
| 182 | parser = DoclingThreadedPdfParser( |
| 183 | parser_config=ThreadedPdfParserConfig( |
| 184 | loglevel="fatal", |
| 185 | threads=2, |
| 186 | max_concurrent_results=4, |
| 187 | boundary_type=PdfPageBoundaryType.CROP_BOX, |
| 188 | ), |
| 189 | decode_config=_make_decode_config(), |
| 190 | ) |
| 191 | |
| 192 | key = parser.load(filename) |
| 193 | assert parser.page_count(key) > 0 |
| 194 | |
| 195 | count = 0 |
| 196 | for result in parser.iterate_results(): |
| 197 | assert result.success, ( |
| 198 | f"Failed to decode page {result.page_number}: {result.error_message}" |
| 199 | ) |
| 200 | assert result.doc_key == key |
| 201 | assert result.page_width > 0 |
| 202 | assert result.page_height > 0 |
| 203 | assert result.get_timings().total() > 0 |
| 204 | assert result.timings.total_s > 0 |
| 205 | assert result.timings.decode_page_s > 0 |
| 206 | count += 1 |
| 207 | |
| 208 | assert count == parser.page_count(key) |
| 209 | |
| 210 | |
| 211 | def test_threaded_results_match_sequential(): |
nothing calls this directly
no test coverage detected