zkxw/backend/tests/test_protocol_sample_payment_status_guard.py

from datetime import datetime
from decimal import Decimal

import pytest
from sqlalchemy import create_engine
from sqlalchemy.orm import Query, sessionmaker

from app.crud.protocol_sample_payment import (
    replace_schedules,
    save_sample_payments_for_protocol,
    sync_schedules_and_transfers,
)
from app.models.base import Base
from app.models.models import (
    BizProtocolPaymentSchedule,
    BizProtocolPaymentTransfer,
    BizProtocolSamplePayment,
)


def _session():
    engine = create_engine("sqlite:///:memory:")
    Base.metadata.create_all(
        engine,
        tables=[
            BizProtocolSamplePayment.__table__,
            BizProtocolPaymentSchedule.__table__,
            BizProtocolPaymentTransfer.__table__,
        ],
    )
    return sessionmaker(bind=engine)()


def _payment(db):
    row = BizProtocolSamplePayment(
        id="pay-1",
        collection_protocol_id="protocol-1",
        sample_code="S1",
        pay_mode="fq",
        contract_amount=Decimal("100"),
        tenant_no="tenant-a",
        create_dept_id="",
        create_time=datetime(2026, 10, 14),
        update_time=datetime(2026, 10, 14),
        del_flag=0,
    )
    db.add(row)
    db.commit()
    return row


def _schedule(db, *, schedule_no=1, status="cleared"):
    row = BizProtocolPaymentSchedule(
        id=f"schedule-{schedule_no}",
        sample_payment_id="pay-1",
        collection_protocol_id="protocol-1",
        sample_code="S1",
        schedule_no=schedule_no,
        due_amount=Decimal("100"),
        due_date="20261031",
        schedule_status=status,
        tenant_no="tenant-a",
        create_dept_id="",
        create_time=datetime(2026, 10, 14),
        update_time=datetime(2026, 10, 14),
        del_flag=0,
    )
    db.add(row)
    db.commit()
    return row


def _transfer(db, *, schedule_no=1, status="yjf"):
    row = BizProtocolPaymentTransfer(
        id=f"transfer-{schedule_no}",
        schedule_id=f"schedule-{schedule_no}",
        sample_payment_id="pay-1",
        collection_protocol_id="protocol-1",
        sample_code="S1",
        transfer_amount=Decimal("100"),
        transfer_type="bank",
        pay_status=status,
        tenant_no="tenant-a",
        create_dept_id="",
        create_time=datetime(2026, 10, 14),
        update_time=datetime(2026, 10, 14),
        del_flag=0,
    )
    db.add(row)
    db.commit()
    return row


def test_existing_schedule_status_is_not_overwritten_by_payload():
    db = _session()
    _payment(db)
    _schedule(db, status="cleared")

    rows = sync_schedules_and_transfers(
        db,
        sample_payment_id="pay-1",
        tenant_no="tenant-a",
        user_id="u1",
        schedules=[
            {
                "scheduleNo": 1,
                "dueAmount": 100,
                "dueDate": "20261130",
                "scheduleStatus": "void",
            }
        ],
    )

    assert rows[0].schedule_status == "cleared"


@pytest.mark.parametrize("status", [None, "", " "])
def test_non_open_blank_schedule_status_is_preserved(status):
    db = _session()
    _payment(db)
    _schedule(db, status=status)

    rows = sync_schedules_and_transfers(
        db,
        sample_payment_id="pay-1",
        tenant_no="tenant-a",
        user_id="u1",
        schedules=[{"scheduleNo": 1, "dueAmount": 100}],
    )

    assert rows[0].schedule_status == status


def test_new_schedule_ignores_payload_status_and_defaults_open():
    db = _session()
    _payment(db)

    rows = sync_schedules_and_transfers(
        db,
        sample_payment_id="pay-1",
        tenant_no="tenant-a",
        user_id="u1",
        schedules=[
            {
                "scheduleNo": 1,
                "dueAmount": 100,
                "dueDate": "20261031",
                "scheduleStatus": "cleared",
            }
        ],
    )

    assert rows[0].schedule_status == "open"


def test_structure_change_rejects_renumbering_locked_schedule():
    db = _session()
    _payment(db)
    _schedule(db, schedule_no=1, status="cleared")

    with pytest.raises(ValueError, match="锁定批次"):
        sync_schedules_and_transfers(
            db,
            sample_payment_id="pay-1",
            tenant_no="tenant-a",
            user_id="u1",
            schedules=[
                {
                    "scheduleNo": 2,
                    "dueAmount": 100,
                    "dueDate": "20261130",
                }
            ],
        )


