Browse Source

feat: add versioned financial exports

master
Codex 2 months ago
parent
commit
92bc3820ef
  1. 83
      compose.yaml
  2. 1
      config/settings.py
  3. 2
      config/urls.py
  4. 1
      core/context.py
  5. 12
      core/templates/core/exports.html
  6. 15
      core/views.py
  7. 286
      finance/exports.py
  8. 1
      finance/management/__init__.py
  9. 1
      finance/management/commands/__init__.py
  10. 23
      finance/management/commands/export_finances.py
  11. 220
      tests/finance/test_exports.py
  12. 149
      tests/test_export_command.py
  13. 63
      tests/test_export_view.py

83
compose.yaml

@ -0,0 +1,83 @@
name: gizmoduck
x-app: &app
build: .
image: gizmoduck-local
x-runtime: &runtime
DJANGO_SECRET_KEY: ${DJANGO_SECRET_KEY:?Set DJANGO_SECRET_KEY in .env}
DB_NAME: gizmoduck
DB_HOST: db
DB_USER: gizmoduck_app
DB_PASSWORD: ${RUNTIME_PASSWORD:?Set RUNTIME_PASSWORD in .env}
services:
db:
image: postgres:17
environment:
POSTGRES_DB: gizmoduck
POSTGRES_PASSWORD: ${POSTGRES_PASSWORD:?Set POSTGRES_PASSWORD in .env}
MIGRATION_PASSWORD: ${MIGRATION_PASSWORD:?Set MIGRATION_PASSWORD in .env}
RUNTIME_PASSWORD: ${RUNTIME_PASSWORD:?Set RUNTIME_PASSWORD in .env}
volumes:
- postgres_data:/var/lib/postgresql/data
- ./docker/postgres/init.sh:/docker-entrypoint-initdb.d/10-roles.sh:ro
healthcheck:
test: ["CMD-SHELL", "pg_isready -h 127.0.0.1 -U postgres -d gizmoduck"]
interval: 2s
timeout: 3s
retries: 30
ports:
- "127.0.0.1:5432:5432"
web:
<<: *app
environment: *runtime
ports: ["127.0.0.1:8000:8000"]
depends_on:
db: {condition: service_healthy}
manage:
<<: *app
profiles: ["tools"]
environment:
<<: *runtime
DB_USER: gizmoduck_migrate
DB_PASSWORD: ${MIGRATION_PASSWORD:?Set MIGRATION_PASSWORD in .env}
volumes:
- ./imports:/app/imports
- ./exports:/app/exports
entrypoint: ["python", "manage.py"]
command: ["check"]
depends_on:
db: {condition: service_healthy}
test-db:
image: postgres:17
profiles: ["test"]
environment:
POSTGRES_DB: gizmoduck_tests
POSTGRES_PASSWORD: ${TEST_PASSWORD:?Set TEST_PASSWORD in .env}
tmpfs: [/var/lib/postgresql/data]
volumes:
- ./docker/postgres/test-init.sh:/docker-entrypoint-initdb.d/10-test.sh:ro
healthcheck:
test: ["CMD-SHELL", "pg_isready -h 127.0.0.1 -U postgres -d gizmoduck_tests"]
interval: 2s
timeout: 3s
retries: 30
test:
<<: *app
profiles: ["test"]
environment:
DJANGO_SECRET_KEY: ${DJANGO_SECRET_KEY:?Set DJANGO_SECRET_KEY in .env}
DB_HOST: test-db
DB_NAME: gizmoduck_tests
DB_USER: gizmoduck_test
DB_PASSWORD: ${TEST_PASSWORD:?Set TEST_PASSWORD in .env}
command: ["pytest", "-q", "-p", "no:cacheprovider"]
depends_on:
test-db: {condition: service_healthy}
permissions:
<<: *app
profiles: ["tools"]
environment: *runtime
command: ["python", "tests/check_permissions.py"]
depends_on:
db: {condition: service_healthy}
volumes:
postgres_data:

1
config/settings.py

@ -51,3 +51,4 @@ USE_TZ = True
TIME_ZONE = "UTC"
IMPORT_ROOT = "/app/imports"
EXPORT_ROOT = os.environ.get("EXPORT_ROOT", "/app/exports")

2
config/urls.py

