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>
86 lines
2.6 KiB
Python
86 lines
2.6 KiB
Python
# tests/services/test_recurring.py
|
|
import datetime
|
|
from sqlalchemy import create_engine
|
|
from sqlalchemy.orm import Session
|
|
|
|
from src.db import Base
|
|
from src.models import *
|
|
from src.services.recurring import RecurringDetector
|
|
|
|
|
|
def make_session():
|
|
engine = create_engine("sqlite:///:memory:")
|
|
Base.metadata.create_all(engine)
|
|
return Session(engine)
|
|
|
|
|
|
def make_recurring_data(session):
|
|
member = HouseholdMember(name="Andrew", relationship="self")
|
|
session.add(member)
|
|
session.flush()
|
|
account = Account(name="Checking", institution="WF", account_type="checking", owner_id=member.id)
|
|
session.add(account)
|
|
session.flush()
|
|
|
|
# Netflix monthly: ~$19.07 on 3rd/4th of month
|
|
for month in [1, 2]:
|
|
session.add(Transaction(
|
|
date=datetime.date(2026, month, 4),
|
|
amount=-19.07, description="Netflix.com", account_id=account.id,
|
|
))
|
|
|
|
# HelloFresh weekly: ~$142.87
|
|
for day in [1, 8, 15, 22, 29]:
|
|
session.add(Transaction(
|
|
date=datetime.date(2026, 1, day),
|
|
amount=-142.87, description="HELLOFRESH", account_id=account.id,
|
|
))
|
|
|
|
# Random one-off purchase
|
|
session.add(Transaction(
|
|
date=datetime.date(2026, 1, 10),
|
|
amount=-95.38, description="CARL'S GOLFLAND INC", account_id=account.id,
|
|
))
|
|
|
|
session.commit()
|
|
return account
|
|
|
|
|
|
def test_detect_monthly_recurring():
|
|
session = make_session()
|
|
make_recurring_data(session)
|
|
detector = RecurringDetector(session)
|
|
results = detector.detect()
|
|
netflix = [r for r in results if "Netflix" in r["description"]]
|
|
assert len(netflix) == 1
|
|
assert netflix[0]["frequency"] == "monthly"
|
|
assert abs(netflix[0]["typical_amount"] - 19.07) < 0.01
|
|
|
|
|
|
def test_detect_weekly_recurring():
|
|
session = make_session()
|
|
make_recurring_data(session)
|
|
detector = RecurringDetector(session)
|
|
results = detector.detect()
|
|
hello = [r for r in results if "HELLOFRESH" in r["description"]]
|
|
assert len(hello) == 1
|
|
assert hello[0]["frequency"] == "weekly"
|
|
|
|
|
|
def test_one_off_not_detected():
|
|
session = make_session()
|
|
make_recurring_data(session)
|
|
detector = RecurringDetector(session)
|
|
results = detector.detect()
|
|
golf = [r for r in results if "GOLFLAND" in r["description"]]
|
|
assert len(golf) == 0
|
|
|
|
|
|
def test_annual_cost_calculation():
|
|
session = make_session()
|
|
make_recurring_data(session)
|
|
detector = RecurringDetector(session)
|
|
results = detector.detect()
|
|
netflix = [r for r in results if "Netflix" in r["description"]][0]
|
|
assert abs(netflix["annual_cost"] - 19.07 * 12) < 1.0
|