afp-backend / tests /unit /test_csv_stream.py
cdupland
Enhance article import process by adding support for optional logistic fields (`moq`, `origin`) and implementing progress tracking with `total_rows` and `rows_processed`. Update related schemas, repositories, and services to accommodate these changes. Improve documentation to reflect new features and usage. Add tests for new functionality and ensure existing tests cover the updated logic.
860c252
from pathlib import Path
import pytest
from app.domain.csv_stream import iter_mapped_csv_rows
from app.schemas.article_import import ArticleImportMapping, ArticleImportOptions
@pytest.mark.unit
def test_csv_stream_mapped_rows(tmp_path: Path) -> None:
p = tmp_path / "t.csv"
p.write_text("Ref,Libelle FR\nR1,Article 1\nR2,Article 2\n", encoding="utf-8")
mapping = ArticleImportMapping(reference_number="Ref", label_fr="Libelle FR")
options = ArticleImportOptions()
rows = list(iter_mapped_csv_rows(p, mapping, options))
r1 = {
"reference_number": "R1",
"label_fr": "Article 1",
"label_en": None,
"category_id": None,
"moq": None,
"origin": None,
}
r2 = {
"reference_number": "R2",
"label_fr": "Article 2",
"label_en": None,
"category_id": None,
"moq": None,
"origin": None,
}
assert rows == [(2, r1), (3, r2)]