@ -7,6 +7,8 @@ urlpatterns = [
path("", views.home, name="home"),
path("health/", views.health, name="health"),
path("reports/monthly/", views.monthly_report, name="monthly-report"),
path("exports/", views.export_index, name="export-index"),
path("exports/download/", views.export_download, name="export-download"),
path("transactions/", views.transaction_list, name="transaction-list"),
path("transactions/new/", views.transaction_create, name="transaction-create"),
path("transactions/<uuid:transaction_id>/", views.transaction_detail, name="transaction-detail"),

1
core/context.py

@ -6,4 +6,5 @@ def navigation(_request):
("category-list", "Categories"),
("payee-list", "Payees"),
("monthly-report", "Reports"),
("export-index", "Exports"),
)}

12
core/templates/core/exports.html

@ -0,0 +1,12 @@
{% extends "core/base.html" %}
{% block title %}Export financial data · Gizmoduck{% endblock %}
{% block content %}
<h1>Export financial data</h1>
<p>Download a versioned ZIP package containing UTF-8 CSV files and a checksum manifest.</p>
<p>The export includes accounts, payees, categories, card statements, installment plans, transactions, transfers, and source records. It preserves stable identifiers and relationships so the data remains portable.</p>
<p>The download is built from one consistent database snapshot. It does not include audit events or unfinished import reviews.</p>
<form method="post" action="{% url 'export-download' %}">
{% csrf_token %}
<button type="submit">Download export</button>
</form>
{% endblock %}

15
core/views.py

