finance/backend/tests/test_import_holds.py

import uuid
from datetime import date
from pathlib import Path
from types import SimpleNamespace

import pytest
import sqlalchemy as sa
from alembic.config import Config
from alembic.migration import MigrationContext
from alembic.operations import Operations
from fastapi.testclient import TestClient
from sqlalchemy import create_engine
from sqlalchemy.exc import IntegrityError
from sqlalchemy.orm import Session

from alembic import command
from app.budgeting.service import close_period, compute_period_state, get_or_create_period
from app.errors import AppError
from app.identity.models import User
from app.imports import service
from app.imports.adapters.base import record_fingerprint
from app.imports.models import (
    HoldResolution,
    ParseRun,
    SourceRecord,
    SourceRecordLink,
    StatementDocument,
)
from app.ledger.models import Operation
from app.ledger.service import create_account
from app.spaces.models import Membership
from app.spaces.service import create_space

MAY_10 = date(2026, 5, 10)


def seed(db: Session, *, name: str = "Holds"):
    user = User(email=f"{uuid.uuid4().hex}@example.com", password_hash="unused")
    db.add(user)
    db.flush()
    space = create_space(db, user=user, name=name)
    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 make_document(
    db: Session,
    space,
    account,
    *,
    user: User | None = None,
    status: str = "ready",
    adapter_id: str = "alfa_debit_xlsx",
) -> tuple[StatementDocument, ParseRun]:
    document = StatementDocument(
        space_id=space.id,
        account_id=account.id,
        original_filename=f"{uuid.uuid4().hex}.xlsx",
        size_bytes=100,
        content_sha256=uuid.uuid4().hex + uuid.uuid4().hex,
        object_key=uuid.uuid4().hex,
        status=status,
        reconciliation={"full": True, "matched": True},
        uploaded_by_user_id=user.id if user else None,
    )
    db.add(document)
    db.flush()
    run = ParseRun(
        space_id=space.id,
        document_id=document.id,
        adapter_id=adapter_id,
        status="succeeded",
    )
    db.add(run)
    db.flush()
    document.active_parse_run_id = run.id
    document.adapter_id = adapter_id
    return document, run


def add_record(
    db: Session,
    document: StatementDocument,
    run: ParseRun,
    *,
    direction: str = "out",
    amount: int = 12_345,
    occurred_on: date = MAY_10,
    bank_status: str = "posted",
    external_id: str | None = None,
    description: str | None = None,
) -> SourceRecord:
    record = SourceRecord(
        space_id=document.space_id,
        document_id=document.id,
        parse_run_id=run.id,
        external_id=external_id,
        occurred_on=occurred_on,
        amount_kopecks=amount,
        direction=direction,
        bank_status=bank_status,
        description=description,
        row_ref=f"sheet0:{uuid.uuid4().hex[:8]}",
        record_fp=record_fingerprint(occurred_on, amount, description),
    )
    db.add(record)
    db.flush()
    return record


def hold_document(db: Session, *, occurred_on: date = MAY_10, external_id: str | None = None):
    """A confirmed import whose active run contains exactly one unresolved hold."""
    user, space, account = seed(db)
    document, run = make_document(db, space, account, user=user)
    hold = add_record(
        db,
        document,
        run,
        bank_status="hold",
        occurred_on=occurred_on,
        external_id=external_id,
        description="Coffee shop",
    )
    service.confirm_document(db, space_id=space.id, document_id=document.id, user=user)
    db.flush()
    return user, space, account, document, hold


def test_bank_status_defaults_to_posted_and_is_constrained(db: Session) -> None:
    user, space, account = seed(db)
    document, run = make_document(db, space, account, user=user)
    record = SourceRecord(
        space_id=space.id,
        document_id=document.id,
        parse_run_id=run.id,
        amount_kopecks=1_000,
        direction="out",
        row_ref="sheet0:row1",
        record_fp="fp",
    )
    db.add(record)
    db.flush()
    assert record.bank_status == "posted"
    bad = SourceRecord(
        space_id=space.id,
        document_id=document.id,
        parse_run_id=run.id,
        amount_kopecks=1_000,
        direction="out",
        bank_status="pending",
        row_ref="sheet0:row2",
        record_fp="fp2",
    )
    db.add(bad)
    with pytest.raises(IntegrityError):
        db.flush()
    db.rollback()


