Build a fake BatchExtractor with prepare/finalize/batch_provider.
(
prepared_map: dict[str, PreparedFile],
finalize_results: dict[str, ExtractionResult] | None = None,
finalize_error: Exception | None = None,
)
| 64 | |
| 65 | |
| 66 | def _make_extractor( |
| 67 | prepared_map: dict[str, PreparedFile], |
| 68 | finalize_results: dict[str, ExtractionResult] | None = None, |
| 69 | finalize_error: Exception | None = None, |
| 70 | ) -> _FakeExtractor: |
| 71 | """Build a fake BatchExtractor with prepare/finalize/batch_provider.""" |
| 72 | ext = _FakeExtractor() |
| 73 | |
| 74 | def _prepare(gz_path: str) -> PreparedFile: |
| 75 | return prepared_map[gz_path] |
| 76 | |
| 77 | ext.prepare.side_effect = _prepare |
| 78 | |
| 79 | if finalize_error: |
| 80 | |
| 81 | def _finalize(gz_path, prepared, responses): |
| 82 | raise finalize_error |
| 83 | |
| 84 | ext.finalize.side_effect = _finalize |
| 85 | elif finalize_results: |
| 86 | |
| 87 | def _finalize(gz_path, prepared, responses): |
| 88 | return finalize_results[gz_path] |
| 89 | |
| 90 | ext.finalize.side_effect = _finalize |
| 91 | else: |
| 92 | |
| 93 | def _finalize(gz_path, prepared, responses): |
| 94 | return _make_result(gz_path) |
| 95 | |
| 96 | ext.finalize.side_effect = _finalize |
| 97 | |
| 98 | return ext |
| 99 | |
| 100 | |
| 101 | def _make_batch_provider( |
no test coverage detected