andy
db06108d2b
Adds remaining files from parallel development: - Services: analysis, csv_reader, forecasting, normalizer, recurring - UI: recurring_view, settings_view, sidebar, themes (dark/light) - Tests: analysis, csv_reader, forecasting, import_categorize, normalizer, recurring, integration - App entry point (main.py) and CLAUDE.md 52 tests passing across all modules. Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
72 lines
2.4 KiB
Python
72 lines
2.4 KiB
Python
# tests/services/test_import_categorize.py
|
|
from pathlib import Path
|
|
from sqlalchemy import create_engine
|
|
from sqlalchemy.orm import Session
|
|
|
|
from src.db import Base
|
|
from src.models import *
|
|
from src.seed import seed_categories
|
|
from src.services.importer import ImportService
|
|
|
|
RAWDATA = Path(__file__).parent.parent.parent / "rawdata"
|
|
|
|
|
|
def make_session():
|
|
engine = create_engine("sqlite:///:memory:")
|
|
Base.metadata.create_all(engine)
|
|
return Session(engine)
|
|
|
|
|
|
def test_import_applies_categorization_rules():
|
|
session = make_session()
|
|
seed_categories(session)
|
|
member = HouseholdMember(name="Andrew", relationship="self")
|
|
session.add(member)
|
|
session.flush()
|
|
account = Account(name="Chase", institution="Chase", account_type="credit", owner_id=member.id)
|
|
session.add(account)
|
|
session.flush()
|
|
|
|
groceries = session.query(Category).filter_by(name="Groceries").one()
|
|
rule = CategorizationRule(pattern="PUBLIX", category_id=groceries.id, priority=10)
|
|
session.add(rule)
|
|
session.commit()
|
|
|
|
svc = ImportService(session)
|
|
svc.import_csv(
|
|
RAWDATA / "Chase0372_Activity20260101_20260210_20260210.CSV",
|
|
account_id=account.id,
|
|
column_map={"date": "Transaction Date", "amount": "Amount", "description": "Description"},
|
|
amount_logic="signed",
|
|
)
|
|
|
|
publix_txns = session.query(Transaction).filter(Transaction.description.contains("PUBLIX")).all()
|
|
assert len(publix_txns) > 0
|
|
for txn in publix_txns:
|
|
assert txn.category_id == groceries.id
|
|
assert txn.tag == "needs"
|
|
|
|
|
|
def test_uncategorized_transactions_have_no_category():
|
|
session = make_session()
|
|
seed_categories(session)
|
|
member = HouseholdMember(name="Andrew", relationship="self")
|
|
session.add(member)
|
|
session.flush()
|
|
account = Account(name="Chase", institution="Chase", account_type="credit", owner_id=member.id)
|
|
session.add(account)
|
|
session.flush()
|
|
|
|
# No rules defined
|
|
svc = ImportService(session)
|
|
svc.import_csv(
|
|
RAWDATA / "Chase0372_Activity20260101_20260210_20260210.CSV",
|
|
account_id=account.id,
|
|
column_map={"date": "Transaction Date", "amount": "Amount", "description": "Description"},
|
|
amount_logic="signed",
|
|
)
|
|
|
|
uncategorized = session.query(Transaction).filter(Transaction.category_id.is_(None)).count()
|
|
total = session.query(Transaction).count()
|
|
assert uncategorized == total
|