def test_hold_row_creates_no_operation_on_confirm(db: Session) -> None:
    user, space, account = seed(db)
    document, run = make_document(db, space, account, user=user)
    hold = add_record(db, document, run, bank_status="hold")
    add_record(
        db,
        document,
        run,
        bank_status="posted",
        occurred_on=date(2026, 5, 11),
        amount=7_000,
    )
    service.confirm_document(db, space_id=space.id, document_id=document.id, user=user)
    db.flush()
    operations = db.query(Operation).filter_by(space_id=space.id).all()
    assert len(operations) == 1
    assert operations[0].amount_kopecks == 7_000
    assert db.query(SourceRecordLink).filter_by(source_record_id=hold.id).count() == 0
    assert document.status == "confirmed"
    detail = service.document_detail(db, space_id=space.id, document_id=document.id)
    hold_row = next(item for item in detail["records"] if item["id"] == hold.id)
    assert hold_row["bank_status"] == "hold"
    assert hold_row["hold_resolution"] is None
    assert hold_row["critical"] is False


def test_unresolved_hold_blocks_only_its_occurred_month(db: Session) -> None:
    user, space, _account, _document, _hold = hold_document(db)
    april = get_or_create_period(db, space_id=space.id, year=2026, month=4)
    close_period(db, space_id=space.id, user=user, period=april)
    may = get_or_create_period(db, space_id=space.id, year=2026, month=5)
    state = compute_period_state(db, space_id=space.id, period=may)
    assert any(blocker["code"] == "IMPORT_UNRESOLVED_HOLDS" for blocker in state["close_blockers"])
    with pytest.raises(AppError) as exc:
        close_period(db, space_id=space.id, user=user, period=may)
    assert exc.value.code == "BUDGET_CLOSE_BLOCKED"
    june = get_or_create_period(db, space_id=space.id, year=2026, month=6)
    june_state = compute_period_state(db, space_id=space.id, period=june)
    assert not any(
        blocker["code"] == "IMPORT_UNRESOLVED_HOLDS" for blocker in june_state["close_blockers"]
    )


def test_discard_as_cancelled_without_reason_unblocks_month(db: Session) -> None:
    user, space, _account, document, hold = hold_document(db)
    resolution = service.resolve_hold(
        db,
        space_id=space.id,
        document_id=document.id,
        source_record_id=hold.id,
        user=user,
        action="discard_as_cancelled",
    )
    db.flush()
    assert resolution.resolution == "discarded_as_cancelled"
    assert resolution.operation_id is None
    assert resolution.resolved_by_user_id == user.id
    assert db.query(Operation).filter_by(space_id=space.id).count() == 0
    detail = service.document_detail(db, space_id=space.id, document_id=document.id)
    hold_row = next(item for item in detail["records"] if item["id"] == hold.id)
    assert hold_row["hold_resolution"]["reason"] is None
    april = get_or_create_period(db, space_id=space.id, year=2026, month=4)
    close_period(db, space_id=space.id, user=user, period=april)
    may = get_or_create_period(db, space_id=space.id, year=2026, month=5)
    close_period(db, space_id=space.id, user=user, period=may)
    assert may.status == "closed"


def test_confirm_as_posted_accepts_blank_reason(db: Session) -> None:
    user, space, _account, document, hold = hold_document(db)
    resolution = service.resolve_hold(
        db,
        space_id=space.id,
        document_id=document.id,
        source_record_id=hold.id,
        user=user,
        action="confirm_as_posted",
        reason="  ",
    )
    db.flush()
    assert resolution.reason is None
    assert resolution.operation_id is not None


