100 lines
3.3 KiB
Python
100 lines
3.3 KiB
Python
import datetime
|
|
from sqlalchemy import create_engine
|
|
from sqlalchemy.orm import Session
|
|
|
|
from src.db import Base
|
|
from src.models import *
|
|
from src.services.transfer_detector import TransferDetector
|
|
|
|
|
|
def make_session():
|
|
engine = create_engine("sqlite:///:memory:")
|
|
Base.metadata.create_all(engine)
|
|
return Session(engine)
|
|
|
|
|
|
def test_detect_matching_transfer():
|
|
session = make_session()
|
|
member = HouseholdMember(name="Andrew", relationship="self")
|
|
session.add(member)
|
|
session.flush()
|
|
checking = Account(name="Checking", institution="WF", account_type="checking", owner_id=member.id)
|
|
chase = Account(name="Chase", institution="Chase", account_type="credit", owner_id=member.id)
|
|
session.add_all([checking, chase])
|
|
session.flush()
|
|
|
|
# Payment from checking to Chase
|
|
txn1 = Transaction(
|
|
date=datetime.date(2026, 1, 29), amount=-1461.35,
|
|
description="CHASE CREDIT CRD EPAY", raw_description="CHASE CREDIT CRD EPAY 260128 9077835526 ANDREW B CONLON",
|
|
account_id=checking.id,
|
|
)
|
|
txn2 = Transaction(
|
|
date=datetime.date(2026, 1, 28), amount=1461.35,
|
|
description="Payment Thank You - Web", raw_description="Payment Thank You - Web",
|
|
account_id=chase.id,
|
|
)
|
|
session.add_all([txn1, txn2])
|
|
session.commit()
|
|
|
|
detector = TransferDetector(session)
|
|
pairs = detector.detect()
|
|
assert len(pairs) == 1
|
|
assert pairs[0]["amount"] == 1461.35
|
|
|
|
|
|
def test_detect_marks_transfers():
|
|
session = make_session()
|
|
member = HouseholdMember(name="Andrew", relationship="self")
|
|
session.add(member)
|
|
session.flush()
|
|
checking = Account(name="Checking", institution="WF", account_type="checking", owner_id=member.id)
|
|
chase = Account(name="Chase", institution="Chase", account_type="credit", owner_id=member.id)
|
|
session.add_all([checking, chase])
|
|
session.flush()
|
|
|
|
txn1 = Transaction(
|
|
date=datetime.date(2026, 1, 29), amount=-1461.35,
|
|
description="CHASE CREDIT CRD EPAY", account_id=checking.id,
|
|
)
|
|
txn2 = Transaction(
|
|
date=datetime.date(2026, 1, 28), amount=1461.35,
|
|
description="Payment Thank You - Web", account_id=chase.id,
|
|
)
|
|
session.add_all([txn1, txn2])
|
|
session.commit()
|
|
|
|
detector = TransferDetector(session)
|
|
detector.mark_transfers()
|
|
|
|
session.refresh(txn1)
|
|
session.refresh(txn2)
|
|
assert txn1.is_transfer is True
|
|
assert txn2.is_transfer is True
|
|
|
|
|
|
def test_no_false_positive_different_amounts():
|
|
session = make_session()
|
|
member = HouseholdMember(name="Andrew", relationship="self")
|
|
session.add(member)
|
|
session.flush()
|
|
checking = Account(name="Checking", institution="WF", account_type="checking", owner_id=member.id)
|
|
chase = Account(name="Chase", institution="Chase", account_type="credit", owner_id=member.id)
|
|
session.add_all([checking, chase])
|
|
session.flush()
|
|
|
|
txn1 = Transaction(
|
|
date=datetime.date(2026, 1, 29), amount=-1461.35,
|
|
description="CHASE CREDIT CRD EPAY", account_id=checking.id,
|
|
)
|
|
txn2 = Transaction(
|
|
date=datetime.date(2026, 1, 28), amount=500.00,
|
|
description="Payment Thank You - Web", account_id=chase.id,
|
|
)
|
|
session.add_all([txn1, txn2])
|
|
session.commit()
|
|
|
|
detector = TransferDetector(session)
|
|
pairs = detector.detect()
|
|
assert len(pairs) == 0
|