wlfb/bin
SHA256
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: CSV-Import ueber bestehende Pipeline mit dreiwertiger Saldo-Pruefung

Browse Source 
Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
wlfb
2026-07-19 21:45:53 +02:00
parent e2d713e960
commit ea4858c4de
7 changed files with 354 additions and 50 deletions
Download Patch File Download Diff File Expand all files Collapse all files

31
finance/app/routers/imports.py
View File

@@ -11,7 +11,7 @@ from app.auth import require_auth
from app.config import get_settings
from app.db import get_session
from app.models.tables import Statement, Transaction
from app.services.importer import process_pdf
from app.services.importer import process_file
router = APIRouter(prefix="/api/imports", tags=["imports"],
dependencies=[Depends(require_auth)])
@@ -50,22 +50,22 @@ class TransactionOut(BaseModel):
class PreviewOut(BaseModel):
statement: StatementOut
transactions: list[TransactionOut]
balance_ok: bool
balance_ok: bool | None
duplicates: int
@router.post("/upload", response_model=StatementOut, status_code=201)
def upload(file: UploadFile, session: Session = Depends(get_session)):
safe_name = Path(file.filename or "upload.pdf").name
if not safe_name or not safe_name.lower().endswith(".pdf"):
raise HTTPException(400, "Nur PDF-Dateien")
if not safe_name or not safe_name.lower().endswith((".pdf", ".csv")):
raise HTTPException(400, "Nur PDF- oder CSV-Dateien")
settings = get_settings()
inbox_dir = settings.inbox_dir
inbox_dir.mkdir(parents=True, exist_ok=True)
dest = inbox_dir / safe_name
with dest.open("wb") as f:
f.write(file.file.read())
stmt = process_pdf(session, dest)
stmt = process_file(session, dest)
return StatementOut.model_validate(stmt)
@@ -74,9 +74,10 @@ def scan_inbox(session: Session = Depends(get_session)):
settings = get_settings()
inbox_dir = settings.inbox_dir
inbox_dir.mkdir(parents=True, exist_ok=True)
paths = sorted(inbox_dir.glob("*.pdf")) + sorted(inbox_dir.glob("*.csv"))
results = []
for path in sorted(inbox_dir.glob("*.pdf")):
stmt = process_pdf(session, path)
for path in paths:
stmt = process_file(session, path)
results.append(StatementOut.model_validate(stmt))
return results
@@ -96,11 +97,17 @@ def preview(statement_id: int, session: Session = Depends(get_session)):
select(Transaction).where(Transaction.statement_id == statement_id)
.order_by(Transaction.booking_date, Transaction.id)
).scalars().all()
balance_ok = (
stmt.opening_balance is not None and stmt.closing_balance is not None
and (stmt.opening_balance + sum((t.amount for t in txs), Decimal("0"))
- stmt.closing_balance) == 0
)
# Dreiwertig (Ausbaustufe 3 Task 3): manche CSV-Formate (z.B. HVB, DKB)
# liefern keine Saldodaten - None statt False, damit die Vorschau nicht
# faelschlich einen Saldo-Fehler anzeigt, wo schlicht keine Pruefung
# moeglich ist.
if stmt.opening_balance is None or stmt.closing_balance is None:
balance_ok = None
else:
balance_ok = (
stmt.opening_balance + sum((t.amount for t in txs), Decimal("0"))
- stmt.closing_balance
) == 0
duplicates = sum(1 for t in txs if t.is_duplicate)
return PreviewOut(
statement=StatementOut.model_validate(stmt),

118
finance/app/services/importer.py
View File

@@ -1,3 +1,5 @@
from datetime import date
from decimal import Decimal
from pathlib import Path
from sqlalchemy import select
@@ -5,7 +7,8 @@ from sqlalchemy.orm import Session
from app.config import get_settings
from app.models.tables import Account, Statement, Transaction
from app.parsers.base import ParserError
from app.parsers.base import ParsedStatement, ParserError
from app.parsers.csv_formats import parse_csv
from app.parsers.registry import parse_pdf
from app.parsers.validate import balance_difference, dedup_hash
from app.services.categorize import apply_rules
@@ -14,6 +17,17 @@ from app.services.categorize import apply_rules
def _find_or_create_account(session: Session, bank: str, iban: str | None, filename: str) -> Account:
if not iban:
iban = f"UNBEKANNT-{filename}"
if bank == "hvb_csv":
# HVB-CSV traegt nur die Kontonummer (keine IBAN). Konto-Zuordnung
# ueber "endet auf Kontonummer" gegen alle bestehenden IBANs -
# dokumentierte Einschraenkung (siehe Formatreferenz Task 2/3).
for existing in session.execute(select(Account)).scalars():
if existing.iban.endswith(iban):
return existing
acc = Account(bank=bank, iban=iban, name=iban, type="giro")
session.add(acc)
session.flush()
return acc
acc = session.execute(select(Account).where(Account.iban == iban)).scalar()
if acc is not None:
return acc
@@ -23,6 +37,20 @@ def _find_or_create_account(session: Session, bank: str, iban: str | None, filen
return acc
def _apply_anchor_autofill(account: Account, anchor: tuple[date, Decimal] | None) -> None:
"""CSV-Anker-Regel (Ausbaustufe 3 Task 3): der Datei-Kontostand wird nur
uebernommen, wenn er neuer ist als ein evtl. vorhandener Anker (oder noch
keiner existiert). Ein neuerer, manuell per PATCH gesetzter Anker bleibt
unangetastet - so entkommt kein aelterer CSV-Export einem bewusst
gesetzten aktuellen Stand."""
if anchor is None:
return
anchor_date, anchor_balance = anchor
if account.anchor_date is None or anchor_date >= account.anchor_date:
account.anchor_date = anchor_date
account.anchor_balance = anchor_balance
def _error_statement(session: Session, filename: str, message: str) -> Statement:
stmt = Statement(filename=filename, bank="unbekannt", account_id=None,
status="error", error_message=message)
@@ -32,23 +60,18 @@ def _error_statement(session: Session, filename: str, message: str) -> Statement
return stmt
def process_pdf(session: Session, path: Path) -> Statement:
def _finish_statement(session: Session, path: Path, parsed: ParsedStatement,
check_balance: bool,
anchor: tuple[date, Decimal] | None) -> Statement:
"""Gemeinsamer Abschluss fuer PDF- und CSV-Importe, sobald ein
ParsedStatement vorliegt: Konto zuordnen, Statement+Transaktionen als
Entwurf anlegen, optional den Saldo pruefen, Datei aus dem Posteingang
verschieben. `check_balance=False` (CSV-Formate ohne Saldodaten, z.B.
HVB/DKB) laesst den Entwurf OHNE Fehler stehen statt einer
Saldo-Differenz-Pruefung, die auf None-Feldern crashen wuerde."""
settings = get_settings()
filename = path.name
try:
parsed = parse_pdf(path)
except ParserError as exc:
return _error_statement(session, filename, str(exc))
except Exception as exc:
# Fremdbibliotheken (pdfplumber/pdfminer) werfen bei strukturell
# kaputten PDFs eigene Exception-Typen statt ParserError (z.B.
# PdfminerException "No /Root object!"). Ohne diesen Fang wuerde ein
# kaputtes Upload zu einem unbehandelten 500 fuehren statt zu einem
# sauberen Fehler-Import wie bei "Bank nicht erkannt". Die Datei
# bleibt dabei unangetastet im Posteingang.
return _error_statement(session, filename, f"PDF nicht lesbar: {exc}")
account = _find_or_create_account(session, parsed.bank, parsed.iban, filename)
stmt = Statement(
@@ -64,13 +87,14 @@ def process_pdf(session: Session, path: Path) -> Statement:
session.add(stmt)
session.flush()
diff = balance_difference(parsed)
if diff != 0:
stmt.status = "error"
stmt.error_message = f"Saldo-Differenz {diff} EUR"
session.commit()
session.refresh(stmt)
return stmt
if check_balance:
diff = balance_difference(parsed)
if diff != 0:
stmt.status = "error"
stmt.error_message = f"Saldo-Differenz {diff} EUR"
session.commit()
session.refresh(stmt)
return stmt
drafts = []
for t in parsed.transactions:
@@ -100,10 +124,11 @@ def process_pdf(session: Session, path: Path) -> Statement:
apply_rules(session, drafts)
stmt.status = "draft"
_apply_anchor_autofill(account, anchor)
# Invariant: a committed draft implies the PDF left the inbox. Move the
# file before committing so a failed move can never leave a committed
# draft with the PDF still sitting in the inbox.
# Invariant: a committed draft implies the source file left the inbox.
# Move the file before committing so a failed move can never leave a
# committed draft with the file still sitting in the inbox.
uploads_dir = settings.uploads_dir
uploads_dir.mkdir(parents=True, exist_ok=True)
try:
@@ -116,3 +141,46 @@ def process_pdf(session: Session, path: Path) -> Statement:
session.refresh(stmt)
return stmt
def process_pdf(session: Session, path: Path) -> Statement:
filename = path.name
try:
parsed = parse_pdf(path)
except ParserError as exc:
return _error_statement(session, filename, str(exc))
except Exception as exc:
# Fremdbibliotheken (pdfplumber/pdfminer) werfen bei strukturell
# kaputten PDFs eigene Exception-Typen statt ParserError (z.B.
# PdfminerException "No /Root object!"). Ohne diesen Fang wuerde ein
# kaputtes Upload zu einem unbehandelten 500 fuehren statt zu einem
# sauberen Fehler-Import wie bei "Bank nicht erkannt". Die Datei
# bleibt dabei unangetastet im Posteingang.
return _error_statement(session, filename, f"PDF nicht lesbar: {exc}")
return _finish_statement(session, path, parsed, check_balance=True, anchor=None)
def process_csv(session: Session, path: Path) -> Statement:
filename = path.name
try:
parsed_csv = parse_csv(path)
except ParserError as exc:
return _error_statement(session, filename, str(exc))
except Exception as exc:
# Analog zu process_pdf: eine strukturell kaputte/unlesbare CSV darf
# nicht als unbehandelter 500 durchschlagen.
return _error_statement(session, filename, f"CSV nicht lesbar: {exc}")
return _finish_statement(session, path, parsed_csv.statement,
check_balance=parsed_csv.balance_checkable,
anchor=parsed_csv.anchor)
def process_file(session: Session, path: Path) -> Statement:
"""Dateiendungs-Weiche fuer die Import-Pipeline (Ausbaustufe 3 Task 3):
.csv laeuft ueber den CSV-Formatparser, alles andere weiterhin ueber den
PDF-Parser."""
if path.suffix.lower() == ".csv":
return process_csv(session, path)
return process_pdf(session, path)

6
finance/app/templates/_preview_table.html
View File

@@ -18,8 +18,10 @@
</table>
<p>
Saldo-Status:
{% if balance_ok %}
<span class="badge badge-ok">Saldo stimmt</span>
{% if balance_ok is none %}
<span class="badge">Saldo-Prüfung: nicht verfügbar (Format ohne Saldodaten)</span>
{% elif balance_ok %}
<span class="badge badge-ok">Saldo plausibel</span>
{% else %}
<span class="badge badge-error">Saldo-Differenz – bitte prüfen</span>
{% endif %}

25
finance/app/templates/hilfe.html
View File

@@ -16,15 +16,24 @@
Über die Übersicht umbenennbar (z. B. "DKB Giro").
</li>
<li>
<strong>Kontoauszug-Import:</strong> PDF in die Drop-Zone auf der
Import-Seite ziehen (oder in die Inbox
<strong>Kontoauszug-Import:</strong> primärer Weg ist der
Kontoumsatz-CSV-Export der Bank (DKB, VR-Bank, HypoVereinsbank) – in die
Drop-Zone auf der Import-Seite ziehen (oder in die Inbox
<code>~/.local/share/finance_pod/data/inbox/</code> legen und "Inbox
scannen"). Jeder Import wird geprüft (Anfangssaldo + Buchungen = Endsaldo)
und landet als Entwurf: erst nach Kontrolle der Vorschau "Übernehmen"
klicken. Duplikate werden erkannt und nicht doppelt übernommen. Bestätigte
Importe lassen sich als Ganzes zurückrollen (Auszug + alle seine
Buchungen), z. B. um einen Auszug mit einem verbesserten Parser neu zu
importieren; einzelne Buchungen bleiben unlöschbar.
scannen"). PDF-Kontoauszüge funktionieren weiterhin als Fallback, falls
kein CSV-Export vorliegt. Jeder Import landet als Entwurf: erst nach
Kontrolle der Vorschau "Übernehmen" klicken. Der Saldo-Status in der
Vorschau ist dreiwertig – "Saldo plausibel" (Anfangssaldo + Buchungen =
Endsaldo bzw. lückenlose Saldo-Kette), eine rote Saldo-Differenz, oder
"Saldo-Prüfung nicht verfügbar", wenn das Format (z. B. HVB-CSV) keine
Saldodaten liefert – dort lohnt ein Blick in die Zeilenzahl und
Stichproben vor dem Übernehmen. CSV-Importe mit Datei-Kontostand (DKB)
oder lückenloser Saldo-Kette (VR) aktualisieren beim Import automatisch
den Konto-Saldo-Anker, sofern er dadurch nicht älter würde als ein bereits
vorhandener. Duplikate werden erkannt und nicht doppelt übernommen.
Bestätigte Importe lassen sich als Ganzes zurückrollen (Auszug + alle
seine Buchungen), z. B. um einen Auszug mit einem verbesserten Parser neu
zu importieren; einzelne Buchungen bleiben unlöschbar.
</li>
<li>
<strong>Kategorien &amp; Regeln:</strong> Buchungen lassen sich kategorisieren;

6
finance/app/templates/import.html
View File

@@ -4,14 +4,14 @@
<h1>Import</h1>
<div id="dropzone">
Kontoauszug-PDF hier ablegen oder klicken zum Hochladen
<input type="file" id="file-input" accept="application/pdf" style="display:none">
Kontoumsatz-CSV oder Kontoauszug-PDF hier ablegen oder klicken zum Hochladen
<input type="file" id="file-input" accept=".csv,.pdf,application/pdf,text/csv" style="display:none">
</div>
<button type="button" hx-post="/api/imports/scan-inbox" hx-swap="none"
hx-on::after-request="if(event.detail.successful){htmx.trigger(document.getElementById('imports'), 'refresh')}">
Inbox scannen
</button>
<p class="muted">Durchsucht <code>~/.local/share/finance_pod/data/inbox/</code> nach neuen PDF-Kontoauszügen.</p>
<p class="muted">Durchsucht <code>~/.local/share/finance_pod/data/inbox/</code> nach neuen CSV- oder PDF-Kontoauszügen.</p>
<p id="upload-message"></p>
<h2>Importe</h2>

35
finance/tests/test_gui.py
View File

@@ -62,6 +62,41 @@ def test_import_list_fragment_renders_after_login(client):
assert "Noch keine Imports" in r.text
def test_preview_fragment_shows_balance_ok_none_for_format_without_saldo(
client, tmp_path, monkeypatch):
# UX-Regel: Saldo-Status ist IMMER sichtbar, auch wenn das Format (hier
# HVB-CSV) keine Saldodaten liefert - dreiwertig statt faelschlich "rot".
monkeypatch.setenv("FB_INBOX_DIR", str(tmp_path / "inbox"))
monkeypatch.setenv("FB_UPLOADS_DIR", str(tmp_path / "uploads"))
from app.config import get_settings
get_settings.cache_clear()
from tests.test_import_api import _hvb_csv_bytes
sid = client.post("/api/imports/upload", headers=H,
files={"file": ("HVB_gui.csv", _hvb_csv_bytes(), "text/csv")}
).json()["id"]
client.post("/login", data={"username": "admin", "password": "geheim"})
r = client.get(f"/import/{sid}/preview-fragment")
assert r.status_code == 200
assert "Saldo-Prüfung: nicht verfügbar (Format ohne Saldodaten)" in r.text
def test_preview_fragment_shows_balance_ok_true_for_vr_csv(client, tmp_path, monkeypatch):
monkeypatch.setenv("FB_INBOX_DIR", str(tmp_path / "inbox"))
monkeypatch.setenv("FB_UPLOADS_DIR", str(tmp_path / "uploads"))
from app.config import get_settings
get_settings.cache_clear()
from tests.test_import_api import _vr_csv_bytes
sid = client.post("/api/imports/upload", headers=H,
files={"file": ("VR_gui.csv", _vr_csv_bytes(), "text/csv")}
).json()["id"]
client.post("/login", data={"username": "admin", "password": "geheim"})
r = client.get(f"/import/{sid}/preview-fragment")
assert r.status_code == 200
assert "Saldo plausibel" in r.text
def test_version_visible(client):
from pathlib import Path
version = (Path(__file__).resolve().parent.parent / "VERSION").read_text().strip()

183
finance/tests/test_import_api.py
View File

@@ -1,3 +1,5 @@
import csv
import io
import re
from datetime import date
from decimal import Decimal
@@ -295,3 +297,184 @@ def test_upload_corrupt_pdf_produces_error_statement_not_500(client, tmp_path, m
# Datei bleibt im Posteingang liegen, wird nicht ins Uploads-Verzeichnis verschoben.
assert (tmp_path / "inbox" / "kaputt.pdf").exists()
assert not (tmp_path / "uploads" / "kaputt.pdf").exists()
# ---------------------------------------------------------------------------
# CSV-Import (Ausbaustufe 3, Task 3): rein SYNTHETISCHE Mini-CSVs, nach der
# in docs/superpowers/plans/2026-07-19-ausbaustufe-3.md dokumentierten
# Formatreferenz konstruiert. KEINE echten Kontodaten.
# ---------------------------------------------------------------------------
def _dkb_csv_bytes(kontostand_date="30.06.2026", kontostand_amount="1.000,00 €") -> bytes:
buf = io.StringIO()
w = csv.writer(buf, delimiter=";", quoting=csv.QUOTE_ALL)
w.writerow(["Girokonto", "DE12500105170001234567"])
w.writerow(["Zeitraum:", "01.06.2026 - 30.06.2026"])
w.writerow([f"Kontostand vom {kontostand_date}:", kontostand_amount])
w.writerow([])
w.writerow(["Buchungsdatum", "Wertstellung", "Status", "Zahlungspflichtige*r",
"Zahlungsempfänger*in", "Verwendungszweck", "Umsatztyp", "IBAN",
"Betrag (€)", "Gläubiger-ID", "Mandatsreferenz", "Kundenreferenz"])
# Datei ist absteigend sortiert (neueste Buchung zuerst).
w.writerow(["15.06.26", "15.06.26", "Gebucht", "Max Mustermann", "Beispiel GmbH",
"Testzweck Miete", "Lastschrift", "DE12500105170001234567",
"-50,00", "", "", ""])
w.writerow(["10.06.26", "10.06.26", "Storniert", "Max Mustermann", "Nicht Gebucht GmbH",
"sollte ignoriert werden", "Lastschrift", "DE12500105170001234567",
"-999,00", "", "", ""])
w.writerow(["01.06.26", "01.06.26", "Gebucht", "Arbeitgeber XY", "Max Mustermann",
"Testzweck Gehalt", "Gutschrift", "DE12500105170001234567",
"1500,00", "", "", ""])
return ("" + buf.getvalue()).encode("utf-8")
def _vr_csv_bytes() -> bytes:
buf = io.StringIO()
w = csv.writer(buf, delimiter=";")
w.writerow(["Bezeichnung Auftragskonto", "IBAN Auftragskonto", "BIC Auftragskonto",
"Bankname Auftragskonto", "Buchungstag", "Valutadatum",
"Name Zahlungsbeteiligter", "IBAN Zahlungsbeteiligter",
"BIC (SWIFT-Code) Zahlungsbeteiligter", "Buchungstext", "Verwendungszweck",
"Betrag", "Waehrung", "Saldo nach Buchung", "Bemerkung",
"Gekennzeichneter Umsatz", "Glaeubiger ID", "Mandatsreferenz"])
# Datei ist absteigend sortiert (neueste Buchung zuerst).
w.writerow(["Girokonto", "DE02600501010000123456", "SOLADEST600", "VR Bank Test",
"15.06.2026", "15.06.2026", "Test Empfaenger", "DE00000000000000000000",
"", "Ueberweisung", "Testzweck 2", "100,00", "EUR", "1.100,00", "", "", "", ""])
w.writerow(["Girokonto", "DE02600501010000123456", "SOLADEST600", "VR Bank Test",
"01.06.2026", "01.06.2026", "Test Arbeitgeber", "DE00000000000000000000",
"", "Gutschrift", "Testzweck 1", "1000,00", "EUR", "1.000,00", "", "", "", ""])
text = ("" + buf.getvalue()).replace("\n", "\r\n")
return text.encode("utf-8")
def _hvb_csv_bytes(kontonummer="1234567") -> bytes:
buf = io.StringIO()
w = csv.writer(buf, delimiter=";")
w.writerow(["Kontonummer", "Buchungsdatum", "Valuta", "Verwendungszweck", "Betrag",
"Waehrung"])
# Datei ist aufsteigend sortiert (aelteste Buchung zuerst).
w.writerow([kontonummer, "01.06.2026", "01.06.2026", "Testzweck A", "500,00", "EUR"])
w.writerow([kontonummer, "15.06.2026", "15.06.2026", "Testzweck B", "-20,00", "EUR"])
return buf.getvalue().encode("utf-16")
def _upload_csv(client, name: str, content: bytes):
return client.post("/api/imports/upload", headers=H,
files={"file": (name, content, "text/csv")})
def test_upload_csv_vr_balance_ok_true_and_confirm(client, tmp_path, monkeypatch):
monkeypatch.setenv("FB_INBOX_DIR", str(tmp_path / "inbox"))
monkeypatch.setenv("FB_UPLOADS_DIR", str(tmp_path / "uploads"))
from app.config import get_settings
get_settings.cache_clear()
r = _upload_csv(client, "VR_test.csv", _vr_csv_bytes())
assert r.status_code == 201
body = r.json()
assert body["status"] == "draft"
assert body["bank"] == "vr_csv"
sid = body["id"]
prev = client.get(f"/api/imports/{sid}/preview", headers=H).json()
assert prev["balance_ok"] is True
assert len(prev["transactions"]) == 2
assert client.post(f"/api/imports/{sid}/confirm", headers=H).status_code == 200
accounts = client.get("/api/accounts", headers=H).json()
acc = next(a for a in accounts if a["iban"] == "DE02600501010000123456")
assert acc["anchor_balance"] == "1100.00"
assert acc["anchor_date"] == "2026-06-15"
def test_upload_csv_hvb_balance_ok_none(client, tmp_path, monkeypatch):
monkeypatch.setenv("FB_INBOX_DIR", str(tmp_path / "inbox"))
monkeypatch.setenv("FB_UPLOADS_DIR", str(tmp_path / "uploads"))
from app.config import get_settings
get_settings.cache_clear()
r = _upload_csv(client, "HVB_test.csv", _hvb_csv_bytes())
assert r.status_code == 201
body = r.json()
assert body["status"] == "draft"
assert body["bank"] == "hvb_csv"
sid = body["id"]
prev = client.get(f"/api/imports/{sid}/preview", headers=H).json()
assert prev["balance_ok"] is None
assert len(prev["transactions"]) == 2
assert client.post(f"/api/imports/{sid}/confirm", headers=H).status_code == 200
# HVB traegt die Kontonummer als iban im Konto (dokumentierte Einschraenkung).
accounts = client.get("/api/accounts", headers=H).json()
acc = next(a for a in accounts if a["iban"] == "1234567")
assert acc["anchor_date"] is None # HVB liefert keinen Anker
def test_upload_csv_dkb_balance_ok_none_and_sets_account_anchor(client, tmp_path, monkeypatch):
monkeypatch.setenv("FB_INBOX_DIR", str(tmp_path / "inbox"))
monkeypatch.setenv("FB_UPLOADS_DIR", str(tmp_path / "uploads"))
from app.config import get_settings
get_settings.cache_clear()
r = _upload_csv(client, "DKB_test.csv", _dkb_csv_bytes())
assert r.status_code == 201
body = r.json()
assert body["status"] == "draft"
assert body["bank"] == "dkb_csv"
sid = body["id"]
prev = client.get(f"/api/imports/{sid}/preview", headers=H).json()
assert prev["balance_ok"] is None
# Storno-Zeile ("Storniert") darf nicht importiert werden.
assert len(prev["transactions"]) == 2
assert client.post(f"/api/imports/{sid}/confirm", headers=H).status_code == 200
accounts = client.get("/api/accounts", headers=H).json()
acc = next(a for a in accounts if a["iban"] == "DE12500105170001234567")
assert acc["anchor_date"] == "2026-06-30"
assert acc["anchor_balance"] == "1000.00"
def test_upload_csv_dkb_does_not_downgrade_newer_manual_anchor(client, db, tmp_path, monkeypatch):
monkeypatch.setenv("FB_INBOX_DIR", str(tmp_path / "inbox"))
monkeypatch.setenv("FB_UPLOADS_DIR", str(tmp_path / "uploads"))
from app.config import get_settings
get_settings.cache_clear()
from app.models.tables import Account
# Konto existiert bereits mit einem manuellen Anker NACH dem CSV-Kontostand-Datum
# (30.06.2026) - dieser neuere, manuell gesetzte Anker darf durch den CSV-Import
# NICHT ueberschrieben/zurueckdatiert werden.
acc = Account(bank="dkb_csv", iban="DE12500105170001234567", name="DKB Giro",
type="giro", anchor_date=date(2026, 7, 10), anchor_balance=Decimal("42.00"))
db.add(acc)
db.commit()
r = _upload_csv(client, "DKB_test.csv", _dkb_csv_bytes())
assert r.status_code == 201
sid = r.json()["id"]
assert client.post(f"/api/imports/{sid}/confirm", headers=H).status_code == 200
accounts = client.get("/api/accounts", headers=H).json()
acc_after = next(a for a in accounts if a["iban"] == "DE12500105170001234567")
assert acc_after["anchor_date"] == "2026-07-10"
assert acc_after["anchor_balance"] == "42.00"
def test_upload_rejects_non_pdf_non_csv(client, tmp_path, monkeypatch):
monkeypatch.setenv("FB_INBOX_DIR", str(tmp_path / "inbox"))
monkeypatch.setenv("FB_UPLOADS_DIR", str(tmp_path / "uploads"))
from app.config import get_settings
get_settings.cache_clear()
r = client.post("/api/imports/upload", headers=H,
files={"file": ("x.txt", b"not a csv or pdf", "text/plain")})
assert r.status_code == 400
assert "PDF" in r.text and "CSV" in r.text
def test_scan_inbox_picks_up_csv(client, tmp_path, monkeypatch):
inbox = tmp_path / "inbox"
monkeypatch.setenv("FB_INBOX_DIR", str(inbox))
monkeypatch.setenv("FB_UPLOADS_DIR", str(tmp_path / "uploads"))
from app.config import get_settings
get_settings.cache_clear()
inbox.mkdir(parents=True)
(inbox / "VR_scan.csv").write_bytes(_vr_csv_bytes())
r = client.post("/api/imports/scan-inbox", headers=H)
assert r.status_code == 200
results = r.json()
assert len(results) == 1
assert results[0]["bank"] == "vr_csv"
assert results[0]["status"] == "draft"