def test_hold_cannot_be_resolved_before_document_confirmation(db: Session) -> None:
    user, space, account = seed(db)
    document, run = make_document(db, space, account, user=user)
    hold = add_record(db, document, run, bank_status="hold")

    with pytest.raises(AppError) as exc:
        service.resolve_hold(
            db,
            space_id=space.id,
            document_id=document.id,
            source_record_id=hold.id,
            user=user,
            action="confirm_as_posted",
            reason="Not confirmed yet",
        )

    assert exc.value.code == "IMPORT_HOLD_DOCUMENT_NOT_CONFIRMED"
    assert db.query(Operation).filter_by(space_id=space.id).count() == 0
    assert db.query(HoldResolution).count() == 0


def test_manual_confirm_rejects_bad_action_and_non_hold_row(db: Session) -> None:
    user, space, account = seed(db)
    document, run = make_document(db, space, account, user=user)
    hold = add_record(db, document, run, bank_status="hold")
    posted = add_record(
        db, document, run, bank_status="posted", occurred_on=date(2026, 5, 11), amount=900
    )
    service.confirm_document(db, space_id=space.id, document_id=document.id, user=user)
    with pytest.raises(AppError) as exc:
        service.resolve_hold(
            db,
            space_id=space.id,
            document_id=document.id,
            source_record_id=hold.id,
            user=user,
            action="delete_forever",
            reason="why not",
        )
    assert exc.value.code == "IMPORT_HOLD_ACTION_INVALID"
    with pytest.raises(AppError) as exc:
        service.resolve_hold(
            db,
            space_id=space.id,
            document_id=document.id,
            source_record_id=posted.id,
            user=user,
            action="discard_as_cancelled",
            reason="posted row",
        )
    assert exc.value.code == "IMPORT_HOLD_INVALID"
    with pytest.raises(AppError) as exc:
        service.resolve_hold(
            db,
            space_id=space.id,
            document_id=document.id,
            source_record_id=uuid.uuid4(),
            user=user,
            action="discard_as_cancelled",
            reason="missing row",
        )
    assert exc.value.code == "IMPORT_HOLD_NOT_FOUND"


def test_manual_confirm_creates_one_operation_and_late_import_dedupes(db: Session) -> None:
    user, space, account, document, hold = hold_document(db, external_id="H-1")
    resolution = service.resolve_hold(
        db,
        space_id=space.id,
        document_id=document.id,
        source_record_id=hold.id,
        user=user,
        action="confirm_as_posted",
        reason="Verified against the paper receipt",
        account_id=account.id,
    )
    db.flush()
    operations = db.query(Operation).filter_by(space_id=space.id).all()
    assert len(operations) == 1
    operation = operations[0]
    assert operation.type == "expense"
    assert operation.amount_kopecks == 12_345
    assert operation.occurred_on == MAY_10
    assert resolution.operation_id == operation.id
    link = db.query(SourceRecordLink).filter_by(source_record_id=hold.id).one_or_none()
    assert link is not None and link.operation_id == operation.id
    with pytest.raises(AppError) as exc:
        service.resolve_hold(
            db,
            space_id=space.id,
            document_id=document.id,
            source_record_id=hold.id,
            user=user,
            action="discard_as_cancelled",
            reason="second try",
        )
    assert exc.value.code == "IMPORT_HOLD_ALREADY_RESOLVED"

    later, later_run = make_document(db, space, account, user=user)
    late_posted = add_record(
        db,
        later,
        later_run,
        bank_status="posted",
        external_id="H-1",
        description="Coffee shop",
    )
    service.confirm_document(db, space_id=space.id, document_id=later.id, user=user)
    db.flush()
    assert db.query(Operation).filter_by(space_id=space.id).count() == 1
    late_link = db.query(SourceRecordLink).filter_by(source_record_id=late_posted.id).one_or_none()
    assert late_link is not None
    assert late_link.operation_id == operation.id
    assert late_link.kind == "duplicate"


def test_manual_confirm_of_income_hold_creates_income(db: Session) -> None:
    user, space, account = seed(db)
    document, run = make_document(db, space, account, user=user)
    hold = add_record(
        db,
        document,
        run,
        direction="in",
        amount=45_000,
        bank_status="hold",
        description="Refund hold",
    )
    service.confirm_document(db, space_id=space.id, document_id=document.id, user=user)
    service.resolve_hold(
        db,
        space_id=space.id,
        document_id=document.id,
        source_record_id=hold.id,
        user=user,
        action="confirm_as_posted",
        reason="Money arrived",
    )
    db.flush()
    operation = db.query(Operation).filter_by(space_id=space.id).one()
    assert operation.type == "income"
    assert operation.amount_kopecks == 45_000


