256 lines
9.4 KiB
Python
256 lines
9.4 KiB
Python
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
|