@ -1,12 +1,13 @@
from django.contrib import messages
from django.core.paginator import Paginator
from django.db import DatabaseError, connection
from django.http import JsonResponse
from django.http import HttpResponse, JsonResponse
from django.shortcuts import get_object_or_404, redirect, render
from django.utils import timezone
from django.views.decorators.http import require_POST
from finance.models import Account, Category, Payee, Transaction
from finance.exports import build_export
from finance.queries import (
QueryInputError, TransactionFilters, filtered_transactions,
listed_accounts, listed_categories, listed_payees,
@ -55,6 +56,18 @@ def monthly_report(request):
return _monthly_report(request, "core/reports/monthly.html")
def export_index(request):
return render(request, "core/exports.html")
@require_POST
def export_download(request):
artifact = build_export()
response = HttpResponse(artifact.content, content_type="application/zip")
response["Content-Disposition"] = f'attachment; filename="{artifact.filename}"'
return response
def transaction_list(request):
filter_errors = {}
try:

286
finance/exports.py

@ -0,0 +1,286 @@
import csv
import json
import os
import re
import tempfile
from dataclasses import dataclass
from datetime import date, datetime, timezone as datetime_timezone
from decimal import Decimal
from hashlib import sha256
from io import BytesIO, StringIO
from pathlib import Path
from uuid import UUID
from zipfile import ZIP_DEFLATED, ZipFile, ZipInfo
from django.conf import settings
from django.db import connection, transaction
from django.utils import timezone
from finance.models import (
Account,
CardStatement,
Category,
InstallmentPlan,
Payee,
SourceRecord,
Transaction,
Transfer,
)
EXPORT_FORMAT = "gizmoduck-export"
EXPORT_VERSION = 1
APPLICATION_VERSION = "0.1.0"
@dataclass(frozen=True)
class ExportArtifact:
filename: str
content: bytes
version: int
row_counts: dict[str, int]
@dataclass(frozen=True)
class ExportResult:
name: str
path: Path
version: int
row_counts: dict[str, int]
class ExportError(Exception):
def __init__(self, code):
self.code = code
super().__init__(code)
_PLAIN_ZIP_NAME = re.compile(r"\A[A-Za-z0-9][A-Za-z0-9._-]{0,198}\.zip\Z")
SCHEMAS = (
(
"accounts.csv",
Account,
("id", "name", "kind", "currency", "archived", "created_at", "updated_at"),
),
(
"payees.csv",
Payee,
("id", "name", "archived", "created_at", "updated_at"),
),
(
"categories.csv",
Category,
("id", "name", "kind", "archived", "created_at", "updated_at"),
),
(
"statements.csv",
CardStatement,
(
"id",
"account_id",
"period_start",
"period_end",
"due_date",
"statement_balance",
"created_at",
"updated_at",
),
),
(
"installment_plans.csv",
InstallmentPlan,
("id", "account_id", "description", "total_installments", "created_at", "updated_at"),
),
(
"transactions.csv",
Transaction,
(
"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",
Transfer,
("id", "outgoing_id", "incoming_id", "created_at", "updated_at"),
),
(
"source_records.csv",
SourceRecord,
(
"id",
"transaction_id",
"document_hash",
"source_format",
"parser_version",
"item_locator",
"created_at",
),
),
)
def build_export() -> ExportArtifact:
with transaction.atomic():
with connection.cursor() as cursor:
cursor.execute("SET TRANSACTION ISOLATION LEVEL REPEATABLE READ")
created_at = timezone.now()
csv_files = {
filename: _csv_content(model, fields)
for filename, model, fields in SCHEMAS
}
file_manifest = {
filename: {
"rows": row_count,
"sha256": sha256(content).hexdigest(),
}
for (filename, _model, _fields), (content, row_count) in zip(
SCHEMAS, csv_files.values(), strict=True
)
}
manifest = {
"format": EXPORT_FORMAT,
"version": EXPORT_VERSION,
"created_at": _value(created_at),
"application_version": APPLICATION_VERSION,
"files": file_manifest,
}
package = BytesIO()
with ZipFile(package, "w") as archive:
for filename, (content, _row_count) in csv_files.items():
_write_member(archive, filename, content)
manifest_content = (
json.dumps(manifest, ensure_ascii=False, separators=(",", ":")) + "\n"
).encode("utf-8")
_write_member(archive, "manifest.json", manifest_content)
return ExportArtifact(
filename=f"gizmoduck-export-{created_at.astimezone(datetime_timezone.utc):%Y%m%dT%H%M%SZ}.zip",
content=package.getvalue(),
version=EXPORT_VERSION,
row_counts={name: details["rows"] for name, details in file_manifest.items()},
)
def write_export(name: str | None = None) -> ExportResult:
if name is None:
name = f"gizmoduck-export-{timezone.now().astimezone(datetime_timezone.utc):%Y%m%dT%H%M%SZ}.zip"
if not isinstance(name, str) or not _PLAIN_ZIP_NAME.fullmatch(name):
raise ExportError("invalid_name")
try:
root = Path(settings.EXPORT_ROOT).resolve(strict=True)
except (OSError, RuntimeError, TypeError):
raise ExportError("invalid_root") from None
if not root.is_dir():
raise ExportError("invalid_root")
destination = root / name
if destination.is_symlink():
raise ExportError("invalid_name")
reservation_fd = None
owns_reservation = False
temporary_path = None
published = False
phase = "reservation"
try:
try:
reservation_fd = os.open(
destination,
os.O_WRONLY | os.O_CREAT | os.O_EXCL,
0o600,
)
owns_reservation = True
except FileExistsError:
raise ExportError("target_exists") from None
finally:
if reservation_fd is not None:
os.close(reservation_fd)
reservation_fd = None
phase = "snapshot"
artifact = build_export()
phase = "publication"
temporary_fd, temporary_name = tempfile.mkstemp(
prefix=".gizmoduck-export-", suffix=".tmp", dir=root
)
temporary_path = Path(temporary_name)
with os.fdopen(temporary_fd, "wb") as stream:
stream.write(artifact.content)
stream.flush()
os.fsync(stream.fileno())
os.replace(temporary_path, destination)
temporary_path = None
published = True
return ExportResult(
name=name,
path=destination,
version=artifact.version,
row_counts=artifact.row_counts,
)
except ExportError:
raise
except Exception:
code = "export_failed" if phase == "snapshot" else "publication_failed"
raise ExportError(code) from None
finally:
if temporary_path is not None:
try:
temporary_path.unlink(missing_ok=True)
except OSError:
pass
if owns_reservation and not published:
try:
destination.unlink(missing_ok=True)
except OSError:
pass
def _csv_content(model, fields):
stream = StringIO(newline="")
writer = csv.writer(stream, lineterminator="\n")
writer.writerow(fields)
row_count = 0
for row in model.objects.order_by("id").values_list(*fields):
writer.writerow(_value(value) for value in row)
row_count += 1
return stream.getvalue().encode("utf-8"), row_count
def _value(value):
if value is None:
return ""
if isinstance(value, bool):
return str(value).lower()
if isinstance(value, Decimal):
return format(value, "f")
if isinstance(value, UUID):
return str(value)
if isinstance(value, (date, datetime)):
return value.isoformat()
return str(value)
def _write_member(archive, filename, content):
info = ZipInfo(filename, date_time=(1980, 1, 1, 0, 0, 0))
info.compress_type = ZIP_DEFLATED
info.create_system = 3
info.external_attr = 0o600 << 16
archive.writestr(info, content, compress_type=ZIP_DEFLATED, compresslevel=9)

1
finance/management/__init__.py

@ -0,0 +1 @@

1
finance/management/commands/__init__.py

@ -0,0 +1 @@

23
finance/management/commands/export_finances.py

@ -0,0 +1,23 @@
import json
from django.core.management.base import BaseCommand, CommandError
from finance.exports import ExportError, write_export
class Command(BaseCommand):
help = "Create a versioned portable financial export."
def add_arguments(self, parser):
parser.add_argument("--output", dest="name")
def handle(self, *args, **options):
try:
result = write_export(options["name"])
except ExportError as exc:
raise CommandError(exc.code) from None
self.stdout.write(json.dumps({
"name": result.name,
"version": result.version,
"rows": result.row_counts,
}, separators=(",", ":")))

220
tests/finance/test_exports.py

@ -0,0 +1,220 @@
import csv
import json
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 finance.exports import build_export
from finance.models import (
Account,
AuditEvent,
CardStatement,
Category,
InstallmentPlan,
Payee,
SourceRecord,
Transaction,
Transfer,
)
pytestmark = pytest.mark.django_db(transaction=True)
EXPECTED_FILES = [
"accounts.csv",
"payees.csv",
"categories.csv",
"statements.csv",
"installment_plans.csv",
"transactions.csv",
"transfers.csv",
"source_records.csv",
]
EXPECTED_TRANSACTION_HEADER = (
"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"
)
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",
)
category = Category.objects.create(
id=UUID("30000000-0000-4000-8000-000000000001"),
name="Synthetic Food",
kind=Category.Kind.EXPENSE,
)
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
assert package.read("transactions.csv").decode("utf-8").splitlines()[0] == EXPECTED_TRANSACTION_HEADER
for name in EXPECTED_FILES:
content = package.read(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

149
tests/test_export_command.py

@ -0,0 +1,149 @@
import json
from datetime import datetime, timezone as datetime_timezone
from io import StringIO
import pytest
from django.core.management import call_command
from django.core.management.base import CommandError
from finance.exports import ExportArtifact, ExportError, write_export
def synthetic_artifact(filename="gizmoduck-export-20260907T120000Z.zip"):
return ExportArtifact(
filename=filename,
content=b"synthetic zip bytes",
version=1,
row_counts={"accounts.csv": 2, "transactions.csv": 4},
)
@pytest.fixture
def export_root(tmp_path, settings):
settings.EXPORT_ROOT = tmp_path
return tmp_path
def test_write_export_generates_a_plain_name_and_publishes_complete_content(
export_root, monkeypatch
):
now = datetime(2026, 9, 7, 12, 0, tzinfo=datetime_timezone.utc)
monkeypatch.setattr("finance.exports.timezone.now", lambda: now)
monkeypatch.setattr("finance.exports.build_export", synthetic_artifact)
result = write_export()
assert result.name == "gizmoduck-export-20260907T120000Z.zip"
assert result.path == export_root / result.name
assert result.path.read_bytes() == b"synthetic zip bytes"
assert list(export_root.iterdir()) == [result.path]
@pytest.mark.parametrize(
"name",
[
"../outside.zip",
"nested/outside.zip",
"nested\\outside.zip",
"/tmp/outside.zip",
"C:\\outside.zip",
"export.txt",
".zip",
],
)
def test_write_export_rejects_non_plain_zip_names(export_root, name):
with pytest.raises(ExportError, match="invalid_name"):
write_export(name)
assert list(export_root.iterdir()) == []
def test_write_export_rejects_existing_target_before_snapshot(export_root, monkeypatch):
existing = export_root / "existing.zip"
existing.write_bytes(b"original")
def unexpected_snapshot():
raise AssertionError("snapshot must not be built")
monkeypatch.setattr("finance.exports.build_export", unexpected_snapshot)
with pytest.raises(ExportError, match="target_exists"):
write_export(existing.name)
assert existing.read_bytes() == b"original"
def test_write_export_rejects_destination_symlink(export_root, monkeypatch):
outside = export_root.parent / "outside-export.zip"
outside.write_bytes(b"outside original")
link = export_root / "linked.zip"
try:
link.symlink_to(outside)
except OSError:
pytest.skip("Symlink creation unavailable")
monkeypatch.setattr("finance.exports.build_export", synthetic_artifact)
with pytest.raises(ExportError, match="invalid_name"):
write_export(link.name)
assert outside.read_bytes() == b"outside original"
assert link.is_symlink()
def test_write_export_removes_reservation_and_temporary_file_on_snapshot_failure(
export_root, monkeypatch
):
def failed_snapshot():
raise RuntimeError("SYNTHETIC PRIVATE DATABASE DETAIL")
monkeypatch.setattr("finance.exports.build_export", failed_snapshot)
with pytest.raises(ExportError, match="export_failed") as caught:
write_export("failed.zip")
assert "PRIVATE" not in str(caught.value)
assert list(export_root.iterdir()) == []
def test_write_export_cleans_up_if_atomic_publication_fails(export_root, monkeypatch):
monkeypatch.setattr("finance.exports.build_export", synthetic_artifact)
def failed_replace(_source, _destination):
raise OSError("SYNTHETIC PRIVATE FILESYSTEM DETAIL")
monkeypatch.setattr("finance.exports.os.replace", failed_replace)
with pytest.raises(ExportError, match="publication_failed") as caught:
write_export("failed.zip")
assert "PRIVATE" not in str(caught.value)
assert list(export_root.iterdir()) == []
def test_export_command_prints_only_safe_package_metadata(export_root, monkeypatch):
monkeypatch.setattr("finance.exports.build_export", synthetic_artifact)
output = StringIO()
call_command("export_finances", "--output", "synthetic.zip", stdout=output)
payload = json.loads(output.getvalue())
assert payload == {
"name": "synthetic.zip",
"version": 1,
"rows": {"accounts.csv": 2, "transactions.csv": 4},
}
assert str(export_root) not in output.getvalue()
assert "synthetic zip bytes" not in output.getvalue()
def test_export_command_converts_private_failure_to_safe_code(export_root, monkeypatch):
def failed_snapshot():
raise RuntimeError("SYNTHETIC PRIVATE DATABASE DETAIL")
monkeypatch.setattr("finance.exports.build_export", failed_snapshot)
with pytest.raises(CommandError, match="export_failed") as caught:
call_command("export_finances", "--output", "failed.zip")
assert "PRIVATE" not in str(caught.value)
assert list(export_root.iterdir()) == []

63
tests/test_export_view.py

@ -0,0 +1,63 @@
import json
from io import BytesIO
from zipfile import ZipFile
import pytest
from django.test import Client
from django.urls import reverse
pytestmark = pytest.mark.django_db(transaction=True)
def test_export_index_explains_the_portable_package(client):
response = client.get(reverse("export-index"))
assert response.status_code == 200
body = response.content.decode()
assert "Export financial data" in body
assert "accounts" in body
assert "transactions" in body
assert "source records" in body
assert "ZIP" in body
assert f'action="{reverse("export-download")}"' in body
def test_export_download_posts_an_in_memory_versioned_zip(client, settings, tmp_path):
settings.EXPORT_ROOT = tmp_path
response = client.post(
reverse("export-download"),
{"output": "../../outside.zip", "path": str(tmp_path / "chosen.zip")},
)
assert response.status_code == 200
assert response["Content-Type"] == "application/zip"
assert response["Content-Disposition"].startswith(
'attachment; filename="gizmoduck-export-'
)
assert response["Content-Disposition"].endswith('.zip"')
with ZipFile(BytesIO(response.content)) as package:
manifest = json.loads(package.read("manifest.json"))
assert manifest["format"] == "gizmoduck-export"
assert manifest["version"] == 1
assert list(tmp_path.iterdir()) == []
def test_export_download_requires_post(client):
assert client.get(reverse("export-download")).status_code == 405
def test_export_download_requires_csrf():
guarded = Client(enforce_csrf_checks=True)
response = guarded.post(reverse("export-download"))
assert response.status_code == 403
def test_exports_are_linked_from_other_pages(client):
response = client.get(reverse("transaction-list"))
assert response.status_code == 200
assert f'href="{reverse("export-index")}">Exports</a>' in response.content.decode()
Loading…
Cancel
Save