You can not select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
264 lines
11 KiB
264 lines
11 KiB
"""Synthetic acceptance cases for explicit review and atomic import persistence.""" |
|
|
|
from datetime import date |
|
from decimal import Decimal |
|
from hashlib import sha256 |
|
|
|
import pytest |
|
|
|
from finance.models import AuditEvent, InstallmentPlan, SourceRecord, Transaction, Transfer |
|
from finance.services import Actor, DomainError, archive, create_account, create_category, create_payee, record_transaction |
|
from ingestion import services |
|
|
|
pytestmark = pytest.mark.django_db(transaction=True) |
|
ACTOR = Actor(kind="AGENT", identifier="synthetic-import-test") |
|
HEADER = "Data,Documento,Historico,Agencia,Lote,Valor\n" |
|
ROW = "15/01/2026,1,Synthetic purchase,0,0,-12.50\n" |
|
DATA = (HEADER + ROW).encode() |
|
|
|
|
|
def account(name="Synthetic import bank"): |
|
return create_account(name=name, kind="BANK", actor=ACTOR) |
|
|
|
|
|
def preview(bank, data=DATA): |
|
return services.create_preview(data=data, profile="BANK_CSV_V1", section_mapping={"1": str(bank.pk)}, actor=ACTOR) |
|
|
|
|
|
def decide(batch, **fields): |
|
return services.apply_decisions(batch_id=batch.pk, revision=batch.revision, actor=ACTOR, |
|
decisions=[{"locator": item.locator, "action": "CREATE", **fields} |
|
for item in batch.items.all()]) |
|
|
|
|
|
def commit(batch): |
|
return services.commit_preview(batch_id=batch.pk, revision=batch.revision, actor=ACTOR) |
|
|
|
|
|
def assert_no_financial_rows(): |
|
for model in (Transaction, SourceRecord, InstallmentPlan, Transfer): |
|
assert model.objects.count() == 0 |
|
|
|
|
|
def test_preview_and_decisions_are_inert_then_commit_preserves_source_and_audit(): |
|
batch = preview(account()) |
|
assert batch.state == "PREVIEW" |
|
assert batch.revision == 1 |
|
assert_no_financial_rows() |
|
review = services.export_review(batch_id=batch.pk) |
|
assert review["batch_id"] == str(batch.pk) |
|
assert review["revision"] == 1 |
|
assert len(review["items"]) == 1 |
|
assert review["items"][0]["candidate"]["amount"] == "-12.50" |
|
batch = decide(batch) |
|
assert batch.revision == 2 |
|
assert_no_financial_rows() |
|
previous_audits = set(AuditEvent.objects.values_list("pk", flat=True)) |
|
result = commit(batch) |
|
assert result.state == "COMMITTED" |
|
assert result.revision == batch.revision |
|
row = Transaction.objects.get() |
|
assert row.amount == Decimal("-12.50") |
|
assert row.posting_date == row.occurrence_date == date(2026, 1, 15) |
|
source = SourceRecord.objects.get() |
|
assert source.transaction_id == row.pk |
|
assert source.document_hash == sha256(DATA).hexdigest() |
|
assert source.item_locator == batch.items.get().locator |
|
events = list(AuditEvent.objects.exclude(pk__in=previous_audits)) |
|
assert len(events) == 1 |
|
assert events[0].outcome == "SUCCESS" |
|
assert events[0].actor_identifier == ACTOR.identifier |
|
audit_count = AuditEvent.objects.count() |
|
repeated = commit(batch) |
|
assert repeated.result == result.result |
|
assert Transaction.objects.count() == SourceRecord.objects.count() == 1 |
|
assert AuditEvent.objects.count() == audit_count |
|
|
|
|
|
def test_every_row_requires_explicit_decision_and_rejected_commit_is_audited(): |
|
batch = preview(account()) |
|
previous_audits = set(AuditEvent.objects.values_list("pk", flat=True)) |
|
with pytest.raises(DomainError) as caught: |
|
commit(batch) |
|
assert caught.value.code |
|
assert_no_financial_rows() |
|
batch.refresh_from_db() |
|
assert batch.state == "PREVIEW" |
|
event = AuditEvent.objects.exclude(pk__in=previous_audits).get() |
|
assert event.outcome == "FAILURE" |
|
assert event.error_code == caught.value.code |
|
assert event.actor_identifier == ACTOR.identifier |
|
|
|
|
|
def test_malformed_row_can_only_be_omitted_with_explicit_skip(): |
|
batch = preview(account(), (HEADER + "31/02/2026,1,Synthetic invalid date,0,0,-12.50\n").encode()) |
|
assert batch.items.get().issues |
|
with pytest.raises(DomainError): |
|
commit(batch) |
|
batch = decide(batch, action="SKIP", reason="MALFORMED") |
|
assert commit(batch).state == "COMMITTED" |
|
assert_no_financial_rows() |
|
|
|
|
|
def test_exact_reimport_requires_already_imported_skip(): |
|
bank = account() |
|
commit(decide(preview(bank))) |
|
batch = preview(bank) |
|
assert "ALREADY_IMPORTED" in str(batch.items.get().duplicates) |
|
with pytest.raises(DomainError): |
|
commit(batch) |
|
batch = decide(batch, action="SKIP", reason="ALREADY_IMPORTED") |
|
assert commit(batch).state == "COMMITTED" |
|
assert Transaction.objects.count() == SourceRecord.objects.count() == 1 |
|
|
|
|
|
def test_semantic_duplicates_require_acknowledgement_but_can_be_legitimate(): |
|
bank = account() |
|
commit(decide(preview(bank))) |
|
# The document ID changes the physical source without changing financial fields. |
|
batch = preview(bank, DATA.replace(b",1,", b",2,")) |
|
assert batch.items.get().duplicates |
|
batch = decide(batch) |
|
with pytest.raises(DomainError): |
|
commit(batch) |
|
batch = decide(batch, acknowledge_duplicate=True) |
|
commit(batch) |
|
assert Transaction.objects.count() == SourceRecord.objects.count() == 2 |
|
|
|
|
|
def test_match_attaches_provenance_without_changing_existing_annotations(): |
|
bank = account() |
|
category = create_category(name="Synthetic category", kind="EXPENSE", actor=ACTOR) |
|
payee = create_payee(name="Synthetic payee", actor=ACTOR) |
|
row = record_transaction(account_id=bank.pk, amount=Decimal("-12.50"), kind="EXPENSE", |
|
posting_date=date(2026, 1, 15), occurrence_date=date(2026, 1, 15), |
|
description="Synthetic purchase", category_id=category.pk, |
|
payee_id=payee.pk, actor=ACTOR) |
|
before = (row.description, row.category_id, row.payee_id, row.updated_at) |
|
batch = decide(preview(bank), action="MATCH", transaction_id=str(row.pk)) |
|
commit(batch) |
|
row.refresh_from_db() |
|
assert (row.description, row.category_id, row.payee_id, row.updated_at) == before |
|
assert Transaction.objects.count() == 1 |
|
assert SourceRecord.objects.get().transaction_id == row.pk |
|
|
|
|
|
def test_cancellation_and_stale_revision_cannot_commit(): |
|
batch = preview(account()) |
|
old_revision = batch.revision |
|
batch = decide(batch) |
|
with pytest.raises(DomainError): |
|
services.commit_preview(batch_id=batch.pk, revision=old_revision, actor=ACTOR) |
|
batch = services.cancel_preview(batch_id=batch.pk, revision=batch.revision, actor=ACTOR) |
|
assert batch.state == "CANCELLED" |
|
with pytest.raises(DomainError): |
|
commit(batch) |
|
assert_no_financial_rows() |
|
|
|
|
|
def test_new_duplicate_makes_review_stale_and_refresh_requires_new_consent(): |
|
bank = account() |
|
batch = decide(preview(bank)) |
|
record_transaction(account_id=bank.pk, amount=Decimal("-12.50"), kind="EXPENSE", |
|
posting_date=date(2026, 1, 15), occurrence_date=date(2026, 1, 15), |
|
description="Synthetic purchase", actor=ACTOR) |
|
with pytest.raises(DomainError) as caught: |
|
commit(batch) |
|
assert caught.value.code == "STALE_PREVIEW" |
|
assert SourceRecord.objects.count() == 0 |
|
refreshed = services.refresh_preview(batch_id=batch.pk, revision=batch.revision, actor=ACTOR) |
|
assert refreshed.revision == batch.revision + 1 |
|
assert all(not item.decision for item in refreshed.items.all()) |
|
with pytest.raises(DomainError): |
|
commit(refreshed) |
|
commit(decide(refreshed, acknowledge_duplicate=True)) |
|
assert Transaction.objects.count() == 2 |
|
assert SourceRecord.objects.count() == 1 |
|
|
|
|
|
def test_archiving_destination_invalidates_review(): |
|
bank = account() |
|
batch = decide(preview(bank)) |
|
archive(bank, actor=ACTOR) |
|
with pytest.raises(DomainError) as caught: |
|
commit(batch) |
|
assert caught.value.code == "STALE_PREVIEW" |
|
assert_no_financial_rows() |
|
|
|
|
|
def test_later_row_failure_rolls_back_posting_sources_and_success_audit(monkeypatch): |
|
batch = decide(preview(account(), (HEADER + ROW + "16/01/2026,2,Synthetic second purchase,0,0,-7.25\n").encode())) |
|
original = services.record |
|
calls = [] |
|
|
|
def fail_second(**kwargs): |
|
calls.append(kwargs) |
|
if len(calls) == 2: |
|
assert Transaction.objects.count() == SourceRecord.objects.count() == 1 |
|
raise DomainError("synthetic_later_row_failure") |
|
return original(**kwargs) |
|
|
|
monkeypatch.setattr(services, "record", fail_second) |
|
previous_audits = set(AuditEvent.objects.values_list("pk", flat=True)) |
|
with pytest.raises(DomainError) as caught: |
|
commit(batch) |
|
assert len(calls) == 2 |
|
assert caught.value.code == "synthetic_later_row_failure" |
|
assert_no_financial_rows() |
|
batch.refresh_from_db() |
|
assert batch.state == "PREVIEW" |
|
event = AuditEvent.objects.exclude(pk__in=previous_audits).get() |
|
assert event.outcome == "FAILURE" |
|
assert event.error_code == caught.value.code |
|
|
|
|
|
@pytest.mark.parametrize("extra", [{"amount": "-1.00"}, {"description": "Replacement"}, {"unexpected": True}]) |
|
def test_decisions_cannot_change_source_evidence_or_accept_unknown_fields(extra): |
|
batch = preview(account()) |
|
with pytest.raises(DomainError): |
|
decide(batch, **extra) |
|
assert_no_financial_rows() |
|
|
|
|
|
def test_duplicate_locators_are_rejected(): |
|
batch = preview(account()) |
|
decision = {"locator": batch.items.get().locator, "action": "CREATE"} |
|
with pytest.raises(DomainError): |
|
services.apply_decisions(batch_id=batch.pk, revision=batch.revision, decisions=[decision, decision], actor=ACTOR) |
|
assert_no_financial_rows() |
|
|
|
|
|
def test_installments_share_explicit_new_plan_without_creating_unobserved_rows(): |
|
card = create_account(name="Synthetic installment card", kind="CREDIT_CARD", actor=ACTOR) |
|
data = (f"15.01.2026{'Synthetic purchase PARC 2/6':37}BR 12,50 0,00\n" |
|
f"15.02.2026{'Synthetic purchase PARC 3/6':37}BR 12,50 0,00\n").encode() |
|
batch = services.create_preview(data=data, profile="OUROCARD_FULL_V1", |
|
section_mapping={"1": str(card.pk)}, actor=ACTOR) |
|
batch = decide(batch, new_plan="synthetic-purchase") |
|
assert_no_financial_rows() |
|
commit(batch) |
|
plan = InstallmentPlan.objects.get() |
|
assert plan.account_id == card.pk |
|
assert plan.total_installments == 6 |
|
assert set(Transaction.objects.values_list("installment_number", flat=True)) == {2, 3} |
|
assert set(Transaction.objects.values_list("installment_plan_id", flat=True)) == {plan.pk} |
|
assert SourceRecord.objects.count() == 2 |
|
|
|
|
|
def test_applying_decisions_replaces_complete_set_and_missing_rows_block_commit(): |
|
batch = decide(preview(account(), (HEADER + ROW + "16/01/2026,2,Synthetic second purchase,0,0,-7.25\n").encode())) |
|
first = batch.items.order_by("locator").first() |
|
batch = services.apply_decisions(batch_id=batch.pk, revision=batch.revision, actor=ACTOR, |
|
decisions=[{"locator": first.locator, "action": "CREATE"}]) |
|
assert not batch.items.exclude(pk=first.pk).get().decision |
|
with pytest.raises(DomainError): |
|
commit(batch) |
|
assert_no_financial_rows() |
|
|
|
|
|
def test_source_already_attached_to_another_account_cannot_be_claimed(): |
|
commit(decide(preview(account()))) |
|
batch = decide(preview(account("Synthetic second import bank"))) |
|
with pytest.raises(DomainError): |
|
commit(batch) |
|
assert Transaction.objects.count() == SourceRecord.objects.count() == 1
|
|
|