def test_manual_confirm_honors_closed_period(db: Session) -> None:
    user, space, account = seed(db)
    february = get_or_create_period(db, space_id=space.id, year=2026, month=2)
    close_period(db, space_id=space.id, user=user, period=february)
    march = get_or_create_period(db, space_id=space.id, year=2026, month=3)
    close_period(db, space_id=space.id, user=user, period=march)
    document, run = make_document(db, space, account, user=user)
    hold = add_record(db, document, run, bank_status="hold", occurred_on=date(2026, 3, 15))
    service.confirm_document(db, space_id=space.id, document_id=document.id, user=user)
    with pytest.raises(AppError) as exc:
        service.resolve_hold(
            db,
            space_id=space.id,
            document_id=document.id,
            source_record_id=hold.id,
            user=user,
            action="confirm_as_posted",
            reason="Book it",
        )
    assert exc.value.code == "BUDGET_PERIOD_CLOSED"
    db.flush()
    assert db.query(Operation).filter_by(space_id=space.id).count() == 0
    assert db.query(HoldResolution).count() == 0
    resolution = service.resolve_hold(
        db,
        space_id=space.id,
        document_id=document.id,
        source_record_id=hold.id,
        user=user,
        action="discard_as_cancelled",
        reason="Never posted",
    )
    assert resolution.operation_id is None


def test_unique_posted_record_resolves_hold_on_confirmation(db: Session) -> None:
    user, space, account = seed(db)
    first, first_run = make_document(db, space, account, user=user)
    hold = add_record(
        db,
        first,
        first_run,
        bank_status="hold",
        external_id="ALFA-1",
        description="Coffee shop",
    )
    service.confirm_document(db, space_id=space.id, document_id=first.id, user=user)
    assert db.query(HoldResolution).count() == 0

    second, second_run = make_document(db, space, account, user=user)
    posted = add_record(
        db,
        second,
        second_run,
        bank_status="posted",
        external_id="ALFA-1",
        description="Coffee shop",
    )
    service.confirm_document(db, space_id=space.id, document_id=second.id, user=user)
    db.flush()
    resolution = db.query(HoldResolution).one()
    assert resolution.hold_source_record_id == hold.id
    assert resolution.resolving_source_record_id == posted.id
    assert resolution.resolution == "matched_posted"
    assert resolution.resolved_by_user_id is None
    operations = db.query(Operation).filter_by(space_id=space.id).all()
    assert len(operations) == 1
    assert operations[0].amount_kopecks == 12_345
    may = get_or_create_period(db, space_id=space.id, year=2026, month=5)
    state = compute_period_state(db, space_id=space.id, period=may)
    assert not any(
        blocker["code"] == "IMPORT_UNRESOLVED_HOLDS" for blocker in state["close_blockers"]
    )


def test_external_id_match_survives_description_drift(db: Session) -> None:
    user, space, account = seed(db)
    first, first_run = make_document(db, space, account, user=user)
    hold = add_record(
        db,
        first,
        first_run,
        bank_status="hold",
        external_id="X-9",
        description="HOLD COFFEE",
        occurred_on=MAY_10,
    )
    service.confirm_document(db, space_id=space.id, document_id=first.id, user=user)
    second, second_run = make_document(db, space, account, user=user)
    posted = add_record(
        db,
        second,
        second_run,
        bank_status="posted",
        external_id="X-9",
        description="Coffee llc prague",
        occurred_on=date(2026, 5, 12),
    )
    service.confirm_document(db, space_id=space.id, document_id=second.id, user=user)
    db.flush()
    resolution = db.query(HoldResolution).one()
    assert resolution.hold_source_record_id == hold.id
    assert resolution.resolving_source_record_id == posted.id


