481 lines
22 KiB
Python
481 lines
22 KiB
Python
import csv
|
||
import io
|
||
import re
|
||
from datetime import date
|
||
from decimal import Decimal
|
||
from pathlib import Path
|
||
|
||
import pytest
|
||
|
||
from app.parsers.base import ParsedStatement, ParsedTransaction
|
||
|
||
H = {"Authorization": "Bearer test-key"}
|
||
|
||
FAKE = ParsedStatement(
|
||
bank="dkb", iban="DE02120300000000202051",
|
||
period_start=date(2026, 6, 1), period_end=date(2026, 6, 30),
|
||
opening_balance=Decimal("100.00"), closing_balance=Decimal("40.00"),
|
||
transactions=[ParsedTransaction(date(2026, 6, 3), date(2026, 6, 3),
|
||
Decimal("-60.00"), "Miete Juni", "Vermieter")])
|
||
|
||
|
||
@pytest.fixture
|
||
def fake_parse(monkeypatch):
|
||
monkeypatch.setattr("app.services.importer.parse_pdf", lambda p: FAKE)
|
||
|
||
|
||
# Ein Auszug mit ZWEI identischen Buchungen (gleicher dedup_hash). Saldo bleibt
|
||
# konsistent (100 - 60 - 60 = -20). Beide muessen bestaetigt werden koennen.
|
||
FAKE_TWINS = ParsedStatement(
|
||
bank="dkb", iban="DE02120300000000202051",
|
||
period_start=date(2026, 6, 1), period_end=date(2026, 6, 30),
|
||
opening_balance=Decimal("100.00"), closing_balance=Decimal("-20.00"),
|
||
transactions=[
|
||
ParsedTransaction(date(2026, 6, 3), date(2026, 6, 3),
|
||
Decimal("-60.00"), "Miete Juni", "Vermieter"),
|
||
ParsedTransaction(date(2026, 6, 3), date(2026, 6, 3),
|
||
Decimal("-60.00"), "Miete Juni", "Vermieter"),
|
||
])
|
||
|
||
|
||
@pytest.fixture
|
||
def fake_parse_twins(monkeypatch):
|
||
monkeypatch.setattr("app.services.importer.parse_pdf", lambda p: FAKE_TWINS)
|
||
|
||
|
||
def _upload(client, name="auszug.pdf"):
|
||
return client.post("/api/imports/upload", headers=H,
|
||
files={"file": (name, b"%PDF-fake", "application/pdf")})
|
||
|
||
|
||
def test_upload_preview_confirm(client, fake_parse, 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(client)
|
||
assert r.status_code == 201
|
||
sid = r.json()["id"]
|
||
prev = client.get(f"/api/imports/{sid}/preview", headers=H).json()
|
||
assert prev["balance_ok"] is True
|
||
assert len(prev["transactions"]) == 1
|
||
assert client.post(f"/api/imports/{sid}/confirm", headers=H).status_code == 200
|
||
txs = client.get("/api/transactions", headers=H).json()
|
||
assert len(txs) == 1 and txs[0]["status"] == "confirmed"
|
||
# Zweiter Import desselben Auszugs: alles Duplikate
|
||
sid2 = _upload(client, "auszug2.pdf").json()["id"]
|
||
prev2 = client.get(f"/api/imports/{sid2}/preview", headers=H).json()
|
||
assert prev2["duplicates"] == 1
|
||
client.post(f"/api/imports/{sid2}/confirm", headers=H)
|
||
assert len(client.get("/api/transactions", headers=H).json()) == 1
|
||
|
||
|
||
def test_confirm_rechecks_duplicates_across_two_drafts(client, fake_parse, tmp_path, monkeypatch):
|
||
# Beide Auszuege werden importiert, bevor einer bestaetigt ist -> zum
|
||
# Importzeitpunkt ist keine Buchung ein Duplikat (dedup prueft nur gegen
|
||
# bestaetigte). Erst beim zweiten Confirm faellt auf, dass die Buchung nun
|
||
# bereits bestaetigt existiert. Der Re-Check muss sie ueberspringen, sonst
|
||
# wird dieselbe Buchung doppelt bestaetigt.
|
||
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()
|
||
sid1 = _upload(client, "a1.pdf").json()["id"]
|
||
sid2 = _upload(client, "a2.pdf").json()["id"]
|
||
assert client.post(f"/api/imports/{sid1}/confirm", headers=H).status_code == 200
|
||
assert client.post(f"/api/imports/{sid2}/confirm", headers=H).status_code == 200
|
||
# Zweites Confirm darf die Buchung nicht erneut bestaetigen.
|
||
assert len(client.get("/api/transactions", headers=H).json()) == 1
|
||
# Zweites Confirm eines bereits bestaetigten Imports -> 409.
|
||
assert client.post(f"/api/imports/{sid1}/confirm", headers=H).status_code == 409
|
||
|
||
|
||
def test_confirm_keeps_identical_twins_within_same_statement(client, fake_parse_twins, tmp_path, monkeypatch):
|
||
# Zwei identische Buchungen im SELBEN Auszug teilen sich den dedup_hash. Der
|
||
# Confirm-Re-Check darf nur statement-uebergreifende Duplikate abfangen,
|
||
# sonst wuerde die erste bestaetigte Buchung ihre Zwillingsbuchung als
|
||
# Duplikat markieren und den vom Preview geprueften Saldo zerstoeren.
|
||
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()
|
||
sid = _upload(client).json()["id"]
|
||
prev = client.get(f"/api/imports/{sid}/preview", headers=H).json()
|
||
assert prev["balance_ok"] is True
|
||
assert prev["duplicates"] == 0
|
||
assert client.post(f"/api/imports/{sid}/confirm", headers=H).status_code == 200
|
||
txs = client.get("/api/transactions", headers=H).json()
|
||
assert len(txs) == 2
|
||
assert all(t["status"] == "confirmed" for t in txs)
|
||
|
||
|
||
def test_upload_sanitizes_path_traversal_filename(client, fake_parse, 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": ("../evil.pdf", b"%PDF-fake", "application/pdf")})
|
||
assert r.status_code == 201
|
||
# Der Dateiname darf keine Pfad-Anteile aus dem Client uebernehmen: nur
|
||
# der Basisname landet im Uploads-Verzeichnis, nichts entkommt nach oben.
|
||
assert not (tmp_path / "evil.pdf").exists()
|
||
assert (tmp_path / "uploads" / "evil.pdf").exists()
|
||
assert not (tmp_path / "inbox" / "evil.pdf").exists()
|
||
|
||
|
||
def test_upload_rejects_non_pdf(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 pdf", "text/plain")})
|
||
assert r.status_code == 400
|
||
|
||
|
||
def test_delete_error_import_returns_204(client, db):
|
||
# Ein fehlgeschlagener Import (status="error") muss verworfen werden
|
||
# koennen - nur bestaetigte Importe sind vor dem Loeschen geschuetzt.
|
||
from app.models.tables import Statement
|
||
db.add(Statement(filename="kaputt.pdf", bank="dkb", status="error",
|
||
error_message="PDF nicht lesbar"))
|
||
db.commit()
|
||
sid = db.query(Statement).one().id
|
||
assert client.delete(f"/api/imports/{sid}", headers=H).status_code == 204
|
||
|
||
|
||
def test_rollback_confirmed_import_deletes_transactions_and_statement(client, fake_parse, tmp_path, monkeypatch):
|
||
# Ein bestaetigter Import laesst sich als Ganzes zurueckrollen: alle seine
|
||
# Buchungen UND der Auszug selbst verschwinden, das Konto bleibt bestehen
|
||
# (Name/IBAN fuer einen Re-Import mit verbessertem Parser).
|
||
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()
|
||
sid = _upload(client).json()["id"]
|
||
assert client.post(f"/api/imports/{sid}/confirm", headers=H).status_code == 200
|
||
account_id = client.get("/api/accounts", headers=H).json()[0]["id"]
|
||
|
||
r = client.post(f"/api/imports/{sid}/rollback", headers=H)
|
||
assert r.status_code == 200
|
||
body = r.json()
|
||
assert body == {"deleted_transactions": 1, "statement_id": sid}
|
||
|
||
txs = client.get("/api/transactions", headers=H).json()
|
||
assert len(txs) == 0
|
||
imports = client.get("/api/imports", headers=H).json()
|
||
assert all(s["id"] != sid for s in imports)
|
||
accounts = client.get("/api/accounts", headers=H).json()
|
||
assert any(a["id"] == account_id for a in accounts)
|
||
|
||
|
||
def test_rollback_then_reimport_restores_transactions(client, fake_parse, tmp_path, monkeypatch):
|
||
# Der eigentliche Zweck des Rollbacks: denselben Auszug erneut importieren
|
||
# koennen (z.B. nach einem verbesserten Parser), ohne dass alte Buchungen
|
||
# als Duplikate im Weg stehen.
|
||
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()
|
||
sid = _upload(client).json()["id"]
|
||
client.post(f"/api/imports/{sid}/confirm", headers=H)
|
||
client.post(f"/api/imports/{sid}/rollback", headers=H)
|
||
|
||
sid2 = _upload(client, "auszug_reimport.pdf").json()["id"]
|
||
prev = client.get(f"/api/imports/{sid2}/preview", headers=H).json()
|
||
assert prev["balance_ok"] is True
|
||
assert prev["duplicates"] == 0
|
||
assert client.post(f"/api/imports/{sid2}/confirm", headers=H).status_code == 200
|
||
txs = client.get("/api/transactions", headers=H).json()
|
||
assert len(txs) == 1 and txs[0]["status"] == "confirmed"
|
||
|
||
|
||
def test_rollback_draft_import_returns_409(client, fake_parse, 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()
|
||
sid = _upload(client).json()["id"]
|
||
r = client.post(f"/api/imports/{sid}/rollback", headers=H)
|
||
assert r.status_code == 409
|
||
|
||
|
||
def test_rollback_unknown_import_returns_404(client):
|
||
r = client.post("/api/imports/999999/rollback", headers=H)
|
||
assert r.status_code == 404
|
||
|
||
|
||
def _row_for(html: str, filename: str) -> str:
|
||
"""Isoliert den <tr>-Block eines bestimmten Import-Statements anhand seines
|
||
Dateinamens, damit Button-Zustaende eindeutig dem richtigen Statement
|
||
zugeordnet werden (statt nur global auf Disabled-Buttons zu zaehlen)."""
|
||
for row in html.split("<tr>"):
|
||
if filename in row:
|
||
return row
|
||
raise AssertionError(f"keine Zeile fuer {filename!r} gefunden")
|
||
|
||
|
||
def _has_active_button(row: str, label: str) -> bool:
|
||
for tag in re.finditer(r"<button[^>]*>" + re.escape(label) + r"</button>", row):
|
||
if "disabled" not in tag.group(0):
|
||
return True
|
||
return False
|
||
|
||
|
||
def _has_disabled_button(row: str, label: str) -> bool:
|
||
return bool(re.search(r"<button[^>]*disabled[^>]*>" + re.escape(label) + r"</button>", row))
|
||
|
||
|
||
def test_import_list_button_matrix(client, tmp_path, monkeypatch):
|
||
# UX-Regel: Bedienelemente sind immer sichtbar, ggf. ausgegraut (disabled),
|
||
# statt je nach Status komplett zu fehlen. Drei Statements mit den drei
|
||
# moeglichen Status erzeugen, dann pro Zeile die Button-Matrix pruefen.
|
||
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()
|
||
|
||
# parse_pdf direkt (nicht ueber die fake_parse-Fixture) patchen/zuruecksetzen,
|
||
# damit der dritte Upload weiter unten wieder den echten Parser durchlaeuft
|
||
# und an kaputten Bytes wie in test_upload_corrupt_pdf_produces_error_statement_not_500
|
||
# scheitert - monkeypatch.undo() wuerde hier auch die setenv-Aenderungen
|
||
# oben zuruecknehmen.
|
||
import app.services.importer as importer_mod
|
||
original_parse_pdf = importer_mod.parse_pdf
|
||
importer_mod.parse_pdf = lambda p: FAKE
|
||
try:
|
||
sid_draft = _upload(client, "entwurf.pdf").json()["id"]
|
||
sid_confirmed = _upload(client, "bestaetigt.pdf").json()["id"]
|
||
assert client.post(f"/api/imports/{sid_confirmed}/confirm", headers=H).status_code == 200
|
||
finally:
|
||
importer_mod.parse_pdf = original_parse_pdf
|
||
|
||
r_err = client.post("/api/imports/upload", headers=H,
|
||
files={"file": ("fehler.pdf", b"not a pdf", "application/pdf")})
|
||
assert r_err.status_code == 201
|
||
assert r_err.json()["status"] == "error"
|
||
|
||
client.post("/login", data={"username": "admin", "password": "geheim"})
|
||
r = client.get("/import/list")
|
||
assert r.status_code == 200
|
||
|
||
draft_row = _row_for(r.text, "entwurf.pdf")
|
||
assert _has_active_button(draft_row, "Übernehmen")
|
||
assert _has_active_button(draft_row, "Verwerfen")
|
||
assert _has_disabled_button(draft_row, "Import zurückrollen")
|
||
assert not _has_active_button(draft_row, "Import zurückrollen")
|
||
|
||
confirmed_row = _row_for(r.text, "bestaetigt.pdf")
|
||
assert _has_disabled_button(confirmed_row, "Übernehmen")
|
||
assert _has_disabled_button(confirmed_row, "Verwerfen")
|
||
assert _has_active_button(confirmed_row, "Import zurückrollen")
|
||
assert not _has_disabled_button(confirmed_row, "Import zurückrollen")
|
||
|
||
error_row = _row_for(r.text, "fehler.pdf")
|
||
assert _has_active_button(error_row, "Verwerfen")
|
||
assert _has_disabled_button(error_row, "Übernehmen")
|
||
assert _has_disabled_button(error_row, "Import zurückrollen")
|
||
|
||
|
||
def test_upload_corrupt_pdf_produces_error_statement_not_500(client, tmp_path, monkeypatch):
|
||
# Datei besteht die Endungspruefung (.pdf), ist aber strukturell kein
|
||
# gueltiges PDF -> pdfplumber wirft eine eigene Exception (keine
|
||
# ParserError). Das darf nicht als unbehandelter 500 durchschlagen,
|
||
# sondern muss wie "Bank nicht erkannt" als sauberer Fehler-Import
|
||
# sichtbar werden.
|
||
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": ("kaputt.pdf", b"not a pdf", "application/pdf")})
|
||
assert r.status_code == 201
|
||
body = r.json()
|
||
assert body["status"] == "error"
|
||
assert "PDF nicht lesbar" in body["error_message"]
|
||
# 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"
|