feat: CSV-Parser fuer DKB/VR/HVB-Kontoumsatz-Exporte
Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
245
finance/app/parsers/csv_formats.py
Normal file
245
finance/app/parsers/csv_formats.py
Normal file
@@ -0,0 +1,245 @@
|
||||
"""Parser für Kontoumsatz-CSV-Exporte (DKB, VR, HVB).
|
||||
|
||||
Siehe „CSV-Formatreferenz" in docs/superpowers/plans/2026-07-19-ausbaustufe-3.md
|
||||
für die verbindliche Spalten-/Formatspezifikation je Bank.
|
||||
"""
|
||||
import csv
|
||||
import io
|
||||
import re
|
||||
from dataclasses import dataclass
|
||||
from datetime import date
|
||||
from decimal import Decimal
|
||||
from pathlib import Path
|
||||
|
||||
from app.parsers.base import (ParsedStatement, ParsedTransaction, ParserError,
|
||||
parse_german_amount, parse_german_date)
|
||||
|
||||
RE_KONTOSTAND = re.compile(r"Kontostand vom (\d{2}\.\d{2}\.\d{4}):")
|
||||
RE_ZEITRAUM = re.compile(r"(\d{2}\.\d{2}\.\d{4})\s*-\s*(\d{2}\.\d{2}\.\d{4})")
|
||||
|
||||
|
||||
@dataclass
|
||||
class ParsedCsv:
|
||||
statement: ParsedStatement
|
||||
anchor: tuple[date, Decimal] | None
|
||||
balance_checkable: bool
|
||||
|
||||
|
||||
def _detect_encoding(raw: bytes) -> str:
|
||||
if raw.startswith(b"\xef\xbb\xbf"):
|
||||
return "utf-8-sig"
|
||||
if raw.startswith(b"\xff\xfe"):
|
||||
return "utf-16-le"
|
||||
if raw.startswith(b"\xfe\xff"):
|
||||
return "utf-16-be"
|
||||
sample = raw[:64]
|
||||
if b"\x00" in sample:
|
||||
even_zeros = sum(1 for i in range(0, len(sample), 2) if sample[i] == 0)
|
||||
odd_zeros = sum(1 for i in range(1, len(sample), 2) if i < len(sample) and sample[i] == 0)
|
||||
return "utf-16-be" if even_zeros > odd_zeros else "utf-16-le"
|
||||
return "utf-8"
|
||||
|
||||
|
||||
def _decode(raw: bytes) -> str:
|
||||
enc = _detect_encoding(raw)
|
||||
text = raw.decode(enc, errors="ignore")
|
||||
if text.startswith(""):
|
||||
text = text[1:]
|
||||
return text
|
||||
|
||||
|
||||
def _strip_currency(s: str) -> str:
|
||||
return s.replace("€", "").replace("EUR", "").replace("\xa0", " ").strip()
|
||||
|
||||
|
||||
def _parse_amount(s: str) -> Decimal:
|
||||
return parse_german_amount(_strip_currency(s))
|
||||
|
||||
|
||||
def _parse_two_digit_year_date(s: str) -> date:
|
||||
m = re.fullmatch(r"(\d{2})\.(\d{2})\.(\d{2})", s.strip())
|
||||
if not m:
|
||||
raise ParserError(f"DKB-CSV: Kein Datum im Format TT.MM.JJ: {s!r}")
|
||||
day, month, yy = int(m.group(1)), int(m.group(2)), int(m.group(3))
|
||||
year = 2000 + yy if yy <= 69 else 1900 + yy
|
||||
try:
|
||||
return date(year, month, day)
|
||||
except ValueError:
|
||||
raise ParserError(f"DKB-CSV: Ungueltiges Datum: {s!r}")
|
||||
|
||||
|
||||
def detect_csv_format(head_bytes: bytes) -> str | None:
|
||||
try:
|
||||
text = _decode(head_bytes)
|
||||
except (LookupError, UnicodeError):
|
||||
return None
|
||||
lines = text.splitlines()
|
||||
if not lines:
|
||||
return None
|
||||
first = lines[0]
|
||||
if first.startswith("Kontonummer;") and "Buchungsdatum" in first and "Valuta" in first \
|
||||
and "Betrag" in first:
|
||||
return "hvb_csv"
|
||||
if first.startswith("Bezeichnung Auftragskonto;") and "IBAN Auftragskonto" in first:
|
||||
return "vr_csv"
|
||||
head = "\n".join(lines[:6])
|
||||
if "Kontostand vom" in head and "Buchungsdatum" in head and "Zahlungsempf" in head:
|
||||
return "dkb_csv"
|
||||
return None
|
||||
|
||||
|
||||
def parse_csv(path: Path) -> ParsedCsv:
|
||||
raw = path.read_bytes()
|
||||
fmt = detect_csv_format(raw)
|
||||
if fmt is None:
|
||||
raise ParserError("CSV-Format nicht erkannt")
|
||||
text = _decode(raw)
|
||||
if fmt == "dkb_csv":
|
||||
return _parse_dkb(text)
|
||||
if fmt == "vr_csv":
|
||||
return _parse_vr(text)
|
||||
return _parse_hvb(text)
|
||||
|
||||
|
||||
def _rows(text: str) -> list[list[str]]:
|
||||
return list(csv.reader(io.StringIO(text), delimiter=";", quotechar='"'))
|
||||
|
||||
|
||||
def _parse_dkb(text: str) -> ParsedCsv:
|
||||
rows = _rows(text)
|
||||
if len(rows) < 2:
|
||||
raise ParserError("DKB-CSV: Datei zu kurz")
|
||||
iban = rows[0][1].strip() if len(rows[0]) > 1 else None
|
||||
|
||||
period_start = period_end = None
|
||||
kontostand_date: date | None = None
|
||||
kontostand_amount: Decimal | None = None
|
||||
header_idx = None
|
||||
for i, row in enumerate(rows):
|
||||
if row and row[0].strip() == "Zeitraum:" and len(row) > 1:
|
||||
m = RE_ZEITRAUM.search(row[1])
|
||||
if m:
|
||||
period_start = parse_german_date(m.group(1))
|
||||
period_end = parse_german_date(m.group(2))
|
||||
if row and RE_KONTOSTAND.search(row[0] or "") and len(row) > 1:
|
||||
m = RE_KONTOSTAND.search(row[0])
|
||||
kontostand_date = parse_german_date(m.group(1))
|
||||
kontostand_amount = _parse_amount(row[1])
|
||||
if row and row[0].strip() == "Buchungsdatum":
|
||||
header_idx = i
|
||||
break
|
||||
if header_idx is None:
|
||||
raise ParserError("DKB-CSV: Spaltenkopf nicht gefunden")
|
||||
if kontostand_date is None or kontostand_amount is None:
|
||||
raise ParserError("DKB-CSV: Kontostand-Zeile nicht gefunden")
|
||||
|
||||
header = [h.strip() for h in rows[header_idx]]
|
||||
idx = {name: i for i, name in enumerate(header)}
|
||||
required = ["Buchungsdatum", "Wertstellung", "Status", "Zahlungspflichtige*r",
|
||||
"Zahlungsempfänger*in", "Verwendungszweck", "Umsatztyp", "Betrag (€)"]
|
||||
for col in required:
|
||||
if col not in idx:
|
||||
raise ParserError(f"DKB-CSV: Spalte fehlt: {col}")
|
||||
|
||||
txs: list[ParsedTransaction] = []
|
||||
for row in rows[header_idx + 1:]:
|
||||
if not row or len(row) < len(header):
|
||||
continue
|
||||
if row[idx["Status"]].strip() != "Gebucht":
|
||||
continue
|
||||
amount = _parse_amount(row[idx["Betrag (€)"]])
|
||||
counterparty = (row[idx["Zahlungsempfänger*in"]] if amount < 0
|
||||
else row[idx["Zahlungspflichtige*r"]]).strip()
|
||||
purpose = f"{row[idx['Umsatztyp']].strip()} {row[idx['Verwendungszweck']].strip()}".strip()
|
||||
booking = _parse_two_digit_year_date(row[idx["Buchungsdatum"]])
|
||||
value = _parse_two_digit_year_date(row[idx["Wertstellung"]])
|
||||
txs.append(ParsedTransaction(booking_date=booking, value_date=value, amount=amount,
|
||||
purpose=purpose, counterparty=counterparty))
|
||||
txs.reverse() # file is sorted descending -> chronologically ascending
|
||||
|
||||
stmt = ParsedStatement(bank="dkb_csv", iban=iban, period_start=period_start,
|
||||
period_end=period_end, opening_balance=None,
|
||||
closing_balance=None, transactions=txs)
|
||||
return ParsedCsv(statement=stmt, anchor=(kontostand_date, kontostand_amount),
|
||||
balance_checkable=False)
|
||||
|
||||
|
||||
def _parse_vr(text: str) -> ParsedCsv:
|
||||
rows = _rows(text)
|
||||
if not rows:
|
||||
raise ParserError("VR-CSV: Datei leer")
|
||||
header = [h.strip() for h in rows[0]]
|
||||
idx = {name: i for i, name in enumerate(header)}
|
||||
required = ["IBAN Auftragskonto", "Buchungstag", "Valutadatum", "Name Zahlungsbeteiligter",
|
||||
"Buchungstext", "Verwendungszweck", "Betrag", "Saldo nach Buchung"]
|
||||
for col in required:
|
||||
if col not in idx:
|
||||
raise ParserError(f"VR-CSV: Spalte fehlt: {col}")
|
||||
|
||||
data_rows = [(i, row) for i, row in enumerate(rows[1:], start=2) if row and len(row) >= len(header)]
|
||||
if not data_rows:
|
||||
raise ParserError("VR-CSV: keine Buchungszeilen gefunden")
|
||||
|
||||
iban = rows[1][idx["IBAN Auftragskonto"]].strip() if len(rows) > 1 else None
|
||||
|
||||
# File is sorted descending (newest first) -> reverse for chronological ascending.
|
||||
asc = list(reversed(data_rows))
|
||||
|
||||
txs: list[ParsedTransaction] = []
|
||||
saldi: list[Decimal] = []
|
||||
prev_saldo: Decimal | None = None
|
||||
for line_no, row in asc:
|
||||
amount = _parse_amount(row[idx["Betrag"]])
|
||||
saldo = _parse_amount(row[idx["Saldo nach Buchung"]])
|
||||
if prev_saldo is not None and prev_saldo + amount != saldo:
|
||||
raise ParserError(f"VR-CSV: Saldo-Kette inkonsistent bei Zeile {line_no}")
|
||||
prev_saldo = saldo
|
||||
booking = parse_german_date(row[idx["Buchungstag"]])
|
||||
value = parse_german_date(row[idx["Valutadatum"]])
|
||||
purpose = f"{row[idx['Buchungstext']].strip()} {row[idx['Verwendungszweck']].strip()}".strip()
|
||||
counterparty = row[idx["Name Zahlungsbeteiligter"]].strip()
|
||||
txs.append(ParsedTransaction(booking_date=booking, value_date=value, amount=amount,
|
||||
purpose=purpose, counterparty=counterparty))
|
||||
saldi.append(saldo)
|
||||
|
||||
opening_balance = saldi[0] - txs[0].amount
|
||||
closing_balance = saldi[-1]
|
||||
anchor = (txs[-1].booking_date, closing_balance)
|
||||
|
||||
stmt = ParsedStatement(bank="vr_csv", iban=iban,
|
||||
period_start=txs[0].booking_date, period_end=txs[-1].booking_date,
|
||||
opening_balance=opening_balance, closing_balance=closing_balance,
|
||||
transactions=txs)
|
||||
return ParsedCsv(statement=stmt, anchor=anchor, balance_checkable=True)
|
||||
|
||||
|
||||
def _parse_hvb(text: str) -> ParsedCsv:
|
||||
rows = _rows(text)
|
||||
if not rows:
|
||||
raise ParserError("HVB-CSV: Datei leer")
|
||||
header = [h.strip() for h in rows[0]]
|
||||
idx = {name: i for i, name in enumerate(header)}
|
||||
required = ["Kontonummer", "Buchungsdatum", "Valuta", "Verwendungszweck", "Betrag"]
|
||||
for col in required:
|
||||
if col not in idx:
|
||||
raise ParserError(f"HVB-CSV: Spalte fehlt: {col}")
|
||||
|
||||
data_rows = [row for row in rows[1:] if row and len(row) >= len(header)]
|
||||
if not data_rows:
|
||||
raise ParserError("HVB-CSV: keine Buchungszeilen gefunden")
|
||||
|
||||
kontonummer = data_rows[0][idx["Kontonummer"]].strip()
|
||||
|
||||
txs: list[ParsedTransaction] = []
|
||||
for row in data_rows: # already chronologically ascending
|
||||
booking = parse_german_date(row[idx["Buchungsdatum"]])
|
||||
value = parse_german_date(row[idx["Valuta"]])
|
||||
amount = _parse_amount(row[idx["Betrag"]])
|
||||
purpose = row[idx["Verwendungszweck"]].strip()
|
||||
txs.append(ParsedTransaction(booking_date=booking, value_date=value, amount=amount,
|
||||
purpose=purpose, counterparty=""))
|
||||
|
||||
stmt = ParsedStatement(bank="hvb_csv", iban=kontonummer,
|
||||
period_start=txs[0].booking_date, period_end=txs[-1].booking_date,
|
||||
opening_balance=None, closing_balance=None, transactions=txs)
|
||||
return ParsedCsv(statement=stmt, anchor=None, balance_checkable=False)
|
||||
255
finance/tests/test_csv_formats.py
Normal file
255
finance/tests/test_csv_formats.py
Normal file
@@ -0,0 +1,255 @@
|
||||
import json
|
||||
from datetime import date
|
||||
from decimal import Decimal
|
||||
from pathlib import Path
|
||||
|
||||
import pytest
|
||||
|
||||
from app.parsers.base import ParserError
|
||||
from app.parsers.csv_formats import detect_csv_format, parse_csv
|
||||
from app.parsers.validate import balance_difference
|
||||
|
||||
FIXTURES = Path(__file__).parent / "fixtures"
|
||||
|
||||
FIXTURE_CASES = [
|
||||
("DKB_2025.csv", "dkb_csv"),
|
||||
("DKB_2026.csv", "dkb_csv"),
|
||||
("VR_2025.csv", "vr_csv"),
|
||||
("VR_2026.csv", "vr_csv"),
|
||||
("HVB_2025.csv", "hvb_csv"),
|
||||
("HVB_2026.csv", "hvb_csv"),
|
||||
]
|
||||
|
||||
|
||||
def _read_bytes(filename: str) -> bytes:
|
||||
return (FIXTURES / filename).read_bytes()
|
||||
|
||||
|
||||
def _gebucht_count(raw: bytes) -> int:
|
||||
"""Independently count 'Gebucht' rows without keeping/printing any row content."""
|
||||
# DKB is UTF-8 (with BOM); Status column is ASCII, safe to count on raw text.
|
||||
text = raw.decode("utf-8-sig")
|
||||
return text.count(';"Gebucht";')
|
||||
|
||||
|
||||
@pytest.mark.parametrize("filename,fmt", FIXTURE_CASES)
|
||||
def test_fixture_rowcount_and_ascending(filename, fmt):
|
||||
path = FIXTURES / filename
|
||||
if not path.exists():
|
||||
pytest.skip(f"Fixture {filename} fehlt")
|
||||
parsed = parse_csv(path)
|
||||
stmt = parsed.statement
|
||||
assert stmt.bank == fmt
|
||||
assert len(stmt.transactions) > 0
|
||||
dates = [t.booking_date for t in stmt.transactions]
|
||||
assert dates == sorted(dates), "Transaktionen muessen chronologisch aufsteigend sein"
|
||||
for t in stmt.transactions:
|
||||
assert isinstance(t.amount, Decimal)
|
||||
|
||||
|
||||
@pytest.mark.parametrize("filename,fmt", [c for c in FIXTURE_CASES if c[1] == "vr_csv"])
|
||||
def test_vr_fixture_balance_chain_and_anchor(filename, fmt):
|
||||
path = FIXTURES / filename
|
||||
if not path.exists():
|
||||
pytest.skip(f"Fixture {filename} fehlt")
|
||||
parsed = parse_csv(path)
|
||||
assert parsed.balance_checkable is True
|
||||
assert balance_difference(parsed.statement) == Decimal("0")
|
||||
assert parsed.anchor is not None
|
||||
anchor_date, anchor_balance = parsed.anchor
|
||||
assert isinstance(anchor_date, date)
|
||||
assert isinstance(anchor_balance, Decimal)
|
||||
|
||||
|
||||
@pytest.mark.parametrize("filename,fmt", [c for c in FIXTURE_CASES if c[1] == "dkb_csv"])
|
||||
def test_dkb_fixture_anchor_and_gebucht_filter(filename, fmt):
|
||||
path = FIXTURES / filename
|
||||
if not path.exists():
|
||||
pytest.skip(f"Fixture {filename} fehlt")
|
||||
raw = _read_bytes(filename)
|
||||
parsed = parse_csv(path)
|
||||
assert parsed.anchor is not None
|
||||
assert parsed.balance_checkable is False
|
||||
assert parsed.statement.opening_balance is None
|
||||
assert parsed.statement.closing_balance is None
|
||||
# Filtering: parsed transaction count must equal independently-counted
|
||||
# "Gebucht" data rows in the raw file (counts only, no content asserted).
|
||||
assert len(parsed.statement.transactions) == _gebucht_count(raw)
|
||||
for t in parsed.statement.transactions:
|
||||
assert t.counterparty != ""
|
||||
|
||||
|
||||
@pytest.mark.parametrize("filename,fmt", [c for c in FIXTURE_CASES if c[1] == "hvb_csv"])
|
||||
def test_hvb_fixture_no_balance_check(filename, fmt):
|
||||
path = FIXTURES / filename
|
||||
if not path.exists():
|
||||
pytest.skip(f"Fixture {filename} fehlt")
|
||||
parsed = parse_csv(path)
|
||||
assert parsed.balance_checkable is False
|
||||
assert parsed.anchor is None
|
||||
assert parsed.statement.opening_balance is None
|
||||
assert parsed.statement.closing_balance is None
|
||||
for t in parsed.statement.transactions:
|
||||
assert t.counterparty == ""
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Synthetic tests: minimal, fictitious CSVs constructed as bytes in-test.
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def _dkb_row(booking, value, status, payer, payee, purpose, umsatztyp, iban, amount):
|
||||
return (
|
||||
f'"{booking}";"{value}";"{status}";"{payer}";"{payee}";'
|
||||
f'"{purpose}";"{umsatztyp}";"{iban}";"{amount}";"";"";""\r\n'
|
||||
)
|
||||
|
||||
|
||||
def _build_dkb_csv() -> bytes:
|
||||
header = (
|
||||
'"Girokonto";"DE00123456780000000042"\r\n'
|
||||
'"Zeitraum:";"01.01.2025 - 31.01.2025"\r\n'
|
||||
'"Kontostand vom 31.01.2025:";"1.000,00\xa0€"\r\n'
|
||||
'""\r\n'
|
||||
'"Buchungsdatum";"Wertstellung";"Status";"Zahlungspflichtige*r";'
|
||||
'"Zahlungsempfänger*in";"Verwendungszweck";"Umsatztyp";"IBAN";"Betrag (€)";'
|
||||
'"Gläubiger-ID";"Mandatsreferenz";"Kundenreferenz"\r\n'
|
||||
)
|
||||
rows = (
|
||||
_dkb_row("15.01.25", "15.01.25", "Gebucht", "Max Muster", "Erika Beispiel",
|
||||
"Testzweck A", "Ausgang", "DE00TESTIBAN1", "-10,00")
|
||||
+ _dkb_row("10.01.25", "10.01.25", "Vorgemerkt", "Max Muster", "Erika Beispiel",
|
||||
"Testzweck B (nicht gebucht)", "Ausgang", "DE00TESTIBAN1", "-99,00")
|
||||
+ _dkb_row("05.01.25", "05.01.25", "Gebucht", "Erika Beispiel", "Max Muster",
|
||||
"Testzweck C", "Eingang", "DE00TESTIBAN1", "50,00")
|
||||
)
|
||||
return (header + rows).encode("utf-8-sig")
|
||||
|
||||
|
||||
def _build_vr_csv(saldi_consistent: bool = True) -> bytes:
|
||||
header = (
|
||||
"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\r\n"
|
||||
)
|
||||
# File order is descending (newest first), matches real format.
|
||||
saldo_newest = "120,00" if saldi_consistent else "999,00"
|
||||
rows = (
|
||||
"Testkonto;DE00TESTVR000000000001;GENODETEST;Test-Bank;15.01.2025;15.01.2025;"
|
||||
"Beispiel Gegenpartei;DE00GEGEN0001;GENODETEST;Buchung;Testzweck A;20,00;EUR;"
|
||||
f"{saldo_newest};;;;\r\n"
|
||||
"Testkonto;DE00TESTVR000000000001;GENODETEST;Test-Bank;05.01.2025;05.01.2025;"
|
||||
"Beispiel Gegenpartei 2;DE00GEGEN0002;GENODETEST;Buchung;Testzweck B;100,00;EUR;"
|
||||
"100,00;;;;\r\n"
|
||||
)
|
||||
return (header + rows).encode("utf-8-sig")
|
||||
|
||||
|
||||
def _build_hvb_csv() -> bytes:
|
||||
header = "Kontonummer;Buchungsdatum;Valuta;Verwendungszweck;Betrag;Waehrung\r\n"
|
||||
rows = (
|
||||
"123456789;05.01.2025;05.01.2025;Testzweck mit Umlaut ä ö ü ß;-30,00;EUR\r\n"
|
||||
"123456789;15.01.2025;15.01.2025;Testzweck zwei;40,00;EUR\r\n"
|
||||
)
|
||||
return (header + rows).encode("utf-16")
|
||||
|
||||
|
||||
def test_detect_dkb_csv():
|
||||
assert detect_csv_format(_build_dkb_csv()) == "dkb_csv"
|
||||
|
||||
|
||||
def test_detect_vr_csv():
|
||||
assert detect_csv_format(_build_vr_csv()) == "vr_csv"
|
||||
|
||||
|
||||
def test_detect_hvb_csv():
|
||||
assert detect_csv_format(_build_hvb_csv()) == "hvb_csv"
|
||||
|
||||
|
||||
def test_detect_garbage_returns_none():
|
||||
assert detect_csv_format(b"this,is,not;a\nrecognised\tformat garbage 1234") is None
|
||||
|
||||
|
||||
def test_dkb_synthetic_skips_non_gebucht_rows(tmp_path):
|
||||
p = tmp_path / "dkb_synth.csv"
|
||||
p.write_bytes(_build_dkb_csv())
|
||||
parsed = parse_csv(p)
|
||||
assert len(parsed.statement.transactions) == 2
|
||||
for t in parsed.statement.transactions:
|
||||
assert t.purpose != "" and "nicht gebucht" not in t.purpose
|
||||
|
||||
|
||||
def test_dkb_synthetic_two_digit_year_parsed(tmp_path):
|
||||
p = tmp_path / "dkb_synth.csv"
|
||||
p.write_bytes(_build_dkb_csv())
|
||||
parsed = parse_csv(p)
|
||||
dates = [t.booking_date for t in parsed.statement.transactions]
|
||||
assert date(2025, 1, 5) in dates
|
||||
assert date(2025, 1, 15) in dates
|
||||
|
||||
|
||||
def test_dkb_synthetic_anchor(tmp_path):
|
||||
p = tmp_path / "dkb_synth.csv"
|
||||
p.write_bytes(_build_dkb_csv())
|
||||
parsed = parse_csv(p)
|
||||
assert parsed.anchor == (date(2025, 1, 31), Decimal("1000.00"))
|
||||
assert parsed.balance_checkable is False
|
||||
|
||||
|
||||
def test_vr_synthetic_balance_chain_ok(tmp_path):
|
||||
p = tmp_path / "vr_synth.csv"
|
||||
p.write_bytes(_build_vr_csv(saldi_consistent=True))
|
||||
parsed = parse_csv(p)
|
||||
assert parsed.balance_checkable is True
|
||||
assert balance_difference(parsed.statement) == Decimal("0")
|
||||
assert parsed.anchor == (date(2025, 1, 15), Decimal("120.00"))
|
||||
|
||||
|
||||
def test_vr_synthetic_balance_chain_break_raises(tmp_path):
|
||||
p = tmp_path / "vr_synth_broken.csv"
|
||||
p.write_bytes(_build_vr_csv(saldi_consistent=False))
|
||||
with pytest.raises(ParserError):
|
||||
parse_csv(p)
|
||||
|
||||
|
||||
def test_hvb_synthetic_utf16_decode(tmp_path):
|
||||
p = tmp_path / "hvb_synth.csv"
|
||||
p.write_bytes(_build_hvb_csv())
|
||||
parsed = parse_csv(p)
|
||||
stmt = parsed.statement
|
||||
assert len(stmt.transactions) == 2
|
||||
assert stmt.transactions[0].booking_date == date(2025, 1, 5)
|
||||
assert stmt.transactions[1].booking_date == date(2025, 1, 15)
|
||||
assert "ä" in stmt.transactions[0].purpose
|
||||
assert all(t.counterparty == "" for t in stmt.transactions)
|
||||
assert parsed.balance_checkable is False
|
||||
assert parsed.anchor is None
|
||||
|
||||
|
||||
def test_parse_csv_unknown_format_raises(tmp_path):
|
||||
p = tmp_path / "garbage.csv"
|
||||
p.write_bytes(b"not,a;known\nformat\n")
|
||||
with pytest.raises(ParserError):
|
||||
parse_csv(p)
|
||||
|
||||
|
||||
CSV_FILES = ["DKB_2025", "DKB_2026", "VR_2025", "VR_2026", "HVB_2025", "HVB_2026"]
|
||||
|
||||
|
||||
@pytest.mark.parametrize("name", CSV_FILES)
|
||||
def test_csv_expected_values(name):
|
||||
path = FIXTURES / f"{name}.csv"
|
||||
exp_path = FIXTURES / f"expected_csv_{name}.json"
|
||||
if not path.exists() or not exp_path.exists():
|
||||
pytest.skip("Fixture oder expected-Datei fehlt")
|
||||
result = parse_csv(path)
|
||||
exp = json.loads(exp_path.read_text())
|
||||
txs = result.statement.transactions
|
||||
assert len(txs) == exp["transaction_count"]
|
||||
for c in exp["spot_checks"]:
|
||||
t = txs[c["index"]]
|
||||
assert str(t.booking_date) == c["booking_date"]
|
||||
assert str(t.amount) == c["amount"]
|
||||
assert c["counterparty_contains"] in t.counterparty
|
||||
assert c["purpose_contains"] in t.purpose
|
||||
Reference in New Issue
Block a user