def test_fingerprint_match_without_external_id_resolves(db: Session) -> None:
    user, space, account = seed(db)
    first, first_run = make_document(db, space, account, user=user)
    hold = add_record(
        db,
        first,
        first_run,
        bank_status="hold",
        description="Coffee shop",
    )
    service.confirm_document(db, space_id=space.id, document_id=first.id, user=user)
    second, second_run = make_document(db, space, account, user=user)
    posted = add_record(
        db,
        second,
        second_run,
        bank_status="posted",
        description="Coffee shop",
    )
    matches = service._match_hold_resolutions(db, document=second, run=second_run)
    assert matches == {hold.id: posted.id}
    service.confirm_document(db, space_id=space.id, document_id=second.id, user=user)
    db.flush()
    assert db.query(HoldResolution).count() == 1


def test_ambiguous_overlap_remains_unresolved(db: Session) -> None:
    user, space, account = seed(db)
    first, first_run = make_document(db, space, account, user=user)
    hold_a = add_record(
        db,
        first,
        first_run,
        bank_status="hold",
        external_id="A-1",
        description="Coffee shop",
    )
    hold_b = add_record(
        db,
        first,
        first_run,
        bank_status="hold",
        external_id="B-2",
        description="Coffee shop",
    )
    service.confirm_document(db, space_id=space.id, document_id=first.id, user=user)
    assert hold_a.id != hold_b.id
    second, second_run = make_document(db, space, account, user=user)
    add_record(
        db,
        second,
        second_run,
        bank_status="posted",
        description="Coffee shop",
    )
    service.confirm_document(db, space_id=space.id, document_id=second.id, user=user)
    db.flush()
    assert db.query(HoldResolution).count() == 0
    may = get_or_create_period(db, space_id=space.id, year=2026, month=5)
    state = compute_period_state(db, space_id=space.id, period=may)
    assert any(blocker["code"] == "IMPORT_UNRESOLVED_HOLDS" for blocker in state["close_blockers"])


def test_matcher_rejects_two_posted_candidates_for_one_hold(db: Session) -> None:
    user, space, account = seed(db)
    first, first_run = make_document(db, space, account, user=user)
    add_record(db, first, first_run, bank_status="hold", description="Coffee shop")
    service.confirm_document(db, space_id=space.id, document_id=first.id, user=user)
    second, second_run = make_document(db, space, account, user=user)
    add_record(db, second, second_run, bank_status="posted", description="Coffee shop")
    add_record(db, second, second_run, bank_status="posted", description="Coffee shop")
    matches = service._match_hold_resolutions(db, document=second, run=second_run)
    assert matches == {}
    assert db.query(HoldResolution).count() == 0


def test_discarded_posted_record_does_not_resolve_hold(db: Session) -> None:
    user, space, account = seed(db)
    first, first_run = make_document(db, space, account, user=user)
    hold = add_record(
        db,
        first,
        first_run,
        bank_status="hold",
        external_id="DISCARDED-1",
        description="Coffee shop",
    )
    service.confirm_document(db, space_id=space.id, document_id=first.id, user=user)

    second, second_run = make_document(db, space, account, user=user)
    posted = add_record(
        db,
        second,
        second_run,
        bank_status="posted",
        external_id="DISCARDED-1",
        description="Coffee shop",
    )
    posted.review_decision = "discard"
    service.confirm_document(db, space_id=space.id, document_id=second.id, user=user)
    db.flush()

    assert db.query(HoldResolution).filter_by(hold_source_record_id=hold.id).count() == 0
    assert db.query(Operation).filter_by(space_id=space.id).count() == 0
    may = get_or_create_period(db, space_id=space.id, year=2026, month=5)
    state = compute_period_state(db, space_id=space.id, period=may)
    assert any(blocker["code"] == "IMPORT_UNRESOLVED_HOLDS" for blocker in state["close_blockers"])