def test_locked_schedule_cannot_be_omitted():
    db = _session()
    _payment(db)
    _schedule(db, status="cleared")

    with pytest.raises(ValueError, match="锁定批次"):
        sync_schedules_and_transfers(
            db,
            sample_payment_id="pay-1",
            tenant_no="tenant-a",
            user_id="u1",
            schedules=[],
        )

    db.expire_all()
    assert db.get(BizProtocolPaymentSchedule, "schedule-1").del_flag == 0


@pytest.mark.parametrize("pay_status", ["jfqrz", "yjf"])
@pytest.mark.parametrize("payload", [[], [{"scheduleNo": 2, "dueAmount": 100}]])
def test_locked_transfer_schedule_cannot_be_omitted_or_renumbered(pay_status, payload):
    db = _session()
    _payment(db)
    _schedule(db, status="open")
    _transfer(db, status=pay_status)

    with pytest.raises(ValueError, match="锁定批次"):
        sync_schedules_and_transfers(
            db,
            sample_payment_id="pay-1",
            tenant_no="tenant-a",
            user_id="u1",
            schedules=payload,
        )

    db.expire_all()
    transfer = db.get(BizProtocolPaymentTransfer, "transfer-1")
    assert transfer.schedule_id == "schedule-1"
    assert transfer.pay_status == pay_status
    assert transfer.del_flag == 0


def test_replace_schedules_never_falls_back_locked_transfer_to_first_new_period():
    db = _session()
    _payment(db)
    _schedule(db, status="open")
    _transfer(db, status="yjf")

    with pytest.raises(ValueError, match="锁定批次"):
        replace_schedules(
            db,
            sample_payment_id="pay-1",
            tenant_no="tenant-a",
            user_id="u1",
            schedules=[{"scheduleNo": 2, "dueAmount": 100}],
        )

    db.expire_all()
    assert db.get(BizProtocolPaymentTransfer, "transfer-1").schedule_id == "schedule-1"


def test_locked_transfer_id_cannot_be_submitted_under_another_schedule():
    db = _session()
    _payment(db)
    _schedule(db, schedule_no=1, status="open")
    _schedule(db, schedule_no=2, status="open")
    _transfer(db, schedule_no=1, status="yjf")

    with pytest.raises(ValueError, match="转账笔.*批次"):
        sync_schedules_and_transfers(
            db,
            sample_payment_id="pay-1",
            tenant_no="tenant-a",
            user_id="u1",
            schedules=[
                {"scheduleNo": 1, "dueAmount": 50, "transfers": []},
                {
                    "scheduleNo": 2,
                    "dueAmount": 50,
                    "transfers": [{"id": "transfer-1", "transferAmount": 999}],
                },
            ],
        )

    db.expire_all()
    transfer = db.get(BizProtocolPaymentTransfer, "transfer-1")
    assert transfer.schedule_id == "schedule-1"
    assert transfer.transfer_amount == Decimal("100")


def test_locked_sample_payment_cannot_be_omitted_from_protocol_save():
    db = _session()
    _payment(db)
    _schedule(db, status="cleared")

    with pytest.raises(ValueError, match="锁定缴费方案"):
        save_sample_payments_for_protocol(
            db,
            collection_protocol_id="protocol-1",
            tenant_no="tenant-a",
            user_id="u1",
            sample_payments=[],
        )

    db.expire_all()
    assert db.get(BizProtocolSamplePayment, "pay-1").del_flag == 0
    assert db.get(BizProtocolPaymentSchedule, "schedule-1").del_flag == 0


def test_all_draft_structure_can_still_be_rebuilt():
    db = _session()
    _payment(db)
    _schedule(db, status="open")
    _transfer(db, status="dfj")

    rows = sync_schedules_and_transfers(
        db,
        sample_payment_id="pay-1",
        tenant_no="tenant-a",
        user_id="u1",
        schedules=[{"scheduleNo": 2, "dueAmount": 100}],
    )

    assert [row.schedule_no for row in rows] == [2]
    db.expire_all()
    assert db.get(BizProtocolPaymentSchedule, "schedule-1").del_flag == 1
    assert db.get(BizProtocolPaymentTransfer, "transfer-1").del_flag == 1


def test_schedule_sync_locks_payment_schedule_and_transfer_rows(monkeypatch):
    db = _session()
    _payment(db)
    _schedule(db, status="open")
    _transfer(db, status="dfj")
    calls = []
    original = Query.with_for_update

    def tracking_with_for_update(query, *args, **kwargs):
        calls.append(query)
        return original(query, *args, **kwargs)

    monkeypatch.setattr(Query, "with_for_update", tracking_with_for_update)
    sync_schedules_and_transfers(
        db,
        sample_payment_id="pay-1",
        tenant_no="tenant-a",
        user_id="u1",
        schedules=[{"scheduleNo": 1, "dueAmount": 100}],
    )

    assert len(calls) >= 3