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.
 
 
 
 
 

320 lines
10 KiB

"""Synthetic end-to-end checks for the owner reconciliation pages."""
from datetime import date, datetime
from decimal import Decimal
import re
import pytest
from django.test import Client
from django.urls import reverse
from django.utils import timezone
from finance.models import AuditEvent, ReconciliationCheckpoint
from finance.services import (
Actor,
archive,
create_account,
create_reconciliation_checkpoint,
record_transaction,
)
pytestmark = pytest.mark.django_db(transaction=True)
SETUP_ACTOR = Actor(kind="AGENT", identifier="synthetic-reconciliation-view-test")
AS_OF_DATE = date(2026, 9, 1)
def account(name="Synthetic reconciliation bank"):
return create_account(
name=name,
kind="BANK",
currency="BRL",
actor=SETUP_ACTOR,
)
def post(target, amount, *, posting_date=AS_OF_DATE, kind="EXPENSE"):
return record_transaction(
account_id=target.pk,
amount=Decimal(amount),
posting_date=posting_date,
kind=kind,
actor=SETUP_ACTOR,
)
def checkpoint(target, institution_balance):
return create_reconciliation_checkpoint(
account_id=target.pk,
as_of_date=AS_OF_DATE,
institution_balance=Decimal(institution_balance),
actor=SETUP_ACTOR,
)
def preview_token(response):
match = re.search(r'name="preview_token" value="([^"]+)"', response.content.decode())
return match.group(1) if match else ""
@pytest.mark.parametrize("field", ["account", "as_of_date", "institution_balance"])
def test_confirm_requires_new_preview_after_input_changes(client, field):
bank = account()
other = account("Synthetic other reconciliation bank")
values = {
"account": bank.pk,
"as_of_date": "2026-09-01",
"institution_balance": "0.00",
}
url = reverse("reconciliation-index")
preview = client.post(url, {"action": "preview", **values})
changed = {**values, field: {
"account": other.pk,
"as_of_date": "2026-09-02",
"institution_balance": "12.50",
}[field]}
response = client.post(url, {
"action": "confirm", "preview_token": preview_token(preview), **changed,
})
assert ReconciliationCheckpoint.objects.count() == 0
assert response.status_code == 200
assert response.context["form"].non_field_errors()
assert 'name="action" value="confirm"' not in response.content.decode()
refreshed = client.post(url, {"action": "preview", **changed})
confirmed = client.post(url, {
"action": "confirm", "preview_token": preview_token(refreshed), **changed,
})
assert confirmed.status_code == 302
saved = ReconciliationCheckpoint.objects.get()
assert str(saved.account_id) == str(changed["account"])
assert saved.as_of_date.isoformat() == changed["as_of_date"]
assert saved.institution_balance == Decimal(changed["institution_balance"])
@pytest.mark.parametrize("token_kind", ["missing", "tampered"])
def test_confirm_rejects_missing_or_tampered_preview_token(client, token_kind):
bank = account()
values = {
"account": bank.pk,
"as_of_date": "2026-09-01",
"institution_balance": "0.00",
}
url = reverse("reconciliation-index")
preview = client.post(url, {"action": "preview", **values})
token = "" if token_kind == "missing" else preview_token(preview) + "tampered"
response = client.post(url, {
"action": "confirm", "preview_token": token, **values,
})
assert ReconciliationCheckpoint.objects.count() == 0
assert response.status_code == 200
assert response.context["form"].non_field_errors()
def test_history_exposes_creation_timestamps_for_same_date_corrections(client):
bank = account()
first = checkpoint(bank, "0.00")
correction = checkpoint(bank, "1.00")
first_time = timezone.make_aware(datetime(2026, 9, 2, 10, 11, 12))
correction_time = timezone.make_aware(datetime(2026, 9, 2, 10, 11, 13))
ReconciliationCheckpoint.objects.filter(pk=first.pk).update(created_at=first_time)
ReconciliationCheckpoint.objects.filter(pk=correction.pk).update(created_at=correction_time)
response = client.get(reverse("reconciliation-history", args=[bank.pk]))
body = response.content.decode()
assert "Created at" in body
assert "2026-09-02 10:11:12 UTC" in body
assert "2026-09-02 10:11:13 UTC" in body
assert body.index("10:11:13 UTC") < body.index("10:11:12 UTC")
assert [row.created_at for row in response.context["history"]] == [
correction_time, first_time,
]
def test_reconciliation_is_linked_and_landing_form_lists_only_active_accounts(client):
active = account("Synthetic active bank")
archived = account("Synthetic archived bank")
checkpoint(active, "0.00")
archive(archived, actor=SETUP_ACTOR)
navigation_page = client.get(reverse("transaction-list"))
response = client.get(reverse("reconciliation-index"))
assert navigation_page.status_code == 200
assert f'href="{reverse("reconciliation-index")}">Reconciliation</a>' in navigation_page.content.decode()
assert response.status_code == 200
body = response.content.decode()
assert "<h1>Reconciliation</h1>" in body
assert f'<option value="{active.pk}">Synthetic active bank</option>' in body
assert str(archived.pk) not in body
assert "Latest reconciliation status" in body
assert "Reconciled" in body
assert reverse("reconciliation-history", args=[active.pk]) in body
def test_preview_renders_ledger_balance_and_difference_without_writing(client):
bank = account()
post(bank, "100.00", kind="OPENING", posting_date=date(2026, 8, 31))
post(bank, "-40.00")
response = client.post(
reverse("reconciliation-index"),
{
"action": "preview",
"account": bank.pk,
"as_of_date": "2026-09-01",
"institution_balance": "0.00",
},
)
assert response.status_code == 200
body = response.content.decode()
assert "Preview" in body
assert "Ledger balance" in body and "60.00 BRL" in body
assert "Difference" in body and "-60.00 BRL" in body
assert 'name="action" value="confirm"' in body
assert ReconciliationCheckpoint.objects.count() == 0
def test_confirm_revalidates_and_service_snapshots_fresh_ledger_then_redirects(client):
bank = account()
post(bank, "-40.00")
preview = client.post(
reverse("reconciliation-index"),
{
"action": "preview",
"account": bank.pk,
"as_of_date": "2026-09-01",
"institution_balance": "-40.00",
},
)
assert preview.status_code == 200
assert "0.00 BRL" in preview.content.decode()
post(bank, "-2.00")
response = client.post(
reverse("reconciliation-index"),
{
"action": "confirm",
"account": bank.pk,
"as_of_date": "2026-09-01",
"institution_balance": "-40.00",
"ledger_balance": "999999.00",
"difference": "999999.00",
"preview_token": preview_token(preview),
},
)
saved = ReconciliationCheckpoint.objects.get()
assert response.status_code == 302
assert response.url == reverse("reconciliation-history", args=[bank.pk])
assert saved.ledger_balance_at_creation == Decimal("-42.00")
assert saved.difference_at_creation == Decimal("2.00")
event = AuditEvent.objects.get(
operation="reconciliation_checkpoint.created",
target_id=saved.pk,
)
assert (event.actor_kind, event.actor_identifier) == ("HUMAN", "local-owner")
@pytest.mark.parametrize(
("field", "value", "expected_error"),
[
("as_of_date", "not-a-date", "Enter a valid date."),
("institution_balance", "not-money", "Enter a number."),
("institution_balance", "1.001", "Ensure that there are no more than 2 decimal places."),
],
)
def test_malformed_values_show_field_errors_without_writing(
client, field, value, expected_error
):
bank = account()
payload = {
"action": "confirm",
"account": bank.pk,
"as_of_date": "2026-09-01",
"institution_balance": "0.00",
}
payload[field] = value
response = client.post(reverse("reconciliation-index"), payload)
assert response.status_code == 200
assert expected_error in response.content.decode()
assert field in response.context["form"].errors
assert ReconciliationCheckpoint.objects.count() == 0
def test_reconciliation_confirmation_requires_csrf():
bank = account()
guarded = Client(enforce_csrf_checks=True)
response = guarded.post(
reverse("reconciliation-index"),
{
"action": "confirm",
"account": bank.pk,
"as_of_date": "2026-09-01",
"institution_balance": "0.00",
},
)
assert response.status_code == 403
assert ReconciliationCheckpoint.objects.count() == 0
def test_archived_account_confirmation_fails_safely_without_writing(client):
archived = account()
archive(archived, actor=SETUP_ACTOR)
response = client.post(
reverse("reconciliation-index"),
{
"action": "confirm",
"account": archived.pk,
"as_of_date": "2026-09-01",
"institution_balance": "0.00",
},
)
assert response.status_code == 200
body = response.content.decode()
assert "Select a valid choice." in body
assert "synthetic_private" not in body
assert ReconciliationCheckpoint.objects.count() == 0
def test_history_keeps_corrected_same_date_snapshots_and_shows_live_drift(client):
bank = account()
post(bank, "-40.00")
first = checkpoint(bank, "-39.25")
correction = checkpoint(bank, "-40.00")
post(bank, "-2.00")
archive(bank, actor=SETUP_ACTOR)
response = client.get(reverse("reconciliation-history", args=[bank.pk]))
assert response.status_code == 200
body = response.content.decode()
assert "Synthetic reconciliation bank" in body
assert "Archived" in body
assert "Institution balance" in body
assert "Snapshot ledger balance" in body
assert "Snapshot difference" in body
assert "Live ledger balance" in body
assert "Live difference" in body
assert "Changed since reconciliation" in body
assert "Difference" in body
assert "-42.00 BRL" in body
assert "2.00 BRL" in body
assert "2.75 BRL" in body
assert [row.checkpoint_id for row in response.context["history"]] == [
correction.pk,
first.pk,
]