def test_hold_matching_is_tenant_scoped(db: Session) -> None:
    user1, space1, account1 = seed(db, name="Space One")
    user2, space2, account2 = seed(db, name="Space Two")
    first, first_run = make_document(db, space1, account1, user=user1)
    hold = add_record(
        db,
        first,
        first_run,
        bank_status="hold",
        external_id="T-1",
        description="Coffee shop",
    )
    service.confirm_document(db, space_id=space1.id, document_id=first.id, user=user1)
    second, second_run = make_document(db, space2, account2, user=user2)
    add_record(
        db,
        second,
        second_run,
        bank_status="posted",
        external_id="T-1",
        description="Coffee shop",
    )
    service.confirm_document(db, space_id=space2.id, document_id=second.id, user=user2)
    db.flush()
    assert db.query(HoldResolution).count() == 0
    may_two = get_or_create_period(db, space_id=space2.id, year=2026, month=5)
    state_two = compute_period_state(db, space_id=space2.id, period=may_two)
    assert not any(
        blocker["code"] == "IMPORT_UNRESOLVED_HOLDS" for blocker in state_two["close_blockers"]
    )
    may_one = get_or_create_period(db, space_id=space1.id, year=2026, month=5)
    state_one = compute_period_state(db, space_id=space1.id, period=may_one)
    assert any(
        blocker["code"] == "IMPORT_UNRESOLVED_HOLDS" for blocker in state_one["close_blockers"]
    )
    with pytest.raises(AppError) as exc:
        service.resolve_hold(
            db,
            space_id=space2.id,
            document_id=first.id,
            source_record_id=hold.id,
            user=user2,
            action="discard_as_cancelled",
            reason="cross tenant",
        )
    assert exc.value.code == "IMPORT_NOT_FOUND"


def test_revert_clears_automatic_match_and_reblocks_month(db: Session) -> None:
    user, space, account = seed(db)
    first, first_run = make_document(db, space, account, user=user)
    hold = add_record(
        db,
        first,
        first_run,
        bank_status="hold",
        external_id="R-1",
        description="Coffee shop",
    )
    service.confirm_document(db, space_id=space.id, document_id=first.id, user=user)
    second, second_run = make_document(db, space, account, user=user)
    add_record(
        db,
        second,
        second_run,
        bank_status="posted",
        external_id="R-1",
        description="Coffee shop",
    )
    service.confirm_document(db, space_id=space.id, document_id=second.id, user=user)
    db.flush()
    assert db.query(HoldResolution).count() == 1
    service.revert_document(db, space_id=space.id, document_id=second.id, user=user)
    db.flush()
    assert db.query(HoldResolution).count() == 0
    assert db.query(Operation).filter_by(space_id=space.id).count() == 0
    may = get_or_create_period(db, space_id=space.id, year=2026, month=5)
    state = compute_period_state(db, space_id=space.id, period=may)
    assert any(blocker["code"] == "IMPORT_UNRESOLVED_HOLDS" for blocker in state["close_blockers"])
    assert hold.bank_status == "hold"


def test_revert_preserves_manual_hold_operation_and_provenance(db: Session) -> None:
    user, space, account, document, hold = hold_document(db)
    resolution = service.resolve_hold(
        db,
        space_id=space.id,
        document_id=document.id,
        source_record_id=hold.id,
        user=user,
        action="confirm_as_posted",
        reason="Verified in the bank app",
        account_id=account.id,
    )
    db.flush()
    operation_id = resolution.operation_id

    service.revert_document(db, space_id=space.id, document_id=document.id, user=user)
    db.flush()

    assert document.status == "ready"
    assert operation_id is not None
    assert db.get(Operation, operation_id) is not None
    assert db.get(HoldResolution, resolution.id) is not None
    link = db.query(SourceRecordLink).filter_by(source_record_id=hold.id).one()
    assert link.kind == "hold_confirmed"
    assert link.operation_id == operation_id


def test_hold_resolution_rejects_unknown_kind(db: Session) -> None:
    user, space, _account, _document, hold = hold_document(db)
    db.add(
        HoldResolution(
            space_id=space.id,
            hold_source_record_id=hold.id,
            resolution="resolved_somehow",
        )
    )
    with pytest.raises(IntegrityError):
        db.flush()
    db.rollback()


