Spaces:
Sleeping
Sleeping
File size: 2,147 Bytes
96d8d92 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 | from pathlib import Path
from types import SimpleNamespace
from unittest.mock import AsyncMock
import pytest
from app.repositories.article_bulk_repository import REFERENCE_MAIN_CONFLICT
from app.repositories.import_repository import ArticleUpsertRecord
from app.services.article_import_processor import ArticleImportProcessor
@pytest.mark.unit
@pytest.mark.asyncio
async def test_process_records_reference_rename_conflict(tmp_path: Path) -> None:
csv_path = tmp_path / "job.csv"
csv_path.write_text(
"Ref,New\nR1,R2\n",
encoding="utf-8",
)
import_id = "test-import-id"
object_key = f"catalogues/{import_id}.csv"
job_row = SimpleNamespace(
type="articles",
mapping={"reference_number": "Ref", "reference_new": "New"},
options={"has_header": True, "delimiter": ",", "encoding": "utf-8"},
path=object_key,
)
import_repo = AsyncMock()
import_repo.get_by_id.return_value = job_row
error_repo = AsyncMock()
bulk_repo = AsyncMock()
bulk_repo.upsert_batch.return_value = [(2, REFERENCE_MAIN_CONFLICT)]
async def fake_download(_key: str, iid: str) -> Path:
dest = tmp_path / f"{iid}.csv"
dest.write_bytes(csv_path.read_bytes())
return dest
file_storage = AsyncMock()
file_storage.download_to_temp.side_effect = fake_download
processor = ArticleImportProcessor(import_repo, error_repo, bulk_repo, file_storage)
await processor.process(import_id)
bulk_repo.upsert_batch.assert_awaited_once()
call_records = bulk_repo.upsert_batch.await_args[0][0]
assert len(call_records) == 1
assert isinstance(call_records[0], ArticleUpsertRecord)
assert call_records[0].reference_new == "R2"
error_repo.create_many.assert_awaited()
created_errors = error_repo.create_many.await_args[0][0]
assert any(e.error_code == REFERENCE_MAIN_CONFLICT for e in created_errors)
import_repo.set_finished.assert_awaited_once()
_args, kwargs = import_repo.set_finished.call_args
assert kwargs["status"] == "failed"
assert kwargs["row_succeeded"] == 0
assert kwargs["row_failed"] == 1
|