27a1f22be4
UploadedFile already holds the whole attachment in memory; getvalue() made a second full copy of each of the 230 files before writing. Copy through the file interface instead. Smoke tests pin the progress callback contract: once per XML file, including unreadable ones. Refs #1 Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
60 lines
2.0 KiB
Python
60 lines
2.0 KiB
Python
from __future__ import annotations
|
|
|
|
import zipfile
|
|
from pathlib import Path
|
|
|
|
from app.pipeline import service
|
|
from app.pipeline.service import run_batch
|
|
from tests.smoke.test_batch_smoke import _patch_opencloud, _valid_xml
|
|
|
|
|
|
def test_progress_callback_is_called_once_per_xml_file(
|
|
tmp_path: Path, monkeypatch
|
|
) -> None:
|
|
_patch_opencloud(monkeypatch)
|
|
zip_path = tmp_path / "batch.zip"
|
|
names = [f"SKO115FZ_01_123456789_20260616_X0000{i}.xml" for i in range(1, 4)]
|
|
with zipfile.ZipFile(zip_path, "w") as archive:
|
|
for name in names:
|
|
archive.writestr(name, _valid_xml())
|
|
loose = tmp_path / "SKO115FZ_01_123456789_20260616_X00009.xml"
|
|
loose.write_text(_valid_xml(), encoding="utf-8")
|
|
|
|
calls: list[tuple[int, int, str]] = []
|
|
result = run_batch(
|
|
input_paths=[zip_path, loose],
|
|
launcher="tester",
|
|
opencloud_username="test-user",
|
|
opencloud_access_token="token-123",
|
|
progress_callback=lambda done, total, name: calls.append((done, total, name)),
|
|
)
|
|
|
|
assert result.status == "success"
|
|
assert [done for done, _, _ in calls] == [1, 2, 3, 4]
|
|
assert {total for _, total, _ in calls} == {4}
|
|
assert sorted(name for _, _, name in calls) == sorted([*names, loose.name])
|
|
|
|
|
|
def test_progress_callback_is_called_for_unreadable_file_too(
|
|
tmp_path: Path, monkeypatch
|
|
) -> None:
|
|
_patch_opencloud(monkeypatch)
|
|
loose = tmp_path / "SKO115FZ_01_123456789_20260616_X00001.xml"
|
|
loose.write_text(_valid_xml(), encoding="utf-8")
|
|
|
|
def _boom(self: Path) -> bytes:
|
|
raise OSError("disk gone")
|
|
|
|
monkeypatch.setattr(service.Path, "read_bytes", _boom)
|
|
calls: list[tuple[int, int, str]] = []
|
|
result = run_batch(
|
|
input_paths=[loose],
|
|
launcher="tester",
|
|
opencloud_username="test-user",
|
|
opencloud_access_token="token-123",
|
|
progress_callback=lambda done, total, name: calls.append((done, total, name)),
|
|
)
|
|
|
|
assert result.status == "error"
|
|
assert calls == [(1, 1, loose.name)]
|