src/ingest/b4w.py

@@ -2,7 +2,6 @@ from pathlib import Path
 
 from sqlalchemy.orm import Session
 
-from src.db import get_session
 from src.ingest._common import _reader_raw, _upsert
 from src.models.b4w import B4wDb, B4wExport, Category, CategoryDb
 
@@ -64,9 +63,8 @@ def upsert_b4w(
     )
 
 
-def ingest_b4w(path: Path):
+def ingest_b4w(path: Path, session: Session):
     data = reader_b4w(path)
-    with get_session() as session:
     for row in data:
         upsert_b4w(session, row, do_commit=False)
     session.commit()

tests/test_ingest_b4w.py

@@ -0,0 +1,45 @@
+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)