def test_manual_resolution_accepts_no_reason_at_database_level(db: Session) -> None:
    _user, space, _account, _document, hold = hold_document(db)
    db.add(
        HoldResolution(
            space_id=space.id,
            hold_source_record_id=hold.id,
            resolution="confirmed_as_posted",
            reason=None,
        )
    )
    db.flush()
    assert db.query(HoldResolution).filter_by(hold_source_record_id=hold.id).one().reason is None
    db.rollback()
    _user, space, _account, _document, hold = hold_document(db)
    db.add(
        HoldResolution(
            space_id=space.id,
            hold_source_record_id=hold.id,
            resolution="matched_posted",
            reason=None,
        )
    )
    db.flush()


def test_upgrade_drops_legacy_manual_reason_constraint(tmp_path: Path, monkeypatch) -> None:
    from app.config import get_settings

    database_url = f"sqlite:///{tmp_path / 'hold-reason.db'}"
    monkeypatch.setenv("DATABASE_URL", database_url)
    get_settings.cache_clear()
    config = Config(str(Path(__file__).parents[1] / "alembic.ini"))
    config.set_main_option("script_location", str(Path(__file__).parents[1] / "alembic"))
    try:
        command.upgrade(config, "0003")
        engine = create_engine(database_url)
        try:
            with engine.begin() as connection:
                operations = Operations(MigrationContext.configure(connection))
                with operations.batch_alter_table("hold_resolutions") as batch_op:
                    batch_op.create_check_constraint(
                        "ck_hold_resolutions_manual_reason_required",
                        "NOT (resolution IN ('confirmed_as_posted', 'discarded_as_cancelled') "
                        "AND (reason IS NULL OR reason = ''))",
                    )

            command.upgrade(config, "0005")
            constraints = sa.inspect(engine).get_check_constraints("hold_resolutions")
            assert not any(
                (constraint.get("name") or "").endswith("manual_reason_required")
                for constraint in constraints
            )
            with Session(engine) as db:
                db.add(
                    HoldResolution(
                        space_id=uuid.uuid4(),
                        hold_source_record_id=uuid.uuid4(),
                        resolution="discarded_as_cancelled",
                        reason=None,
                    )
                )
                db.flush()
        finally:
            engine.dispose()
    finally:
        get_settings.cache_clear()


def _draft(rows, *, opening=1_000, closing=1_200, **declared):
    return SimpleNamespace(
        currency="RUB",
        opening_balance_kopecks=opening,
        closing_balance_kopecks=closing,
        declared_inflow_kopecks=declared.get("inflow"),
        declared_outflow_kopecks=declared.get("outflow"),
        declared_hold_kopecks=declared.get("hold"),
        rows=tuple(rows),
    )


def _row(amount, direction, bank_status="posted"):
    return SimpleNamespace(
        amount_kopecks=amount,
        direction=direction,
        bank_status=bank_status,
    )


def test_reconciliation_separates_posted_and_hold_totals() -> None:
    summary = service._reconciliation_summary(
        _draft(
            [
                _row(50_000, "in"),
                _row(30_000, "out"),
                _row(20_000, "out", bank_status="hold"),
            ],
            opening=100_000,
            closing=120_000,
            inflow=50_000,
            outflow=30_000,
            hold=20_000,
        )
    )
    assert summary["parsed_posted_inflow_kopecks"] == 50_000
    assert summary["parsed_posted_outflow_kopecks"] == 30_000
    assert summary["parsed_hold_total_kopecks"] == 20_000
    assert summary["posted_totals_matched"] is True
    assert summary["hold_totals_matched"] is True
    assert summary["balance_matched"] is True
    assert summary["matched"] is True
    assert summary["computed_closing_kopecks"] == 120_000


def test_reconciliation_declared_outflow_mismatch_blocks() -> None:
    summary = service._reconciliation_summary(
        _draft(
            [_row(30_000, "out")],
            opening=100_000,
            closing=70_000,
            inflow=None,
            outflow=40_000,
        )
    )
    assert summary["posted_totals_matched"] is False
    assert summary["hold_totals_matched"] is None
    assert summary["matched"] is False


