k1nq
2117029071
- Updated timestamps and hashes for backend/app/imports/router.py and backend/app/imports/service.py to reflect recent changes. - Updated metadata for backend/tests/test_imports.py with new timestamps and hashes. - Added new entries for .kilo/plugins/graphify.js and AGENTS.md, including their respective timestamps and hashes.
539 lines
18 KiB
Python
539 lines
18 KiB
Python
import uuid
|
|
from contextlib import contextmanager
|
|
from datetime import date
|
|
from types import SimpleNamespace
|
|
|
|
import pytest
|
|
from sqlalchemy.orm import Session
|
|
|
|
from app.errors import AppError
|
|
from app.identity.models import User
|
|
from app.imports import router as imports_router
|
|
from app.imports import service
|
|
from app.imports.adapters.base import SourceRow
|
|
from app.imports.models import (
|
|
DedupCandidate,
|
|
ParseRun,
|
|
SourceRecord,
|
|
SourceRecordLink,
|
|
StatementDocument,
|
|
)
|
|
from app.ledger.models import Operation
|
|
from app.ledger.service import create_account
|
|
from app.spaces.service import create_space
|
|
|
|
|
|
def seed(db: Session):
|
|
user = User(email=f"{uuid.uuid4().hex}@example.com", password_hash="unused")
|
|
db.add(user)
|
|
db.flush()
|
|
space = create_space(db, user=user, name="Imports")
|
|
account = create_account(
|
|
db,
|
|
space_id=space.id,
|
|
user=user,
|
|
name="Debit",
|
|
kind="debit",
|
|
opened_on=date(2026, 1, 1),
|
|
)
|
|
return user, space, account
|
|
|
|
|
|
def ready_document(db: Session):
|
|
user, space, account = seed(db)
|
|
document = StatementDocument(
|
|
space_id=space.id,
|
|
account_id=account.id,
|
|
original_filename="synthetic.xlsx",
|
|
size_bytes=100,
|
|
content_sha256=uuid.uuid4().hex + uuid.uuid4().hex,
|
|
object_key=uuid.uuid4().hex,
|
|
status="ready",
|
|
reconciliation={"full": True, "matched": True},
|
|
uploaded_by_user_id=user.id,
|
|
)
|
|
db.add(document)
|
|
db.flush()
|
|
run = ParseRun(
|
|
space_id=space.id,
|
|
document_id=document.id,
|
|
adapter_id="alfa_debit_xlsx",
|
|
status="succeeded",
|
|
)
|
|
db.add(run)
|
|
db.flush()
|
|
document.active_parse_run_id = run.id
|
|
source = SourceRecord(
|
|
space_id=space.id,
|
|
document_id=document.id,
|
|
parse_run_id=run.id,
|
|
occurred_on=date(2026, 1, 5),
|
|
amount_kopecks=12_345,
|
|
direction="out",
|
|
row_ref="sheet0:row2",
|
|
record_fp=uuid.uuid4().hex,
|
|
)
|
|
db.add(source)
|
|
db.flush()
|
|
return user, space, document
|
|
|
|
|
|
def test_confirm_and_revert_are_idempotent(db: Session) -> None:
|
|
user, space, document = ready_document(db)
|
|
service.confirm_document(db, space_id=space.id, document_id=document.id, user=user)
|
|
service.confirm_document(db, space_id=space.id, document_id=document.id, user=user)
|
|
db.flush()
|
|
assert db.query(Operation).filter_by(space_id=space.id).count() == 1
|
|
assert db.query(SourceRecordLink).filter_by(space_id=space.id).count() == 1
|
|
|
|
service.revert_document(db, space_id=space.id, document_id=document.id, user=user)
|
|
service.revert_document(db, space_id=space.id, document_id=document.id, user=user)
|
|
db.flush()
|
|
assert db.query(Operation).filter_by(space_id=space.id).count() == 0
|
|
assert db.query(SourceRecordLink).filter_by(space_id=space.id).count() == 0
|
|
|
|
|
|
def test_confirm_rejects_arithmetic_mismatch(db: Session) -> None:
|
|
user, space, document = ready_document(db)
|
|
document.reconciliation = {"full": True, "matched": False}
|
|
with pytest.raises(AppError) as exc:
|
|
service.confirm_document(db, space_id=space.id, document_id=document.id, user=user)
|
|
assert exc.value.code == "IMPORT_RECONCILIATION_MISMATCH"
|
|
|
|
|
|
def test_confirmed_import_must_be_reverted_before_reparse(db: Session) -> None:
|
|
user, space, document = ready_document(db)
|
|
service.confirm_document(db, space_id=space.id, document_id=document.id, user=user)
|
|
with pytest.raises(AppError) as exc:
|
|
service.prepare_reparse(db, space_id=space.id, document_id=document.id, user=user)
|
|
assert exc.value.code == "IMPORT_REPARSE_CONFIRMED"
|
|
|
|
|
|
def test_invalid_direction_is_critical() -> None:
|
|
source = service._source_from_row(
|
|
space_id=uuid.uuid4(),
|
|
document_id=uuid.uuid4(),
|
|
run_id=uuid.uuid4(),
|
|
row=SourceRow(
|
|
external_id=None,
|
|
occurred_on=date(2026, 1, 1),
|
|
posted_on=None,
|
|
amount_kopecks=100,
|
|
direction="sideways",
|
|
description=None,
|
|
mcc=None,
|
|
bank_category=None,
|
|
location=None,
|
|
row_ref="row1",
|
|
),
|
|
)
|
|
assert source.critical is True
|
|
assert source.field_warnings == {"codes": ["direction_invalid"]}
|
|
|
|
|
|
def test_unexpected_parser_failure_reaches_terminal_state(db: Session, monkeypatch) -> None:
|
|
_user, space, account = seed(db)
|
|
document = StatementDocument(
|
|
space_id=space.id,
|
|
account_id=account.id,
|
|
original_filename="synthetic.pdf",
|
|
size_bytes=10,
|
|
content_sha256=uuid.uuid4().hex + uuid.uuid4().hex,
|
|
object_key=uuid.uuid4().hex,
|
|
dek_wrapped=b"key",
|
|
dek_wrap_nonce=b"nonce",
|
|
content_nonce=b"nonce",
|
|
status="queued",
|
|
)
|
|
db.add(document)
|
|
db.flush()
|
|
|
|
@contextmanager
|
|
def existing_session(_space_id):
|
|
yield db
|
|
|
|
class BrokenStore:
|
|
def get(self, _key):
|
|
raise RuntimeError("storage unavailable")
|
|
|
|
monkeypatch.setattr(service, "session_scope", existing_session)
|
|
monkeypatch.setattr(service, "get_blob_store", lambda: BrokenStore())
|
|
service.process_statement_document(document.id, space.id)
|
|
assert document.status == "failed"
|
|
assert document.error_code == "IMPORT_PARSE_FAILED"
|
|
|
|
|
|
def test_duplicate_candidate_can_reuse_unique_operation_without_client_id(db: Session) -> None:
|
|
user, space, first_document = ready_document(db)
|
|
service.confirm_document(db, space_id=space.id, document_id=first_document.id, user=user)
|
|
first_source = (
|
|
db.query(SourceRecord).filter_by(parse_run_id=first_document.active_parse_run_id).one()
|
|
)
|
|
second = StatementDocument(
|
|
space_id=space.id,
|
|
account_id=first_document.account_id,
|
|
original_filename="second.xlsx",
|
|
size_bytes=100,
|
|
content_sha256=uuid.uuid4().hex + uuid.uuid4().hex,
|
|
object_key=uuid.uuid4().hex,
|
|
status="review_required",
|
|
reconciliation={"full": True, "matched": True},
|
|
)
|
|
db.add(second)
|
|
db.flush()
|
|
run = ParseRun(
|
|
space_id=space.id,
|
|
document_id=second.id,
|
|
adapter_id="alfa_debit_xlsx",
|
|
status="succeeded",
|
|
)
|
|
db.add(run)
|
|
db.flush()
|
|
second.active_parse_run_id = run.id
|
|
duplicate_source = SourceRecord(
|
|
space_id=space.id,
|
|
document_id=second.id,
|
|
parse_run_id=run.id,
|
|
occurred_on=first_source.occurred_on,
|
|
amount_kopecks=first_source.amount_kopecks,
|
|
direction=first_source.direction,
|
|
row_ref="sheet0:row9",
|
|
record_fp=first_source.record_fp,
|
|
)
|
|
db.add(duplicate_source)
|
|
db.flush()
|
|
assert service._build_candidates(db, document=second, run=run)
|
|
candidate = db.query(DedupCandidate).filter_by(source_record_id=duplicate_source.id).one()
|
|
assert candidate.candidate_kind == "duplicate"
|
|
service.decide_source_record(
|
|
db,
|
|
space_id=space.id,
|
|
document_id=second.id,
|
|
source_record_id=duplicate_source.id,
|
|
decision="duplicate",
|
|
duplicate_operation_id=None,
|
|
user=user,
|
|
)
|
|
assert second.status == "ready"
|
|
service.confirm_document(db, space_id=space.id, document_id=second.id, user=user)
|
|
assert db.query(Operation).filter_by(space_id=space.id).count() == 1
|
|
|
|
|
|
def test_duplicate_candidates_ignore_rows_from_same_parse_run(db: Session) -> None:
|
|
user, space, document = ready_document(db)
|
|
service.confirm_document(db, space_id=space.id, document_id=document.id, user=user)
|
|
first_source = (
|
|
db.query(SourceRecord).filter_by(parse_run_id=document.active_parse_run_id).one()
|
|
)
|
|
existing_link = db.query(SourceRecordLink).filter_by(source_record_id=first_source.id).one()
|
|
second_source = SourceRecord(
|
|
space_id=first_source.space_id,
|
|
document_id=document.id,
|
|
parse_run_id=document.active_parse_run_id,
|
|
occurred_on=first_source.occurred_on,
|
|
amount_kopecks=first_source.amount_kopecks,
|
|
direction=first_source.direction,
|
|
row_ref="sheet0:row3",
|
|
record_fp=first_source.record_fp,
|
|
)
|
|
db.add(second_source)
|
|
db.flush()
|
|
db.add(
|
|
SourceRecordLink(
|
|
space_id=space.id,
|
|
source_record_id=second_source.id,
|
|
operation_id=existing_link.operation_id,
|
|
kind="duplicate",
|
|
)
|
|
)
|
|
db.flush()
|
|
|
|
run = db.get(ParseRun, document.active_parse_run_id)
|
|
assert run is not None
|
|
assert service._build_candidates(db, document=document, run=run) is False
|
|
assert db.query(DedupCandidate).filter_by(space_id=document.space_id).count() == 0
|
|
|
|
|
|
def test_duplicate_candidates_ignore_inactive_parse_runs(db: Session) -> None:
|
|
user, space, source_document = ready_document(db)
|
|
service.confirm_document(db, space_id=space.id, document_id=source_document.id, user=user)
|
|
active_source = (
|
|
db.query(SourceRecord).filter_by(parse_run_id=source_document.active_parse_run_id).one()
|
|
)
|
|
active_link = db.query(SourceRecordLink).filter_by(source_record_id=active_source.id).one()
|
|
inactive_run = ParseRun(
|
|
space_id=space.id,
|
|
document_id=source_document.id,
|
|
adapter_id="alfa_debit_xlsx",
|
|
status="succeeded",
|
|
)
|
|
db.add(inactive_run)
|
|
db.flush()
|
|
inactive_source = SourceRecord(
|
|
space_id=space.id,
|
|
document_id=source_document.id,
|
|
parse_run_id=inactive_run.id,
|
|
occurred_on=active_source.occurred_on,
|
|
amount_kopecks=active_source.amount_kopecks,
|
|
direction=active_source.direction,
|
|
row_ref="sheet0:old-row",
|
|
record_fp="inactive-run-fingerprint",
|
|
)
|
|
db.add(inactive_source)
|
|
db.flush()
|
|
db.add(
|
|
SourceRecordLink(
|
|
space_id=space.id,
|
|
source_record_id=inactive_source.id,
|
|
operation_id=active_link.operation_id,
|
|
kind="duplicate",
|
|
)
|
|
)
|
|
target_document = StatementDocument(
|
|
space_id=space.id,
|
|
account_id=source_document.account_id,
|
|
original_filename="target.xlsx",
|
|
size_bytes=100,
|
|
content_sha256=uuid.uuid4().hex + uuid.uuid4().hex,
|
|
object_key=uuid.uuid4().hex,
|
|
status="review_required",
|
|
reconciliation={"full": True, "matched": True},
|
|
)
|
|
db.add(target_document)
|
|
db.flush()
|
|
target_run = ParseRun(
|
|
space_id=space.id,
|
|
document_id=target_document.id,
|
|
adapter_id="alfa_debit_xlsx",
|
|
status="succeeded",
|
|
)
|
|
db.add(target_run)
|
|
db.flush()
|
|
target_document.active_parse_run_id = target_run.id
|
|
db.add(
|
|
SourceRecord(
|
|
space_id=space.id,
|
|
document_id=target_document.id,
|
|
parse_run_id=target_run.id,
|
|
occurred_on=inactive_source.occurred_on,
|
|
amount_kopecks=inactive_source.amount_kopecks,
|
|
direction=inactive_source.direction,
|
|
row_ref="sheet0:row2",
|
|
record_fp=inactive_source.record_fp,
|
|
)
|
|
)
|
|
db.flush()
|
|
|
|
assert service._build_candidates(db, document=target_document, run=target_run) is False
|
|
assert db.query(DedupCandidate).filter_by(space_id=space.id).count() == 0
|
|
|
|
|
|
def test_duplicate_candidates_ignore_unconfirmed_documents(db: Session) -> None:
|
|
_user, space, first_document = ready_document(db)
|
|
first_source = (
|
|
db.query(SourceRecord).filter_by(parse_run_id=first_document.active_parse_run_id).one()
|
|
)
|
|
second_document = StatementDocument(
|
|
space_id=space.id,
|
|
account_id=first_document.account_id,
|
|
original_filename="second.xlsx",
|
|
size_bytes=100,
|
|
content_sha256=uuid.uuid4().hex + uuid.uuid4().hex,
|
|
object_key=uuid.uuid4().hex,
|
|
status="review_required",
|
|
reconciliation={"full": True, "matched": True},
|
|
)
|
|
db.add(second_document)
|
|
db.flush()
|
|
second_run = ParseRun(
|
|
space_id=space.id,
|
|
document_id=second_document.id,
|
|
adapter_id="alfa_debit_xlsx",
|
|
status="succeeded",
|
|
)
|
|
db.add(second_run)
|
|
db.flush()
|
|
second_document.active_parse_run_id = second_run.id
|
|
db.add(
|
|
SourceRecord(
|
|
space_id=space.id,
|
|
document_id=second_document.id,
|
|
parse_run_id=second_run.id,
|
|
occurred_on=first_source.occurred_on,
|
|
amount_kopecks=first_source.amount_kopecks,
|
|
direction=first_source.direction,
|
|
row_ref="sheet0:row2",
|
|
record_fp=first_source.record_fp,
|
|
)
|
|
)
|
|
db.flush()
|
|
|
|
assert service._build_candidates(db, document=second_document, run=second_run) is False
|
|
assert db.query(DedupCandidate).filter_by(space_id=space.id).count() == 0
|
|
|
|
|
|
def test_duplicate_candidates_are_retained_before_account_mapping(db: Session) -> None:
|
|
user, space, first_document = ready_document(db)
|
|
service.confirm_document(db, space_id=space.id, document_id=first_document.id, user=user)
|
|
first_source = (
|
|
db.query(SourceRecord).filter_by(parse_run_id=first_document.active_parse_run_id).one()
|
|
)
|
|
second_document = StatementDocument(
|
|
space_id=space.id,
|
|
account_id=None,
|
|
original_filename="unmapped.xlsx",
|
|
size_bytes=100,
|
|
content_sha256=uuid.uuid4().hex + uuid.uuid4().hex,
|
|
object_key=uuid.uuid4().hex,
|
|
status="review_required",
|
|
reconciliation={"full": True, "matched": True},
|
|
)
|
|
db.add(second_document)
|
|
db.flush()
|
|
second_run = ParseRun(
|
|
space_id=space.id,
|
|
document_id=second_document.id,
|
|
adapter_id="alfa_debit_xlsx",
|
|
status="succeeded",
|
|
)
|
|
db.add(second_run)
|
|
db.flush()
|
|
second_document.active_parse_run_id = second_run.id
|
|
source = SourceRecord(
|
|
space_id=space.id,
|
|
document_id=second_document.id,
|
|
parse_run_id=second_run.id,
|
|
occurred_on=first_source.occurred_on,
|
|
amount_kopecks=first_source.amount_kopecks,
|
|
direction=first_source.direction,
|
|
row_ref="sheet0:row2",
|
|
record_fp=first_source.record_fp,
|
|
)
|
|
db.add(source)
|
|
db.flush()
|
|
|
|
assert service._build_candidates(db, document=second_document, run=second_run) is True
|
|
candidate = db.query(DedupCandidate).filter_by(source_record_id=source.id).one()
|
|
assert candidate.existing_source_record_id == first_source.id
|
|
|
|
|
|
def test_import_detail_exposes_pending_candidate_and_blocker(db: Session) -> None:
|
|
_user, space, document = ready_document(db)
|
|
document.status = "review_required"
|
|
source = db.query(SourceRecord).filter_by(parse_run_id=document.active_parse_run_id).one()
|
|
candidate = DedupCandidate(
|
|
space_id=space.id,
|
|
source_record_id=source.id,
|
|
candidate_kind="duplicate",
|
|
confidence="possible",
|
|
)
|
|
db.add(candidate)
|
|
db.flush()
|
|
|
|
response = imports_router.detail(document.id, db, SimpleNamespace(space=space))
|
|
|
|
assert response["rows"][0]["decision"] == "pending"
|
|
assert response["rows"][0]["candidates"] == [
|
|
{
|
|
"id": candidate.id,
|
|
"kind": "duplicate",
|
|
"confidence": "possible",
|
|
"resolution": "pending",
|
|
"existingOperationId": None,
|
|
}
|
|
]
|
|
assert response["blockers"] == [
|
|
{
|
|
"code": "IMPORT_DECISION_REQUIRED",
|
|
"message": "Possible duplicate or transfer requires a decision",
|
|
"rowId": str(source.id),
|
|
}
|
|
]
|
|
assert response["totals"]["pending"] == 1
|
|
|
|
|
|
def test_confirm_reports_pending_candidate_instead_of_generic_not_ready(db: Session) -> None:
|
|
user, space, document = ready_document(db)
|
|
document.status = "review_required"
|
|
source = db.query(SourceRecord).filter_by(parse_run_id=document.active_parse_run_id).one()
|
|
db.add(
|
|
DedupCandidate(
|
|
space_id=space.id,
|
|
source_record_id=source.id,
|
|
candidate_kind="duplicate",
|
|
confidence="possible",
|
|
)
|
|
)
|
|
db.flush()
|
|
|
|
with pytest.raises(AppError) as exc:
|
|
service.confirm_document(db, space_id=space.id, document_id=document.id, user=user)
|
|
|
|
assert exc.value.code == "IMPORT_DECISIONS_REQUIRED"
|
|
|
|
|
|
def test_transfer_merge_revert_restores_counterpart_operation(db: Session) -> None:
|
|
user, space, first_document = ready_document(db)
|
|
service.confirm_document(db, space_id=space.id, document_id=first_document.id, user=user)
|
|
first_source = (
|
|
db.query(SourceRecord).filter_by(parse_run_id=first_document.active_parse_run_id).one()
|
|
)
|
|
target = create_account(
|
|
db,
|
|
space_id=space.id,
|
|
user=user,
|
|
name="Second debit",
|
|
kind="debit",
|
|
opened_on=date(2026, 1, 1),
|
|
)
|
|
second = StatementDocument(
|
|
space_id=space.id,
|
|
account_id=target.id,
|
|
original_filename="target.xlsx",
|
|
size_bytes=100,
|
|
content_sha256=uuid.uuid4().hex + uuid.uuid4().hex,
|
|
object_key=uuid.uuid4().hex,
|
|
status="review_required",
|
|
reconciliation={"full": True, "matched": True},
|
|
)
|
|
db.add(second)
|
|
db.flush()
|
|
run = ParseRun(
|
|
space_id=space.id,
|
|
document_id=second.id,
|
|
adapter_id="alfa_debit_xlsx",
|
|
status="succeeded",
|
|
)
|
|
db.add(run)
|
|
db.flush()
|
|
second.active_parse_run_id = run.id
|
|
incoming = SourceRecord(
|
|
space_id=space.id,
|
|
document_id=second.id,
|
|
parse_run_id=run.id,
|
|
occurred_on=first_source.occurred_on,
|
|
amount_kopecks=first_source.amount_kopecks,
|
|
direction="in",
|
|
row_ref="sheet0:row10",
|
|
record_fp=uuid.uuid4().hex,
|
|
)
|
|
db.add(incoming)
|
|
db.flush()
|
|
assert service._build_candidates(db, document=second, run=run)
|
|
candidate = db.query(DedupCandidate).filter_by(source_record_id=incoming.id).one()
|
|
assert candidate.candidate_kind == "transfer"
|
|
service.decide_source_record(
|
|
db,
|
|
space_id=space.id,
|
|
document_id=second.id,
|
|
source_record_id=incoming.id,
|
|
decision="transfer",
|
|
duplicate_operation_id=None,
|
|
user=user,
|
|
)
|
|
service.confirm_document(db, space_id=space.id, document_id=second.id, user=user)
|
|
assert db.query(Operation).filter_by(space_id=space.id, type="transfer").count() == 1
|
|
service.revert_document(db, space_id=space.id, document_id=second.id, user=user)
|
|
assert db.query(Operation).filter_by(space_id=space.id, type="transfer").count() == 0
|
|
assert db.query(Operation).filter_by(space_id=space.id, type="expense").count() == 1
|