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.
316 lines
15 KiB
316 lines
15 KiB
"""Synthetic acceptance cases for the public, audited write boundary.""" |
|
|
|
from contextlib import contextmanager |
|
from datetime import date |
|
from decimal import Decimal |
|
|
|
import pytest |
|
from django.db import transaction |
|
from django.db.models import Sum |
|
|
|
from finance.models import AuditEvent, SourceRecord, Transaction, Transfer |
|
from finance.services import ( |
|
Actor, DomainError, annotate_transaction, archive, attach_source, |
|
create_account, create_category, create_payee, create_plan, create_statement, |
|
create_transfer, match_transfer, record_transaction, |
|
) |
|
|
|
pytestmark = pytest.mark.django_db(transaction=True) |
|
ACTOR = Actor(kind="AGENT", identifier="synthetic-test-agent") |
|
DAY = date(2026, 1, 15) |
|
|
|
|
|
def account(name="Synthetic Bank", kind="BANK", currency="BRL"): |
|
return create_account(name=name, kind=kind, currency=currency, actor=ACTOR) |
|
|
|
|
|
def post(account, amount="-120.00", kind="EXPENSE", **fields): |
|
return record_transaction( |
|
account_id=account.pk, amount=Decimal(amount), kind=kind, |
|
posting_date=DAY, actor=ACTOR, **fields, |
|
) |
|
|
|
|
|
@contextmanager |
|
def rejected_without_posting(): |
|
"""A rejected write must leave movement intact and retain a failure audit.""" |
|
before = Transaction.objects.count() |
|
previous_audits = set(AuditEvent.objects.values_list("pk", flat=True)) |
|
with pytest.raises(DomainError) as caught: |
|
yield |
|
assert caught.value.code |
|
assert Transaction.objects.count() == before |
|
events = list(AuditEvent.objects.exclude(pk__in=previous_audits)) |
|
assert len(events) == 1 |
|
assert events[0].outcome == "FAILURE" |
|
assert events[0].error_code == caught.value.code |
|
assert events[0].actor_kind == "AGENT" |
|
assert events[0].actor_identifier == ACTOR.identifier |
|
|
|
|
|
def test_opening_and_expense_derive_bank_balance(): |
|
bank = account() |
|
post(bank, "1000.00", "OPENING") |
|
expense = post(bank, occurrence_date=date(2026, 1, 12)) |
|
assert Transaction.objects.filter(account=bank).aggregate(total=Sum("amount"))["total"] == Decimal("880.00") |
|
assert expense.posting_date == DAY |
|
assert expense.occurrence_date == date(2026, 1, 12) |
|
with rejected_without_posting(): |
|
post(bank, "1.00", "OPENING") |
|
|
|
|
|
@pytest.mark.parametrize("amount", [0.1, Decimal("0"), Decimal("NaN"), Decimal("Infinity"), Decimal("-Infinity"), Decimal("-0.001"), Decimal("10000000000000000"), Decimal("1e999999999")]) |
|
def test_invalid_money_is_rejected_without_rounding(amount): |
|
bank = account() |
|
with rejected_without_posting(): |
|
record_transaction(account_id=bank.pk, amount=amount, kind="EXPENSE", posting_date=DAY, actor=ACTOR) |
|
|
|
|
|
@pytest.mark.parametrize("kind,amount", [("INCOME", "-1"), ("EXPENSE", "1"), ("REFUND", "-1"), ("UNKNOWN", "1")]) |
|
def test_transaction_sign_matches_kind(kind, amount): |
|
bank = account() |
|
with rejected_without_posting(): |
|
post(bank, amount, kind) |
|
|
|
|
|
@pytest.mark.parametrize("currency", ["brl", "BR", "BRLL", "B1L", "ÉUR"]) |
|
def test_account_currency_requires_three_uppercase_ascii_letters(currency): |
|
with rejected_without_posting(): |
|
account(currency=currency) |
|
|
|
|
|
def test_category_kind_and_archived_assignments_are_validated(): |
|
bank = account() |
|
income = create_category(name="Synthetic Salary", kind="INCOME", actor=ACTOR) |
|
expense = create_category(name="Synthetic Food", kind="EXPENSE", actor=ACTOR) |
|
payee = create_payee(name="Synthetic Merchant", actor=ACTOR) |
|
with rejected_without_posting(): |
|
post(bank, category_id=income.pk) |
|
with rejected_without_posting(): |
|
post(bank, "10", "INCOME", category_id=expense.pk) |
|
historical = post(bank, category_id=expense.pk, payee_id=payee.pk) |
|
archive(expense, actor=ACTOR) |
|
archive(payee, actor=ACTOR) |
|
for fields in ({"category_id": expense.pk}, {"payee_id": payee.pk}): |
|
with rejected_without_posting(): |
|
post(bank, **fields) |
|
updated = annotate_transaction(transaction_id=historical.pk, description="Synthetic revised label", actor=ACTOR) |
|
assert updated.category_id == expense.pk |
|
assert updated.payee_id == payee.pk |
|
assert updated.description == "Synthetic revised label" |
|
archive(bank, actor=ACTOR) |
|
with rejected_without_posting(): |
|
post(bank) |
|
historical.refresh_from_db() |
|
assert historical.amount == Decimal("-120.00") |
|
|
|
|
|
def test_annotations_change_only_permitted_fields_and_audit_names(): |
|
bank = account() |
|
row = post(bank) |
|
category = create_category(name="Synthetic Food", kind="EXPENSE", actor=ACTOR) |
|
payee = create_payee(name="Synthetic Merchant", actor=ACTOR) |
|
updated = annotate_transaction(transaction_id=row.pk, description="Synthetic sensitive label", payee_id=payee.pk, category_id=category.pk, actor=ACTOR) |
|
assert updated.description == "Synthetic sensitive label" |
|
assert updated.category_id == category.pk |
|
assert updated.payee_id == payee.pk |
|
event = AuditEvent.objects.filter(target_id=row.pk).latest("created_at") |
|
assert event.outcome == "SUCCESS" |
|
assert set(event.changed_fields) == {"description", "payee", "category"} |
|
assert "Synthetic sensitive label" not in str(event.__dict__) |
|
cleared = annotate_transaction(transaction_id=row.pk, payee_id=None, category_id=None, actor=ACTOR) |
|
assert cleared.payee_id is None and cleared.category_id is None |
|
|
|
|
|
@pytest.mark.parametrize("field,value", [("amount", Decimal("-99")), ("account_id", None), ("posting_date", date(2026, 2, 1)), ("kind", "INCOME"), ("refund_of_id", None), ("installment_number", 1)]) |
|
def test_annotation_api_rejects_financial_changes(field, value): |
|
row = post(account()) |
|
with pytest.raises((TypeError, DomainError)): |
|
annotate_transaction(transaction_id=row.pk, actor=ACTOR, **{field: value}) |
|
row.refresh_from_db() |
|
assert row.amount == Decimal("-120.00") |
|
assert row.posting_date == DAY |
|
assert row.kind == "EXPENSE" |
|
|
|
|
|
def test_card_payment_adds_no_spending_and_preserves_settlement_dates(): |
|
bank = account() |
|
card = account("Synthetic Card", "CREDIT_CARD") |
|
post(bank, "1000", "OPENING") |
|
post(card, "-300") |
|
payment = create_transfer(outgoing_account_id=bank.pk, incoming_account_id=card.pk, amount=Decimal("300"), outgoing_date=DAY, incoming_date=date(2026, 1, 17), actor=ACTOR) |
|
assert payment.outgoing.amount == Decimal("-300") |
|
assert payment.incoming.amount == Decimal("300") |
|
assert payment.outgoing.posting_date == DAY |
|
assert payment.incoming.posting_date == date(2026, 1, 17) |
|
assert Transaction.objects.filter(kind="EXPENSE").aggregate(total=Sum("amount"))["total"] == Decimal("-300") |
|
assert Transaction.objects.filter(account=card).aggregate(total=Sum("amount"))["total"] == Decimal("0") |
|
assert Transaction.objects.count() == 4 |
|
|
|
|
|
@pytest.mark.parametrize("scenario", ["same_account", "different_currency", "negative_amount"]) |
|
def test_invalid_complete_transfer_is_atomic(scenario): |
|
bank = account() |
|
other = bank if scenario == "same_account" else account("Synthetic Other", currency="USD" if scenario == "different_currency" else "BRL") |
|
with rejected_without_posting(): |
|
create_transfer(outgoing_account_id=bank.pk, incoming_account_id=other.pk, amount=Decimal("-10" if scenario == "negative_amount" else "10"), outgoing_date=DAY, incoming_date=DAY, actor=ACTOR) |
|
assert Transfer.objects.count() == 0 |
|
|
|
|
|
def test_match_existing_transfer_legs_creates_no_extra_transactions(): |
|
outgoing = post(account(), "-25", "TRANSFER") |
|
incoming = post(account("Synthetic Other"), "25", "TRANSFER") |
|
match = match_transfer(outgoing_id=outgoing.pk, incoming_id=incoming.pk, actor=ACTOR) |
|
assert match.outgoing_id == outgoing.pk |
|
assert match.incoming_id == incoming.pk |
|
assert Transaction.objects.count() == 2 |
|
with rejected_without_posting(): |
|
match_transfer(outgoing_id=outgoing.pk, incoming_id=incoming.pk, actor=ACTOR) |
|
assert Transfer.objects.count() == 1 |
|
|
|
|
|
@pytest.mark.parametrize("scenario", ["same_account", "currency", "sign", "magnitude", "kind"]) |
|
def test_matching_validates_both_existing_legs(scenario): |
|
bank = account() |
|
other = bank if scenario == "same_account" else account("Synthetic Other", currency="USD" if scenario == "currency" else "BRL") |
|
outgoing = post(bank, "-25", "EXPENSE" if scenario == "kind" else "TRANSFER") |
|
incoming = post(other, "-25" if scenario == "sign" else "26" if scenario == "magnitude" else "25", "TRANSFER") |
|
with rejected_without_posting(): |
|
match_transfer(outgoing_id=outgoing.pk, incoming_id=incoming.pk, actor=ACTOR) |
|
assert Transfer.objects.count() == 0 |
|
|
|
|
|
def test_refunds_retain_purchase_and_allow_cross_account_partial_adjustments(): |
|
bank = account() |
|
category = create_category(name="Synthetic Food", kind="EXPENSE", actor=ACTOR) |
|
purchase = post(bank, category_id=category.pk) |
|
other = account("Synthetic Other") |
|
refund = post(other, "50", "REFUND", refund_of_id=purchase.pk, category_id=category.pk) |
|
assert refund.refund_of_id == purchase.pk |
|
assert Transaction.objects.filter(kind__in=["EXPENSE", "REFUND"]).aggregate(total=Sum("amount"))["total"] == Decimal("-70") |
|
post(other, "100", "REFUND", refund_of_id=purchase.pk) |
|
unlinked = post(other, "5", "REFUND") |
|
assert unlinked.refund_of_id is None |
|
purchase.refresh_from_db() |
|
assert purchase.amount == Decimal("-120") |
|
|
|
|
|
@pytest.mark.parametrize("scenario", ["currency", "target_kind", "wrong_row_kind"]) |
|
def test_invalid_refund_links_are_rejected(scenario): |
|
bank = account() |
|
original = post(bank, "20" if scenario == "target_kind" else "-20", "INCOME" if scenario == "target_kind" else "EXPENSE") |
|
other = account("Synthetic Other", currency="USD" if scenario == "currency" else "BRL") |
|
with rejected_without_posting(): |
|
post(other, "5", "INCOME" if scenario == "wrong_row_kind" else "REFUND", refund_of_id=original.pk) |
|
|
|
|
|
def test_only_observed_installments_are_posted_and_numbers_are_unique_per_plan(): |
|
card = account("Synthetic Card", "CREDIT_CARD") |
|
first = create_plan(account_id=card.pk, description="Synthetic purchase", total_installments=10, actor=ACTOR) |
|
second = create_plan(account_id=card.pk, description="Synthetic second purchase", total_installments=10, actor=ACTOR) |
|
row = post(card, "-30.01", installment_plan_id=first.pk, installment_number=3) |
|
assert Transaction.objects.count() == 1 |
|
with rejected_without_posting(): |
|
post(card, "-30.01", installment_plan_id=first.pk, installment_number=3) |
|
post(card, "-30.00", installment_plan_id=first.pk, installment_number=4) |
|
post(card, "-30.01", installment_plan_id=second.pk, installment_number=3) |
|
refund = post(card, "5", "REFUND", refund_of_id=row.pk) |
|
assert refund.refund_of_id == row.pk |
|
assert Transaction.objects.filter(kind="EXPENSE").count() == 3 |
|
|
|
|
|
@pytest.mark.parametrize("scenario", ["account", "zero", "overflow", "missing_number", "missing_plan", "kind"]) |
|
def test_invalid_installment_membership_is_rejected(scenario): |
|
bank = account() |
|
plan = create_plan(account_id=bank.pk, description="Synthetic purchase", total_installments=10, actor=ACTOR) |
|
destination = account("Synthetic Other") if scenario == "account" else bank |
|
fields = {"installment_plan_id": plan.pk, "installment_number": 3} |
|
if scenario == "missing_number": |
|
fields.pop("installment_number") |
|
elif scenario == "missing_plan": |
|
fields.pop("installment_plan_id") |
|
elif scenario in {"zero", "overflow"}: |
|
fields["installment_number"] = 0 if scenario == "zero" else 11 |
|
with rejected_without_posting(): |
|
post(destination, "10" if scenario == "kind" else "-10", "INCOME" if scenario == "kind" else "EXPENSE", **fields) |
|
|
|
|
|
def test_statement_metadata_does_not_manufacture_transactions_or_reconciliation(): |
|
card = account("Synthetic Card", "CREDIT_CARD") |
|
statement = create_statement(account_id=card.pk, period_start=date(2026, 1, 1), period_end=date(2026, 1, 31), due_date=date(2026, 2, 10), statement_balance=Decimal("-900"), actor=ACTOR) |
|
assert Transaction.objects.count() == 0 |
|
row = post(card, "-30", statement_id=statement.pk) |
|
assert row.statement_id == statement.pk |
|
statement.refresh_from_db() |
|
assert statement.statement_balance == Decimal("-900") |
|
for destination in (account(), account("Synthetic Other Card", "CREDIT_CARD")): |
|
with rejected_without_posting(): |
|
post(destination, statement_id=statement.pk) |
|
|
|
|
|
@pytest.mark.parametrize("scenario", ["bank", "period", "due"]) |
|
def test_statement_creation_validates_account_and_dates(scenario): |
|
target = account(kind="BANK" if scenario == "bank" else "CREDIT_CARD") |
|
with rejected_without_posting(): |
|
create_statement(account_id=target.pk, period_start=date(2026, 2, 1) if scenario == "period" else date(2026, 1, 1), period_end=date(2026, 1, 31), due_date=date(2026, 1, 30) if scenario == "due" else date(2026, 2, 10), actor=ACTOR) |
|
|
|
|
|
def test_source_identity_is_physical_item_and_supports_multiple_sources_per_transaction(): |
|
row = post(account()) |
|
other = post(row.account, "-40") |
|
metadata = dict(document_hash="a" * 64, source_format="BANK_CSV", parser_version="synthetic-v1", item_locator="row:2", actor=ACTOR) |
|
original = attach_source(transaction_id=row.pk, **metadata) |
|
assert original.transaction_id == row.pk |
|
with rejected_without_posting(): |
|
attach_source(transaction_id=other.pk, **{**metadata, "parser_version": "synthetic-v2"}) |
|
attach_source(transaction_id=row.pk, **{**metadata, "item_locator": "row:3"}) |
|
attach_source(transaction_id=row.pk, **{**metadata, "document_hash": "b" * 64}) |
|
assert SourceRecord.objects.filter(transaction=row).count() == 3 |
|
assert SourceRecord.objects.filter(transaction=other).count() == 0 |
|
assert Transaction.objects.count() == 2 |
|
|
|
|
|
@pytest.mark.parametrize("fields", [{"document_hash": "invalid"}, {"source_format": "OFX"}, {"parser_version": " "}, {"item_locator": " "}]) |
|
def test_invalid_source_metadata_is_rejected(fields): |
|
row = post(account()) |
|
values = dict(document_hash="a" * 64, source_format="BANK_CSV", parser_version="synthetic-v1", item_locator="row:2") |
|
with rejected_without_posting(): |
|
attach_source(transaction_id=row.pk, actor=ACTOR, **{**values, **fields}) |
|
assert SourceRecord.objects.count() == 0 |
|
|
|
|
|
def test_public_writes_refuse_caller_owned_atomic_blocks(): |
|
bank = account() |
|
before = Transaction.objects.count() |
|
with transaction.atomic(): |
|
with pytest.raises((DomainError, RuntimeError)): |
|
post(bank) |
|
assert Transaction.objects.count() == before |
|
|
|
|
|
def test_complete_transfer_preserves_amount_under_low_decimal_precision(): |
|
from decimal import localcontext |
|
|
|
bank = account() |
|
other = account("Synthetic Other") |
|
with localcontext() as context: |
|
context.prec = 4 |
|
payment = create_transfer(outgoing_account_id=bank.pk, incoming_account_id=other.pk, |
|
amount=Decimal("123.45"), outgoing_date=DAY, |
|
incoming_date=DAY, actor=ACTOR) |
|
payment.outgoing.refresh_from_db() |
|
payment.incoming.refresh_from_db() |
|
assert payment.outgoing.amount == Decimal("-123.45") |
|
assert payment.incoming.amount == Decimal("123.45") |
|
|
|
|
|
def test_matching_rejects_amounts_that_only_match_after_context_rounding(): |
|
from decimal import localcontext |
|
|
|
outgoing = post(account(), "-123.40", "TRANSFER") |
|
incoming = post(account("Synthetic Other"), "123.45", "TRANSFER") |
|
with localcontext() as context: |
|
context.prec = 4 |
|
with rejected_without_posting(): |
|
match_transfer(outgoing_id=outgoing.pk, incoming_id=incoming.pk, actor=ACTOR) |
|
assert Transfer.objects.count() == 0
|
|
|