From e3f3da604a5058770347c8ae5aa239d86f8d67d2 Mon Sep 17 00:00:00 2001 From: suguanYang Date: Tue, 25 Aug 2026 10:38:41 +0800 Subject: [PATCH 1/2] fix: upload local MinerU shards when HeadObject returns 403 Temp shard keys are not in S3 yet. A forbidden HeadObject must not abort URL mode; upload the local shard and continue with a presigned URL. Co-authored-by: Cursor --- .../providers/mineru/pdf_service.py | 27 +++++ .../tests/unit/test_mineru_pdf_service.py | 105 ++++++++++++++++++ 2 files changed, 132 insertions(+) diff --git a/apps/worker/app/services/document_parser/providers/mineru/pdf_service.py b/apps/worker/app/services/document_parser/providers/mineru/pdf_service.py index 01af82ea..d330e128 100644 --- a/apps/worker/app/services/document_parser/providers/mineru/pdf_service.py +++ b/apps/worker/app/services/document_parser/providers/mineru/pdf_service.py @@ -36,6 +36,21 @@ def _should_use_mineru_s3_url_mode(s3_key: Optional[str]) -> bool: return not settings.MINERU_UPLOAD_MODE_ENABLED and s3_key is not None +def _is_unverified_s3_object(exc: BaseException) -> bool: + """True when HeadObject cannot distinguish missing from forbidden. + + Fresh temp shard keys are not uploaded yet. S3 often returns 403 instead + of 404 when the caller lacks ListBucket on that prefix. That must not + abort URL mode if a local file can still be uploaded. + """ + text = str(exc).lower() + return ( + "403" in str(exc) + or "forbidden" in text + or "accessdenied" in text.replace(" ", "") + ) + + def _log_mineru_url_mode_storage_fallback( operation: str, s3_key: str, @@ -117,6 +132,18 @@ def _inspect_mineru_source_s3_key(s3_key: Optional[str]) -> tuple[Optional[str], try: existing_file = JobFileStorage().verify_upload_exists(s3_key) except Exception as exc: + if _is_unverified_s3_object(exc): + mineru_logger( + "url_mode_source_unverified", + operation="verify_source_object", + source_s3_key=s3_key, + error_type=type(exc).__name__, + error_message=truncate_log_value(exc), + ).info( + "S3 source existence could not be verified; " + "will upload a local file if one is available" + ) + return None, True _log_mineru_url_mode_storage_fallback( operation="verify_source_object", s3_key=s3_key, diff --git a/apps/worker/tests/unit/test_mineru_pdf_service.py b/apps/worker/tests/unit/test_mineru_pdf_service.py index 152ea5e6..960b8d16 100644 --- a/apps/worker/tests/unit/test_mineru_pdf_service.py +++ b/apps/worker/tests/unit/test_mineru_pdf_service.py @@ -17,6 +17,7 @@ from shared.core.exceptions.domain_exceptions import ( MinerUTaskFailedException, PDFParsingException, + StorageServiceException, TimeoutException, ) @@ -279,3 +280,107 @@ def test_url_poll_other_pdf_error_does_not_direct_upload( poll_mineru_task.assert_called_once() request_upload_target.assert_not_called() upload_file.assert_not_called() + + +_HEAD_OBJECT_FORBIDDEN = StorageServiceException( + internal_message=( + "Storage file verification failed: S3 check object exists failed: " + "An error occurred (403) when calling the HeadObject operation: Forbidden" + ), + operation="verify_exists", +) + + +def _configure_source_storage( + monkeypatch: pytest.MonkeyPatch, + *, + verify_side_effect: object, +) -> Mock: + monkeypatch.setattr(pdf_service.settings, "MINERU_UPLOAD_MODE_ENABLED", False) + storage = Mock() + storage.verify_upload_exists.side_effect = verify_side_effect + storage.upload_source_file.return_value = {"etag": "ok"} + monkeypatch.setattr(pdf_service, "JobFileStorage", Mock(return_value=storage)) + return storage + + +def test_resolve_uploads_local_shard_when_head_object_is_forbidden( + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, +) -> None: + storage = _configure_source_storage( + monkeypatch, + verify_side_effect=_HEAD_OBJECT_FORBIDDEN, + ) + local_pdf = tmp_path / "shard_0.pdf" + local_pdf.write_bytes(b"%PDF-1.4") + s3_key = "tmp/mineru-shards/job_abc/shard_0.pdf" + + resolved = pdf_service.resolve_mineru_source_s3_key( + s3_key, + local_file_path=str(local_pdf), + ) + + assert resolved == s3_key + storage.upload_source_file.assert_called_once_with(str(local_pdf), s3_key) + + +def test_resolve_skips_url_mode_when_head_forbidden_and_no_local_file( + monkeypatch: pytest.MonkeyPatch, +) -> None: + storage = _configure_source_storage( + monkeypatch, + verify_side_effect=_HEAD_OBJECT_FORBIDDEN, + ) + + resolved = pdf_service.resolve_mineru_source_s3_key( + "tmp/mineru-shards/job_abc/shard_0.pdf", + ) + + assert resolved is None + storage.upload_source_file.assert_not_called() + + +def test_resolve_still_falls_back_on_other_verify_errors( + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, +) -> None: + storage = _configure_source_storage( + monkeypatch, + verify_side_effect=StorageServiceException( + internal_message="S3 check object exists failed: timeout", + operation="verify_exists", + ), + ) + local_pdf = tmp_path / "shard_0.pdf" + local_pdf.write_bytes(b"%PDF-1.4") + + resolved = pdf_service.resolve_mineru_source_s3_key( + "tmp/mineru-shards/job_abc/shard_0.pdf", + local_file_path=str(local_pdf), + ) + + assert resolved is None + storage.upload_source_file.assert_not_called() + + +def test_resolve_reuses_existing_object_without_reupload( + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, +) -> None: + storage = _configure_source_storage( + monkeypatch, + verify_side_effect=[{"exists": True, "size": 12}], + ) + local_pdf = tmp_path / "source.pdf" + local_pdf.write_bytes(b"%PDF-1.4") + s3_key = "uploads/job_abc.pdf" + + resolved = pdf_service.resolve_mineru_source_s3_key( + s3_key, + local_file_path=str(local_pdf), + ) + + assert resolved == s3_key + storage.upload_source_file.assert_not_called() + From b07e93f98ecbd7fdb63fb6b42244ce69adb8a380 Mon Sep 17 00:00:00 2001 From: suguanYang Date: Tue, 25 Aug 2026 10:49:22 +0800 Subject: [PATCH 2/2] fix: upload split MinerU shards to S3 before URL mode HeadObject 403 happened because split shards were never uploaded. Put them in S3 and confirm they exist after split, then start MinerU, instead of treating 403 as a later upload fallback. Co-authored-by: Cursor --- .../document_parser/formats/pdf/parser.py | 42 +++- .../providers/mineru/pdf_service.py | 27 -- .../contract/test_parse_task_contract.py | 11 + .../tests/unit/test_mineru_pdf_service.py | 93 +------ .../tests/unit/test_pdf_shard_s3_upload.py | 237 ++++++++++++++++++ 5 files changed, 290 insertions(+), 120 deletions(-) create mode 100644 apps/worker/tests/unit/test_pdf_shard_s3_upload.py diff --git a/apps/worker/app/services/document_parser/formats/pdf/parser.py b/apps/worker/app/services/document_parser/formats/pdf/parser.py index 24188dac..a937b1c6 100755 --- a/apps/worker/app/services/document_parser/formats/pdf/parser.py +++ b/apps/worker/app/services/document_parser/formats/pdf/parser.py @@ -13,6 +13,7 @@ from loguru import logger from shared.core.config import settings +from shared.core.exceptions.domain_exceptions import StorageServiceException from shared.services.storage.job_file_storage import JobFileStorage @@ -98,10 +99,11 @@ def _parse_pdf_via_shards( 1. PROFILE → shard plan + TOC 2. map_agent_shards → 1:1 MinerU shards 3. split_pdf (exclude TOC pages) - 4. MinerU per shard (parallel) - 5. **Per-shard heading prediction** (parallel) - 6. Merge lines_with_heading + images - 7. parse_md Phase B (skip TOC detection + heading prediction) + 4. Upload split shards to S3 and confirm HeadObject succeeds + 5. MinerU per shard (parallel) + 6. **Per-shard heading prediction** (parallel) + 7. Merge lines_with_heading + images + 8. parse_md Phase B (skip TOC detection + heading prediction) """ from concurrent.futures import ThreadPoolExecutor, as_completed from dataclasses import dataclass @@ -200,7 +202,11 @@ def _parse_pdf_via_shards( for shard_index, _shard_pdf_path in enumerate(shard_pdf_paths) ] - # 5. Parse each shard via MinerU (parallel) + with stage_timer("pdf.upload_shards", filename=filename): + _upload_temp_shard_pdfs(shard_pdf_paths, temp_shard_s3_keys) + + # 5. Parse each shard via MinerU (parallel) only after S3 has the + # split objects. URL mode HeadObject then reuses those keys. shard_output_dirs = [None] * len(shard_pdf_paths) def _parse_single_shard(shard_idx, shard_pdf): @@ -355,6 +361,32 @@ def _predict_shard_headings_with_tracking( _cleanup_temp_shard_s3_assets(temp_shard_s3_keys) _cleanup_local_shard_workspace(work_dir) + +def _upload_temp_shard_pdfs( + shard_pdf_paths: list[str], + shard_s3_keys: list[str], +) -> None: + """Upload split shards to S3 and confirm each object exists before MinerU.""" + storage = JobFileStorage() + for shard_index, (shard_path, shard_s3_key) in enumerate( + zip(shard_pdf_paths, shard_s3_keys, strict=True) + ): + logger.info( + f" ☁️ Uploading MinerU shard_{shard_index} to S3 ({shard_s3_key})" + ) + storage.upload_source_file(shard_path, shard_s3_key) + existing_file = storage.verify_upload_exists(shard_s3_key) + if not existing_file.get("exists"): + raise StorageServiceException( + internal_message=( + "Temporary MinerU shard was uploaded but is not visible " + f"in S3: {shard_s3_key}" + ), + operation="verify_source_object", + ) + logger.info(f" ✅ Uploaded MinerU shard_{shard_index}: {shard_s3_key}") + + def _build_temp_shard_s3_key( *, source_s3_key: str | None, diff --git a/apps/worker/app/services/document_parser/providers/mineru/pdf_service.py b/apps/worker/app/services/document_parser/providers/mineru/pdf_service.py index d330e128..01af82ea 100644 --- a/apps/worker/app/services/document_parser/providers/mineru/pdf_service.py +++ b/apps/worker/app/services/document_parser/providers/mineru/pdf_service.py @@ -36,21 +36,6 @@ def _should_use_mineru_s3_url_mode(s3_key: Optional[str]) -> bool: return not settings.MINERU_UPLOAD_MODE_ENABLED and s3_key is not None -def _is_unverified_s3_object(exc: BaseException) -> bool: - """True when HeadObject cannot distinguish missing from forbidden. - - Fresh temp shard keys are not uploaded yet. S3 often returns 403 instead - of 404 when the caller lacks ListBucket on that prefix. That must not - abort URL mode if a local file can still be uploaded. - """ - text = str(exc).lower() - return ( - "403" in str(exc) - or "forbidden" in text - or "accessdenied" in text.replace(" ", "") - ) - - def _log_mineru_url_mode_storage_fallback( operation: str, s3_key: str, @@ -132,18 +117,6 @@ def _inspect_mineru_source_s3_key(s3_key: Optional[str]) -> tuple[Optional[str], try: existing_file = JobFileStorage().verify_upload_exists(s3_key) except Exception as exc: - if _is_unverified_s3_object(exc): - mineru_logger( - "url_mode_source_unverified", - operation="verify_source_object", - source_s3_key=s3_key, - error_type=type(exc).__name__, - error_message=truncate_log_value(exc), - ).info( - "S3 source existence could not be verified; " - "will upload a local file if one is available" - ) - return None, True _log_mineru_url_mode_storage_fallback( operation="verify_source_object", s3_key=s3_key, diff --git a/apps/worker/tests/contract/test_parse_task_contract.py b/apps/worker/tests/contract/test_parse_task_contract.py index 9cd005d3..1137efe7 100644 --- a/apps/worker/tests/contract/test_parse_task_contract.py +++ b/apps/worker/tests/contract/test_parse_task_contract.py @@ -982,8 +982,18 @@ def test_oversized_pdf_happy_path_uses_shard_pipeline_without_external_services( calls: dict[str, object] = {} parse_s3_keys: list[str | None] = [] deleted_s3_keys: list[str] = [] + uploaded_s3_keys: list[str] = [] class _FakeJobFileStorage: + def upload_source_file(self, local_file_path: str, storage_key: str) -> dict[str, str]: + assert Path(local_file_path).exists() + uploaded_s3_keys.append(storage_key) + return {"etag": "test"} + + def verify_upload_exists(self, storage_key: str) -> dict[str, bool]: + assert storage_key in uploaded_s3_keys + return {"exists": True} + def delete_upload_file(self, storage_key: str) -> bool: deleted_s3_keys.append(storage_key) return True @@ -1100,6 +1110,7 @@ def _identity_eval_md_headings( "tmp/mineru-shards/job-oversized/shard_1.pdf", ] assert parse_s3_keys == expected_s3_keys + assert uploaded_s3_keys == expected_s3_keys assert deleted_s3_keys == expected_s3_keys assert not (output_dir / "_shards").exists() assert list(df["type"]) == ["PTXT", "PTXT"] diff --git a/apps/worker/tests/unit/test_mineru_pdf_service.py b/apps/worker/tests/unit/test_mineru_pdf_service.py index 960b8d16..4a8231c0 100644 --- a/apps/worker/tests/unit/test_mineru_pdf_service.py +++ b/apps/worker/tests/unit/test_mineru_pdf_service.py @@ -17,7 +17,6 @@ from shared.core.exceptions.domain_exceptions import ( MinerUTaskFailedException, PDFParsingException, - StorageServiceException, TimeoutException, ) @@ -282,99 +281,17 @@ def test_url_poll_other_pdf_error_does_not_direct_upload( upload_file.assert_not_called() -_HEAD_OBJECT_FORBIDDEN = StorageServiceException( - internal_message=( - "Storage file verification failed: S3 check object exists failed: " - "An error occurred (403) when calling the HeadObject operation: Forbidden" - ), - operation="verify_exists", -) - - -def _configure_source_storage( - monkeypatch: pytest.MonkeyPatch, - *, - verify_side_effect: object, -) -> Mock: - monkeypatch.setattr(pdf_service.settings, "MINERU_UPLOAD_MODE_ENABLED", False) - storage = Mock() - storage.verify_upload_exists.side_effect = verify_side_effect - storage.upload_source_file.return_value = {"etag": "ok"} - monkeypatch.setattr(pdf_service, "JobFileStorage", Mock(return_value=storage)) - return storage - - -def test_resolve_uploads_local_shard_when_head_object_is_forbidden( - monkeypatch: pytest.MonkeyPatch, - tmp_path: Path, -) -> None: - storage = _configure_source_storage( - monkeypatch, - verify_side_effect=_HEAD_OBJECT_FORBIDDEN, - ) - local_pdf = tmp_path / "shard_0.pdf" - local_pdf.write_bytes(b"%PDF-1.4") - s3_key = "tmp/mineru-shards/job_abc/shard_0.pdf" - - resolved = pdf_service.resolve_mineru_source_s3_key( - s3_key, - local_file_path=str(local_pdf), - ) - - assert resolved == s3_key - storage.upload_source_file.assert_called_once_with(str(local_pdf), s3_key) - - -def test_resolve_skips_url_mode_when_head_forbidden_and_no_local_file( - monkeypatch: pytest.MonkeyPatch, -) -> None: - storage = _configure_source_storage( - monkeypatch, - verify_side_effect=_HEAD_OBJECT_FORBIDDEN, - ) - - resolved = pdf_service.resolve_mineru_source_s3_key( - "tmp/mineru-shards/job_abc/shard_0.pdf", - ) - - assert resolved is None - storage.upload_source_file.assert_not_called() - - -def test_resolve_still_falls_back_on_other_verify_errors( - monkeypatch: pytest.MonkeyPatch, - tmp_path: Path, -) -> None: - storage = _configure_source_storage( - monkeypatch, - verify_side_effect=StorageServiceException( - internal_message="S3 check object exists failed: timeout", - operation="verify_exists", - ), - ) - local_pdf = tmp_path / "shard_0.pdf" - local_pdf.write_bytes(b"%PDF-1.4") - - resolved = pdf_service.resolve_mineru_source_s3_key( - "tmp/mineru-shards/job_abc/shard_0.pdf", - local_file_path=str(local_pdf), - ) - - assert resolved is None - storage.upload_source_file.assert_not_called() - - def test_resolve_reuses_existing_object_without_reupload( monkeypatch: pytest.MonkeyPatch, tmp_path: Path, ) -> None: - storage = _configure_source_storage( - monkeypatch, - verify_side_effect=[{"exists": True, "size": 12}], - ) + monkeypatch.setattr(pdf_service.settings, "MINERU_UPLOAD_MODE_ENABLED", False) + storage = Mock() + storage.verify_upload_exists.return_value = {"exists": True, "size": 12} + monkeypatch.setattr(pdf_service, "JobFileStorage", Mock(return_value=storage)) local_pdf = tmp_path / "source.pdf" local_pdf.write_bytes(b"%PDF-1.4") - s3_key = "uploads/job_abc.pdf" + s3_key = "tmp/mineru-shards/job_abc/shard_0.pdf" resolved = pdf_service.resolve_mineru_source_s3_key( s3_key, diff --git a/apps/worker/tests/unit/test_pdf_shard_s3_upload.py b/apps/worker/tests/unit/test_pdf_shard_s3_upload.py new file mode 100644 index 00000000..9c638a54 --- /dev/null +++ b/apps/worker/tests/unit/test_pdf_shard_s3_upload.py @@ -0,0 +1,237 @@ +"""Split shards must be uploaded to S3 before MinerU URL mode starts.""" + +from __future__ import annotations + +import os +from pathlib import Path +from types import SimpleNamespace +from unittest.mock import Mock, call + +import pytest + +os.environ.setdefault("DATABASE_URL", "postgresql+asyncpg://test:test@localhost/test") +os.environ.setdefault("TMP_PATH", "/tmp/knowhere-test") +os.environ.setdefault("S3_BUCKET_NAME", "test-uploads") +os.environ.setdefault("S3_ACCESS_KEY_ID", "test") +os.environ.setdefault("S3_SECRET_ACCESS_KEY", "test") +os.environ.setdefault("S3_TEMP_PATH", "/tmp") + +from app.services.document_agent.manifest import ( + PageAnatomyMap, + Shard, + ShardPlan, + TocResult, +) +from app.services.document_parser.formats.pdf import parser as pdf_parser +from shared.core.exceptions.domain_exceptions import StorageServiceException + + +def test_upload_temp_shard_pdfs_writes_each_local_file( + monkeypatch: pytest.MonkeyPatch, +) -> None: + storage = Mock() + storage.verify_upload_exists.return_value = {"exists": True} + monkeypatch.setattr(pdf_parser, "JobFileStorage", Mock(return_value=storage)) + + pdf_parser._upload_temp_shard_pdfs( + ["/tmp/shard_0.pdf", "/tmp/shard_1.pdf"], + [ + "tmp/mineru-shards/job_abc/shard_0.pdf", + "tmp/mineru-shards/job_abc/shard_1.pdf", + ], + ) + + assert storage.upload_source_file.call_args_list == [ + call("/tmp/shard_0.pdf", "tmp/mineru-shards/job_abc/shard_0.pdf"), + call("/tmp/shard_1.pdf", "tmp/mineru-shards/job_abc/shard_1.pdf"), + ] + assert storage.verify_upload_exists.call_args_list == [ + call("tmp/mineru-shards/job_abc/shard_0.pdf"), + call("tmp/mineru-shards/job_abc/shard_1.pdf"), + ] + + +def _toc_excluded_profile(tmp_path: Path, job_id: str) -> SimpleNamespace: + return SimpleNamespace( + anatomy=PageAnatomyMap( + job_id=job_id, + file_path=str(tmp_path / "doc.pdf"), + page_count=2, + page_features=[], + page_labels=[], + toc_result=TocResult(toc_pages=[1], method="vlm_batch"), + shard_plan=ShardPlan( + enabled=True, + reason="too_large", + shards=[ + Shard( + shard_index=0, + page_start=1, + page_end=2, + page_offset=0, + anchor_type="toc_leaf_boundary", + anchor_evidence="Intro", + ) + ], + ), + ) + ) + + +def test_shard_pipeline_uploads_before_mineru( + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, +) -> None: + events: list[str] = [] + + def fake_split_pdf(pdf_path, shards, work_dir, exclude_pages=None): + events.append("split") + shard_path = Path(work_dir) / "shard_0.pdf" + shard_path.write_bytes(b"%PDF-1.4") + return [str(shard_path)], None + + class _FakeStorage: + def upload_source_file( + self, local_file_path: str, storage_key: str + ) -> dict[str, str]: + events.append(f"upload:{storage_key}") + assert Path(local_file_path).exists() + return {"etag": "ok"} + + def verify_upload_exists(self, storage_key: str) -> dict[str, bool]: + events.append(f"verify:{storage_key}") + return {"exists": True} + + def delete_upload_file(self, storage_key: str) -> bool: + events.append(f"delete:{storage_key}") + return True + + def fake_parse_via_full(shard_pdf, shard_filename, shard_out, s3_key=None): + events.append(f"parse:{s3_key}") + Path(shard_out).mkdir(parents=True, exist_ok=True) + (Path(shard_out) / "full.md").write_text("# Intro\nBody\n", encoding="utf-8") + + monkeypatch.setattr( + "app.services.document_parser.formats.pdf.shard_splitter.split_pdf", + fake_split_pdf, + ) + monkeypatch.setattr(pdf_parser, "JobFileStorage", _FakeStorage) + monkeypatch.setattr(pdf_parser, "parse_via_full", fake_parse_via_full) + monkeypatch.setattr( + "app.services.document_parser.formats.markdown.parser.eval_md_headings", + lambda md_lines, *args, **kwargs: list(md_lines), + ) + monkeypatch.setattr( + pdf_parser, + "parse_md", + lambda *_args, **kwargs: {"ok": True, "lines": kwargs["lines_with_heading"]}, + ) + monkeypatch.setattr( + "app.services.document_parser.formats.pdf.shard_merger.merge_images", + lambda *_args, **_kwargs: None, + ) + + pdf_parser._parse_pdf_via_shards( + str(tmp_path / "doc.pdf"), + "doc.pdf", + str(tmp_path / "out"), + {"smart_title_parse": False, "model_name": "test-model"}, + profile=_toc_excluded_profile(tmp_path, "job-upload-first"), + s3_key="uploads/job-upload-first.pdf", + job_id="job-upload-first", + ) + + assert events[:4] == [ + "split", + "upload:tmp/mineru-shards/job-upload-first/shard_0.pdf", + "verify:tmp/mineru-shards/job-upload-first/shard_0.pdf", + "parse:tmp/mineru-shards/job-upload-first/shard_0.pdf", + ] + + +def test_failed_shard_upload_does_not_start_mineru( + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, +) -> None: + parse_via_full = Mock() + + def fake_split_pdf(pdf_path, shards, work_dir, exclude_pages=None): + shard_path = Path(work_dir) / "shard_0.pdf" + shard_path.write_bytes(b"%PDF-1.4") + return [str(shard_path)], None + + class _FailingStorage: + def upload_source_file( + self, local_file_path: str, storage_key: str + ) -> dict[str, str]: + raise StorageServiceException( + internal_message="S3 upload failed", + operation="upload_local_file", + ) + + def delete_upload_file(self, storage_key: str) -> bool: + return False + + monkeypatch.setattr( + "app.services.document_parser.formats.pdf.shard_splitter.split_pdf", + fake_split_pdf, + ) + monkeypatch.setattr(pdf_parser, "JobFileStorage", _FailingStorage) + monkeypatch.setattr(pdf_parser, "parse_via_full", parse_via_full) + + with pytest.raises(StorageServiceException): + pdf_parser._parse_pdf_via_shards( + str(tmp_path / "doc.pdf"), + "doc.pdf", + str(tmp_path / "out"), + {"smart_title_parse": False, "model_name": "test-model"}, + profile=_toc_excluded_profile(tmp_path, "job-upload-fail"), + s3_key="uploads/job-upload-fail.pdf", + job_id="job-upload-fail", + ) + + parse_via_full.assert_not_called() + + +def test_unverified_shard_upload_does_not_start_mineru( + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, +) -> None: + parse_via_full = Mock() + + def fake_split_pdf(pdf_path, shards, work_dir, exclude_pages=None): + shard_path = Path(work_dir) / "shard_0.pdf" + shard_path.write_bytes(b"%PDF-1.4") + return [str(shard_path)], None + + class _InvisibleStorage: + def upload_source_file( + self, local_file_path: str, storage_key: str + ) -> dict[str, str]: + return {"etag": "ok"} + + def verify_upload_exists(self, storage_key: str) -> dict[str, bool]: + return {"exists": False} + + def delete_upload_file(self, storage_key: str) -> bool: + return True + + monkeypatch.setattr( + "app.services.document_parser.formats.pdf.shard_splitter.split_pdf", + fake_split_pdf, + ) + monkeypatch.setattr(pdf_parser, "JobFileStorage", _InvisibleStorage) + monkeypatch.setattr(pdf_parser, "parse_via_full", parse_via_full) + + with pytest.raises(StorageServiceException, match="not visible"): + pdf_parser._parse_pdf_via_shards( + str(tmp_path / "doc.pdf"), + "doc.pdf", + str(tmp_path / "out"), + {"smart_title_parse": False, "model_name": "test-model"}, + profile=_toc_excluded_profile(tmp_path, "job-upload-invisible"), + s3_key="uploads/job-upload-invisible.pdf", + job_id="job-upload-invisible", + ) + + parse_via_full.assert_not_called()