"""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