mirror of
https://github.com/escalante29/WealthySmart.git
synced 2026-07-17 16:08:47 +02:00
ce3cc69846
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>
132 lines
4.0 KiB
Python
132 lines
4.0 KiB
Python
"""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")
|