Files
finanz-cockpit/tests/test_ingest_b4w.py
T
2026-09-13 12:57:05 +02:00

46 lines
1.9 KiB
Python

import re
from datetime import date
from decimal import Decimal
from pathlib import Path
from sqlalchemy import select
from sqlalchemy.orm import Session
from src.constants import IBAN_PATTERN, Currency
from src.ingest.b4w import ingest_b4w
from src.models.b4w import B4wDb
b4w_db_column_names = [col.key for col in B4wDb.__mapper__.columns]
# TODO: Unit tests for other funtions in src/ingest/b4w.py
# TODO: Similar file for Comdirect (more work due to several tables)
def test_ingest_b4w_e2e(db_session: Session):
path = Path("tests/test_data/b4w_export_dummy_01.csv")
ingest_b4w(path, db_session)
result = db_session.execute(select(B4wDb)).scalars().all()
assert isinstance(result, list)
assert all(isinstance(row.id_b4w, int) for row in result)
assert all(isinstance(row.date_book, date) for row in result)
assert all(isinstance(row.other_name, str) for row in result)
assert all(isinstance(row.other_iban, str) for row in result)
assert all(re.findall(IBAN_PATTERN, str(row.other_iban)) for row in result)
assert all(
(isinstance(row.category_id, int) or row.category_id is None) for row in result
)
assert all(isinstance(row.notes, str) for row in result)
assert all(isinstance(row.id, int) for row in result)
assert all(isinstance(row.hash_id, str) for row in result)
assert all(isinstance(row.account_iban, str) for row in result)
assert all(re.findall(IBAN_PATTERN, str(row.account_iban)) for row in result)
assert all(isinstance(row.account_currency, str) for row in result)
assert all(isinstance(row.date_valid, date) for row in result)
assert all(isinstance(row.amount, Decimal) for row in result)
assert all(isinstance(row.amount_currency, str) for row in result)
assert all(row.amount_currency in list(Currency) for row in result)
assert all(isinstance(row.text, str) for row in result)
print(b4w_db_column_names)