From e18cae224b5b9b3a14c55dc7198848244feea690 Mon Sep 17 00:00:00 2001 From: Codex Date: Wed, 9 Sep 2026 20:32:46 -0300 Subject: [PATCH] feat: prepare guided imports --- ingestion/guided_imports.py | 173 ++++++++++++++- ingestion/parsers.py | 26 ++- tests/test_guided_import_workflow.py | 310 +++++++++++++++++++++++++++ tests/test_import_parsers.py | 18 ++ 4 files changed, 513 insertions(+), 14 deletions(-) create mode 100644 tests/test_guided_import_workflow.py diff --git a/ingestion/guided_imports.py b/ingestion/guided_imports.py index 0e2091b..18ef358 100644 --- a/ingestion/guided_imports.py +++ b/ingestion/guided_imports.py @@ -2,13 +2,20 @@ import os import stat -from dataclasses import dataclass +from dataclasses import dataclass, replace from hashlib import sha256 from pathlib import Path from django.conf import settings +from django.core.exceptions import ValidationError -from .parsers import MAX_BYTES, ParseError, parse_document +from finance.models import Account +from finance.services import Actor, DomainError +from finance.services.base import identity + +from . import services +from .models import ImportBatch +from .parsers import MAX_BYTES, ParseError, parse_document, partial_context_from_creation_line @dataclass(frozen=True) @@ -85,24 +92,32 @@ def _open_import_root() -> int | None: return None -def _read_discovered_file(name: str, root_descriptor: int, expected: os.stat_result) -> DiscoveredFile: +def _read_file_bytes(name: str, root_descriptor: int, expected: os.stat_result) -> bytes: descriptor = None try: - flags = os.O_RDONLY | getattr(os, "O_BINARY", 0) + flags = os.O_RDONLY | getattr(os, "O_BINARY", 0) | getattr(os, "O_NONBLOCK", 0) if no_follow := getattr(os, "O_NOFOLLOW", 0): flags |= no_follow descriptor = os.open(name, flags, dir_fd=root_descriptor) opened = os.fstat(descriptor) if not stat.S_ISREG(opened.st_mode) or not os.path.samestat(expected, opened): - return _unreadable(name) + raise ParseError("UNREADABLE_FILE") with os.fdopen(descriptor, "rb", closefd=False) as stream: data = stream.read(MAX_BYTES + 1) except OSError: - return _unreadable(name) + raise ParseError("UNREADABLE_FILE") from None finally: _close(descriptor) if len(data) > MAX_BYTES: - return DiscoveredFile(name, "", None, (), "INPUT_TOO_LARGE") + raise ParseError("INPUT_TOO_LARGE") + return data + + +def _read_discovered_file(name: str, root_descriptor: int, expected: os.stat_result) -> DiscoveredFile: + try: + data = _read_file_bytes(name, root_descriptor, expected) + except ParseError as error: + return DiscoveredFile(name, "", None, (), error.code) document_hash = sha256(data).hexdigest() if Path(name).suffix.casefold() not in _PROFILES_BY_SUFFIX: return DiscoveredFile(name, document_hash, None, (), "UNSUPPORTED_FILE") @@ -134,3 +149,147 @@ def discover_imports() -> tuple[DiscoveredFile, ...]: return tuple(sorted(discovered, key=lambda row: row.name)) finally: _close(root_descriptor) + + +@dataclass(frozen=True) +class PreparedImport: + name: str + document_hash: str + profile: str | None + sections: tuple[str, ...] + state: str + batch_id: str | None + revision: int | None + counts: dict[str, int] + code: str | None + + +_SAFE_CODES = frozenset({ + "UNREADABLE_FILE", "INPUT_TOO_LARGE", "UNSUPPORTED_FILE", "UNRECOGNIZED_LAYOUT", + "FILE_CHANGED", "INVALID_SECTION_MAPPING", "MISSING_CREATION_DATE", "INVALID_CREATION_DATE", + "STALE_REVISION", "STALE_PREVIEW", "BATCH_NOT_FOUND", "BATCH_NOT_PREVIEW", + "REVIEW_REQUIRED", "INVALID_PREPARED_IMPORT", +}) + + +def _verified_bytes(discovered: DiscoveredFile | PreparedImport) -> bytes: + name = discovered.name + if (not isinstance(name, str) or not name or name in {".", ".."} + or any(char in name for char in ("/", "\\", "\x00", ":"))): + raise ParseError("UNREADABLE_FILE") + root_descriptor = _open_import_root() + if root_descriptor is None: + raise ParseError("UNREADABLE_FILE") + try: + expected = os.stat(name, dir_fd=root_descriptor, follow_symlinks=False) + if not stat.S_ISREG(expected.st_mode): + raise ParseError("UNREADABLE_FILE") + data = _read_file_bytes(name, root_descriptor, expected) + except OSError: + raise ParseError("UNREADABLE_FILE") from None + finally: + _close(root_descriptor) + if sha256(data).hexdigest() != discovered.document_hash: + raise ParseError("FILE_CHANGED") + profile, sections = _profile_and_sections(Path(name), data) + if profile != discovered.profile or sections != discovered.sections or profile is None: + raise ParseError("FILE_CHANGED") + return data + + +def _validated_mapping(discovered, section_mapping): + if (not isinstance(section_mapping, dict) or set(section_mapping) != set(discovered.sections) + or any(not isinstance(value, str) for value in section_mapping.values())): + raise DomainError("INVALID_SECTION_MAPPING") + try: + mapping = {key: str(identity(value)) for key, value in section_mapping.items()} + except DomainError: + raise DomainError("INVALID_SECTION_MAPPING") from None + active = set(Account.objects.filter(pk__in=mapping.values(), archived=False, currency="BRL") + .values_list("pk", flat=True)) + if {str(pk) for pk in active} != set(mapping.values()): + raise DomainError("INVALID_SECTION_MAPPING") + return mapping + + +def _cancel_unfinished(batch_id, actor): + # Cancellation must use the latest revision after a stale-review failure. + for _ in range(3): + batch = ImportBatch.objects.filter(pk=batch_id, state="PREVIEW").first() + if batch is None: + return + try: + services.cancel_preview(batch_id=batch.pk, revision=batch.revision, actor=actor) + return + except DomainError as error: + if error.code != "STALE_REVISION": + raise + raise DomainError("CANCELLATION_FAILED") + + +def _attention(prepared, error, actor, *, cancel=True): + code = getattr(error, "code", None) + code = code if isinstance(code, str) and code in _SAFE_CODES else "IMPORT_FAILED" + if cancel and prepared.batch_id: + try: + _cancel_unfinished(prepared.batch_id, actor) + except DomainError: + code = "CANCELLATION_FAILED" + return replace(prepared, state="NEEDS_ATTENTION", counts={"created": 0, "matched": 0, "skipped": 0}, code=code) + + +def prepare_import(*, discovered: DiscoveredFile, section_mapping: dict[str, str], actor: Actor) -> PreparedImport: + """Prepare complete deterministic decisions through the audited lifecycle.""" + prepared = PreparedImport(discovered.name, discovered.document_hash, discovered.profile, + discovered.sections, "NEEDS_ATTENTION", None, None, + {"created": 0, "matched": 0, "skipped": 0}, None) + try: + if discovered.code: + raise ParseError(discovered.code) + data = _verified_bytes(discovered) + mapping = _validated_mapping(discovered, section_mapping) + context = partial_context_from_creation_line(data) if discovered.profile == "OUROCARD_PARTIAL_V1" else {} + batch = services.create_preview(data=data, profile=discovered.profile, + section_mapping=mapping, context=context, actor=actor) + prepared = replace(prepared, batch_id=str(batch.pk), revision=batch.revision) + decisions = [] + counts = {"created": 0, "matched": 0, "skipped": 0} + for item in batch.items.order_by("locator"): + if item.issues == ["ALREADY_IMPORTED"]: + decisions.append({"locator": item.locator, "action": "SKIP", "reason": "ALREADY_IMPORTED"}) + counts["skipped"] += 1 + elif not item.issues and not item.duplicates["existing"] and not item.duplicates["within"]: + decisions.append({"locator": item.locator, "action": "CREATE"}) + counts["created"] += 1 + else: + raise DomainError("REVIEW_REQUIRED") + if not decisions: + raise DomainError("REVIEW_REQUIRED") + batch = services.apply_decisions(batch_id=batch.pk, revision=batch.revision, + decisions=decisions, actor=actor) + return replace(prepared, state="READY" if counts["created"] else "ALREADY_IMPORTED", + revision=batch.revision, counts=counts) + except (DomainError, ParseError, ValidationError) as error: + return _attention(prepared, error, actor) + + +def commit_prepared_import(*, prepared: PreparedImport, actor: Actor) -> PreparedImport: + """Revalidate source identity and delegate the atomic commit to persistence.""" + try: + batch_id = identity(prepared.batch_id) + except DomainError: + return _attention(prepared, DomainError("INVALID_PREPARED_IMPORT"), actor, cancel=False) + try: + batch = ImportBatch.objects.filter(pk=batch_id).first() + if batch is None: + raise DomainError("BATCH_NOT_FOUND") + if batch.document_hash != prepared.document_hash or batch.profile != prepared.profile: + return _attention(prepared, DomainError("INVALID_PREPARED_IMPORT"), actor, cancel=False) + if prepared.state not in {"READY", "ALREADY_IMPORTED", "COMMITTED"} or type(prepared.revision) is not int: + raise DomainError("INVALID_PREPARED_IMPORT") + _verified_bytes(prepared) + batch = services.commit_preview(batch_id=batch.pk, revision=prepared.revision, actor=actor) + return replace(prepared, state="COMMITTED", revision=batch.revision, + counts={key: batch.result[key] for key in ("created", "matched", "skipped")}, code=None) + except (DomainError, ParseError, ValidationError) as error: + return _attention(prepared, error, actor) diff --git a/ingestion/parsers.py b/ingestion/parsers.py index a2bc0c3..4b8a011 100644 --- a/ingestion/parsers.py +++ b/ingestion/parsers.py @@ -94,6 +94,14 @@ def _txt_items(text, profile, skipped): section, section_count = '1', 0 for number, line in enumerate(text.splitlines(), 1): locator = f'line:{number}' + if partial and number == 21: + try: + _creation_date(line) + except ParseError: + pass + else: + skipped['metadata'] += 1 + continue if not line.strip(): skipped['blank'] += 1 elif re.match(r'^\s*\d\s*-\s*[A-Z]', line): @@ -128,19 +136,23 @@ def _decode_source(data: bytes) -> tuple[str, str]: return 'latin-1', data.decode('latin-1') +def _creation_date(line: str) -> date: + match = re.fullmatch(r'DATA\s+DE\s+EMISS(?:A|Ã)O\s*:\s*(\d{2}/\d{2}/\d{4})', line.strip(), re.I) + if match is None: + raise ParseError('INVALID_CREATION_DATE') + try: + return date.fromisoformat('-'.join(reversed(match[1].split('/')))) + except ValueError: + raise ParseError('INVALID_CREATION_DATE') from None + + def partial_context_from_creation_line(data: bytes) -> dict[str, str]: """Build the bounded partial-statement interval from its line-21 issue date.""" _, text = _decode_source(data) lines = text.splitlines() if len(lines) < 21: raise ParseError('MISSING_CREATION_DATE') - match = re.fullmatch(r'DATA\s+DE\s+EMISS(?:A|Ã)O\s*:\s*(\d{2}/\d{2}/\d{4})', lines[20].strip(), re.I) - if match is None: - raise ParseError('INVALID_CREATION_DATE') - try: - creation = date.fromisoformat('-'.join(reversed(match[1].split('/')))) - except ValueError: - raise ParseError('INVALID_CREATION_DATE') from None + creation = _creation_date(lines[20]) start_year = creation.year if creation.month == 12 else creation.year - 1 start_month = creation.month % 12 + 1 start = date(start_year, start_month, 1) diff --git a/tests/test_guided_import_workflow.py b/tests/test_guided_import_workflow.py new file mode 100644 index 0000000..1dd960c --- /dev/null +++ b/tests/test_guided_import_workflow.py @@ -0,0 +1,310 @@ +"""Synthetic integration checks for deterministic guided imports.""" + +from dataclasses import asdict, replace +from datetime import date +from uuid import uuid4 + +import pytest + +from finance.models import Account, AuditEvent, SourceRecord, Transaction +from finance.services import Actor, DomainError, archive, create_account, record_transaction +from ingestion import guided_imports, services +from ingestion.models import ImportBatch + +pytestmark = pytest.mark.django_db(transaction=True) +ACTOR = Actor(kind="HUMAN", identifier="synthetic-guided-owner") +HEADER = "Data,Documento,Historico,Agencia,Lote,Valor\n" +ROW = "15/01/2026,1,Synthetic purchase,0,0,-12.50\n" +DATA = (HEADER + ROW + "16/01/2026,2,Synthetic second purchase,0,0,-7.25\n").encode() + + +@pytest.fixture +def bank(): + return create_account(name="Synthetic guided bank", kind="BANK", actor=ACTOR) + + +@pytest.fixture +def source(tmp_path, settings): + settings.IMPORT_ROOT = tmp_path + + def write(data=DATA, name="bank.csv"): + (tmp_path / name).write_bytes(data) + return next(row for row in guided_imports.discover_imports() if row.name == name) + + return write + + +def prepare(discovered, bank): + return guided_imports.prepare_import(discovered=discovered, section_mapping={"1": str(bank.pk)}, actor=ACTOR) + + +def commit(prepared): + return guided_imports.commit_prepared_import(prepared=prepared, actor=ACTOR) + + +def assert_no_open_preview(): + assert not ImportBatch.objects.filter(state="PREVIEW").exists() + + +def test_clean_preflight_is_inert_and_complete_commit_preserves_audit_and_provenance(source, bank): + discovered = source() + prepared = prepare(discovered, bank) + assert prepared.state == "READY" + assert prepared.code is None + assert prepared.document_hash == discovered.document_hash + assert prepared.counts == {"created": 2, "matched": 0, "skipped": 0} + assert not Transaction.objects.exists() + assert not SourceRecord.objects.exists() + batch = ImportBatch.objects.get(pk=prepared.batch_id) + assert batch.revision == prepared.revision == 2 + assert list(batch.items.values_list("decision", flat=True)) == [{"action": "CREATE"}] * 2 + result = commit(prepared) + assert result.state == "COMMITTED" + assert result.counts == {"created": 2, "matched": 0, "skipped": 0} + assert Transaction.objects.count() == SourceRecord.objects.count() == 2 + assert set(SourceRecord.objects.values_list("document_hash", flat=True)) == {discovered.document_hash} + assert set(AuditEvent.objects.filter(operation__startswith="import_").values_list("actor_identifier", flat=True)) == {ACTOR.identifier} + assert "Synthetic purchase" not in str(asdict(result)) + assert_no_open_preview() + + +def test_exact_second_import_only_skips_and_commit_is_idempotent(source, bank): + discovered = source() + first = prepare(discovered, bank) + assert commit(first).counts["created"] == 2 + assert commit(first).counts["created"] == 2 + second = prepare(discovered, bank) + assert second.state == "ALREADY_IMPORTED" + assert second.counts == {"created": 0, "matched": 0, "skipped": 2} + assert all(item.decision == {"action": "SKIP", "reason": "ALREADY_IMPORTED"} + for item in ImportBatch.objects.get(pk=second.batch_id).items.all()) + assert commit(second).counts["skipped"] == 2 + assert Transaction.objects.count() == SourceRecord.objects.count() == 2 + assert_no_open_preview() + + +def test_ambiguous_existing_duplicate_cancels_whole_file(source, bank): + commit(prepare(source(), bank)) + ambiguous = prepare(source(DATA.replace(b",1,", b",3,")), bank) + assert ambiguous.state == "NEEDS_ATTENTION" + assert ambiguous.code == "REVIEW_REQUIRED" + assert Transaction.objects.count() == 2 + assert_no_open_preview() + + +@pytest.mark.parametrize("data", [ + (HEADER + ROW + ROW).encode(), + (HEADER + ROW + "31/02/2026,2,Synthetic invalid date,0,0,-2.00\n").encode(), + (HEADER + ROW + "synthetic malformed row\n").encode(), +]) +def test_uncertain_or_malformed_rows_cancel_without_any_decisions(source, bank, data): + prepared = prepare(source(data), bank) + assert prepared.state == "NEEDS_ATTENTION" + assert prepared.code == "REVIEW_REQUIRED" + assert not Transaction.objects.exists() + assert all(not item.decision for item in ImportBatch.objects.get().items.all()) + assert_no_open_preview() + + +def test_exceptional_file_does_not_block_an_unrelated_ready_file(source, bank): + exceptional = source((HEADER + ROW + ROW).encode(), "ambiguous.csv") + ready = source() + results = [prepare(row, bank) for row in (exceptional, ready)] + assert [row.state for row in results] == ["NEEDS_ATTENTION", "READY"] + assert commit(results[1]).counts["created"] == 2 + assert_no_open_preview() + + +@pytest.mark.parametrize("mapping", [{}, {"2": "ACCOUNT"}, {"1": "ACCOUNT", "2": "ACCOUNT"}, + {"1": None}, {"1": "not-an-id"}, {"1": str(uuid4())}]) +def test_bad_mapping_is_rejected_before_preview(source, bank, mapping): + mapping = {key: str(bank.pk) if value == "ACCOUNT" else value for key, value in mapping.items()} + prepared = guided_imports.prepare_import(discovered=source(), section_mapping=mapping, actor=ACTOR) + assert prepared.state == "NEEDS_ATTENTION" + assert prepared.code == "INVALID_SECTION_MAPPING" + assert not ImportBatch.objects.exists() + + +@pytest.mark.parametrize("change", ["archived", "currency"]) +def test_only_active_brl_accounts_can_be_mapped(source, bank, change): + if change == "archived": + archive(bank, actor=ACTOR) + else: + Account.objects.filter(pk=bank.pk).update(currency="USD") + prepared = prepare(source(), bank) + assert prepared.state == "NEEDS_ATTENTION" + assert prepared.code == "INVALID_SECTION_MAPPING" + assert not ImportBatch.objects.exists() + + +def test_wrong_account_kind_requires_attention_and_cancels(source): + card = create_account(name="Synthetic card", kind="CREDIT_CARD", actor=ACTOR) + prepared = prepare(source(), card) + assert prepared.state == "NEEDS_ATTENTION" + assert_no_open_preview() + + +@pytest.mark.parametrize("stage", ["prepare", "commit"]) +def test_changed_file_cannot_use_scan_hash(source, bank, tmp_path, stage): + discovered = source() + prepared = prepare(discovered, bank) if stage == "commit" else None + (tmp_path / discovered.name).write_bytes(DATA.replace(b"-12.50", b"-13.50")) + result = commit(prepared) if prepared else prepare(discovered, bank) + assert result.state == "NEEDS_ATTENTION" + assert result.code == "FILE_CHANGED" + assert not Transaction.objects.exists() + assert_no_open_preview() + + +@pytest.mark.parametrize("name", ["../outside.csv", "/outside.csv", "nested/file.csv", "..\\outside.csv"]) +def test_tampered_filename_is_rejected(source, bank, name): + result = prepare(replace(source(), name=name), bank) + assert result.state == "NEEDS_ATTENTION" + assert result.code == "UNREADABLE_FILE" + assert not ImportBatch.objects.exists() + + +def test_symlink_replacement_is_rejected_before_commit(source, bank, tmp_path): + discovered = source() + prepared = prepare(discovered, bank) + original = tmp_path / discovered.name + other = tmp_path / "other.csv" + other.write_bytes(DATA) + original.unlink() + original.symlink_to(other) + result = commit(prepared) + assert result.state == "NEEDS_ATTENTION" + assert result.code == "UNREADABLE_FILE" + assert not Transaction.objects.exists() + assert_no_open_preview() + + +@pytest.mark.parametrize("field,value", [("profile", "OUROCARD_FULL_V1"), ("sections", ("2",))]) +def test_altered_scan_metadata_is_rejected(source, bank, field, value): + result = prepare(replace(source(), **{field: value}), bank) + assert result.state == "NEEDS_ATTENTION" + assert result.code == "FILE_CHANGED" + assert not ImportBatch.objects.exists() + + +def test_stale_revision_cancels_unfinished_preview(source, bank): + prepared = prepare(source(), bank) + services.refresh_preview(batch_id=prepared.batch_id, revision=prepared.revision, actor=ACTOR) + result = commit(prepared) + assert result.state == "NEEDS_ATTENTION" + assert result.code == "STALE_REVISION" + assert not Transaction.objects.exists() + assert_no_open_preview() + + +def test_new_duplicate_after_preparation_requires_new_review(source, bank): + prepared = prepare(source(), bank) + record_transaction(account_id=bank.pk, amount="-12.50", kind="EXPENSE", + posting_date=date(2026, 1, 15), occurrence_date=date(2026, 1, 15), + description="Synthetic purchase", actor=ACTOR) + result = commit(prepared) + assert result.state == "NEEDS_ATTENTION" + assert result.code == "STALE_PREVIEW" + assert Transaction.objects.count() == 1 + assert not SourceRecord.objects.exists() + assert_no_open_preview() + + +def test_domain_failure_is_safe_and_rolls_back_then_cancels(source, bank, monkeypatch): + prepared = prepare(source(), bank) + original = services.record + calls = 0 + + def fail_second(**kwargs): + nonlocal calls + calls += 1 + if calls == 2: + raise DomainError("Synthetic source text must never be disclosed") + return original(**kwargs) + + monkeypatch.setattr(services, "record", fail_second) + result = commit(prepared) + assert result.state == "NEEDS_ATTENTION" + assert result.code == "IMPORT_FAILED" + assert not Transaction.objects.exists() + assert not SourceRecord.objects.exists() + assert "Synthetic source text" not in str(asdict(result)) + assert_no_open_preview() + + +def test_multi_section_card_creates_only_after_every_mapping_is_complete(source): + first = create_account(name="Synthetic first card", kind="CREDIT_CARD", actor=ACTOR) + second = create_account(name="Synthetic second card", kind="CREDIT_CARD", actor=ACTOR) + data = ("1 - SYNTHETIC FIRST\n" + f"15.01.2026{'Synthetic first':37}BR 12,50 0,00\n" + + "2 - SYNTHETIC SECOND\n" + f"16.01.2026{'Synthetic second':37}BR 7,25 0,00\n").encode() + discovered = source(data, "cards.txt") + result = guided_imports.prepare_import(discovered=discovered, + section_mapping={"1": str(first.pk), "2": str(second.pk)}, actor=ACTOR) + assert result.state == "READY" + assert commit(result).counts["created"] == 2 + assert set(Transaction.objects.values_list("account_id", flat=True)) == {first.pk, second.pk} + + +def test_partial_missing_context_has_safe_code_and_no_preview(source): + card = create_account(name="Synthetic partial card", kind="CREDIT_CARD", actor=ACTOR) + discovered = source(f"31/12{'Synthetic card row':43}BR 12,50 0,00\n".encode(), "partial.txt") + result = prepare(discovered, card) + assert result.state == "NEEDS_ATTENTION" + assert result.code == "MISSING_CREATION_DATE" + assert not ImportBatch.objects.exists() + + +def test_partial_valid_context_resolves_new_year_and_commits(source): + card = create_account(name="Synthetic partial card", kind="CREDIT_CARD", actor=ACTOR) + data = ("\n" * 20 + "Data de emissão: 20/01/2026\n" + + f"31/12{'Synthetic prior year':43}BR 12,50 0,00\n" + + f"05/01{'Synthetic current year':43}BR 7,25 0,00\n").encode() + result = prepare(source(data, "partial.txt"), card) + assert result.state == "READY" + batch = ImportBatch.objects.get(pk=result.batch_id) + assert batch.context == {"start_date": "2025-02-01", "end_date": "2026-01-31"} + assert commit(result).counts["created"] == 2 + assert set(Transaction.objects.values_list("posting_date", flat=True)) == {date(2025, 12, 31), date(2026, 1, 5)} + + +def test_tampered_batch_identifier_returns_safe_result_without_cancelling_another_batch(source, bank): + first = prepare(source(), bank) + different = prepare(source(DATA.replace(b"-12.50", b"-13.50"), "second.csv"), bank) + result = commit(replace(first, batch_id=different.batch_id)) + assert result.state == "NEEDS_ATTENTION" + assert result.code == "INVALID_PREPARED_IMPORT" + assert ImportBatch.objects.get(pk=different.batch_id).state == "PREVIEW" + assert not Transaction.objects.exists() + + +def test_invalid_batch_identifier_is_safe(source, bank): + result = commit(replace(prepare(source(), bank), batch_id="invalid-id")) + assert result.state == "NEEDS_ATTENTION" + assert result.code == "INVALID_PREPARED_IMPORT" + + +def test_altered_prepared_source_cannot_cancel_unrelated_preview(source, bank, tmp_path): + first = prepare(source(), bank) + different = prepare(source(DATA.replace(b"-12.50", b"-13.50"), "second.csv"), bank) + (tmp_path / first.name).unlink() + result = commit(replace(first, batch_id=different.batch_id)) + assert result.state == "NEEDS_ATTENTION" + assert result.code == "INVALID_PREPARED_IMPORT" + assert ImportBatch.objects.get(pk=different.batch_id).state == "PREVIEW" + + +def test_partially_imported_source_skips_exact_rows_and_creates_only_new_rows(source, bank): + discovered = source() + batch = services.create_preview(data=DATA, profile=discovered.profile, + section_mapping={"1": str(bank.pk)}, actor=ACTOR) + first, second = batch.items.order_by("locator") + batch = services.apply_decisions(batch_id=batch.pk, revision=batch.revision, actor=ACTOR, decisions=[ + {"locator": first.locator, "action": "CREATE"}, + {"locator": second.locator, "action": "SKIP", "reason": "USER_EXCLUDED"}, + ]) + services.commit_preview(batch_id=batch.pk, revision=batch.revision, actor=ACTOR) + prepared = prepare(discovered, bank) + assert prepared.state == "READY" + assert commit(prepared).counts == {"created": 1, "matched": 0, "skipped": 1} + assert Transaction.objects.count() == SourceRecord.objects.count() == 2 + assert_no_open_preview() diff --git a/tests/test_import_parsers.py b/tests/test_import_parsers.py index 688a17b..c873030 100644 --- a/tests/test_import_parsers.py +++ b/tests/test_import_parsers.py @@ -164,3 +164,21 @@ def test_synthetic_fixture_profiles(filename, profile, amounts): rows = normalize(parse_document(data, profile, {'start_date':'2025-12-01','end_date':'2026-01-31'})) assert [row.amount for row in rows] == amounts assert all(not row.errors for row in rows) + + +@pytest.mark.parametrize("line,position,profile,skipped", [ + ("Data de emissão: 20/01/2026", 21, "OUROCARD_PARTIAL_V1", True), + ("Data de emissão: 31/02/2026", 21, "OUROCARD_PARTIAL_V1", False), + ("Data de emissão: 20/01/2026 trailing text", 21, "OUROCARD_PARTIAL_V1", False), + ("Data de emissão: 20/01/2026", 20, "OUROCARD_PARTIAL_V1", False), + ("Data de emissão: 20/01/2026", 21, "OUROCARD_FULL_V1", False), +]) +def test_only_valid_line_21_partial_creation_metadata_is_excluded(line, position, profile, skipped): + data = ("\n" * (position - 1) + line + "\n").encode() + parsed = parse_document(data, profile) + if skipped: + assert parsed.items == () + assert parsed.skipped["metadata"] == 1 + else: + assert len(parsed.items) == 1 + assert parsed.items[0].errors == ("UNRECOGNIZED_LINE",)