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.
115 lines
3.5 KiB
115 lines
3.5 KiB
"""Synthetic checks for immutable, audited reconciliation checkpoints.""" |
|
|
|
from datetime import date |
|
from decimal import Decimal |
|
|
|
import pytest |
|
|
|
from finance.models import AuditEvent, ReconciliationCheckpoint |
|
from finance.services import ( |
|
Actor, |
|
DomainError, |
|
archive, |
|
create_account, |
|
create_reconciliation_checkpoint, |
|
record_transaction, |
|
) |
|
|
|
|
|
pytestmark = pytest.mark.django_db(transaction=True) |
|
ACTOR = Actor(kind="HUMAN", identifier="synthetic-local-owner") |
|
AS_OF_DATE = date(2026, 9, 1) |
|
|
|
|
|
def account(name="Synthetic reconciliation account"): |
|
return create_account(name=name, kind="BANK", actor=ACTOR) |
|
|
|
|
|
def post(account, amount, posting_date): |
|
return record_transaction( |
|
account_id=account.pk, |
|
amount=Decimal(amount), |
|
kind="EXPENSE", |
|
posting_date=posting_date, |
|
actor=ACTOR, |
|
) |
|
|
|
|
|
def test_checkpoint_snapshots_inclusive_ledger_balance_and_audit_metadata(): |
|
bank = account() |
|
post(bank, "-40.00", date(2026, 8, 31)) |
|
post(bank, "-2.50", AS_OF_DATE) |
|
post(bank, "-7.50", date(2026, 9, 2)) |
|
|
|
checkpoint = create_reconciliation_checkpoint( |
|
account_id=bank.pk, |
|
as_of_date=AS_OF_DATE, |
|
institution_balance=Decimal("0.00"), |
|
actor=ACTOR, |
|
) |
|
|
|
assert checkpoint.institution_balance == Decimal("0.00") |
|
assert checkpoint.ledger_balance_at_creation == Decimal("-42.50") |
|
assert checkpoint.difference_at_creation == Decimal("42.50") |
|
event = AuditEvent.objects.get( |
|
operation="reconciliation_checkpoint.created", |
|
target_id=checkpoint.pk, |
|
) |
|
assert event.outcome == "SUCCESS" |
|
assert event.actor_kind == "HUMAN" |
|
assert event.actor_identifier == "synthetic-local-owner" |
|
assert event.changed_fields == [ |
|
"account", |
|
"as_of_date", |
|
"institution_balance", |
|
"ledger_balance_at_creation", |
|
"difference_at_creation", |
|
] |
|
|
|
|
|
def test_checkpoint_correction_retains_same_date_history(): |
|
bank = account() |
|
post(bank, "-40.00", AS_OF_DATE) |
|
|
|
first = create_reconciliation_checkpoint( |
|
account_id=bank.pk, |
|
as_of_date=AS_OF_DATE, |
|
institution_balance=Decimal("-39.00"), |
|
actor=ACTOR, |
|
) |
|
correction = create_reconciliation_checkpoint( |
|
account_id=bank.pk, |
|
as_of_date=AS_OF_DATE, |
|
institution_balance=Decimal("-40.00"), |
|
actor=ACTOR, |
|
) |
|
|
|
assert first.pk != correction.pk |
|
assert ReconciliationCheckpoint.objects.filter( |
|
account=bank, |
|
as_of_date=AS_OF_DATE, |
|
).count() == 2 |
|
assert correction.ledger_balance_at_creation == Decimal("-40.00") |
|
assert correction.difference_at_creation == Decimal("0.00") |
|
|
|
|
|
def test_archived_account_rejects_checkpoint_without_creating_one(): |
|
bank = account() |
|
archive(bank, actor=ACTOR) |
|
checkpoint_count = ReconciliationCheckpoint.objects.count() |
|
audit_count = AuditEvent.objects.count() |
|
|
|
with pytest.raises(DomainError, match="archived_reference"): |
|
create_reconciliation_checkpoint( |
|
account_id=bank.pk, |
|
as_of_date=AS_OF_DATE, |
|
institution_balance=Decimal("0.00"), |
|
actor=ACTOR, |
|
) |
|
|
|
assert ReconciliationCheckpoint.objects.count() == checkpoint_count |
|
failed = AuditEvent.objects.exclude(operation="archive").latest("created_at") |
|
assert AuditEvent.objects.count() == audit_count + 1 |
|
assert failed.operation == "reconciliation_checkpoint.created" |
|
assert failed.outcome == "FAILURE" |
|
assert failed.error_code == "archived_reference"
|
|
|