def test_reconciliation_hold_mismatch_is_review_blocking() -> None:
    summary = service._reconciliation_summary(
        _draft([_row(25_000, "out", bank_status="hold")], opening=None, closing=None, hold=10_000)
    )
    assert summary["full"] is False
    assert summary["hold_totals_matched"] is False
    assert summary["matched"] is False
    assert summary["balance_matched"] is None


def test_reconciliation_balance_equation_ignores_holds() -> None:
    summary = service._reconciliation_summary(
        _draft(
            [
                _row(50_000, "in"),
                _row(30_000, "out"),
                _row(20_000, "out", bank_status="hold"),
            ],
            opening=100_000,
            closing=140_000,
        )
    )
    assert summary["balance_matched"] is False
    assert summary["matched"] is False
    assert summary["computed_closing_kopecks"] == 120_000


def test_reconciliation_without_declared_totals_keeps_legacy_semantics() -> None:
    draft = SimpleNamespace(
        currency="RUB",
        opening_balance_kopecks=100,
        closing_balance_kopecks=300,
        rows=(_row(200, "in"),),
    )
    summary = service._reconciliation_summary(draft)
    assert summary["full"] is True
    assert summary["posted_totals_matched"] is None
    assert summary["hold_totals_matched"] is None
    assert summary["balance_matched"] is True
    assert summary["matched"] is True


def test_source_row_without_bank_status_defaults_posted() -> None:
    from app.imports.adapters.base import SourceRow

    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, 5, 10),
            posted_on=None,
            amount_kopecks=1_000,
            direction="out",
            description=None,
            mcc=None,
            bank_category=None,
            location=None,
            row_ref="row1",
        ),
    )
    assert source.bank_status == "posted"
    assert source.critical is False


def test_resolve_endpoint_requires_csrf_and_resolves_hold(client: TestClient, db: Session) -> None:
    invite = client.post(
        "/api/v1/admin/invitations",
        headers={"X-Admin-Token": "test-admin-token"},
        json={"email": "holds-user@example.com"},
    )
    assert invite.status_code == 201, invite.text
    accepted = client.post(
        "/api/v1/auth/invitations/accept",
        json={
            "token": invite.json()["token"],
            "password": "a-long-password-for-tests",
            "fullName": "Holds User",
            "locale": "en",
        },
    )
    assert accepted.status_code == 201, accepted.text
    csrf = client.cookies["csrf_token"]
    db.commit()

    user = db.query(User).filter_by(email="holds-user@example.com").one()
    space_id = db.query(Membership).filter_by(user_id=user.id).one().space_id
    account = create_account(
        db,
        space_id=space_id,
        user=user,
        name="Debit",
        kind="debit",
        opened_on=date(2026, 1, 1),
    )
    space = SimpleNamespace(id=space_id)
    document, run = make_document(db, space, account, user=user)
    hold = add_record(db, document, run, bank_status="hold")
    hold.mcc = 5411
    service.confirm_document(db, space_id=space_id, document_id=document.id, user=user)
    db.commit()

    rejected = client.post(
        f"/api/v1/imports/{document.id}/rows/{hold.id}/resolve",
        json={"action": "confirm_as_posted", "reason": "no csrf"},
    )
    assert rejected.status_code == 403

    resolved = client.post(
        f"/api/v1/imports/{document.id}/rows/{hold.id}/resolve",
        headers={"X-CSRF-Token": csrf},
        json={"action": "confirm_as_posted"},
    )
    assert resolved.status_code == 200, resolved.text
    rows = resolved.json()["rows"]
    row = next(item for item in rows if item["id"] == str(hold.id))
    assert row["bankStatus"] == "hold"
    assert row["mcc"] == 5411
    assert row["mccName"] == "Продуктовые магазины и супермаркеты"
    assert row["holdResolution"]["resolution"] == "confirmed_as_posted"
    assert row["holdResolution"]["reason"] is None
    assert row["holdResolution"]["operationId"] is not None
    operations = db.query(Operation).filter_by(space_id=space_id).all()
    assert len(operations) == 1
    assert operations[0].category_id is not None