mirror of
https://github.com/escalante29/WealthySmart.git
synced 2026-07-17 16:08:47 +02:00
386 lines
13 KiB
Python
386 lines
13 KiB
Python
"""Tasa Cero installment plans: rounding, scheduling, budget exclusion,
|
|
auto-detection at ingestion, regeneration, unconvert, and cascade behavior.
|
|
|
|
Rounding and scheduling cases are pinned against real BAC Financiamientos
|
|
data (2026-05/07). Endpoint functions are exercised directly with a session,
|
|
same style as test_import_review_and_bulk.py.
|
|
"""
|
|
|
|
from datetime import datetime
|
|
from decimal import Decimal
|
|
|
|
import pytest
|
|
from fastapi import HTTPException
|
|
|
|
from app.api.v1.endpoints.installments import (
|
|
create_installment_plan,
|
|
delete_installment_plan,
|
|
get_installment_plan,
|
|
list_installment_plans,
|
|
update_installment_plan,
|
|
)
|
|
from app.api.v1.endpoints.transactions import (
|
|
BulkActionRequest,
|
|
bulk_action,
|
|
create_transaction,
|
|
delete_transaction,
|
|
update_transaction,
|
|
)
|
|
from app.models.models import (
|
|
Category,
|
|
InstallmentPlan,
|
|
InstallmentPlanCreate,
|
|
InstallmentPlanUpdate,
|
|
Transaction,
|
|
TransactionCreate,
|
|
TransactionSource,
|
|
TransactionType,
|
|
TransactionUpdate,
|
|
)
|
|
from app.services.budget_projection import (
|
|
compute_actuals_by_source,
|
|
get_cycle_range,
|
|
)
|
|
from app.services.installments import (
|
|
compute_installment_amounts,
|
|
compute_installment_dates,
|
|
is_tasa_cero_merchant,
|
|
)
|
|
from sqlmodel import select
|
|
|
|
|
|
def make_cc_tx(
|
|
session,
|
|
merchant="CC CONSTRUPLAZA",
|
|
amount="111053.60",
|
|
date=datetime(2026, 6, 25, 15, 43),
|
|
reference="75669363",
|
|
source=TransactionSource.CREDIT_CARD,
|
|
**kw,
|
|
):
|
|
tx = Transaction(
|
|
amount=Decimal(amount),
|
|
merchant=merchant,
|
|
date=date,
|
|
reference=reference,
|
|
source=source,
|
|
**kw,
|
|
)
|
|
session.add(tx)
|
|
session.commit()
|
|
session.refresh(tx)
|
|
return tx
|
|
|
|
|
|
def convert(session, tx, n=3, first=None):
|
|
return create_installment_plan(
|
|
InstallmentPlanCreate(
|
|
transaction_id=tx.id, num_installments=n, first_installment_date=first
|
|
),
|
|
session=session,
|
|
_user="t",
|
|
)
|
|
|
|
|
|
def get_cuotas(session, plan_id):
|
|
return session.exec(
|
|
select(Transaction)
|
|
.where(Transaction.installment_plan_id == plan_id)
|
|
.order_by(Transaction.date)
|
|
).all()
|
|
|
|
|
|
class TestRounding:
|
|
"""Pinned against real BAC plans."""
|
|
|
|
@pytest.mark.parametrize(
|
|
"total,n,per,last",
|
|
[
|
|
("111053.60", 3, "37017.80", "37018.00"),
|
|
("425800.00", 6, "70966.60", "70967.00"),
|
|
("98632.00", 3, "32877.30", "32877.40"),
|
|
("190864.80", 3, "63621.60", "63621.60"), # divides exactly
|
|
],
|
|
)
|
|
def test_real_bac_plans(self, total, n, per, last):
|
|
amounts = compute_installment_amounts(Decimal(total), n)
|
|
assert amounts[:-1] == [Decimal(per)] * (n - 1)
|
|
assert amounts[-1] == Decimal(last)
|
|
|
|
def test_sum_invariant(self):
|
|
for total, n in [("100.00", 3), ("214800.00", 3), ("99999.99", 7)]:
|
|
amounts = compute_installment_amounts(Decimal(total), n)
|
|
assert sum(amounts) == Decimal(total)
|
|
assert len(amounts) == n
|
|
|
|
|
|
class TestScheduling:
|
|
def test_first_before_cut_stays_in_purchase_cycle(self):
|
|
# 14/06 purchase (cycle 18/05-18/06) -> cuotas 19/06, 19/07
|
|
dates = compute_installment_dates(datetime(2026, 6, 14, 19, 22), 3)
|
|
assert dates == [
|
|
datetime(2026, 6, 14, 19, 22),
|
|
datetime(2026, 6, 19),
|
|
datetime(2026, 7, 19),
|
|
]
|
|
|
|
def test_first_after_cut(self):
|
|
# 27/06 purchase (cycle 18/06-18/07) -> cuotas 19/07, 19/08
|
|
dates = compute_installment_dates(datetime(2026, 6, 27, 19, 43), 3)
|
|
assert dates == [
|
|
datetime(2026, 6, 27, 19, 43),
|
|
datetime(2026, 7, 19),
|
|
datetime(2026, 8, 19),
|
|
]
|
|
|
|
def test_december_wrap(self):
|
|
dates = compute_installment_dates(datetime(2026, 12, 20), 3)
|
|
assert dates == [
|
|
datetime(2026, 12, 20),
|
|
datetime(2027, 1, 19),
|
|
datetime(2027, 2, 19),
|
|
]
|
|
|
|
def test_consecutive_cycles(self):
|
|
"""Each cuota lands in its own consecutive billing cycle."""
|
|
dates = compute_installment_dates(datetime(2026, 5, 5, 11, 19), 6)
|
|
cycles = []
|
|
for d in dates:
|
|
if d.day >= 18:
|
|
cy, cm = d.year, d.month
|
|
else:
|
|
cy, cm = (d.year, d.month - 1) if d.month > 1 else (d.year - 1, 12)
|
|
start, end = get_cycle_range(cy, cm)
|
|
assert start <= d < end
|
|
cycles.append((cy, cm))
|
|
assert cycles == [(2026, m) for m in range(4, 10)]
|
|
|
|
|
|
class TestDetection:
|
|
@pytest.mark.parametrize(
|
|
"merchant,expected",
|
|
[
|
|
("CC CONSTRUPLAZA", True),
|
|
("CC TILO.CO", True),
|
|
("CONSTRUPLAZA S", False),
|
|
("ALMACENES SIMAN", False),
|
|
("MERCCADO", False),
|
|
],
|
|
)
|
|
def test_merchant_regex(self, merchant, expected):
|
|
assert is_tasa_cero_merchant(merchant) is expected
|
|
|
|
def test_ingestion_auto_creates_plan(self, session):
|
|
tx = create_transaction(
|
|
TransactionCreate(
|
|
amount=Decimal("301614.60"),
|
|
merchant="CC CONSTRUPLAZA",
|
|
date=datetime(2026, 6, 27, 19, 43),
|
|
reference="24724909",
|
|
),
|
|
session=session,
|
|
_user="t",
|
|
)
|
|
assert tx.is_installment_anchor is True
|
|
plan = session.exec(
|
|
select(InstallmentPlan).where(
|
|
InstallmentPlan.anchor_transaction_id == tx.id
|
|
)
|
|
).one()
|
|
assert plan.num_installments == 3
|
|
assert plan.first_installment_date == tx.date
|
|
cuotas = get_cuotas(session, plan.id)
|
|
assert [c.amount for c in cuotas] == [
|
|
Decimal("100538.20"),
|
|
Decimal("100538.20"),
|
|
Decimal("100538.20"),
|
|
]
|
|
assert cuotas[0].merchant == "CC CONSTRUPLAZA CUOTA:01/03"
|
|
assert cuotas[0].reference == "24724909-C01"
|
|
assert all(c.installment_plan_id == plan.id for c in cuotas)
|
|
|
|
@pytest.mark.parametrize(
|
|
"kw",
|
|
[
|
|
{"merchant": "WALMART"},
|
|
{"transaction_type": TransactionType.DEVOLUCION},
|
|
{"source": TransactionSource.CASH},
|
|
],
|
|
)
|
|
def test_ingestion_ignores_non_tasa_cero(self, session, kw):
|
|
data = dict(
|
|
amount=Decimal("1000"),
|
|
merchant="CC CONSTRUPLAZA",
|
|
date=datetime(2026, 6, 1),
|
|
)
|
|
data.update(kw)
|
|
tx = create_transaction(
|
|
TransactionCreate(**data), session=session, _user="t"
|
|
)
|
|
assert tx.is_installment_anchor is False
|
|
assert session.exec(select(InstallmentPlan)).all() == []
|
|
|
|
def test_duplicate_reference_still_409_and_creates_nothing(self, session):
|
|
make_cc_tx(session)
|
|
with pytest.raises(HTTPException) as exc:
|
|
create_transaction(
|
|
TransactionCreate(
|
|
amount=Decimal("1"),
|
|
merchant="CC CONSTRUPLAZA",
|
|
date=datetime(2026, 7, 1),
|
|
reference="75669363",
|
|
),
|
|
session=session,
|
|
_user="t",
|
|
)
|
|
assert exc.value.status_code == 409
|
|
assert session.exec(select(InstallmentPlan)).all() == []
|
|
|
|
|
|
class TestBudgetExclusion:
|
|
def test_anchor_excluded_cuotas_counted_per_month(self, session):
|
|
tx = make_cc_tx(session, date=datetime(2026, 6, 25))
|
|
convert(session, tx) # cuotas: 25/06, 19/07, 19/08
|
|
# Budget month = cycle ENDING on its 18th: 25/06 -> July, 19/07 -> Aug,
|
|
# 19/08 -> Sep. The full 111,053.60 must appear nowhere.
|
|
by_month = {
|
|
m: compute_actuals_by_source(session, 2026, m)["CREDIT_CARD"]["net"]
|
|
for m in (6, 7, 8, 9, 10)
|
|
}
|
|
assert by_month[6] == 0
|
|
assert by_month[7] == pytest.approx(37017.80)
|
|
assert by_month[8] == pytest.approx(37017.80)
|
|
assert by_month[9] == pytest.approx(37018.00)
|
|
assert by_month[10] == 0
|
|
|
|
def test_cuotas_inherit_anchor_category(self, session):
|
|
cat = Category(name="Ferretería")
|
|
session.add(cat)
|
|
session.commit()
|
|
tx = make_cc_tx(session, category_id=cat.id)
|
|
plan = convert(session, tx)
|
|
assert all(
|
|
c.category_id == cat.id for c in get_cuotas(session, plan.id)
|
|
)
|
|
|
|
|
|
class TestPlanLifecycle:
|
|
def test_manual_convert_validations(self, session):
|
|
tx = make_cc_tx(session)
|
|
convert(session, tx)
|
|
# already converted
|
|
with pytest.raises(HTTPException) as exc:
|
|
convert(session, tx)
|
|
assert exc.value.status_code == 409
|
|
# a cuota cannot itself be converted
|
|
cuota = session.exec(
|
|
select(Transaction).where(Transaction.installment_plan_id.is_not(None))
|
|
).first()
|
|
with pytest.raises(HTTPException) as exc:
|
|
convert(session, cuota)
|
|
assert exc.value.status_code == 400
|
|
# non-CC rejected
|
|
cash = make_cc_tx(
|
|
session, reference="r-cash", source=TransactionSource.CASH
|
|
)
|
|
with pytest.raises(HTTPException) as exc:
|
|
convert(session, cash)
|
|
assert exc.value.status_code == 400
|
|
|
|
def test_regenerate_3_to_6(self, session):
|
|
tx = make_cc_tx(session, amount="425800.00", reference="58710871")
|
|
plan = convert(session, tx)
|
|
updated = update_installment_plan(
|
|
plan.id,
|
|
InstallmentPlanUpdate(
|
|
num_installments=6,
|
|
first_installment_date=datetime(2026, 5, 19),
|
|
),
|
|
session=session,
|
|
_user="t",
|
|
)
|
|
assert updated.num_installments == 6
|
|
assert updated.installment_amount == pytest.approx(70966.60)
|
|
assert updated.last_installment_amount == pytest.approx(70967.00)
|
|
cuotas = get_cuotas(session, plan.id)
|
|
assert len(cuotas) == 6
|
|
assert cuotas[0].date == datetime(2026, 5, 19)
|
|
assert cuotas[-1].date == datetime(2026, 10, 19)
|
|
assert cuotas[0].reference == "58710871-C01"
|
|
assert cuotas[-1].merchant == "CC CONSTRUPLAZA CUOTA:06/06"
|
|
|
|
def test_unconvert_restores_anchor(self, session):
|
|
tx = make_cc_tx(session)
|
|
plan = convert(session, tx)
|
|
delete_installment_plan(plan.id, session=session, _user="t")
|
|
session.refresh(tx)
|
|
assert tx.is_installment_anchor is False
|
|
assert get_cuotas(session, plan.id) == []
|
|
assert session.exec(select(InstallmentPlan)).all() == []
|
|
|
|
def test_delete_anchor_tears_down_plan(self, session):
|
|
tx = make_cc_tx(session)
|
|
plan = convert(session, tx)
|
|
delete_transaction(tx.id, session=session, _user="t")
|
|
assert session.exec(select(InstallmentPlan)).all() == []
|
|
assert get_cuotas(session, plan.id) == []
|
|
|
|
def test_bulk_delete_anchor_tears_down_plan(self, session):
|
|
tx = make_cc_tx(session)
|
|
convert(session, tx)
|
|
bulk_action(
|
|
BulkActionRequest(ids=[tx.id], action="delete"),
|
|
session=session,
|
|
_user="t",
|
|
)
|
|
assert session.exec(select(InstallmentPlan)).all() == []
|
|
assert session.exec(select(Transaction)).all() == []
|
|
|
|
def test_list_and_detail(self, session):
|
|
tx = make_cc_tx(session, date=datetime(2026, 5, 12, 15, 41))
|
|
convert(session, tx, first=datetime(2026, 5, 19))
|
|
resp = list_installment_plans(session=session, _user="t")
|
|
assert len(resp.plans) == 1
|
|
plan = resp.plans[0]
|
|
assert plan.merchant == "CC CONSTRUPLAZA"
|
|
assert plan.total_amount == pytest.approx(111053.60)
|
|
assert plan.remaining_amount + plan.paid_amount == pytest.approx(
|
|
111053.60
|
|
)
|
|
assert resp.total_remaining == pytest.approx(plan.remaining_amount)
|
|
detail = get_installment_plan(plan.id, session=session, _user="t")
|
|
assert len(detail.cuotas) == 3
|
|
|
|
|
|
class TestDeferredGuards:
|
|
def test_patch_cuota_deferred_rejected(self, session):
|
|
tx = make_cc_tx(session)
|
|
plan = convert(session, tx)
|
|
cuota = get_cuotas(session, plan.id)[0]
|
|
with pytest.raises(HTTPException) as exc:
|
|
update_transaction(
|
|
cuota.id,
|
|
TransactionUpdate(deferred_to_next_cycle=True),
|
|
session=session,
|
|
_user="t",
|
|
)
|
|
assert exc.value.status_code == 400
|
|
|
|
def test_bulk_set_deferred_skips_cuotas(self, session):
|
|
tx = make_cc_tx(session)
|
|
plan = convert(session, tx)
|
|
cuota = get_cuotas(session, plan.id)[0]
|
|
normal = make_cc_tx(session, merchant="WALMART", reference="w1")
|
|
resp = bulk_action(
|
|
BulkActionRequest(
|
|
ids=[cuota.id, normal.id], action="set_deferred", deferred=True
|
|
),
|
|
session=session,
|
|
_user="t",
|
|
)
|
|
assert resp["affected"] == 1
|
|
session.refresh(cuota)
|
|
session.refresh(normal)
|
|
assert cuota.deferred_to_next_cycle is False
|
|
assert normal.deferred_to_next_cycle is True
|