import csv import json import threading from datetime import date, datetime, timezone as datetime_timezone from decimal import Decimal from hashlib import sha256 from io import BytesIO, StringIO from uuid import UUID from zipfile import ZipFile import pytest from django.db import connections, transaction import finance.exports as exports from finance.exports import build_export from finance.models import ( Account, AuditEvent, CardStatement, Category, CategoryGroup, InstallmentPlan, Payee, SourceRecord, Transaction, Transfer, ) pytestmark = pytest.mark.django_db(transaction=True) EXPECTED_FILES = [ "accounts.csv", "payees.csv", "category_groups.csv", "categories.csv", "statements.csv", "installment_plans.csv", "transactions.csv", "transfers.csv", "source_records.csv", ] EXPECTED_HEADERS = { "accounts.csv": "id,name,kind,currency,archived,created_at,updated_at", "payees.csv": "id,name,archived,created_at,updated_at", "category_groups.csv": "id,name,archived,created_at,updated_at", "categories.csv": "id,name,kind,group_id,archived,created_at,updated_at", "statements.csv": ( "id,account_id,period_start,period_end,due_date,statement_balance,created_at,updated_at" ), "installment_plans.csv": ( "id,account_id,description,total_installments,created_at,updated_at" ), "transactions.csv": ( "id,account_id,amount,posting_date,occurrence_date,description,kind,payee_id," "category_id,statement_id,refund_of_id,installment_plan_id,installment_number," "created_at,updated_at" ), "transfers.csv": "id,outgoing_id,incoming_id,created_at,updated_at", "source_records.csv": ( "id,transaction_id,document_hash,source_format,parser_version,item_locator,created_at" ), } def rows(package, name): content = package.read(name).decode("utf-8") return list(csv.DictReader(StringIO(content))) def seed_complete_export(): account = Account.objects.create( id=UUID("10000000-0000-4000-8000-000000000002"), name="Synthetic Card", kind=Account.Kind.CREDIT_CARD, currency="BRL", archived=True, ) Account.objects.create( id=UUID("10000000-0000-4000-8000-000000000001"), name="Synthetic Bank", kind=Account.Kind.BANK, currency="USD", ) payee = Payee.objects.create( id=UUID("20000000-0000-4000-8000-000000000001"), name="Synthetic Merchant", ) group = CategoryGroup.objects.create( id=UUID("30000000-0000-4000-8000-000000000002"), name="Synthetic Household", ) category = Category.objects.create( id=UUID("30000000-0000-4000-8000-000000000001"), name="Synthetic Food", kind=Category.Kind.EXPENSE, group=group, ) statement = CardStatement.objects.create( id=UUID("40000000-0000-4000-8000-000000000001"), account=account, period_start=date(2026, 1, 1), period_end=date(2026, 1, 31), due_date=date(2026, 2, 10), statement_balance=Decimal("1234.50"), ) plan = InstallmentPlan.objects.create( id=UUID("50000000-0000-4000-8000-000000000001"), account=account, description="Synthetic appliance", total_installments=3, ) purchase = Transaction.objects.create( id=UUID("60000000-0000-4000-8000-000000000001"), account=account, amount=Decimal("-411.50"), posting_date=date(2026, 1, 12), occurrence_date=date(2026, 1, 10), description="Synthetic purchase, first part", kind=Transaction.Kind.EXPENSE, payee=payee, category=category, statement=statement, installment_plan=plan, installment_number=1, ) refund = Transaction.objects.create( id=UUID("60000000-0000-4000-8000-000000000002"), account=account, amount=Decimal("11.50"), posting_date=date(2026, 2, 1), description="", kind=Transaction.Kind.REFUND, category=category, refund_of=purchase, ) transfer_out = Transaction.objects.create( id=UUID("60000000-0000-4000-8000-000000000003"), account_id=UUID("10000000-0000-4000-8000-000000000001"), amount=Decimal("-20.00"), posting_date=date(2026, 2, 2), kind=Transaction.Kind.TRANSFER, ) transfer_in = Transaction.objects.create( id=UUID("60000000-0000-4000-8000-000000000004"), account=account, amount=Decimal("20.00"), posting_date=date(2026, 2, 2), kind=Transaction.Kind.TRANSFER, ) Transfer.objects.create( id=UUID("70000000-0000-4000-8000-000000000001"), outgoing=transfer_out, incoming=transfer_in, ) SourceRecord.objects.create( id=UUID("80000000-0000-4000-8000-000000000001"), transaction=purchase, document_hash="a" * 64, source_format=SourceRecord.Format.OUROCARD_TXT, parser_version="synthetic-v1", item_locator="line:2", ) AuditEvent.objects.create( actor_kind=AuditEvent.ActorKind.HUMAN, actor_identifier="synthetic-owner", operation="synthetic_operation", outcome=AuditEvent.Outcome.SUCCESS, ) return account, purchase, refund def test_export_has_versioned_manifest_and_stable_csv_contract(): seed_complete_export() artifact = build_export() with ZipFile(BytesIO(artifact.content)) as package: assert package.namelist() == [*EXPECTED_FILES, "manifest.json"] manifest = json.loads(package.read("manifest.json")) assert manifest["format"] == "gizmoduck-export" assert manifest["version"] == 1 assert manifest["application_version"] == "0.1.0" assert list(manifest["files"]) == EXPECTED_FILES for name in EXPECTED_FILES: content = package.read(name) assert content.decode("utf-8").splitlines()[0] == EXPECTED_HEADERS[name] assert manifest["files"][name]["sha256"] == sha256(content).hexdigest() assert manifest["files"][name]["rows"] == len(rows(package, name)) def test_export_retains_portable_values_relationships_and_provenance(): account, purchase, refund = seed_complete_export() artifact = build_export() with ZipFile(BytesIO(artifact.content)) as package: account_rows = rows(package, "accounts.csv") assert [row["id"] for row in account_rows] == [ "10000000-0000-4000-8000-000000000001", "10000000-0000-4000-8000-000000000002", ] assert account_rows[1]["archived"] == "true" assert account_rows[1]["currency"] == "BRL" statement_row = rows(package, "statements.csv")[0] assert statement_row["statement_balance"] == "1234.50" assert statement_row["due_date"] == "2026-02-10" transaction_rows = rows(package, "transactions.csv") purchase_row = next(row for row in transaction_rows if row["id"] == str(purchase.pk)) assert purchase_row["amount"] == "-411.50" assert purchase_row["occurrence_date"] == "2026-01-10" assert purchase_row["description"] == "Synthetic purchase, first part" assert purchase_row["account_id"] == str(account.pk) assert purchase_row["installment_number"] == "1" assert "T" in purchase_row["created_at"] refund_row = next(row for row in transaction_rows if row["id"] == str(refund.pk)) assert refund_row["occurrence_date"] == "" assert refund_row["payee_id"] == "" assert refund_row["refund_of_id"] == str(purchase.pk) source_row = rows(package, "source_records.csv")[0] assert source_row == { "id": "80000000-0000-4000-8000-000000000001", "transaction_id": str(purchase.pk), "document_hash": "a" * 64, "source_format": "OUROCARD_TXT", "parser_version": "synthetic-v1", "item_locator": "line:2", "created_at": source_row["created_at"], } assert "audit_events.csv" not in package.namelist() def test_export_is_byte_stable_for_the_same_snapshot(monkeypatch): seed_complete_export() fixed_now = datetime(2026, 9, 7, 12, 30, tzinfo=datetime_timezone.utc) monkeypatch.setattr("finance.exports.timezone.now", lambda: fixed_now) first = build_export() second = build_export() assert first.filename == second.filename == "gizmoduck-export-20260907T123000Z.zip" assert first.content == second.content def test_export_uses_one_repeatable_read_snapshot_across_related_files(monkeypatch): seed_complete_export() accounts_read = threading.Event() writer_finished = threading.Event() writer_errors = [] new_account_id = UUID("10000000-0000-4000-8000-000000000003") new_payee_id = UUID("20000000-0000-4000-8000-000000000002") new_transaction_id = UUID("60000000-0000-4000-8000-000000000005") new_source_id = UUID("80000000-0000-4000-8000-000000000002") original_csv_content = exports._csv_content def pause_after_accounts(model, fields): content = original_csv_content(model, fields) if model is Account: accounts_read.set() assert writer_finished.wait(10), "separate writer did not finish" return content def commit_related_rows(): try: assert accounts_read.wait(10), "export did not read accounts" with transaction.atomic(): account = Account.objects.create( id=new_account_id, name="Synthetic Concurrent Account", kind=Account.Kind.BANK, currency="BRL", ) payee = Payee.objects.create( id=new_payee_id, name="Synthetic Concurrent Payee", ) row = Transaction.objects.create( id=new_transaction_id, account=account, amount=Decimal("-7.25"), posting_date=date(2026, 3, 1), description="Synthetic concurrent transaction", kind=Transaction.Kind.EXPENSE, payee=payee, ) SourceRecord.objects.create( id=new_source_id, transaction=row, document_hash="b" * 64, source_format=SourceRecord.Format.BANK_CSV, parser_version="synthetic-v2", item_locator="line:3", ) except BaseException as exc: writer_errors.append(exc) finally: connections.close_all() writer_finished.set() monkeypatch.setattr(exports, "_csv_content", pause_after_accounts) writer = threading.Thread(target=commit_related_rows) writer.start() artifact = build_export() writer.join(10) assert not writer.is_alive() assert writer_errors == [] with ZipFile(BytesIO(artifact.content)) as package: assert str(new_account_id) not in {row["id"] for row in rows(package, "accounts.csv")} assert str(new_payee_id) not in {row["id"] for row in rows(package, "payees.csv")} assert str(new_transaction_id) not in { row["id"] for row in rows(package, "transactions.csv") } assert str(new_source_id) not in { row["id"] for row in rows(package, "source_records.csv") }