Import dedup transparency + override; bulk transaction actions

Paste-import now returns each skipped duplicate WITH the matched
existing row (id, merchant, date, amount, source) instead of a bare
count, and accepts allow_duplicates to force-import false positives —
the practical fix for same-day identical purchases hashing together
(UX-20; supersedes the BE-08 hash change, which would have orphaned
every existing hash). POST /transactions/bulk applies
delete/set_category/set_deferred to up to 500 ids atomically. 82 tests.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
Carlos Escalante
2026-06-10 20:46:23 -06:00
parent 5612f69f11
commit ce3cc69846
3 changed files with 215 additions and 8 deletions

View File

@@ -0,0 +1,131 @@
"""Paste-import dedup transparency + override (UX-20) and bulk actions (4.4).
These exercise the endpoint functions directly with a session — no HTTP
stack needed.
"""
from datetime import datetime
from decimal import Decimal
import pytest
from fastapi import HTTPException
from app.api.v1.endpoints.import_transactions import (
PasteImportRequest,
paste_import,
)
from app.api.v1.endpoints.transactions import BulkActionRequest, bulk_action
from app.models.models import Category, Transaction, TransactionSource
LINE = "15/04/2026\tWALMART\\SAN JOSE\\CR\t25,000.50 CRC"
def do_import(session, text=LINE, **kw):
return paste_import(
PasteImportRequest(text=text, **kw), session=session, _user="t"
)
class TestImportReview:
def test_duplicate_is_reported_with_matched_row(self, session):
first = do_import(session)
assert first.imported == 1 and first.duplicates == 0
second = do_import(session)
assert second.imported == 0
assert second.duplicates == 1
assert len(second.skipped) == 1
skip = second.skipped[0]
assert skip.merchant == "WALMART"
assert skip.amount == 25000.50
assert skip.existing_merchant == "WALMART"
assert skip.existing_amount == 25000.50
assert skip.existing_id > 0
def test_allow_duplicates_overrides_dedup(self, session):
do_import(session)
forced = do_import(session, allow_duplicates=True)
assert forced.imported == 1
assert forced.duplicates == 0
assert forced.skipped == []
def make_tx(session, merchant="M", category_id=None):
tx = Transaction(
amount=Decimal("10"),
merchant=merchant,
date=datetime(2026, 4, 1),
category_id=category_id,
source=TransactionSource.CREDIT_CARD,
)
session.add(tx)
session.commit()
session.refresh(tx)
return tx
class TestBulkActions:
def test_bulk_set_category(self, session):
cat = Category(name="Bulk")
session.add(cat)
session.commit()
session.refresh(cat)
txs = [make_tx(session, f"m{i}") for i in range(3)]
result = bulk_action(
BulkActionRequest(
ids=[t.id for t in txs], action="set_category", category_id=cat.id
),
session=session,
_user="t",
)
assert result["affected"] == 3
for t in txs:
session.refresh(t)
assert t.category_id == cat.id
def test_bulk_set_deferred_and_clear_category(self, session):
cat = Category(name="X")
session.add(cat)
session.commit()
session.refresh(cat)
tx = make_tx(session, category_id=cat.id)
bulk_action(
BulkActionRequest(ids=[tx.id], action="set_deferred", deferred=True),
session=session,
_user="t",
)
session.refresh(tx)
assert tx.deferred_to_next_cycle is True
bulk_action(
BulkActionRequest(ids=[tx.id], action="set_category", category_id=None),
session=session,
_user="t",
)
session.refresh(tx)
assert tx.category_id is None
def test_bulk_delete(self, session):
txs = [make_tx(session, f"d{i}") for i in range(2)]
result = bulk_action(
BulkActionRequest(ids=[t.id for t in txs], action="delete"),
session=session,
_user="t",
)
assert result["affected"] == 2
assert session.get(Transaction, txs[0].id) is None
@pytest.mark.parametrize(
"req",
[
BulkActionRequest(ids=[], action="delete"),
BulkActionRequest(ids=[1], action="explode"),
BulkActionRequest(ids=[1], action="set_deferred"), # missing flag
BulkActionRequest(ids=[1], action="set_category", category_id=99999),
],
)
def test_invalid_requests_rejected(self, session, req):
with pytest.raises(HTTPException):
bulk_action(req, session=session, _user="t")